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