@@ -16,6 +16,7 @@ #include "ub_transport_lite_impl.h" #include "notify_manager.h" #include "aicpu_hccl_def.h" +#include "ns_recovery/aicpu/ns_recovery_func_lite.h" #include "dlhal_function_v2.h" #include "profiling_command_handle_lite.h" #include "aicpu_daemon_service.h" @@ -27,7 +28,7 @@ constexpr u32 NOTIFY_SIZE_EIGHT = 8; std::function callback); // 临时，后续移动至Op.h HcclResult CollCommAicpu::InitAicpuIndOp(CommAicpuParam *commAicpuParam) { - if (isReady_) { + if (commStatus_ == HcclCommStatus::HCCL_COMM_STATUS_READY) { HCCL_RUN_INFO("[CollCommAicpu][%s]Group[%s] already initialized, skip reinit", __func__, identifier_.c_str()); return HCCL_SUCCESS; @@ -57,9 +58,13 @@ HcclResult CollCommAicpu::InitAicpuIndOp(CommAicpuParam *commAicpuParam) CHK_SMART_PTR_NULL(kfcStatusTransferD2H_); CHK_RET(kfcStatusTransferD2H_->InitDevice(commAicpuParam->kfcStatusTransferD2HParams)); } + + EXECEPTION_CATCH(nsRecoveryLitePtr_ = std::make_shared(), return HCCL_E_PTR); + nsRecoveryLitePtr_->Init(kfcControlTransferH2D_, kfcStatusTransferD2H_); + CHK_RET(Hccl::DlHalFunctionV2::GetInstance().DlHalFunctionInit()); - isReady_ = true; + commStatus_ = HcclCommStatus::HCCL_COMM_STATUS_READY; static std::once_flag onceFlag; std::call_once(onceFlag, [this]() { this->InitBackGroundThread();} ); @@ -69,10 +74,10 @@ HcclResult CollCommAicpu::InitAicpuIndOp(CommAicpuParam *commAicpuParam) return HCCL_SUCCESS; } -void CollCommAicpu::SetIsReady(bool flag) +void CollCommAicpu::SetCommmStatus(HcclCommStatus status) { - HCCL_INFO("[%s]group[%s], flag[%d]", __func__, identifier_.c_str(), flag); - isReady_ = flag; + HCCL_INFO("[%s]group[%s], flag[%d]", __func__, identifier_.c_str(), static_cast(status)); + commStatus_ = status; } HcclResult CollCommAicpu::InitThreads(ThreadMgrAicpuParam *param) @@ -138,10 +143,11 @@ HcclResult CollCommAicpu::AllocChannelResource(HcclChannelUrmaRes *commParam) return HCCL_SUCCESS; } -HcclResult CollCommAicpu::InitUrmaChannel(HcclChannelUrmaRes *commParam) +HcclResult CollCommAicpu::ProcessUrmaRes(HcclChannelUrmaRes *commParam, bool isInit) { HCCL_INFO("[CollCommAicpu][%s] commParam->uniqueIdAddr[%p], commParam->uniqueIdSize[%u]", __func__, commParam->uniqueIdAddr, commParam->uniqueIdSize); + ChannelHandle* channelList = reinterpret_cast(commParam->channelList); for (u32 index = 0; index < commParam->listNum; index++) { std::vector data(commParam->singleUniqueIdSize); @@ -158,14 +164,24 @@ HcclResult CollCommAicpu::InitUrmaChannel(HcclChannelUrmaRes *commParam) HCCL_ERROR("[CollCommAicpu][%s] fail, resType[%d], dataVec size[%u]", __func__, resType, dataVec.size()); return HCCL_E_PARA; } - ChannelHandle channelHandle; - CHK_RET(ParsePackData(dataVec[resType].data, channelHandle)); - // 恢复出的channelHandle回填到commParam中 - ChannelHandle* channelList = reinterpret_cast(commParam->channelList); - channelList[index] = channelHandle; - CHK_RET(RegisterChannelAddDfxTaskInfo(channelHandle)); - HcclCommDfxLite::AddChannelRemoteRankId(identifier_, channelHandle, commParam->remoteRankList[index]); + ChannelHandle channelHandle{0}; + if (isInit) { + CHK_RET(ParsePackData(dataVec[resType].data, channelHandle)); + // 恢复出的channelHandle回填到commParam中 + channelList[index] = channelHandle; + CHK_RET(RegisterChannelAddDfxTaskInfo(channelHandle)); + HcclCommDfxLite::AddChannelRemoteRankId(identifier_, channelHandle, commParam->remoteRankList[index]); + } else { + channelHandle = channelList[index]; + if (!ubTransportMap_.count(channelHandle)) { + HCCL_ERROR("[CollCommAicpu][%s] fail, resType[%d], current ChannelHandle nullptr", __func__, resType); + return HCCL_E_PARA; + } + CHK_RET(ResumePackData(dataVec[resType].data, channelHandle)); + } + + // 打印 HCCL_INFO("[CollCommAicpu][%s] index[%u], currentSrcAddr[%p], singleUniqueIdSize[%u], channelHandle[0x%llx]", __func__, index, currentSrcAddr, commParam->singleUniqueIdSize, channelHandle); } @@ -173,6 +189,11 @@ HcclResult CollCommAicpu::InitUrmaChannel(HcclChannelUrmaRes *commParam) return HCCL_SUCCESS; } +HcclResult CollCommAicpu::InitUrmaChannel(HcclChannelUrmaRes *commParam) +{ + return ProcessUrmaRes(commParam, true); +} + HcclResult CollCommAicpu::ParsePackData(std::vector &data, ChannelHandle &handle) { HCCL_DEBUG("[CollCommAicpu][%s] data: ptr[%p], size[%u]", __func__, data.data(), data.size()); @@ -263,6 +284,44 @@ HcclResult CollCommAicpu::NotifyAlloc(NotifyMgrAicpuParam *param) return HCCL_SUCCESS; } +hccl::NsRecoveryLitePtr CollCommAicpu::GetNsRecoveryLitePtr() +{ + return nsRecoveryLitePtr_; +} + +HcclResult CollCommAicpu::Clean() +{ + for (auto& transPort : ubTransportMap_) { + CHK_RET(transPort.second->Clean()); + } + HCCL_INFO("CollCommAicpu::Clean() finished"); + + return HCCL_SUCCESS; +} + +HcclResult CollCommAicpu::ResumePackData(std::vector &data, ChannelHandle &handle) +{ + Hccl::BinaryStream binaryStream(data); + std::vector transpUniqueId; + binaryStream >> transpUniqueId; + + auto& transPortPtr = ubTransportMap_[handle]; + CHK_RET(transPortPtr->Resume(transpUniqueId)); + return HCCL_SUCCESS; +} + +HcclResult CollCommAicpu::Resume(HcclChannelUrmaRes *commParam) +{ + CHK_PTR_NULL(commParam); + CHK_RET(ProcessUrmaRes(commParam, false)); + nsRecoveryLitePtr_->SetNeedClean(false); + nsRecoveryLitePtr_->ResetErrorReported(); + + commStatus_ = HcclCommStatus::HCCL_COMM_STATUS_READY; + + return HCCL_SUCCESS; +} + void CollCommAicpu::InitBackGroundThread() { static auto commandToBackGroud = Hccl::CommandToBackGroud::Default; @@ -277,6 +336,7 @@ void CollCommAicpu::InitBackGroundThread() hcomm::HcclCommTaskExceptionLite::GetInstance().Init(devId_); Hccl::AicpuDaemonService::GetInstance().Register(&hcomm::HcclCommTaskExceptionLite::GetInstance()); Hccl::AicpuDaemonService::GetInstance().Register(&hccl::CollCommAicpuDestroyFunc::GetInstance()); + Hccl::AicpuDaemonService::GetInstance().Register(&NsRecoveryFuncLite::GetInstance()); // 启动背景线程 if (Hccl::StartMC2MaintenanceThread != nullptr) {