已合并
支持采用HcommChannelDrain实现channel的阻塞完成功能调用 #2771
YantingLiu创建于 15 天前
支持采用HcommChannelDrain实现channel的阻塞完成功能调用 #2771
已合并
共 10 个文件变更+63-13
| @@ -56,6 +56,7 @@ DEFINE_WEAK_FUNC(int32_t, HcommFlush); | |||
| 56 | DEFINE_WEAK_FUNC(int32_t, HcommChannelFence, ChannelHandle channel); | 56 | DEFINE_WEAK_FUNC(int32_t, HcommChannelFence, ChannelHandle channel); |
| 57 | DEFINE_WEAK_FUNC(int32_t, HcommFenceOnThread, ThreadHandle thread); | 57 | DEFINE_WEAK_FUNC(int32_t, HcommFenceOnThread, ThreadHandle thread); |
| 58 | DEFINE_WEAK_FUNC(int32_t, HcommChannelFenceOnThread, ThreadHandle thread, ChannelHandle channel); | 58 | DEFINE_WEAK_FUNC(int32_t, HcommChannelFenceOnThread, ThreadHandle thread, ChannelHandle channel); |
| 59 | +DEFINE_WEAK_FUNC(int32_t, HcommChannelDrainOnThread, ThreadHandle thread, ChannelHandle channel); | ||
| 59 | DEFINE_WEAK_FUNC(HcclResult, HcommThreadJoin, ThreadHandle thread, uint32_t timeout); | 60 | DEFINE_WEAK_FUNC(HcclResult, HcommThreadJoin, ThreadHandle thread, uint32_t timeout); |
| 60 | DEFINE_WEAK_FUNC( | 61 | DEFINE_WEAK_FUNC( |
| 61 | HcclResult, HcclSymWinGetPeerPointer, HcclCommSymWindow winHandle, size_t offset, uint32_t peerRank, void** ptr); | 62 | HcclResult, HcclSymWinGetPeerPointer, HcclCommSymWindow winHandle, size_t offset, uint32_t peerRank, void** ptr); |
| @@ -124,6 +125,7 @@ void HcommPrimitivesDlInit(void* libHcommHandle) | |||
| 124 | INIT_SUPPORT_FLAG(libHcommHandle, HcommChannelFence); | 125 | INIT_SUPPORT_FLAG(libHcommHandle, HcommChannelFence); |
| 125 | INIT_SUPPORT_FLAG(libHcommHandle, HcommFenceOnThread); | 126 | INIT_SUPPORT_FLAG(libHcommHandle, HcommFenceOnThread); |
| 126 | INIT_SUPPORT_FLAG(libHcommHandle, HcommChannelFenceOnThread); | 127 | INIT_SUPPORT_FLAG(libHcommHandle, HcommChannelFenceOnThread); |
| 128 | + INIT_SUPPORT_FLAG(libHcommHandle, HcommChannelDrainOnThread); | ||
| 127 | INIT_SUPPORT_FLAG(libHcommHandle, HcommThreadJoin); | 129 | INIT_SUPPORT_FLAG(libHcommHandle, HcommThreadJoin); |
| 128 | INIT_SUPPORT_FLAG(libHcommHandle, HcclSymWinGetPeerPointer); | 130 | INIT_SUPPORT_FLAG(libHcommHandle, HcclSymWinGetPeerPointer); |
| 129 | INIT_SUPPORT_FLAG(libHcommHandle, HcclCommSymWinGet); | 131 | INIT_SUPPORT_FLAG(libHcommHandle, HcclCommSymWinGet); |
| @@ -107,6 +107,7 @@ DECL_WEAK_FUNC( | |||
| 107 | int32_t, HcommWriteWithNotifyNbiOnThread, ThreadHandle thread, ChannelHandle channel, void* dst, const void* src, | 107 | int32_t, HcommWriteWithNotifyNbiOnThread, ThreadHandle thread, ChannelHandle channel, void* dst, const void* src, |
| 108 | uint64_t len, uint32_t remoteNotifyIdx); | 108 | uint64_t len, uint32_t remoteNotifyIdx); |
| 109 | DECL_WEAK_FUNC(int32_t, HcommChannelFenceOnThread, ThreadHandle thread, ChannelHandle channel); | 109 | DECL_WEAK_FUNC(int32_t, HcommChannelFenceOnThread, ThreadHandle thread, ChannelHandle channel); |
| 110 | +DECL_WEAK_FUNC(int32_t, HcommChannelDrainOnThread, ThreadHandle thread, ChannelHandle channel); | ||
 | |||
| 110 | DECL_SUPPORT_FLAG(HcommBatchTransferOnThread); | 111 | DECL_SUPPORT_FLAG(HcommBatchTransferOnThread); |
| 111 | DECL_SUPPORT_FLAG(HcommThreadResAcquireTimeOut); | 112 | DECL_SUPPORT_FLAG(HcommThreadResAcquireTimeOut); |
| 112 | DECL_SUPPORT_FLAG(HcommSetNotifyWaitTimeOut); | 113 | DECL_SUPPORT_FLAG(HcommSetNotifyWaitTimeOut); |
| @@ -282,7 +282,7 @@ HcclResult InsTempBatchSendRecvDpu::DPUKernelRun( | |||
| 282 | return HCCL_SUCCESS; | 282 | return HCCL_SUCCESS; |
| 283 | } else if (tempAlgParam.opType == BatchSendRecvOpType::FENCE) { | 283 | } else if (tempAlgParam.opType == BatchSendRecvOpType::FENCE) { |
| 284 | ChannelInfo linkRecv = channelIter->second[0]; | 284 | ChannelInfo linkRecv = channelIter->second[0]; |
| 285 | - CHK_RET(static_cast<HcclResult>(HcommChannelFenceOnThread(0, linkRecv.handle))); | 285 | + CHK_RET(HcommChannelDrainOnThreadWithCompat(0, linkRecv.handle)); |
| 286 | CHK_RET(static_cast<HcclResult>(HcommFenceOnThread(0))); | 286 | CHK_RET(static_cast<HcclResult>(HcommFenceOnThread(0))); |
| 287 | } | 287 | } |
| 288 | 288 | ||
| @@ -11,9 +11,43 @@ | |||
| 11 | 11 | ||
| 12 | 12 | ||
| 13 | 13 | ||
| 14 | + | ||
| 15 | + | ||
| 16 | + | ||
| 17 | + | ||
| 18 | + | ||
| 14 | 19 | ||
| 15 | namespace ops_hccl { | 20 | namespace ops_hccl { |
| 16 | 21 | ||
| 22 | + | ||
| 23 | +HcclResult HcommChannelDrainOnThreadWithCompat(ThreadHandle thread, ChannelHandle channel) | ||
| 24 | +{ | ||
| 25 | + if (channel == 0) { | ||
| 26 | + HCCL_ERROR("[HcommChannelDrainOnThread] channel is nullptr."); | ||
| 27 | + return HCCL_E_PTR; | ||
| 28 | + } | ||
| 29 | + | ||
| 30 | + constexpr int hostChannelDrainMinVersion = CANN_VERSION(9, 2, 0, 2); | ||
| 31 | + const int hcommVersion = GetHcommVersion(); | ||
| 32 | + if (hcommVersion < hostChannelDrainMinVersion) { | ||
| 33 | + HCCL_WARNING( | ||
| 34 | + "[HcommChannelDrainOnThread] HCOMM version[%d] does not support Host Channel Drain, " | ||
| 35 | + "fallback to HcommChannelFenceOnThread.", | ||
| 36 | + hcommVersion); | ||
| 37 | + return static_cast<HcclResult>(HcommChannelFenceOnThread(thread, channel)); | ||
| 38 | + } | ||
| 39 | + | ||
| 40 | + const HcclResult drainRet = static_cast<HcclResult>(HcommChannelDrainOnThread(thread, channel)); | ||
| 41 | + if (drainRet == HCCL_E_PTR) { | ||
| 42 | + HCCL_ERROR( | ||
| 43 | + "[HcommChannelDrainOnThread] failed with HCCL_E_PTR, channel[0x%llx], HCOMM version[%d]. " | ||
| 44 | + "Check channel or update HCOMM.", | ||
| 45 | + static_cast<unsigned long long>(channel), hcommVersion); | ||
| 46 | + } | ||
| 47 | + return drainRet; | ||
| 48 | +} | ||
| 49 | + | ||
| 50 | + | ||
| 17 | HcclResult SendRecvWrite(const SendRecvInfo& sendRecvInfo) | 51 | HcclResult SendRecvWrite(const SendRecvInfo& sendRecvInfo) |
| 18 | { | 52 | { |
| 19 | 53 | ||
| @@ -43,7 +77,7 @@ HcclResult SendRecvWrite(const SendRecvInfo& sendRecvInfo) | |||
| 43 | CHK_RET(static_cast<HcclResult>(HcommChannelNotifyRecordOnThread(0, sendChannel.handle, NOTIFY_IDX_FIN_ACK))); | 77 | CHK_RET(static_cast<HcclResult>(HcommChannelNotifyRecordOnThread(0, sendChannel.handle, NOTIFY_IDX_FIN_ACK))); |
| 44 | CHK_RET(static_cast<HcclResult>( | 78 | CHK_RET(static_cast<HcclResult>( |
| 45 | HcommChannelNotifyWaitOnThread(0, recvChannel.handle, NOTIFY_IDX_FIN_ACK, execTimeout))); | 79 | HcommChannelNotifyWaitOnThread(0, recvChannel.handle, NOTIFY_IDX_FIN_ACK, execTimeout))); |
| 46 | - CHK_RET(static_cast<HcclResult>(HcommChannelFenceOnThread(0, sendChannel.handle))); | 80 | + CHK_RET(HcommChannelDrainOnThreadWithCompat(0, sendChannel.handle)); |
| 47 | CHK_RET(static_cast<HcclResult>(HcommFenceOnThread(0))); | 81 | CHK_RET(static_cast<HcclResult>(HcommFenceOnThread(0))); |
| 48 | 82 | ||
| 49 | return HCCL_SUCCESS; | 83 | return HCCL_SUCCESS; |
| @@ -73,7 +107,7 @@ HcclResult SendWrite(const DataInfo& sendInfo) | |||
| 73 | CHK_RET(static_cast<HcclResult>(HcommChannelNotifyRecordOnThread(0, sendChannel.handle, NOTIFY_IDX_FIN_ACK))); | 107 | CHK_RET(static_cast<HcclResult>(HcommChannelNotifyRecordOnThread(0, sendChannel.handle, NOTIFY_IDX_FIN_ACK))); |
| 74 | CHK_RET(static_cast<HcclResult>( | 108 | CHK_RET(static_cast<HcclResult>( |
| 75 | HcommChannelNotifyWaitOnThread(0, sendChannel.handle, NOTIFY_IDX_FIN_ACK, execTimeout))); | 109 | HcommChannelNotifyWaitOnThread(0, sendChannel.handle, NOTIFY_IDX_FIN_ACK, execTimeout))); |
| 76 | - CHK_RET(static_cast<HcclResult>(HcommChannelFenceOnThread(0, sendChannel.handle))); | 110 | + CHK_RET(HcommChannelDrainOnThreadWithCompat(0, sendChannel.handle)); |
B YantingLiu 7 天前 评论： 7 天前 评论： | |||
| 77 | CHK_RET(static_cast<HcclResult>(HcommFenceOnThread(0))); | 111 | CHK_RET(static_cast<HcclResult>(HcommFenceOnThread(0))); |
| 78 | 112 | ||
| 79 | return HCCL_SUCCESS; | 113 | return HCCL_SUCCESS; |
| @@ -99,7 +133,7 @@ HcclResult RecvWrite(const DataInfo& recvInfo) | |||
| 99 | CHK_RET(static_cast<HcclResult>(HcommChannelNotifyRecordOnThread(0, recvChannel.handle, NOTIFY_IDX_FIN_ACK))); | 133 | CHK_RET(static_cast<HcclResult>(HcommChannelNotifyRecordOnThread(0, recvChannel.handle, NOTIFY_IDX_FIN_ACK))); |
| 100 | CHK_RET(static_cast<HcclResult>( | 134 | CHK_RET(static_cast<HcclResult>( |
| 101 | HcommChannelNotifyWaitOnThread(0, recvChannel.handle, NOTIFY_IDX_FIN_ACK, execTimeout))); | 135 | HcommChannelNotifyWaitOnThread(0, recvChannel.handle, NOTIFY_IDX_FIN_ACK, execTimeout))); |
| 102 | - CHK_RET(static_cast<HcclResult>(HcommChannelFenceOnThread(0, recvChannel.handle))); | 136 | + CHK_RET(HcommChannelDrainOnThreadWithCompat(0, recvChannel.handle)); |
| 103 | CHK_RET(static_cast<HcclResult>(HcommFenceOnThread(0))); | 137 | CHK_RET(static_cast<HcclResult>(HcommFenceOnThread(0))); |
| 104 | 138 | ||
| 105 | return HCCL_SUCCESS; | 139 | return HCCL_SUCCESS; |
| @@ -18,6 +18,10 @@ | |||
| 18 | 18 | ||
| 19 | namespace ops_hccl { | 19 | namespace ops_hccl { |
| 20 | 20 | ||
| 21 | + | ||
| 22 | +HcclResult HcommChannelDrainOnThreadWithCompat(ThreadHandle thread, ChannelHandle channel); | ||
| 23 | + | ||
| 24 | + | ||
| 21 | HcclResult SendWrite(const DataInfo& sendInfo); | 25 | HcclResult SendWrite(const DataInfo& sendInfo); |
| 22 | HcclResult RecvWrite(const DataInfo& recvInfo); | 26 | HcclResult RecvWrite(const DataInfo& recvInfo); |
| 23 | HcclResult SendRecvWrite(const SendRecvInfo& sendRecvInfo); | 27 | HcclResult SendRecvWrite(const SendRecvInfo& sendRecvInfo); |
| @@ -9,6 +9,7 @@ | |||
| 9 | */ | 9 | */ |
| 10 | 10 | ||
| 11 | 11 | ||
| 12 | + | ||
| 12 | 13 | ||
| 13 | 14 | ||
| 14 | 15 | ||
| @@ -159,15 +160,15 @@ HcclResult DpuBatchTransfer(std::vector<DpuTransferCtx>& pairs) | |||
| 159 | } | 160 | } |
| 160 | } | 161 | } |
| 161 | } | 162 | } |
| 162 | - // Fence：发送通道 + 接收通道（去重，samePeer 时仅一次） | 163 | + // Drain：发送通道 + 接收通道（去重，samePeer 时仅一次） |
| 163 | for (auto& p : pairs) { | 164 | for (auto& p : pairs) { |
| 164 | - bool txFenced = false; | 165 | + bool txDrained = false; |
| 165 | if (p.hasSend()) { | 166 | if (p.hasSend()) { |
| 166 | - CHK_RET(static_cast<HcclResult>(HcommChannelFenceOnThread(0, p.txCh->handle))); | 167 | + CHK_RET(HcommChannelDrainOnThreadWithCompat(0, p.txCh->handle)); |
| 167 | - txFenced = true; | 168 | + txDrained = true; |
| 168 | } | 169 | } |
| 169 | - if (p.hasRecv() && (!txFenced || p.rxCh->handle != p.txCh->handle)) { | 170 | + if (p.hasRecv() && (!txDrained || p.rxCh->handle != p.txCh->handle)) { |
| 170 | - CHK_RET(static_cast<HcclResult>(HcommChannelFenceOnThread(0, p.rxCh->handle))); | 171 | + CHK_RET(HcommChannelDrainOnThreadWithCompat(0, p.rxCh->handle)); |
| 171 | } | 172 | } |
| 172 | } | 173 | } |
| 173 | CHK_RET(static_cast<HcclResult>(HcommFenceOnThread(0))); | 174 | CHK_RET(static_cast<HcclResult>(HcommFenceOnThread(0))); |
| @@ -58,7 +58,7 @@ struct DpuTransferCtx { | |||
| 58 | /** | 58 | /** |
| 59 | * @brief 三阶段批量传输：前同步 → 写数据 → 后同步 | 59 | * @brief 三阶段批量传输：前同步 → 写数据 → 后同步 |
| 60 | * 调用方负责填充 DpuTransferCtx 列表（含正确的 txDst/txSrc 偏移）， | 60 | * 调用方负责填充 DpuTransferCtx 列表（含正确的 txDst/txSrc 偏移）， |
| 61 | - * 本函数批量执行 Phase 1（step sync）/ Phase 2（HcommWriteNbi）/ Phase 3（data sync + Fence）。 | 61 | + * 本函数批量执行 Phase 1（step sync）/ Phase 2（HcommWriteNbi）/ Phase 3（data sync + Drain）。 |
| 62 | */ | 62 | */ |
| 63 | HcclResult DpuBatchTransfer(std::vector<DpuTransferCtx>& pairs); | 63 | HcclResult DpuBatchTransfer(std::vector<DpuTransferCtx>& pairs); |
| 64 | 64 | ||
| @@ -156,7 +156,7 @@ HcclResult InsTempRecvHostNicDpu::DPUKernelRun( | |||
| 156 | CHK_RET(static_cast<HcclResult>( | 156 | CHK_RET(static_cast<HcclResult>( |
| 157 | HcommChannelNotifyRecordOnThread(0, channels.at(rankIdx)[0].handle, notifyNum))); | 157 | HcommChannelNotifyRecordOnThread(0, channels.at(rankIdx)[0].handle, notifyNum))); |
| 158 | 158 | ||
| 159 | - CHK_RET(static_cast<HcclResult>(HcommChannelFenceOnThread(0, channels.at(rankIdx)[0].handle))); | 159 | + CHK_RET(HcommChannelDrainOnThreadWithCompat(0, channels.at(rankIdx)[0].handle)); |
| 160 | } | 160 | } |
| 161 | } | 161 | } |
| 162 | 162 | ||
| @@ -159,7 +159,7 @@ HcclResult InsTempSendHostNicDpu::DPUKernelRun( | |||
| 159 | // 后同步 | 159 | // 后同步 |
| 160 | CHK_RET(static_cast<HcclResult>( | 160 | CHK_RET(static_cast<HcclResult>( |
| 161 | HcommChannelNotifyWaitOnThread(0, channels.at(rankIdx)[0].handle, 2, timeOutSize))); | 161 | HcommChannelNotifyWaitOnThread(0, channels.at(rankIdx)[0].handle, 2, timeOutSize))); |
| 162 | - CHK_RET(static_cast<HcclResult>(HcommChannelFenceOnThread(0, channels.at(rankIdx)[0].handle))); | 162 | + CHK_RET(HcommChannelDrainOnThreadWithCompat(0, channels.at(rankIdx)[0].handle)); |
| 163 | } | 163 | } |
| 164 | } | 164 | } |
| 165 | 165 | ||
| @@ -907,6 +907,14 @@ int32_t HcommChannelFenceOnThread(ThreadHandle thread, ChannelHandle channel) | |||
| 907 | return 0; | 907 | return 0; |
| 908 | } | 908 | } |
| 909 | 909 | ||
| 910 | +int32_t HcommChannelDrainOnThread(ThreadHandle thread, ChannelHandle channel) | ||
| 911 | +{ | ||
| 912 | + (void)thread; | ||
| 913 | + (void)channel; | ||
| 914 | + HCCL_WARNING("[%s] not support.", __func__); | ||
| 915 | + return 0; | ||
| 916 | +} | ||
| 917 | + | ||
| 910 | int32_t HcommWaitResponse(uint64_t handle, void* dst, size_t sizeByte, uint32_t* msgId) | 918 | int32_t HcommWaitResponse(uint64_t handle, void* dst, size_t sizeByte, uint32_t* msgId) |
| 911 | { | 919 | { |
| 912 | HCCL_WARNING("[%s] not support.", __func__); | 920 | HCCL_WARNING("[%s] not support.", __func__); |
[LOW] Drain 支持标志已生成未声明，与 HcommBatchTransferOnThread 等既有兼容模式不一致
问题分析：
DECL_WEAK_FUNC(int32_t, HcommChannelDrainOnThread, ...)。DEFINE_WEAK_FUNC会自动生成HcommIsSupportHcommChannelDrainOnThread()访问器，:128 的INIT_SUPPORT_FLAG在运行时通过 dlsym 记录符号是否存在；但 dl.h 未加DECL_SUPPORT_FLAG(HcommChannelDrainOnThread);，其他编译单元（如 dpu_alg_data_trans_wrapper.h 的兼容包装）无法调用该标志——当前该标志没有任何使用方，每次初始化的 dlsym 探测结果被浪费。HcommBatchTransferOnThread等均是 "DECL_WEAK_FUNC + DECL_SUPPORT_FLAG" 成对出现（op_common.cc:690 即其消费方）。建议修复： 在第 110 行后补一行声明，供兼容包装做符号预判（与本 PR dpu_alg_data_trans_wrapper.h:22 的
HcommChannelDrainOnThreadWithCompat配合）：DECL_WEAK_FUNC(int32_t, HcommChannelDrainOnThread, ThreadHandle thread, ChannelHandle channel); DECL_SUPPORT_FLAG(HcommChannelDrainOnThread);