| @@ -20,6 +20,7 @@ | |||
| 20 | #include "local_rdma_rma_buffer_manager.h" | 20 | #include "local_rdma_rma_buffer_manager.h" |
| 21 | #include "local_rdma_rma_buffer_v2.h" | 21 | #include "local_rdma_rma_buffer_v2.h" |
| 22 | #include "hccl_one_sided_data.h" | 22 | #include "hccl_one_sided_data.h" |
| 23 | +#include <unistd.h> | ||
| 23 | 24 | ||
| 24 | namespace hcomm_experimental { | 25 | namespace hcomm_experimental { |
| 25 | 26 | ||
| @@ -141,6 +142,11 @@ HcclResult RoceRegedMemMgr::GetMemDesc(const EndpointDesc endpointDesc, Hccl::Lo | |||
| 141 | 142 | ||
| 142 | tempLocalMemDesc.insert(tempLocalMemDesc.end(), tempLocalEndpointDesc.begin(), tempLocalEndpointDesc.end()); | 143 | tempLocalMemDesc.insert(tempLocalMemDesc.end(), tempLocalEndpointDesc.begin(), tempLocalEndpointDesc.end()); |
| 143 | 144 | ||
| 145 | + // 增加本进程pid | ||
| 146 | + const uint32_t processId = static_cast<uint32_t>(getpid()); | ||
| 147 | + const char* processIdBytes = reinterpret_cast<const char*>(&processId); | ||
| 148 | + tempLocalMemDesc.insert(tempLocalMemDesc.end(), processIdBytes, processIdBytes + sizeof(processId)); | ||
| 149 | + | ||
| 144 | // 内存描述符拷贝 | 150 | // 内存描述符拷贝 |
| 145 | localRdmaRmaBuffer->Desc = std::move(tempLocalMemDesc); | 151 | localRdmaRmaBuffer->Desc = std::move(tempLocalMemDesc); |
| 146 | return HCCL_SUCCESS; | 152 | return HCCL_SUCCESS; |
| @@ -162,20 +168,27 @@ RoceRegedMemMgr::MemoryExport(const EndpointDesc endpointDesc, void* memHandle, | |||
| 162 | } | 168 | } |
| 163 | 169 | ||
| 164 | HcclResult RoceRegedMemMgr::GetParamsFromMemDesc( | 170 | HcclResult RoceRegedMemMgr::GetParamsFromMemDesc( |
| 165 | - const void* memDesc, uint32_t descLen, EndpointDesc& endpointDesc, Hccl::ExchangeRdmaBufferDto& dto) | 171 | + const void* memDesc, uint32_t descLen, EndpointDesc& endpointDesc, Hccl::ExchangeRdmaBufferDto& dto, uint32_t& pid) |
| 166 | { | 172 | { |
| 167 | const char* description = static_cast<const char*>(memDesc); | 173 | const char* description = static_cast<const char*>(memDesc); |
| 174 | + constexpr size_t tailLen = sizeof(EndpointDesc) + sizeof(uint32_t); | ||
| 168 | 175 | ||
| 169 | - if (descLen < sizeof(EndpointDesc)) { | 176 | + if (descLen < tailLen) { |
| 170 | HCCL_ERROR( | 177 | HCCL_ERROR( |
| 171 | "[RoceRegedMemMgr][GetParamsFromMemDesc] [%s] descLen[%u] is too small. aim size:[%llu]", __func__, descLen, | 178 | "[RoceRegedMemMgr][GetParamsFromMemDesc] [%s] descLen[%u] is too small. aim size:[%llu]", __func__, descLen, |
| 172 | - sizeof(EndpointDesc)); | 179 | + tailLen); |
| 173 | return HCCL_E_INTERNAL; | 180 | return HCCL_E_INTERNAL; |
| 174 | } | 181 | } |
| 182 | + | ||
| 183 | + // 从memDesc末尾提取pid | ||
| 184 | + if (memcpy_s(&pid, sizeof(uint32_t), description + descLen - sizeof(uint32_t), sizeof(uint32_t)) != EOK) { | ||
| 185 | + HCCL_ERROR( | ||
| 186 | + "[RoceRegedMemMgr][GetParamsFromMemDesc] [%s] pid copy error. aim size:[%llu]", __func__, sizeof(uint32_t)); | ||
| 187 | + return HCCL_E_INTERNAL; | ||
| 188 | + } | ||
| 189 | + | ||
| 175 | // 从memDesc末尾提取EndpointDesc | 190 | // 从memDesc末尾提取EndpointDesc |
| 176 | - if (memcpy_s( | 191 | + if (memcpy_s(&endpointDesc, sizeof(EndpointDesc), description + descLen - tailLen, sizeof(EndpointDesc)) != EOK) { |
| 177 | - &endpointDesc, sizeof(EndpointDesc), description + descLen - sizeof(EndpointDesc), sizeof(EndpointDesc)) | ||
| 178 | - != EOK) { | ||
| 179 | HCCL_ERROR( | 192 | HCCL_ERROR( |
| 180 | "[RoceRegedMemMgr][GetParamsFromMemDesc] [%s] endpointDesc copy error. aim size:[%llu]", __func__, | 193 | "[RoceRegedMemMgr][GetParamsFromMemDesc] [%s] endpointDesc copy error. aim size:[%llu]", __func__, |
| 181 | sizeof(EndpointDesc)); | 194 | sizeof(EndpointDesc)); |
| @@ -185,7 +198,7 @@ HcclResult RoceRegedMemMgr::GetParamsFromMemDesc( | |||
| 185 | // 反序列化 | 198 | // 反序列化 |
| 186 | std::vector<char> tempDesc{}; | 199 | std::vector<char> tempDesc{}; |
| 187 | tempDesc.resize(TRANSPORT_EMD_ESC_SIZE); | 200 | tempDesc.resize(TRANSPORT_EMD_ESC_SIZE); |
| 188 | - tempDesc.assign(description, description + descLen - sizeof(EndpointDesc)); | 201 | + tempDesc.assign(description, description + descLen - tailLen); |
| 189 | Hccl::BinaryStream remoteRdmaRmaBufferStream(tempDesc); | 202 | Hccl::BinaryStream remoteRdmaRmaBufferStream(tempDesc); |
| 190 | dto.Deserialize(remoteRdmaRmaBufferStream); | 203 | dto.Deserialize(remoteRdmaRmaBufferStream); |
| 191 | return HCCL_SUCCESS; | 204 | return HCCL_SUCCESS; |
| @@ -194,27 +207,30 @@ HcclResult RoceRegedMemMgr::GetParamsFromMemDesc( | |||
| 194 | HcclResult RoceRegedMemMgr::MemoryImport(const void* memDesc, uint32_t descLen, HcommMem* outMem) | 207 | HcclResult RoceRegedMemMgr::MemoryImport(const void* memDesc, uint32_t descLen, HcommMem* outMem) |
[MEDIUM] key 维度扩展、加锁、空 Mgr 清理等行为变更无任何 UT 覆盖 第 207 行:HcclResult RoceRegedMemMgr::MemoryImport(const void* memDesc, uint32_t descLen, HcommMem* outMem) 问题分析:本 PR 引入三处行为变更——①remoteRdmaRmaBufferMgrs_ 的 key 从 EndpointDesc 扩展为 (EndpointDesc, pid);②MemoryImport/MemoryUnimport 新增 memMtx_ 互斥保护;③MemoryUnimport 在 Del 后对空 RemoteRdmaRmaBufferMgr 做 erase 清理(roce_mem.cc:273-275)。但 git diff 显示 test/ 目录零变更,PR 模板「补充的UT用例」栏也是空的。experimental/base_comm 参与编译链接(cpu_roce_endpoint.cc 直接构造 RoceRegedMemMgr),上述逻辑的正确性(尤其 Import/Unimport 的 key 对称性与清理时机)完全依赖人工验证,后续重构无回归防护。 修复建议:补充 UT 用例(建议放 test/ut/framework/next/comms/endpoints/reged_mems/ 下,仿 ut_roce_mem_test.cc 的 mockcpp 模式,对 experimental 版另建文件):
![]() ![]() | |||
| 195 | { | 208 | { |
| 196 | HCCL_INFO("[%s] Begin", __FUNCTION__); | 209 | HCCL_INFO("[%s] Begin", __FUNCTION__); |
| 210 | + std::lock_guard<std::mutex> lock(memMtx_); | ||
| 197 | 211 | ||
| 198 | EndpointDesc endpointDesc; | 212 | EndpointDesc endpointDesc; |
| 199 | Hccl::ExchangeRdmaBufferDto dto; | 213 | Hccl::ExchangeRdmaBufferDto dto; |
| 200 | - CHK_RET(GetParamsFromMemDesc(memDesc, descLen, endpointDesc, dto)); | 214 | + uint32_t remotePid = 0; |
| 215 | + CHK_RET(GetParamsFromMemDesc(memDesc, descLen, endpointDesc, dto, remotePid)); | ||
| 216 | + HCCL_INFO("[RoceRegedMemMgr][MemoryImport] remote endpoint pid[%u]", remotePid); | ||
[HIGH] MemoryImport/MemoryUnimport 在 experimental 版中未加锁,存在线程安全风险
建议修复:在 ![]() ![]() | |||
| 201 | 217 | ||
| 202 | // 构造RemoteRdmaRmaBuffer | 218 | // 构造RemoteRdmaRmaBuffer |
| 203 | std::shared_ptr<Hccl::RemoteRdmaRmaBuffer> remoteRdmaRmaBuffer; | 219 | std::shared_ptr<Hccl::RemoteRdmaRmaBuffer> remoteRdmaRmaBuffer; |
| 204 | EXCEPTION_CATCH(remoteRdmaRmaBuffer = std::make_shared<Hccl::RemoteRdmaRmaBuffer>(this->rdmaHandle_, dto), | 220 | EXCEPTION_CATCH(remoteRdmaRmaBuffer = std::make_shared<Hccl::RemoteRdmaRmaBuffer>(this->rdmaHandle_, dto), |
| 205 | return HCCL_E_PTR;); | 221 | return HCCL_E_PTR;); |
| 206 | 222 | ||
| 207 | - // 放到RemoteRdmaRmaBufferMgr_ | 223 | + // key = (endpointDesc, pid),区分同一网卡上的不同远端进程 |
| 208 | Hccl::BufferKey<uintptr_t, u64> tempKey(static_cast<uintptr_t>(dto.addr), dto.size); | 224 | Hccl::BufferKey<uintptr_t, u64> tempKey(static_cast<uintptr_t>(dto.addr), dto.size); |
| 209 | - if (remoteRdmaRmaBufferMgrs_.find(endpointDesc) == remoteRdmaRmaBufferMgrs_.end()) { | 225 | + RemoteRdmaMemKey remoteKey{endpointDesc, remotePid}; |
| 226 | + if (remoteRdmaRmaBufferMgrs_.find(remoteKey) == remoteRdmaRmaBufferMgrs_.end()) { | ||
| 210 | std::unique_ptr<RemoteRdmaRmaBufferMgr> remoteRdmaRmaBufferMgr; | 227 | std::unique_ptr<RemoteRdmaRmaBufferMgr> remoteRdmaRmaBufferMgr; |
| 211 | EXCEPTION_CATCH((remoteRdmaRmaBufferMgr = std::make_unique<RemoteRdmaRmaBufferMgr>()), return HCCL_E_PTR); | 228 | EXCEPTION_CATCH((remoteRdmaRmaBufferMgr = std::make_unique<RemoteRdmaRmaBufferMgr>()), return HCCL_E_PTR); |
| 212 | - CHK_SMART_PTR_NULL(remoteRdmaRmaBufferMgr); | 229 | + remoteRdmaRmaBufferMgrs_[remoteKey] = std::move(remoteRdmaRmaBufferMgr); |
| 213 | - remoteRdmaRmaBufferMgrs_[endpointDesc] = std::move(remoteRdmaRmaBufferMgr); | ||
| 214 | HCCL_INFO("remoteRdmaRmaBufferMgrs_ add remoteRdmaRmaBufferMgr successfully!"); | 230 | HCCL_INFO("remoteRdmaRmaBufferMgrs_ add remoteRdmaRmaBufferMgr successfully!"); |
| 215 | } | 231 | } |
| 216 | 232 | ||
| 217 | - auto resultPair = remoteRdmaRmaBufferMgrs_[endpointDesc]->Add(tempKey, remoteRdmaRmaBuffer); | 233 | + auto resultPair = remoteRdmaRmaBufferMgrs_[remoteKey]->Add(tempKey, remoteRdmaRmaBuffer); |
| 218 | if (!resultPair.second) { | 234 | if (!resultPair.second) { |
| 219 | HCCL_ERROR("[RoceRegedMemMgr][MemoryImport] This memDesc has already been imported!"); | 235 | HCCL_ERROR("[RoceRegedMemMgr][MemoryImport] This memDesc has already been imported!"); |
| 220 | return HCCL_E_AGAIN; | 236 | return HCCL_E_AGAIN; |
| @@ -229,12 +245,15 @@ HcclResult RoceRegedMemMgr::MemoryImport(const void* memDesc, uint32_t descLen, | |||
| 229 | HcclResult RoceRegedMemMgr::MemoryUnimport(const void* memDesc, uint32_t descLen) | 245 | HcclResult RoceRegedMemMgr::MemoryUnimport(const void* memDesc, uint32_t descLen) |
| 230 | { | 246 | { |
| 231 | HCCL_INFO("[%s] Begin", __FUNCTION__); | 247 | HCCL_INFO("[%s] Begin", __FUNCTION__); |
| 248 | + std::lock_guard<std::mutex> lock(memMtx_); | ||
| 232 | 249 | ||
| 233 | EndpointDesc endpointDesc; | 250 | EndpointDesc endpointDesc; |
| 234 | Hccl::ExchangeRdmaBufferDto dto; | 251 | Hccl::ExchangeRdmaBufferDto dto; |
| 235 | - CHK_RET(GetParamsFromMemDesc(memDesc, descLen, endpointDesc, dto)); | 252 | + uint32_t remotePid = 0; |
| 253 | + CHK_RET(GetParamsFromMemDesc(memDesc, descLen, endpointDesc, dto, remotePid)); | ||
| 236 | 254 | ||
| 237 | - if (remoteRdmaRmaBufferMgrs_.find(endpointDesc) == remoteRdmaRmaBufferMgrs_.end()) { | 255 | + RemoteRdmaMemKey remoteKey{endpointDesc, remotePid}; |
| 256 | + if (remoteRdmaRmaBufferMgrs_.find(remoteKey) == remoteRdmaRmaBufferMgrs_.end()) { | ||
| 238 | HCCL_ERROR("[RoceRegedMemMgr][MemoryUnimport] Remote buffer manager Not Found."); | 257 | HCCL_ERROR("[RoceRegedMemMgr][MemoryUnimport] Remote buffer manager Not Found."); |
| 239 | return HCCL_E_NOT_FOUND; | 258 | return HCCL_E_NOT_FOUND; |
| 240 | } | 259 | } |
| @@ -244,13 +263,16 @@ HcclResult RoceRegedMemMgr::MemoryUnimport(const void* memDesc, uint32_t descLen | |||
| 244 | Hccl::BufferKey<uintptr_t, u64> tempKey(static_cast<uintptr_t>(dto.addr), dto.size); | 263 | Hccl::BufferKey<uintptr_t, u64> tempKey(static_cast<uintptr_t>(dto.addr), dto.size); |
| 245 | 264 | ||
| 246 | bool resultPair = false; | 265 | bool resultPair = false; |
| 247 | - EXCEPTION_CATCH(resultPair = remoteRdmaRmaBufferMgrs_[endpointDesc]->Del(tempKey), return HCCL_E_NOT_FOUND); | 266 | + EXCEPTION_CATCH(resultPair = remoteRdmaRmaBufferMgrs_[remoteKey]->Del(tempKey), return HCCL_E_NOT_FOUND); |
| 248 | // 计数器大于1时,返回false,说明框架层有其它设备在使用这段内存,返回HCCL_E_AGAIN | 267 | // 计数器大于1时,返回false,说明框架层有其它设备在使用这段内存,返回HCCL_E_AGAIN |
| 249 | if (!resultPair) { | 268 | if (!resultPair) { |
| 250 | HCCL_INFO("[RoceRegedMemMgr][[MemoryUnimport] Memory reference count is larger than 0 " | 269 | HCCL_INFO("[RoceRegedMemMgr][[MemoryUnimport] Memory reference count is larger than 0 " |
| 251 | "(used by other RemoteRank)."); | 270 | "(used by other RemoteRank)."); |
| 252 | return HCCL_E_AGAIN; | 271 | return HCCL_E_AGAIN; |
| 253 | } | 272 | } |
| 273 | + if (!remoteRdmaRmaBufferMgrs_[remoteKey]->size()) { | ||
| 274 | + remoteRdmaRmaBufferMgrs_.erase(remoteKey); | ||
| 275 | + } | ||
| 254 | return HCCL_SUCCESS; | 276 | return HCCL_SUCCESS; |
| 255 | } | 277 | } |
| 256 | 278 | ||
| @@ -11,7 +11,10 @@ | |||
| 11 | #ifndef HCOMM_EXPERIMENTAL_ROCE_MEM_H | 11 | #ifndef HCOMM_EXPERIMENTAL_ROCE_MEM_H |
| 12 | #define HCOMM_EXPERIMENTAL_ROCE_MEM_H | 12 | #define HCOMM_EXPERIMENTAL_ROCE_MEM_H |
| 13 | 13 | ||
| 14 | +#include <cstdint> | ||
| 15 | +#include <cstring> | ||
| 14 | #include <memory> | 16 | #include <memory> |
| 17 | +#include <mutex> | ||
| 15 | #include <vector> | 18 | #include <vector> |
| 16 | #include <string> | 19 | #include <string> |
| 17 | #include "reged_mem_mgr.h" | 20 | #include "reged_mem_mgr.h" |
| @@ -22,6 +25,39 @@ | |||
| 22 | #include "exchange_rdma_buffer_dto.h" | 25 | #include "exchange_rdma_buffer_dto.h" |
| 23 | 26 | ||
| 24 | namespace hcomm_experimental { | 27 | namespace hcomm_experimental { |
| 28 | + | ||
| 29 | +/** | ||
| 30 | + * @note 远端内存管理key:EndpointDesc标识网卡,pid标识该网卡上的远端进程。 | ||
| 31 | + */ | ||
| 32 | +struct RemoteRdmaMemKey { | ||
| 33 | + EndpointDesc endpointDesc{}; | ||
| 34 | + uint32_t pid{0}; | ||
[MEDIUM] pid 作为远端进程唯一标识存在复用漂移风险,重启进程可能误拒导入或误删他人内存 第 34 行:uint32_t pid{0}; 问题分析:RemoteRdmaMemKey 以 (endpointDesc, pid) 区分远端进程,pid 取自导出方 修复建议:
![]() ![]() | |||
| 35 | +}; | ||
| 36 | + | ||
| 37 | +inline bool operator==(const RemoteRdmaMemKey& lhs, const RemoteRdmaMemKey& rhs) noexcept | ||
| 38 | +{ | ||
| 39 | + return (lhs.pid == rhs.pid) && (std::memcmp(&lhs.endpointDesc, &rhs.endpointDesc, sizeof(EndpointDesc)) == 0); | ||
| 40 | +} | ||
| 41 | + | ||
| 42 | +struct RemoteRdmaMemKeyHash { | ||
| 43 | + size_t operator()(const RemoteRdmaMemKey& key) const noexcept | ||
| 44 | + { | ||
| 45 | + // FNV-1a(64位)对EndpointDesc字节序列与pid做hash | ||
| 46 | + const uint8_t* p = reinterpret_cast<const uint8_t*>(&key); | ||
| 47 | + size_t h | ||
| 48 | + = sizeof(size_t) == 8 ? static_cast<size_t>(14695981039346656037ull) : static_cast<size_t>(2166136261u); | ||
| 49 | + const size_t prime | ||
| 50 | + = sizeof(size_t) == 8 ? static_cast<size_t>(1099511628211ull) : static_cast<size_t>(16777619u); | ||
| 51 | + for (size_t i = 0; i < sizeof(key.endpointDesc); ++i) { | ||
| 52 | + h ^= static_cast<size_t>(p[i]); | ||
| 53 | + h *= prime; | ||
| 54 | + } | ||
| 55 | + h ^= static_cast<size_t>(key.pid); | ||
| 56 | + h *= prime; | ||
| 57 | + return h; | ||
| 58 | + } | ||
| 59 | +}; | ||
| 60 | + | ||
| 25 | /** | 61 | /** |
| 26 | * @note 职责:用于通信设备EndPoint的注册内存信息管理,支持基于RmaBufferMgr类的重叠内存的检测报错等。 | 62 | * @note 职责:用于通信设备EndPoint的注册内存信息管理,支持基于RmaBufferMgr类的重叠内存的检测报错等。 |
| 27 | */ | 63 | */ |
| @@ -44,12 +80,15 @@ public: | |||
| 44 | HcclResult GetAllMemHandles(void** memHandles, uint32_t* memHandleNum) override; | 80 | HcclResult GetAllMemHandles(void** memHandles, uint32_t* memHandleNum) override; |
| 45 | HcclResult GetMemDesc(const EndpointDesc endpointDesc, Hccl::LocalRdmaRmaBuffer* localRdmaRmaBuffer); | 81 | HcclResult GetMemDesc(const EndpointDesc endpointDesc, Hccl::LocalRdmaRmaBuffer* localRdmaRmaBuffer); |
| 46 | HcclResult GetParamsFromMemDesc( | 82 | HcclResult GetParamsFromMemDesc( |
| 47 | - const void* memDesc, uint32_t descLen, EndpointDesc& endpointDesc, Hccl::ExchangeRdmaBufferDto& dto); | 83 | + const void* memDesc, uint32_t descLen, EndpointDesc& endpointDesc, Hccl::ExchangeRdmaBufferDto& dto, |
| 84 | + uint32_t& pid); | ||
| 48 | 85 | ||
| 49 | private: | 86 | private: |
| 50 | std::unique_ptr<LocalRdmaRmaBufferMgr> localRdmaRmaBufferMgr_{}; | 87 | std::unique_ptr<LocalRdmaRmaBufferMgr> localRdmaRmaBufferMgr_{}; |
| 51 | std::vector<std::shared_ptr<Hccl::LocalRdmaRmaBuffer>> allRegisteredBuffers_; | 88 | std::vector<std::shared_ptr<Hccl::LocalRdmaRmaBuffer>> allRegisteredBuffers_; |
| 52 | - std::unordered_map<EndpointDesc, std::unique_ptr<RemoteRdmaRmaBufferMgr>> remoteRdmaRmaBufferMgrs_; | 89 | + std::unordered_map<RemoteRdmaMemKey, std::unique_ptr<RemoteRdmaRmaBufferMgr>, RemoteRdmaMemKeyHash> |
| 90 | + remoteRdmaRmaBufferMgrs_; | ||
| 91 | + mutable std::mutex memMtx_; | ||
| 53 | }; | 92 | }; |
| 54 | } // namespace hcomm_experimental | 93 | } // namespace hcomm_experimental |
| 55 | 94 | ||


[MEDIUM] memDesc 线格式追加 pid 但无版本/魔数标识,旧格式描述符会被静默错位解析
GetMemDesc 将 memDesc 线格式从
[dto][EndpointDesc]扩展为[dto][EndpointDesc][pid],但线格式没有引入任何版本/魔数标识,本处仅以descLen < tailLen做长度下界校验,无法区分新旧两种格式。旧格式(未追加 pid)的 memDesc 只要 dto 长度 >= 4 字节即可通过该校验,随后被静默错位解析:末尾 4 字节(实为 EndpointDesc 尾部)被当作 pid,EndpointDesc 整体向后偏移 4 字节,dto 被截断 4 字节。结果是按错误的 (endpointDesc, pid) 建 key、以错误的 addr/size 导入远端内存,且无任何报错(不满足 fail-fast)。
建议:
descLen == dtoLen + sizeof(EndpointDesc) + sizeof(uint32_t),使旧格式能被干净地拒绝而不是静默错位解析。