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 "base_mem_transport.h"
12 : #include "coll_operator_check.h"
13 :
14 : namespace Hccl {
15 129 : BaseMemTransport::BaseMemTransport(
16 129 : CommonLocRes& commonLocRes, Attribution& attr, const LinkData& linkData, const Socket& socket, TransportType type)
17 129 : : commonLocRes(commonLocRes),
18 129 : attr(attr),
19 129 : linkData(linkData),
20 129 : socket(const_cast<Socket*>(&socket)),
21 516 : transportType(type)
22 : {
23 129 : HcclResult ret = CheckCommonLocRes(commonLocRes);
24 129 : if (ret != HCCL_SUCCESS) {
25 0 : THROW<InvalidParamsException>(
26 0 : StringFormat("[BaseMemTransport::BaseMemTransport] CheckCommonLocRes failed, ret=%d", ret));
27 : }
28 129 : }
29 :
30 8 : BaseMemTransport::BaseMemTransport(
31 : CommonLocRes& commonLocRes, Attribution& attr, const LinkData& linkData, const Socket& socket, TransportType type,
32 8 : std::function<void(u32 streamId, u32 taskId, TaskParam taskParam)> callback)
33 8 : : commonLocRes(commonLocRes),
34 8 : attr(attr),
35 8 : linkData(linkData),
36 8 : socket(const_cast<Socket*>(&socket)),
37 8 : transportType(type),
38 32 : callback(callback)
39 : {
40 8 : HcclResult ret = CheckCommonLocRes(commonLocRes);
41 8 : if (ret != HCCL_SUCCESS) {
42 0 : THROW<InvalidParamsException>(
43 0 : StringFormat("[BaseMemTransport::BaseMemTransport] CheckCommonLocRes failed, ret=%d", ret));
44 : }
45 8 : }
46 :
47 5 : void BaseMemTransport::Establish()
48 : {
49 5 : baseStatus = TransportStatus::INIT;
50 5 : rmtRmaBufferVec.clear();
51 5 : }
52 :
53 4 : void BaseMemTransport::SetBaseStatusReady() { baseStatus = TransportStatus::READY; }
54 :
55 5 : bool BaseMemTransport::IsSocketReady()
56 : {
57 5 : if (socket == nullptr) {
58 0 : HCCL_ERROR("[BaseMemTransport::IsSocketReady] %s socket is nullptr, please check", GetLinkDescInfo().c_str());
59 0 : return false;
60 : }
61 :
62 5 : SocketStatus socketStatus = isHost_ ? socket->GetStatus() : socket->GetAsyncStatus();
63 5 : if (socketStatus == SocketStatus::OK) {
64 5 : baseStatus = TransportStatus::SOCKET_OK;
65 5 : return true;
66 0 : } else if (socketStatus == SocketStatus::TIMEOUT) {
67 0 : baseStatus = TransportStatus::SOCKET_TIMEOUT;
68 0 : return false;
69 : }
70 :
71 0 : return false;
72 : }
73 :
74 1 : void BaseMemTransport::NotifyVecPack(BinaryStream& binaryStream)
75 : {
76 1 : binaryStream << notifyNum;
77 3 : HCCL_INFO("start pack %s notifyVec", transportType.Describe().c_str());
78 1 : u32 pos = 0;
79 2 : for (auto& it : commonLocRes.notifyVec) {
80 1 : binaryStream << pos;
81 1 : std::unique_ptr<Serializable> dto = it->GetExchangeDto();
82 1 : dto->Serialize(binaryStream);
83 3 : HCCL_INFO("pack notify pos=%u, dto %s", pos, dto->Describe().c_str());
84 1 : pos++;
85 1 : }
86 1 : }
87 :
88 1 : void BaseMemTransport::ConnVecPack(BinaryStream& binaryStream)
89 : {
90 1 : binaryStream << connNum;
91 3 : HCCL_INFO("start pack %s connVec", transportType.Describe().c_str());
92 1 : u32 pos = 0;
93 2 : for (auto& it : commonLocRes.connVec) {
94 1 : binaryStream << pos;
95 1 : std::unique_ptr<Serializable> dto = it->GetExchangeDto();
96 1 : dto->Serialize(binaryStream);
97 3 : HCCL_INFO("pack connection pos=%u, dto %s", pos, dto->Describe().c_str());
98 1 : pos++;
99 1 : }
100 1 : }
101 :
102 1 : void BaseMemTransport::HandshakeMsgPack(BinaryStream& binaryStream)
103 : {
104 3 : HCCL_INFO(
105 : "[BaseMemTransport::%s] start pack %s handshakeMsg, size=%zu, accelerator=%s", __func__,
106 : transportType.Describe().c_str(), attr.handshakeMsg.size(), attr.opAcceState.Describe().c_str());
107 1 : binaryStream << static_cast<u32>(attr.opAcceState);
108 1 : binaryStream << attr.handshakeMsg;
109 1 : }
110 :
111 0 : HcclResult BaseMemTransport::HandshakeMsgUnpack(BinaryStream& binaryStream)
112 : {
113 0 : u32 rmtAccelerator{0};
114 0 : binaryStream >> rmtAccelerator;
115 0 : rmtOpAcceState = static_cast<AcceleratorState::Value>(rmtAccelerator);
116 0 : HCCL_INFO(
117 : "[BaseMemTransport::%s] locOpAccelerator[%s], rmtOpAccelerator[%s]", __func__,
118 : attr.opAcceState.Describe().c_str(), rmtOpAcceState.Describe().c_str());
119 0 : if (rmtOpAcceState != attr.opAcceState) {
120 0 : HCCL_ERROR(
121 : "[BaseMemTransport::HandshakeMsgUnpack] Accelerator information check fail. "
122 : "locOpAccelerator[%s], rmtOpAccelerator[%s]",
123 : attr.opAcceState.Describe().c_str(), rmtOpAcceState.Describe().c_str());
124 0 : return HCCL_E_PARA;
125 : }
126 :
127 0 : rmtHandshakeMsg.clear();
128 0 : binaryStream >> rmtHandshakeMsg;
129 :
130 0 : if (attr.handshakeMsg.size() != rmtHandshakeMsg.size()) {
131 0 : HCCL_ERROR(
132 : "[BaseMemTransport::HandshakeMsgUnpack] handshakeMsg size=%zu is not equal to rmt=%zu",
133 : attr.handshakeMsg.size(), rmtHandshakeMsg.size());
134 0 : return HCCL_E_PARA;
135 : }
136 :
137 : // 单边通信情况下，handshakeMsg的size为0
138 0 : if (attr.handshakeMsg.size() == 0) {
139 0 : return HCCL_SUCCESS;
140 : }
141 0 : auto localCollOperator = CollOperator::GetPackedData(attr.handshakeMsg);
142 0 : auto remoteCollOperator = CollOperator::GetPackedData(rmtHandshakeMsg);
143 0 : CheckCollOperator(localCollOperator, remoteCollOperator); // 两端算子参数一致性校验
144 0 : return HCCL_SUCCESS;
145 0 : }
146 :
147 836 : string BaseMemTransport::GetLinkDescInfo()
148 : {
149 : return StringFormat(
150 : "rank[%u], rmtRank[%u] linkData=%s, type=%s", linkData.GetLocalRankId(), linkData.GetRemoteRankId(),
151 836 : linkData.Describe().c_str(), transportType.Describe().c_str());
152 : }
153 :
154 2 : string BaseMemTransport::DescribeSocket() const
155 : {
156 2 : return StringFormat("BaseMemTransport socket=[%s]", socket->Describe().c_str());
157 : }
158 :
159 137 : HcclResult BaseMemTransport::CheckLocNotify(CommonLocRes& res)
160 : {
161 411 : HCCL_INFO("%s notify check start, notifyNum=%zu", GetLinkDescInfo().c_str(), res.notifyVec.size());
162 : // notify 不允许出现空指针情况
163 186 : for (auto& it : res.notifyVec) {
164 49 : if (it == nullptr) {
165 0 : HCCL_ERROR("[BaseMemTransport::CheckLocNotify] %s notify is nullptr", GetLinkDescInfo().c_str());
166 0 : return HCCL_E_PARA;
167 : }
168 147 : HCCL_INFO("locNotify=%s", it->Describe().c_str());
169 : }
170 411 : HCCL_INFO("%s notify check ok, notifyNum=%zu", GetLinkDescInfo().c_str(), res.notifyVec.size());
171 137 : return HCCL_SUCCESS;
172 : }
173 :
174 137 : void BaseMemTransport::CheckLocBuffer(CommonLocRes& res)
175 : {
176 411 : HCCL_INFO("%s buffer check start, bufferNum=%zu", GetLinkDescInfo().c_str(), res.bufferVec.size());
177 137 : u32 bufIndex = 0;
178 203 : for (auto& it : res.bufferVec) {
179 66 : if (it == nullptr) {
180 45 : HCCL_INFO("bufIndex=%u is nullptr", bufIndex);
181 : } else {
182 153 : HCCL_INFO("bufIndex=%u, buf=%s", bufIndex, it->Describe().c_str());
183 : }
184 66 : bufIndex++;
185 : }
186 :
187 411 : HCCL_INFO("%s buffer check ok, bufferNum=%zu", GetLinkDescInfo().c_str(), res.bufferVec.size());
188 137 : }
189 :
190 137 : HcclResult BaseMemTransport::CheckLocConn(CommonLocRes& res)
191 : {
192 411 : HCCL_INFO("%s connection check start, connNum=%zu", GetLinkDescInfo().c_str(), res.connVec.size());
193 199 : for (auto& it : res.connVec) {
194 62 : if (it == nullptr) {
195 0 : HCCL_ERROR("[BaseMemTransport::CheckLocConn] %s conn is nullptr", GetLinkDescInfo().c_str());
196 0 : return HCCL_E_PARA;
197 : }
198 186 : HCCL_INFO("conn=%s", it->Describe().c_str());
199 : }
200 411 : HCCL_INFO("%s connection check ok, connNum=%zu", GetLinkDescInfo().c_str(), res.connVec.size());
201 137 : return HCCL_SUCCESS;
202 : }
203 :
204 137 : HcclResult BaseMemTransport::CheckCommonLocRes(CommonLocRes& res)
205 : {
206 137 : HcclResult ret = CheckLocNotify(res);
207 137 : if (ret != HCCL_SUCCESS) {
208 0 : HCCL_ERROR("[BaseMemTransport::CheckCommonLocRes] CheckLocNotify failed, ret=%d", ret);
209 0 : return ret;
210 : }
211 137 : CheckLocBuffer(res);
212 137 : ret = CheckLocConn(res);
213 137 : if (ret != HCCL_SUCCESS) {
214 0 : HCCL_ERROR("[BaseMemTransport::CheckCommonLocRes] CheckLocConn failed, ret=%d", ret);
215 0 : return ret;
216 : }
217 137 : return HCCL_SUCCESS;
218 : }
219 :
220 : } // namespace Hccl
|
