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 "aicpu_thread_process.h"
      12              : #include <iomanip>
      13              : 
      14              : using namespace hccl;
      15              : 
      16              : std::mutex AicpuThreadProcess::mutex_;
      17              : std::vector<std::shared_ptr<hccl::Thread>> AicpuThreadProcess::threads_;
      18              : 
      19            0 : HcclResult AicpuThreadProcess::InitThreads(ThreadMgrAicpuParam *param)
      20              : {
      21            0 :     CHK_PTR_NULL(param);
      22            0 :     u32 threadNum = param->threadNum;
      23            0 :     std::vector<std::shared_ptr<Thread>> outThreads;
      24            0 :     outThreads.reserve(threadNum);
      25            0 :     std::string hcomId(param->hcomId);
      26            0 :     CHK_RET(AicpuThreadProcess::ResumeThread(param, outThreads, false));
      27              : 
      28            0 :     ThreadHandle *threadArray = static_cast<ThreadHandle*>(param->deviceHandle);
      29              :     // 空指针校验
      30            0 :     CHK_PTR_NULL(threadArray);
      31            0 :     for (size_t i = 0; i < threadNum; ++i) {
      32            0 :         threadArray[i] = reinterpret_cast<ThreadHandle>(outThreads[i].get());  // 拷贝裸指针
      33            0 :         HCCL_INFO("[HcclCommAicpu][%s] threadArray[%zu] = [%lu]", __func__, i, threadArray[i]);
      34              :     }
      35            0 :     threads_.insert(threads_.end(), std::make_move_iterator(outThreads.begin()),
      36              :         std::make_move_iterator(outThreads.end()));
      37            0 :     HCCL_INFO("[HcclCommAicpu][%s] comm identifier[%s], init threads num[%u] success",
      38              :         __func__, hcomId.c_str(), threadNum);
      39            0 :     return HCCL_SUCCESS;
      40            0 : }
      41              : 
      42            0 : HcclResult AicpuThreadProcess::AicpuThreadInit(ThreadMgrAicpuParam *param)
      43              : {
      44            0 :     CHK_RET(hrtSetWorkModeAicpu(true));
      45            0 :     CHK_RET(hrtSetlocalDevice(param->deviceLogicId));
      46            0 :     CHK_RET(hrtSetlocalDeviceType(static_cast<DevType>(param->deviceType)));
      47            0 :     std::lock_guard<std::mutex> addLock(mutex_);
      48            0 :     HcclResult ret = InitThreads(param);
      49            0 :     CHK_PRT_RET(ret != HCCL_SUCCESS,
      50              :         HCCL_ERROR("[AicpuThreadProcess][AicpuIndOpThreadInit]errNo[0x%016llx] Failed to init threads",
      51              :         HCCL_ERROR_CODE(ret)), ret);
      52            0 :     return HCCL_SUCCESS;
      53            0 : }
      54              : 
      55            0 : HcclResult AicpuThreadProcess::AicpuThreadDestroy(ThreadMgrAicpuParam *param)
      56              : {
      57            0 :     HCCL_INFO("[AicpuThreadProcess][%s] threadNum[%u]", __func__, param->threadNum);
      58            0 :     std::lock_guard<std::mutex> addLock(mutex_);
      59            0 :     ThreadHandle *threadArray = static_cast<ThreadHandle*>(param->deviceHandle);
      60            0 :     CHK_PTR_NULL(threadArray);
      61              : 
      62            0 :     for (u32 i = 0; i < param->threadNum; ++i) {
      63            0 :         ThreadHandle handle = threadArray[i];
      64            0 :         auto it = std::find_if(threads_.begin(), threads_.end(),
      65            0 :             [handle](const std::shared_ptr<Thread> &ptr) {
      66            0 :                 return reinterpret_cast<ThreadHandle>(ptr.get()) == handle;
      67              :             });
      68            0 :         if (it == threads_.end()) {
      69            0 :             HCCL_WARNING("[AicpuThreadProcess][%s] thread handle[0x%llx] not found in threads_", __func__, handle);
      70            0 :             continue; // 继续处理其他线程
      71              :         }
      72              :         // 从容器中移除，shared_ptr 自动释放对象
      73            0 :         threads_.erase(it);
      74            0 :         HCCL_DEBUG("[AicpuThreadProcess][%s] destroyed thread handle[0x%llx]", __func__, handle);
      75              :     }
      76              : 
      77            0 :     HCCL_INFO("[AicpuThreadProcess][%s] success", __func__);
      78            0 :     return HCCL_SUCCESS;
      79            0 : }
      80              : 
      81            1 : HcclResult AicpuThreadProcess::ResumeThread(ThreadMgrAicpuParam *param,
      82              :     std::vector<std::shared_ptr<Thread>> &outThreads, bool isSupplementNotify)
      83              : {
      84            1 :     CHK_PTR_NULL(param);
      85            1 :     u32 threadNum = param->threadNum;
      86            1 :     std::string hcomId(param->hcomId);
      87            1 :     ThreadHandle *threadArray = static_cast<ThreadHandle*>(param->deviceHandle);
      88            2 :     for (u32 i = 0; i < threadNum; ++i) {
      89            1 :         std::string thdUniqueId(param->threadParam[i], THREAD_UNIQUE_ID_MAX_SIZE);
      90            1 :         if (UNLIKELY(HcclCheckLogLevel(HCCL_LOG_INFO))) {
      91            1 :             std::ostringstream oss;
      92            1 :             oss << "threadParam[" << i << "] raw bytes: ";
      93            1 :             constexpr u32 HEX_WIDTH = 2;
      94         6001 :             for (u32 j = 0; j < THREAD_UNIQUE_ID_MAX_SIZE; ++j) {
      95         6000 :                 oss << std::hex << std::setw(HEX_WIDTH) << std::setfill('0')
      96         6000 :                     << static_cast<unsigned int>(static_cast<unsigned char>(param->threadParam[i][j])) << " ";
      97              :             }
      98            1 :             HCCL_INFO("[HcclCommAicpu][%s] %s", __func__, oss.str().c_str());
      99            1 :         }
     100            1 :         std::shared_ptr<AicpuTsThread> thread;
     101            1 :         EXCEPTION_CATCH((thread = std::make_shared<AicpuTsThread>(thdUniqueId)), return HCCL_E_PTR);
     102            1 :         u32 notifyNum = 0;
     103            1 :         std::string notifyDesc;
     104            1 :         CHK_RET(thread->GetNotifyByUniqueId(notifyNum, notifyDesc));
     105            1 :         if (isSupplementNotify) {
     106            1 :             AicpuTsThread *threadPtr = reinterpret_cast<AicpuTsThread*>(threadArray[i]);
     107            1 :             CHK_PTR_NULL(threadPtr);
     108            1 :             HCCL_INFO("[%s]threadIdx[%u], threadHandle[%llu], notifyNum[%u], newNotifyNum[%u]", __func__, i,
     109              :                 threadArray[i], threadPtr->GetNotifyNum(), notifyNum);
     110            1 :             CHK_RET(threadPtr->SupplementNotify(notifyNum, notifyDesc));
     111              :         } else {
     112            0 :             HcclResult ret = thread->Init();
     113            0 :             if (ret != HCCL_SUCCESS) {
     114            0 :                 HCCL_ERROR("[HcclCommAicpu][%s] comm identifier[%s], init threads num[%u] failed at index %u",
     115              :                     __func__, hcomId.c_str(), param->threadNum, i);
     116            0 :                 return ret;
     117              :             }
     118            0 :             outThreads.emplace_back(thread);
     119              :         }
     120            1 :     }
     121            1 :     return HCCL_SUCCESS;
     122            1 : }
     123              : 
     124            1 : HcclResult AicpuThreadProcess::AicpuThreadSupplementNotify(ThreadMgrAicpuParam *param)
     125              : {
     126            1 :     CHK_PTR_NULL(param);
     127            1 :     u32 threadNum = param->threadNum;
     128            1 :     std::string hcomId(param->hcomId);
     129            1 :     std::vector<std::shared_ptr<Thread>> outThreads;
     130            1 :     CHK_RET(AicpuThreadProcess::ResumeThread(param, outThreads, true));
     131              : 
     132            1 :     HCCL_INFO("[HcclCommAicpu][%s] comm identifier[%s], init threads num[%u] success",
     133              :         __func__, hcomId.c_str(), threadNum);
     134            1 :     return HCCL_SUCCESS;
     135            1 : }