@@ -35,63 +35,16 @@ #include "hcclCommOp.h" #include "exception_handler.h" #include "param_check_pub.h" +#include "channel_process.h" #include "launch_device.h" namespace hcomm { -static std::unordered_map> g_ChannelMap; -static std::unordered_map g_ChannelD2HMap; static std::unordered_map> g_ThreadMap; static aclrtBinHandle g_BinHandle; -static std::mutex g_ChannelMapMtx; static std::mutex g_BinHandleMtx; } // namespace hcomm -namespace hcomm { - -/** - * @brief 单锁版本：输入任意 channel handle（可能是 device/host），先通过 g_ChannelD2HMap 映射到 - * 真实的 handle，再通过 g_ChannelMap 找到 Channel，并在持锁状态下执行 func。 - * - * @tparam Func 形如：HcclResult func(Channel& ch) - */ -template -static inline HcclResult WithChannelByHandleLocked(ChannelHandle inHandle, Func &&func) -{ - // 单锁：该锁同时保护 g_ChannelD2HMap 和 g_ChannelMap - std::lock_guard lk(hcomm::g_ChannelMapMtx); - - // 1) D2H 映射 - auto itH = hcomm::g_ChannelD2HMap.find(inHandle); - if (itH == hcomm::g_ChannelD2HMap.end()) { - HCCL_ERROR("[%s] handle not found in g_ChannelD2HMap, inHandle[0x%llx].", __func__, inHandle); - return HcclResult::HCCL_E_NOT_FOUND; - } - const ChannelHandle mappedHandle = itH->second; - - // 2) ChannelMap 查找 - auto itC = hcomm::g_ChannelMap.find(mappedHandle); - if (itC == hcomm::g_ChannelMap.end() || !itC->second) { - HCCL_ERROR("[%s] channel not found in g_ChannelMap, inHandle[0x%llx], mappedHandle[0x%llx].", - __func__, - inHandle, - mappedHandle); - return HcclResult::HCCL_E_INTERNAL; - } - - Channel *ch = itC->second.get(); - if (ch == nullptr) { - HCCL_ERROR( - "[%s] null channel pointer, inHandle[0x%llx], mappedHandle[0x%llx].", __func__, inHandle, mappedHandle); - return HcclResult::HCCL_E_INTERNAL; - } - - // 3) 锁内执行用户逻辑（注意：func 内不要做长耗时/阻塞操作） - return std::forward(func)(*ch); -} - -} // namespace hcomm - using namespace hcomm; static HcommEndpointMap g_EndpointMap; @@ -240,332 +193,74 @@ HcclResult HcommMemGetAllMemHandles(EndpointHandle endpointHandle, void **memHan return HCCL_SUCCESS; } -HcclResult HcommChannelCreate(EndpointHandle endpointHandle, CommEngine engine, HcommChannelDesc *channelDescs, - uint32_t channelNum, ChannelHandle *channels) +// 集合通信使用，待归一到HcommChannelCreate +HcclResult HcommCollectiveChannelCreate(EndpointHandle endpointHandle, CommEngine engine, + HcommChannelDesc *channelDescs, uint32_t channelNum, ChannelHandle *channels) { CHK_PTR_NULL(channelDescs); CHK_PTR_NULL(channels); + CHK_PRT_RET((channelNum == 0), HCCL_ERROR("[%s]Invalid channelNum, channelNum[%u]", + __func__, channelNum), HCCL_E_PARA); + HCCL_INFO("[%s] START. endpointHandle[0x%llx], engine[%d], channelNum[%u].", + __func__, endpointHandle, engine, channelNum); - for (uint32_t i = 0; i < channelNum; ++i) { - // 1) 创建对象：不持全局表锁，避免扩大临界区 - std::unique_ptr tmpPtr = nullptr; - CHK_RET(Channel::CreateChannel(endpointHandle, engine, channelDescs[i], tmpPtr)); - CHK_SMART_PTR_NULL(tmpPtr); - - ChannelHandle handle = reinterpret_cast(tmpPtr.get()); - channels[i] = handle; - HCCL_INFO("%s handle[0x%llx], ptr[%p]", __func__, handle, tmpPtr.get()); - - // 2) 仅在修改全局表时持锁（该锁同时保护两张表） - { - std::lock_guard lk(hcomm::g_ChannelMapMtx); - - if (hcomm::g_ChannelMap.find(handle) != hcomm::g_ChannelMap.end()) { - HCCL_ERROR("[%s] channel handle already exists [0x%llx] in ChannelMap", __func__, handle); - return HCCL_E_INTERNAL; - } - if (hcomm::g_ChannelD2HMap.find(handle) != hcomm::g_ChannelD2HMap.end()) { - HCCL_ERROR("[%s] channel handle already exists [0x%llx] in g_ChannelD2HMap", __func__, handle); - return HCCL_E_INTERNAL; - } - - hcomm::g_ChannelMap.emplace(handle, std::move(tmpPtr)); - hcomm::g_ChannelD2HMap.emplace(handle, handle); - } - } - - return HCCL_SUCCESS; + return ChannelProcess::CreateChannelsLoop(endpointHandle, engine, channelDescs, channelNum, channels); } -static HcclResult CombineHostMemory(const std::vector> &hostPackBuffers, hccl::HostMem &hostPackBuf) +HcclResult HcommChannelCreate(EndpointHandle endpointHandle, CommEngine engine, + HcommChannelDesc *channelDescs, uint32_t channelNum, ChannelHandle *channels) { - if (hostPackBuffers.empty()) { - HCCL_ERROR("[%s] hostPackBuffers is empty, please check.", __func__); - return HCCL_E_PARA; - } + CHK_PTR_NULL(channelDescs); + CHK_PTR_NULL(channels); + CHK_PRT_RET((channelNum == 0), HCCL_ERROR("[%s]Invalid channelNum, channelNum[%u]", + __func__, channelNum), HCCL_E_PARA); + HCCL_INFO("[%s] START. endpointHandle[0x%llx], engine[%d], channelNum[%u].", + __func__, endpointHandle, engine, channelNum); - // 将离散数据复制到连续内存中 - u8 *dstPtr = static_cast(hostPackBuf.ptr()); // 目标内存起始地址 - u64 dstMax = hostPackBuf.size(); - u64 packSize = 0; + std::vector hostChannelHandles(channelNum); + ChannelHandle* targetChannels = hostChannelHandles.data(); - for (const auto &mem : hostPackBuffers) { - // 先检查是否会溢出，再执行复制操作 - CHK_PRT_RET((packSize + mem.size()) > dstMax, - HCCL_ERROR("[%s] fail, packSize[%llu] + mem.size[%zu] is bigger than dstMax[%llu]", - __func__, packSize, mem.size(), dstMax), - HCCL_E_PARA); - - CHK_SAFETY_FUNC_RET(memcpy_s(dstPtr, mem.size(), mem.data(), mem.size())); - dstPtr += mem.size(); // 移动目标指针 - packSize += mem.size(); - } - - HCCL_INFO("[%s] end of merging host memory, hostPackBuf.addr[%p], hostPackBuf.size[%zu]", - __func__, - hostPackBuf.ptr(), - hostPackBuf.size()); - - return HCCL_SUCCESS; -} - -static HcclResult FillChannelD2HMap(ChannelHandle *deviceChannelHandles, - ChannelHandle *hostChannelHandles, uint32_t listNum) -{ - std::lock_guard lk(hcomm::g_ChannelMapMtx); - for (uint32_t idx = 0; idx < listNum; idx++) { - auto deviceChannelHandle = deviceChannelHandles[idx]; - auto hostChannelHandle = hostChannelHandles[idx]; - HCCL_INFO("%s deviceChannelHandle[0x%llx], hostChannelHandle[0x%llx]", - __func__, - deviceChannelHandle, - hostChannelHandle); - g_ChannelD2HMap.emplace(deviceChannelHandle, hostChannelHandle); - } - - return HCCL_SUCCESS; -} - -HcclResult HcommChannelKernelLaunch(ChannelHandle *channelHandles, ChannelHandle *hostChannelHandles, uint32_t listNum, - const std::string &commTag, aclrtBinHandle binHandle) -{ - HCCL_RUN_INFO("[%s] listNum[%u], commTag[%s]", __func__, listNum, commTag.c_str()); - std::vector> hostPackBuffers(listNum); - HcclChannelUrmaRes channelParam{}; - CHK_SAFETY_FUNC_RET(memset_s(&channelParam, sizeof(channelParam), 0, sizeof(channelParam))); - - // 获取到host侧序列化的地址 - uint32_t totalListNum = 0; - for (uint32_t index = 0; index < listNum; index++) { - auto aicpuTsUrmaChannel = reinterpret_cast(hostChannelHandles[index]); - CHK_PRT(aicpuTsUrmaChannel->H2DResPack(hostPackBuffers[index])); - totalListNum += hostPackBuffers[index].size(); - } - HCCL_INFO("[%s] totalListNum[%llu]", __func__, totalListNum); - - // 分配连续的host内存，将序列化的地址放入其中 - hccl::HostMem hostPackBuf = hccl::HostMem::alloc(totalListNum); - CHK_PTR_NULL(hostPackBuf.ptr()); - CHK_RET(CombineHostMemory(hostPackBuffers, hostPackBuf)); - hccl::DeviceMem devicePackBuf = hccl::DeviceMem::alloc(totalListNum); - CHK_PTR_NULL(devicePackBuf.ptr()); - - // 将host侧序列化内容拷贝到device侧内存中 - CHK_RET(hrtMemSyncCopy(devicePackBuf.ptr(), - totalListNum, - hostPackBuf.ptr(), - totalListNum, - HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE)); - - // 为device侧的channelList分配内存 - hccl::DeviceMem deviceChannelList = hccl::DeviceMem::alloc(listNum * sizeof(ChannelHandle)); - CHK_PTR_NULL(deviceChannelList.ptr()); - - // channelParam资源参数填充 - s32 sRet = strncpy_s(channelParam.hcomId, HCOMID_MAX_LENGTH, commTag.c_str(), HCOMID_MAX_LENGTH - 1); - CHK_PRT_RET(sRet != EOK, HCCL_ERROR("[%s] str copy fail. return[%d]", __func__, sRet), HCCL_E_INTERNAL); - - channelParam.channelList = static_cast(deviceChannelList.ptr()); - channelParam.listNum = listNum; - channelParam.uniqueIdAddr = static_cast(devicePackBuf.ptr()); - channelParam.uniqueIdSize = totalListNum; - channelParam.singleUniqueIdSize = totalListNum / hostPackBuffers.size(); - hccl::DeviceMem remoteRankList = hccl::DeviceMem::alloc(listNum * sizeof(u32)); - CHK_PTR_NULL(remoteRankList.ptr()); - std::vector remoteRankIdList(listNum); - for ( u32 i = 0; i < listNum; ++i) { - CHK_RET(hccl::HcclCommDfx::GetChannelRemoteRankId(commTag, hostChannelHandles[i], remoteRankIdList[i])); - } - // 通过安全的内存拷贝将主机内存数据传输到设备内存 - CHK_RET(hrtMemSyncCopy(remoteRankList.ptr(), listNum * sizeof(u32), remoteRankIdList.data(), - listNum * sizeof(u32), HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE)); - channelParam.remoteRankList = static_cast(remoteRankList.ptr()); - // 创建局部流 - hccl::Stream localStream(hccl::StreamType::STREAM_TYPE_ONLINE); - constexpr u32 aicpuStreamMode = 1; - CHK_RET(hrtStreamSetMode(localStream.ptr(), aicpuStreamMode)); - - // 下kernel - std::string kernelName = "RunAicpuIndOpChannelInitV2"; - struct InitTask { - u64 context; - bool isCustom; - }; - // 拷贝channelParam到device - hccl::DeviceMem addr = hccl::DeviceMem::alloc(sizeof(channelParam)); - CHK_PTR_NULL(addr.ptr()); - - CHK_RET(hrtMemSyncCopy(addr.ptr(), - sizeof(channelParam), - &channelParam, - sizeof(channelParam), - HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE)); - - InitTask customInitTask = {0}; - customInitTask.context = reinterpret_cast(addr.ptr()); - customInitTask.isCustom = false; - u16 timeOut = NOTIFY_DEFAULT_WAIT_TIME > std::numeric_limits::max() ? - std::numeric_limits::max() : NOTIFY_DEFAULT_WAIT_TIME; - CHK_RET(hccl::AicpuAclKernelLaunch(localStream.ptr(), - reinterpret_cast(&customInitTask), - sizeof(customInitTask), - binHandle, - kernelName, - true, - timeOut)); - - CHK_RET( - hcclStreamSynchronize(localStream.ptr(), hccl::CommConfiger::GetInstance().GetCommConfigExecTimeOut(commTag))); - - // 将device侧的channelList拷贝回host侧的channelList - CHK_RET(hrtMemSyncCopy(channelHandles, - listNum * sizeof(ChannelHandle), - deviceChannelList.ptr(), - listNum * sizeof(ChannelHandle), - HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST)); - - CHK_RET(FillChannelD2HMap(channelHandles, hostChannelHandles, listNum)); - - HCCL_INFO("[%s] channel kernel launch success.", __func__); + CHK_RET(ChannelProcess::CreateChannelsLoop(endpointHandle, engine, channelDescs, channelNum, targetChannels)); + CHK_RET(ChannelProcess::ConnectChannels(targetChannels, channelNum, engine)); + CHK_RET(EnsureKernelBinLoaded(engine)); + CHK_RET(ChannelProcess::SaveChannels(targetChannels, channels, channelNum, engine, g_BinHandle)); return HCCL_SUCCESS; } HcclResult HcommChannelGet(const ChannelHandle channelHandle, void **channel) { - CHK_PTR_NULL(channel); - - const auto &D2HhandleIter = hcomm::g_ChannelD2HMap.find(channelHandle); - if (D2HhandleIter == hcomm::g_ChannelD2HMap.end()) { - HCCL_ERROR("[Hcomm][%s] channel[%llx] not found.", __func__, channelHandle); - return HcclResult::HCCL_E_NOT_FOUND; - } - - const auto handle = D2HhandleIter->second; - const auto &handleIter = hcomm::g_ChannelMap.find(handle); - if (handleIter == hcomm::g_ChannelMap.end()) { - HCCL_ERROR("[Hcomm][%s] channel[%llx] not found.", __func__, handle); - return HcclResult::HCCL_E_NOT_FOUND; - } - *channel = reinterpret_cast(handleIter->second.get()); - return HcclResult::HCCL_SUCCESS; + return ChannelProcess::ChannelGet(channelHandle, channel); } -HcclResult HcommChannelGetStatus(const ChannelHandle *channelList, uint32_t listNum, int32_t *statusList) +HcclResult HcommChannelGetStatus(const ChannelHandle *channelList, uint32_t listNum, int32_t* statusList) { - // 不得随意添加无效日志，可能造成刷屏 + // 当前为非阻塞式建链，直接返回成功 + // 参数校验 CHK_PTR_NULL(channelList); CHK_PTR_NULL(statusList); - - u32 readyCount = 0; - - for (uint32_t i = 0; i < listNum; ++i) { - const ChannelHandle inHandle = channelList[i]; - int32_t status = 0; - - // 单锁：D2H 映射 + 查 map + 锁内调用 GetStatus() - HcclResult ret = hcomm::WithChannelByHandleLocked(inHandle, [&](Channel &channel) -> HcclResult { - status = channel.GetStatus(); // 锁内调用，防止 destroy 并发释放 - return HcclResult::HCCL_SUCCESS; - }); - - if (ret != HcclResult::HCCL_SUCCESS) { - HCCL_ERROR("[%s] Get ChannelHandle failed.", __func__); - return ret; - } - CHK_PRT_RET( - status == ChannelStatus::FAILED, HCCL_ERROR("%s failed, status[%d]", __func__, status), HCCL_E_NETWORK); - - CHK_PRT_RET(status == ChannelStatus::SOCKET_TIMEOUT, - HCCL_ERROR("%s timeout, status[%d]", __func__, status), - HCCL_E_TIMEOUT); - - readyCount += (status == ChannelStatus::READY) ? 1 : 0; - statusList[i] = status; + CHK_PRT_RET((listNum == 0), HCCL_ERROR("[%s]Invalid listNum, listNum[%u]", + __func__, listNum), HCCL_E_PARA); + // 为每个通道设置成功状态 + for (uint32_t i = 0; i < listNum; i++) { + statusList[i] = 0; } - - HcclResult finalRet = (readyCount == listNum) ? HCCL_SUCCESS : HCCL_E_AGAIN; - return finalRet; + return HCCL_SUCCESS; } HcclResult HcommChannelGetNotifyNum(ChannelHandle channelHandle, uint32_t *notifyNum) { - return hcomm::WithChannelByHandleLocked(channelHandle, [&](Channel &channel) -> HcclResult { - // 锁内调用，避免 destroy 并发释放 - channel.GetNotifyNum(notifyNum); - return HcclResult::HCCL_SUCCESS; - }); + return ChannelProcess::ChannelGetNotifyNum(channelHandle, notifyNum); } HcclResult HcommChannelDestroy(const ChannelHandle *channels, uint32_t channelNum) { - CHK_PTR_NULL(channels); - - // 单锁：g_ChannelMapMtx 同时保护 g_ChannelMap + g_ChannelD2HMap - std::lock_guard lk(hcomm::g_ChannelMapMtx); - - for (uint32_t i = 0; i < channelNum; ++i) { - const ChannelHandle inHandle = channels[i]; - - // 1) 先做 D2H 映射（统一销毁入口 handle） - auto itH = hcomm::g_ChannelD2HMap.find(inHandle); - if (itH == hcomm::g_ChannelD2HMap.end()) { - HCCL_ERROR( - "[Hcomm][%s] failed to find handle mapping in g_ChannelD2HMap, inHandle[0x%llx].", __func__, inHandle); - return HcclResult::HCCL_E_NOT_FOUND; - } - const ChannelHandle mappedHandle = itH->second; - - // 2) 从 ChannelMap 删除 channel（以 mappedHandle 为准） - auto itC = hcomm::g_ChannelMap.find(mappedHandle); - if (itC == hcomm::g_ChannelMap.end()) { - HCCL_ERROR("[Hcomm][%s] failed to find channel in g_ChannelMap, inHandle[0x%llx], mappedHandle[0x%llx].", - __func__, - inHandle, - mappedHandle); - return HcclResult::HCCL_E_NOT_FOUND; - } - - HCCL_INFO("[Hcomm][%s] erase channel: inHandle[0x%llx], mappedHandle[0x%llx], ptr[%p]", - __func__, - inHandle, - mappedHandle, - itC->second.get()); - - // 3) 先 erase ChannelMap（unique_ptr 释放对象） - hcomm::g_ChannelMap.erase(itC); - - // 4) 清理 D2HMap 中所有指向 mappedHandle 的映射，避免残留导致后续查到“已销毁” - for (auto it = hcomm::g_ChannelD2HMap.begin(); it != hcomm::g_ChannelD2HMap.end();) { - if (it->second == mappedHandle) { - it = hcomm::g_ChannelD2HMap.erase(it); - } else { - ++it; - } - } - } - - return HcclResult::HCCL_SUCCESS; + return ChannelProcess::ChannelDestroy(channels, channelNum, g_BinHandle); } HcclResult HcommChannelGetRemoteMem(ChannelHandle channelHandle, HcommMem **remoteMem, uint32_t *memNum, char **memTags) { - HcclMem **remoteMemConverted = reinterpret_cast(remoteMem); - - return hcomm::WithChannelByHandleLocked(channelHandle, [&](Channel &channel) -> HcclResult { - // 锁内调用，避免 destroy 并发释放 - channel.GetRemoteMem(remoteMemConverted, memNum, memTags); - return HcclResult::HCCL_SUCCESS; - }); -} - -HcclResult HcommChannelGetUserRemoteMem(ChannelHandle channelHandle, CommMem **remoteMem, char ***memTag, uint32_t *memNum) -{ - return hcomm::WithChannelByHandleLocked(channelHandle, [&](Channel &channel) -> HcclResult { - // 锁内调用，避免 destroy 并发释放 - channel.GetUserRemoteMem(remoteMem, memTag, memNum); - return HcclResult::HCCL_SUCCESS; - }); + return ChannelProcess::ChannelGetRemoteMem(channelHandle, remoteMem, memNum, memTags); } HcclResult HcommThreadAlloc(CommEngine engine, uint32_t threadNum, uint32_t notifyNumPerThread, ThreadHandle *threads) {