Line data    Source code

       1              : /**
       2              :  * Copyright (c) 2025 Huawei Technologies Co., Ltd.
       3              :  * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
       4              :  * CANN Open Software License Agreement Version 2.0 (the "License").
       5              :  * Please refer to the License for details. You may not use this file except in compliance with the License.
       6              :  * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
       7              :  * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
       8              :  * See LICENSE in the root of the software repository for the full text of the License.
       9              :  */
      10              : 
      11              : #include "parallel_task_loader.h"
      12              : #include "profiling_manager_pub.h"
      13              : 
      14              : namespace hccl {
      15          523 : ParallelTaskLoader::ParallelTaskLoader(const s32 deviceLogicId, const HcclDispatcher dispatcher)
      16          523 :     : deviceLogicId_(deviceLogicId),
      17          523 :       dispatcher_(dispatcher),
      18          523 :       taskLoaderNum_(0)
      19          523 : {}
      20              : 
      21          523 : ParallelTaskLoader::~ParallelTaskLoader() {}
      22              : 
      23            0 : HcclResult ParallelTaskLoader::Prepare(std::vector<Stream*> streamsPtr, SubCommInfo level0CommInfo)
      24              : {
      25              :     // 参数保存
      26            0 :     streamsPtr_.resize(streamsPtr.size());
      27            0 :     for (u32 streamIndex = 0; streamIndex < streamsPtr.size(); streamIndex++) {
      28            0 :         streamsPtr_[streamIndex] = streamsPtr[streamIndex];
      29              :     }
      30            0 :     commInfo_ = level0CommInfo;
      31            0 :     HCCL_INFO("[ParallelTaskLoader]Prepare streams size[%u], taskLoaderNum_[%u]", streamsPtr_.size(), taskLoaderNum_);
      32              : 
      33              :     // 当前现有的taskLoader线程可以满足业务多流的使用
      34            0 :     if (taskLoaderNum_ >= streamsPtr_.size()) {
      35            0 :         HCCL_INFO("[ParallelTaskLoader] taskloaderNum [%u]", taskLoaderNum_);
      36            0 :         return HCCL_SUCCESS;
      37              :     }
      38              : 
      39            0 :     streamTaskLoader_.resize(streamsPtr_.size());
      40              :     // 当前现有的taskLoader无法满足业务多流的使用，需要扩展多流资源
      41            0 :     for (u32 streamIndex = taskLoaderNum_; streamIndex < streamsPtr_.size(); streamIndex++) {
      42            0 :         streamTaskLoader_[streamIndex].reset(new (std::nothrow) TaskLoader(deviceLogicId_, dispatcher_));
      43            0 :         CHK_SMART_PTR_NULL(streamTaskLoader_[streamIndex]);
      44            0 :         HcclResult ret = streamTaskLoader_[streamIndex]->Init();
      45            0 :         CHK_PRT_RET(
      46              :             ret != HCCL_SUCCESS,
      47              :             HCCL_ERROR("[ParallelTaskLoader][Init]streamIndex[%u] TaskLoader failed, return[%d]", streamIndex, ret),
      48              :             ret);
      49              :     }
      50            0 :     taskLoaderNum_ = streamsPtr_.size();
      51            0 :     HCCL_INFO("[ParallelTaskLoader] Prepare success taskLoaderNum[%u]", taskLoaderNum_);
      52              : 
      53            0 :     return HCCL_SUCCESS;
      54              : }
      55              : 
      56            0 : HcclResult ParallelTaskLoader::StartTaskLoad()
      57              : {
      58            0 :     tidInfo_.resize(streamsPtr_.size());
      59              : 
      60              :     // 配置线程启动参数
      61            0 :     for (u32 streamIndex = 0; streamIndex < streamsPtr_.size(); streamIndex++) {
      62            0 :         streamTaskLoader_[streamIndex]->Prepare(streamsPtr_[streamIndex], commInfo_);
      63              :         // 获取线程ID
      64            0 :         tidInfo_[streamIndex] = streamTaskLoader_[streamIndex]->GetTid();
      65              :     }
      66              : #ifndef OPEN_HCCL_TEST
      67            0 :     CHK_RET(hccl::ProfilingManagerPub::CallMsprofReportMultiThreadInfo(tidInfo_));
      68              : #endif
      69              : 
      70              :     // 通知流线程执行
      71            0 :     for (u32 streamIndex = 0; streamIndex < streamsPtr_.size(); streamIndex++) {
      72            0 :         streamTaskLoader_[streamIndex]->NotifyStart();
      73              :     }
      74            0 :     return HCCL_SUCCESS;
      75              : }
      76              : 
      77            0 : HcclResult ParallelTaskLoader::WaitTaskLoadFinish()
      78              : {
      79              :     // 等待流线程执行
      80            0 :     for (u32 streamIndex = 0; streamIndex < streamsPtr_.size(); streamIndex++) {
      81            0 :         streamTaskLoader_[streamIndex]->WaitDone();
      82            0 :         CHK_RET(streamTaskLoader_[streamIndex]->GetExecuteResult());
      83              :     }
      84            0 :     return HCCL_SUCCESS;
      85              : }
      86              : 
      87            0 : HcclResult ParallelTaskLoader::ClearTagCommInfo()
      88              : {
      89            0 :     commInfo_ = SubCommInfo{};
      90            0 :     for (u32 streamIndex = 0; streamIndex < streamsPtr_.size(); streamIndex++) {
      91            0 :         CHK_RET(streamTaskLoader_[streamIndex]->ClearTagCommInfo());
      92              :     }
      93            0 :     return HCCL_SUCCESS;
      94              : }
      95              : } // namespace hccl