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 "transport_mem.h"
12 : #include "log.h"
13 : #include "transport_ipc_mem.h"
14 : #include "transport_roce_mem.h"
15 : #ifdef CCL_KERNEL
16 : #include "transport_device_roce_mem.h"
17 : #endif
18 :
19 : namespace hccl {
20 : constexpr u32 INVALID_REMOTE_RANK_ID = 0xFFFFFFFF;
21 0 : TransportMem::TransportMem(
22 : const std::unique_ptr<NotifyPool>& notifyPool, const HcclNetDevCtx& netDevCtx, const HcclDispatcher& dispatcher,
23 0 : AttrInfo& attrInfo)
24 0 : : TransportMem(notifyPool, netDevCtx, dispatcher, attrInfo, false)
25 0 : {}
26 :
27 0 : TransportMem::TransportMem(
28 : const std::unique_ptr<NotifyPool>& notifyPool, const HcclNetDevCtx& netDevCtx, const HcclDispatcher& dispatcher,
29 0 : AttrInfo& attrInfo, bool aicpuUnfoldMode)
30 0 : : notifyPool_(notifyPool),
31 0 : netDevCtx_(netDevCtx),
32 0 : dispatcher_(dispatcher),
33 0 : localRankId_(attrInfo.localRankId),
34 0 : remoteRankId_(attrInfo.remoteRankId),
35 0 : aicpuUnfoldMode_(aicpuUnfoldMode)
36 0 : {}
37 :
38 0 : TransportMem::~TransportMem() {}
39 :
40 : // static
41 0 : std::shared_ptr<TransportMem> TransportMem::Create(
42 : TpType tpType, const std::unique_ptr<NotifyPool>& notifyPool, const HcclNetDevCtx& netDevCtx,
43 : const HcclDispatcher& dispatcher, AttrInfo& attrInfo)
44 : {
45 0 : return Create(tpType, notifyPool, netDevCtx, dispatcher, attrInfo, false);
46 : }
47 :
48 0 : std::shared_ptr<TransportMem> TransportMem::Create(
49 : TpType tpType, [[maybe_unused]] const std::unique_ptr<NotifyPool>& notifyPool,
50 : [[maybe_unused]] const HcclNetDevCtx& netDevCtx, [[maybe_unused]] const HcclDispatcher& dispatcher,
51 : [[maybe_unused]] AttrInfo& attrInfo, [[maybe_unused]] bool aicpuUnfoldMode)
52 : {
53 0 : std::shared_ptr<TransportMem> transportMemPtr;
54 : #if !defined(CCL_KERNEL) || defined(CCL_LLT)
55 0 : CHK_PRT_RET((netDevCtx == nullptr), HCCL_ERROR("[TransportMem][Create]netDevCtx is null"), nullptr);
56 0 : HCCL_DEBUG(
57 : "transportMem create tpType:%u netDevCtx:%p dispatcher:%p localRankId:%u remoteRankId:%u sdid:%u "
58 : "serverId:%u trafficClass:%u serviceLevel:%u",
59 : tpType, netDevCtx, dispatcher, attrInfo.localRankId, attrInfo.remoteRankId, attrInfo.sdid, attrInfo.serverId,
60 : attrInfo.trafficClass, attrInfo.serviceLevel);
61 0 : switch (tpType) {
62 0 : case TpType::ROCE:
63 : transportMemPtr
64 0 : = std::make_unique<TransportRoceMem>(notifyPool, netDevCtx, dispatcher, attrInfo, aicpuUnfoldMode);
65 0 : break;
66 0 : case TpType::IPC:
67 : transportMemPtr
68 0 : = std::make_unique<TransportIpcMem>(notifyPool, netDevCtx, dispatcher, attrInfo, aicpuUnfoldMode);
69 0 : break;
70 0 : default:
71 0 : break;
72 : }
73 : #else
74 : HCCL_ERROR("[TransportMem] The Create interface with qpInfo should be used on the AICPU, tpType[%u]", tpType);
75 : #endif
76 0 : return transportMemPtr;
77 0 : }
78 :
79 0 : std::shared_ptr<TransportMem> TransportMem::Create(
80 : TpType tpType, [[maybe_unused]] const HcclQpInfoV2& qpInfo, const HcclDispatcher& dispatcher, AttrInfo& attrInfo)
81 : {
82 0 : const std::unique_ptr<NotifyPool> notifyPool = nullptr; // dummy for device ibv transport
83 0 : const HcclNetDevCtx netDevCtx = nullptr; // dummy
84 0 : HCCL_DEBUG(
85 : "[TransportMem] create tpType:%u netDevCtx:%p dispatcher:%p localRankId:%u remoteRankId:%u", tpType, netDevCtx,
86 : dispatcher, attrInfo.localRankId, attrInfo.remoteRankId);
87 0 : std::shared_ptr<TransportMem> transportMemPtr;
88 0 : switch (tpType) {
89 0 : case TpType::ROCE_DEVICE:
90 : #ifdef CCL_KERNEL
91 0 : transportMemPtr = std::make_unique<TransportDeviceRoceMem>(
92 0 : notifyPool, netDevCtx, dispatcher, attrInfo, false, qpInfo); // aicpuUnfoldMode is set by host
93 : #else
94 : HCCL_ERROR("[TransportMem] ROCE_DEVICE Only running on the AICPU");
95 : #endif
96 0 : break;
97 0 : default:
98 0 : HCCL_ERROR("[TransportMem] unsupported TpType[%u] on the AICPU", tpType);
99 0 : break;
100 : }
101 0 : return transportMemPtr;
102 0 : }
103 :
104 0 : HcclResult TransportMem::SetDataSocket(const std::shared_ptr<HcclSocket>& socket)
105 : {
106 0 : dataSocket_ = socket;
107 0 : return HCCL_SUCCESS;
108 : }
109 :
110 : HcclResult
111 0 : TransportMem::DoExchangeMemDesc(const RmaMemDescs& localMemDescs, RmaMemDescs& remoteMemDescs, u32& actualNumOfRemote)
112 : {
113 0 : HCCL_INFO(
114 : "[HcclOneSidedConn][ExchangeMemDesc]localRank[%u] exchange memDesc begin, role[%u]", localRankId_,
115 : dataSocket_->GetLocalRole());
116 :
117 0 : if (dataSocket_->GetLocalRole() == HcclSocketRole::SOCKET_ROLE_CLIENT) {
118 : // 先收后发
119 0 : CHK_RET(ReceiveRemoteMemDesc(remoteMemDescs, actualNumOfRemote));
120 0 : CHK_RET(SendLocalMemDesc(localMemDescs));
121 : } else {
122 : // 先发后收
123 0 : CHK_RET(SendLocalMemDesc(localMemDescs));
124 0 : CHK_RET(ReceiveRemoteMemDesc(remoteMemDescs, actualNumOfRemote));
125 : }
126 0 : HCCL_INFO("[HcclOneSidedConn][ExchangeMemDesc]get actualNumOfRemotee[%u]", actualNumOfRemote);
127 : // 校验remoteDescs中的remoteRankId和conn对象中保存的localRankId是否一样
128 0 : for (u32 i = 0; i < actualNumOfRemote; i++) {
129 0 : CHK_PTR_NULL((remoteMemDescs.array) + i);
130 0 : u32 tempRankId = remoteMemDescs.array[i].remoteRankId;
131 0 : HCCL_DEBUG("[TransportMem][ExchangeMemDesc]tempRankId:%u, userRank:%u", tempRankId, localRankId_);
132 0 : if (tempRankId == INVALID_REMOTE_RANK_ID) {
133 0 : HCCL_INFO("[DoExchangeMemDesc] It's unnecessary to check remoteID.");
134 0 : continue;
135 : }
136 0 : if (tempRankId != localRankId_) {
137 0 : HCCL_ERROR(
138 : "[TransportMem][ExchangeMemDesc]localRank[%u] receive remoteMemDesc from wrong localRank[%u], "
139 : "connection is for localRank[%u]",
140 : localRankId_, tempRankId, localRankId_);
141 0 : return HCCL_E_INTERNAL;
142 : }
143 : }
144 0 : return HCCL_SUCCESS;
145 : }
146 :
147 0 : HcclResult TransportMem::SendLocalMemDesc(const RmaMemDescs& localMemDescs)
148 : {
149 0 : HcclResult ret = dataSocket_->Send(&localMemDescs.arrayLength, sizeof(u32));
150 0 : CHK_PRT_RET(
151 : ret != HCCL_SUCCESS,
152 : HCCL_ERROR(
153 : "errNo[0x%016llx] localRank[%u] send localMemDesc.arrayLength to remote "
154 : "failed, ret[%u]",
155 : HCCL_ERROR_CODE(ret), localRankId_, ret),
156 : ret);
157 0 : HCCL_DEBUG("send localMemDescs.arrayLength:%u", localMemDescs.arrayLength);
158 :
159 0 : if (localMemDescs.arrayLength == 0) {
160 0 : HCCL_INFO("localMemDescs.arrayLength[%u], no need to send data", localMemDescs.arrayLength);
161 : } else {
162 0 : HCCL_DEBUG("send descSize:%u", localMemDescs.arrayLength * sizeof(RmaMemDesc));
163 0 : ret = dataSocket_->Send(localMemDescs.array, localMemDescs.arrayLength * sizeof(RmaMemDesc));
164 0 : CHK_PRT_RET(
165 : ret != HCCL_SUCCESS,
166 : HCCL_ERROR(
167 : "errNo[0x%016llx] localRank[%u] send localMemDesc to remote "
168 : "failed, ret[%u]",
169 : HCCL_ERROR_CODE(ret), localRankId_, ret),
170 : ret);
171 : }
172 0 : return HCCL_SUCCESS;
173 : }
174 :
175 0 : HcclResult TransportMem::ReceiveRemoteMemDesc(RmaMemDescs& remoteMemDescs, u32& actualNumOfRemote)
176 : {
177 0 : HcclResult ret = dataSocket_->Recv(&actualNumOfRemote, sizeof(u32));
178 0 : remoteMemDescs.arrayLength = actualNumOfRemote;
179 0 : CHK_PRT_RET(
180 : ret != HCCL_SUCCESS,
181 : HCCL_ERROR(
182 : "errNo[0x%016llx] localRank[%u] receive actualNumOfRemote to remote "
183 : "failed, ret[%u]",
184 : HCCL_ERROR_CODE(ret), localRankId_, ret),
185 : ret);
186 0 : HCCL_DEBUG("receive actualNumOfRemote:%u", actualNumOfRemote);
187 0 : if (actualNumOfRemote == 0) {
188 0 : HCCL_INFO("actualNumOfRemote[%u], no need to receive data", actualNumOfRemote);
189 : } else {
190 0 : HCCL_DEBUG("receive descSize:%u", actualNumOfRemote * sizeof(RmaMemDesc));
191 0 : ret = dataSocket_->Recv(remoteMemDescs.array, actualNumOfRemote * sizeof(RmaMemDesc));
192 0 : CHK_PRT_RET(
193 : ret != HCCL_SUCCESS,
194 : HCCL_ERROR(
195 : "errNo[0x%016llx] localRank[%u] receive remoteMemDesc from remote "
196 : "failed, ret[%u]",
197 : HCCL_ERROR_CODE(ret), localRankId_, ret),
198 : ret);
199 : }
200 0 : return HCCL_SUCCESS;
201 : }
202 : } // namespace hccl
|
