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