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          520 : ParallelTaskLoader::ParallelTaskLoader(const s32 deviceLogicId, const HcclDispatcher dispatcher)
      16          520 :     : deviceLogicId_(deviceLogicId), dispatcher_(dispatcher), taskLoaderNum_(0)
      17          520 : {}
      18              : 
      19          519 : ParallelTaskLoader::~ParallelTaskLoader()
      20          519 : {}
      21              : 
      22            0 : HcclResult ParallelTaskLoader::Prepare(std::vector<Stream *> streamsPtr, SubCommInfo level0CommInfo)
      23              : {
      24              :     // 参数保存
      25            0 :     streamsPtr_.resize(streamsPtr.size());
      26            0 :     for (u32 streamIndex = 0; streamIndex < streamsPtr.size(); streamIndex++) {
      27            0 :         streamsPtr_[streamIndex] = streamsPtr[streamIndex];
      28              :     }
      29            0 :     commInfo_ = level0CommInfo;
      30            0 :     HCCL_INFO("[ParallelTaskLoader]Prepare streams size[%u], taskLoaderNum_[%u]", streamsPtr_.size(), taskLoaderNum_);
      31              : 
      32              :     // 当前现有的taskLoader线程可以满足业务多流的使用
      33            0 :     if (taskLoaderNum_ >= streamsPtr_.size()) {
      34            0 :         HCCL_INFO("[ParallelTaskLoader] taskloaderNum [%u]", taskLoaderNum_);
      35            0 :         return HCCL_SUCCESS;
      36              :     }
      37              : 
      38            0 :     streamTaskLoader_.resize(streamsPtr_.size());
      39              :     // 当前现有的taskLoader无法满足业务多流的使用，需要扩展多流资源
      40            0 :     for (u32 streamIndex = taskLoaderNum_; streamIndex < streamsPtr_.size(); streamIndex++) {
      41            0 :         streamTaskLoader_[streamIndex].reset(new (std::nothrow) TaskLoader(deviceLogicId_, dispatcher_));
      42            0 :         CHK_SMART_PTR_NULL(streamTaskLoader_[streamIndex]);
      43            0 :         HcclResult ret = streamTaskLoader_[streamIndex]->Init();
      44            0 :         CHK_PRT_RET(ret != HCCL_SUCCESS,
      45              :             HCCL_ERROR("[ParallelTaskLoader][Init]streamIndex[%u] TaskLoader failed, return[%d]", streamIndex, ret),
      46              :             ret);
      47              :     }
      48            0 :     taskLoaderNum_ = streamsPtr_.size();
      49            0 :     HCCL_INFO("[ParallelTaskLoader] Prepare success taskLoaderNum[%u]", taskLoaderNum_);
      50              : 
      51            0 :     return HCCL_SUCCESS;
      52              : }
      53              : 
      54            0 : HcclResult ParallelTaskLoader::StartTaskLoad()
      55              : {
      56            0 :     tidInfo_.resize(streamsPtr_.size());
      57              : 
      58              :     // 配置线程启动参数
      59            0 :     for (u32 streamIndex = 0; streamIndex < streamsPtr_.size(); streamIndex++) {
      60            0 :         streamTaskLoader_[streamIndex]->Prepare(streamsPtr_[streamIndex], commInfo_);
      61              :         // 获取线程ID
      62            0 :         tidInfo_[streamIndex] = streamTaskLoader_[streamIndex]->GetTid();
      63              :     }
      64              : #ifndef OPEN_HCCL_TEST
      65            0 :     CHK_RET(hccl::ProfilingManagerPub::CallMsprofReportMultiThreadInfo(tidInfo_));
      66              : #endif
      67              : 
      68              :     // 通知流线程执行
      69            0 :     for (u32 streamIndex = 0; streamIndex < streamsPtr_.size(); streamIndex++) {
      70            0 :         streamTaskLoader_[streamIndex]->NotifyStart();
      71              :     }
      72            0 :     return HCCL_SUCCESS;
      73              : }
      74              : 
      75            0 : HcclResult ParallelTaskLoader::WaitTaskLoadFinish()
      76              : {
      77              :     // 等待流线程执行
      78            0 :     for (u32 streamIndex = 0; streamIndex < streamsPtr_.size(); streamIndex++) {
      79            0 :         streamTaskLoader_[streamIndex]->WaitDone();
      80            0 :         CHK_RET(streamTaskLoader_[streamIndex]->GetExecuteResult());
      81              :     }
      82            0 :     return HCCL_SUCCESS;
      83              : }
      84              : 
      85            0 : HcclResult ParallelTaskLoader::ClearTagCommInfo()
      86              : {
      87            0 :     commInfo_ = SubCommInfo{};
      88            0 :     for (u32 streamIndex = 0; streamIndex < streamsPtr_.size(); streamIndex++) {
      89            0 :         CHK_RET(streamTaskLoader_[streamIndex]->ClearTagCommInfo());
      90              :     }
      91            0 :     return HCCL_SUCCESS;
      92              : }
      93              : }  // namespace hccl