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 <algorithm>
12 : #include <cerrno>
13 : #include <cmath>
14 : #include <cstring>
15 : #include <sys/socket.h>
16 : #include <sys/epoll.h>
17 : #include <unistd.h>
18 : #include <ctime>
19 : #include "sal_pub.h"
20 : #include "topoinfo_exchange_dispatcher.h"
21 :
22 : namespace hccl {
23 16 : TopoInfoExchangeDispather::~TopoInfoExchangeDispather()
24 : {
25 16 : CleanResource();
26 16 : }
27 :
28 13 : HcclResult TopoInfoExchangeDispather::BroadcastRankTable(
29 : const std::map<std::string, std::shared_ptr<HcclSocket>> connectSockets, const RankTable_t &clusterInfo, const std::string &failedAgentIdList)
30 : {
31 13 : CHK_RET(PrepareResource(connectSockets, clusterInfo, failedAgentIdList));
32 13 : CHK_RET(ProcessSend());
33 13 : HCCL_INFO("cluster topo exchange worker broadcast topoinfo success, rankNum[%d], "\
34 : "threadNum[%u]", rankNum_, threadNum_);
35 13 : return HCCL_SUCCESS;
36 : }
37 :
38 0 : HcclResult TopoInfoExchangeDispather::BroadcastGroupLeaderInfo(
39 : const std::map<std::string, std::shared_ptr<HcclSocket>> connectSockets, const GroupLeader_t &leaderInfo)
40 : {
41 0 : CHK_RET(PrepareLeaderResource(connectSockets, leaderInfo));
42 0 : CHK_RET(ProcessSend());
43 0 : HCCL_INFO("cluster topo exchange worker broadcast GroupLeaderInfo success, rankNum[%d], "\
44 : "threadNum[%u]", rankNum_, threadNum_);
45 0 : return HCCL_SUCCESS;
46 : }
47 :
48 13 : void TopoInfoExchangeDispather::InitWorkerThread()
49 : {
50 13 : threadNum_ = std::max(1, std::min(int(rankNum_/RANK_CAPACITY_PER_THREAD), int(MAX_THREAD_NUM)));
51 13 : HCCL_INFO("[TopoInfoExchangeDispather][InitWorkerThread]calculate threadNum[%u], rankNum[%d]",
52 : threadNum_, rankNum_);
53 26 : for (u32 i = 0; i < threadNum_; ++i) {
54 13 : auto th = std::thread(&TopoInfoExchangeDispather::RunWorkerThread, this, i);
55 13 : workerThreads_.emplace_back(std::move(th));
56 13 : HCCL_DEBUG("[TopoInfoExchangeDispather][InitWorkerThread]create thread[%u]", i);
57 13 : }
58 13 : }
59 :
60 13 : void TopoInfoExchangeDispather::WorkerWait(int workId)
61 : {
62 13 : HCCL_DEBUG("[TopoInfoExchangeDispather][WorkerWait]start wait! workId[%d]", workId);
63 13 : std::unique_lock <std::mutex> lck(wakeMutex_);
64 26 : while (!ready_ && !stop_) {
65 13 : wakeManager_.wait(lck);
66 : }
67 13 : HCCL_DEBUG("[TopoInfoExchangeDispather][WorkerWait]finish wait! workId[%d]", workId);
68 13 : }
69 :
70 13 : bool TopoInfoExchangeDispather::GetTask(WorkerTask &workTask)
71 : {
72 13 : auto &taskQueue = taskQueue_;
73 13 : std::unique_lock <std::mutex> lckForGetTask(taskQueueMutex_);
74 13 : if (taskQueue.empty()) {
75 13 : ready_ = false;
76 13 : return false;
77 : }
78 0 : workTask = taskQueue.front();
79 0 : taskQueue.pop();
80 0 : return true;
81 13 : }
82 :
83 13 : void TopoInfoExchangeDispather::RunWorkerThread(int workId)
84 : {
85 : //给当前线程添加名字
86 13 : SetThreadName("Hccl_RunWorker");
87 :
88 26 : while (!stop_) {
89 13 : WorkerWait(workId);
90 : while (true) {
91 13 : WorkerTask task;
92 13 : if (GetTask(task)) {
93 0 : task();
94 : } else {
95 13 : break;
96 : }
97 13 : }
98 : }
99 13 : HCCL_DEBUG("[TopoInfoExchangeDispather][RunWorkerThread]finish thread! workId[%d]", workId);
100 13 : }
101 :
102 13 : HcclResult TopoInfoExchangeDispather::PrepareResource(
103 : const std::map<std::string, std::shared_ptr<HcclSocket>> connectSockets, const RankTable_t &clusterInfo, const std::string &failedAgentIdList)
104 : {
105 13 : rankNum_ = connectSockets.size();
106 13 : InitWorkerThread();
107 :
108 13 : HcclResult ret = hrtRaCreateEventHandle(epollFds_);
109 13 : if (ret != HCCL_SUCCESS) {
110 0 : HCCL_ERROR("[TopoInfoExchangeDispather][PrepareEpollResource]hrtRaCreateEventHandle create"\
111 : " epollFds_ failed, ret[%d]", ret);
112 0 : return HCCL_E_TCP_TRANSFER;
113 : }
114 :
115 13 : nlohmann::json basicJson;
116 13 : CHK_RET(topoInfoExchangeServer_->TopoInfoExchangeBase::Struct2Json(clusterInfo, basicJson));
117 13 : basicJson[PROP_STEP] = topoInfoExchangeServer_->TopoInfoExchangeBase::currentStep_;
118 13 : if (!failedAgentIdList.empty()) {
119 0 : basicJson["fault_info"] = "[" + failedAgentIdList + "]";
120 0 : basicJson["fault_type"] = static_cast<int>(TopoDetectResult::TOPO_CONNECT_FAILED);
121 : }
122 13 : rankTableJson_ = basicJson.dump();
123 :
124 13 : u32 socketIndex = 0; // socket已经经过rankid（or serverip +deviceid排序）
125 13 : for (auto it : connectSockets) {
126 0 : FdContext fdcontext;
127 0 : fdcontext.socket = it.second;
128 0 : if (topoInfoExchangeServer_->TopoInfoExchangeBase::isByMasterInfo_) { // masterInfo场景下无法获取rankid
129 0 : fdcontext.txState.identify = socketIndex;
130 : }
131 0 : fdcontext.txState.bodyLen = rankTableJson_.length();
132 0 : fdcontext.txState.data = &rankTableJson_[0];
133 0 : fdcontext.txState.rankId = socketIndex;
134 0 : socketIndex++;
135 0 : HCCL_DEBUG("[TopoInfoExchangeDispather][PrepareResource]socketIndex:%u, bodyLen:%u, data:%u", socketIndex,
136 : fdcontext.txState.bodyLen, fdcontext.txState.data);
137 0 : fdHandleToFdContextMap_.emplace(it.second->GetFdHandle(), fdcontext);
138 0 : }
139 :
140 13 : HCCL_DEBUG("[TopoInfoExchangeDispather][PrepareEpollResource]fdHandleToFdContextMap_ size[%d]",
141 : fdHandleToFdContextMap_.size());
142 13 : return HCCL_SUCCESS;
143 13 : }
144 :
145 0 : HcclResult TopoInfoExchangeDispather::PrepareLeaderResource(
146 : const std::map<std::string, std::shared_ptr<HcclSocket>> connectSockets, const GroupLeader_t &leaderInfo)
147 : {
148 0 : rankNum_ = connectSockets.size();
149 0 : InitWorkerThread();
150 :
151 0 : HcclResult ret = hrtRaCreateEventHandle(epollFds_);
152 0 : if (ret != HCCL_SUCCESS) {
153 0 : HCCL_ERROR("[TopoInfoExchangeDispather][PrepareEpollResource]hrtRaCreateEventHandle create"\
154 : " epollFds_ failed, ret[%d]", ret);
155 0 : return HCCL_E_TCP_TRANSFER;
156 : }
157 :
158 0 : nlohmann::json basicJson;
159 0 : CHK_RET(topoInfoExchangeServer_->TopoInfoExchangeBase::GrpLeader2Json(leaderInfo, basicJson));
160 0 : basicJson[PROP_STEP] = topoInfoExchangeServer_->TopoInfoExchangeBase::currentStep_;
161 0 : rankTableJson_ = basicJson.dump();
162 :
163 0 : u32 socketIndex = 0; // socket已经经过rankid（or serverip +deviceid排序）
164 0 : for (auto it : connectSockets) {
165 0 : FdContext fdContext;
166 0 : fdContext.socket = it.second;
167 0 : if (topoInfoExchangeServer_->TopoInfoExchangeBase::isByMasterInfo_) { // masterInfo场景下无法获取rankid
168 0 : fdContext.txState.identify = socketIndex;
169 : }
170 0 : fdContext.txState.bodyLen = rankTableJson_.length();
171 0 : fdContext.txState.data = &rankTableJson_[0];
172 0 : fdContext.txState.rankId = socketIndex;
173 0 : socketIndex++;
174 0 : HCCL_DEBUG("[TopoInfoExchangeDispather][PrepareLeaderResource]socketIndex:%u, bodyLen:%u, data:%u", socketIndex,
175 : fdContext.txState.bodyLen, fdContext.txState.data);
176 0 : fdHandleToFdContextMap_.emplace(it.second->GetFdHandle(), fdContext);
177 0 : }
178 :
179 0 : HCCL_DEBUG("[TopoInfoExchangeDispather][PrepareEpollResource]fdHandleToFdContextMap_ size[%d]",
180 : fdHandleToFdContextMap_.size());
181 0 : return HCCL_SUCCESS;
182 0 : }
183 :
184 16 : void TopoInfoExchangeDispather::WakeWoker()
185 : {
186 16 : std::unique_lock <std::mutex> lck(wakeMutex_);
187 16 : ready_ = true;
188 16 : wakeManager_.notify_all();
189 16 : }
190 :
191 16 : void TopoInfoExchangeDispather::CleanResource()
192 : {
193 : // 主线程广播结束，结束从线程（不确定是否存在出于wait状态的线程，统一全部唤醒）
194 16 : stop_ = true;
195 16 : WakeWoker();
196 16 : HCCL_INFO("[TopoInfoExchangeDispather][PrepareEpollResource]wake all workers.");
197 29 : for (auto &th : workerThreads_) {
198 13 : if (th.joinable()) {
199 13 : th.join();
200 : }
201 : }
202 16 : fdHandleToFdContextMap_.clear();
203 16 : workerThreads_.clear();
204 16 : }
205 :
206 13 : HcclResult TopoInfoExchangeDispather::CloseEpollFd()
207 : {
208 13 : HcclResult ret = hrtRaDestroyEventHandle(epollFds_);
209 13 : if (ret != HCCL_SUCCESS) {
210 0 : HCCL_ERROR("[TopoInfoExchangeDispather][CloseEpollFd]DestroyEventHandle destroy "\
211 : "epollFds_ failed, ret[%d]", ret);
212 0 : return HCCL_E_TCP_TRANSFER;
213 : }
214 13 : return HCCL_SUCCESS;
215 : }
216 :
217 0 : HcclResult TopoInfoExchangeDispather::ProcessOneSendEvent(s32 epollFd, FdHandle &fdHandle)
218 : {
219 0 : std::unique_lock <std::mutex> lckForMap(fdHandleMapMutex_);
220 0 : if (fdHandleToFdContextMap_.find(fdHandle) == fdHandleToFdContextMap_.end()) {
221 0 : HCCL_ERROR("[TopoInfoExchangeDispather][ProcessOneSendEvent]no fdhandle[%p]", fdHandle);
222 0 : stop_ = true;
223 0 : return HCCL_E_INTERNAL;
224 : }
225 0 : auto ctx = &(fdHandleToFdContextMap_.at(fdHandle));
226 0 : if (ctx->txState.Send(ctx->socket) != 0) {
227 0 : HCCL_ERROR("[TopoInfoExchangeDispather][ProcessOneSendEvent]send data to rank[%u] failed.", ctx->txState.rankId);
228 0 : stop_ = true;
229 0 : return HCCL_E_INTERNAL;
230 : }
231 :
232 0 : int ctlType = EPOLL_CTL_DEL;
233 0 : if (ctx->txState.IsOk()) {
234 0 : sendDoneCount_++;
235 : } else {
236 0 : ctlType = EPOLL_CTL_MOD;
237 : }
238 : // EPOLLOUT_LET_ONESHOT -> EPOLLOUT | EPOLLET | EPOLLONESHOT, 防止多个线程同时操作同一个fd（fd重复触发）
239 0 : HcclResult ret = hrtRaCtlEventHandle(epollFds_, fdHandle, ctlType, HcclEpollEvent::HCCL_EPOLLOUT_LET_ONESHOT);
240 0 : if (ret != HCCL_SUCCESS) {
241 0 : HCCL_ERROR("[TopoInfoExchangeDispather][ProcessOneSendEvent]epoll_ctl delete/modify "\
242 : "failed, ctlType[%d]", ctlType);
243 0 : stop_ = true;
244 0 : return HCCL_E_INTERNAL;
245 : }
246 0 : return HCCL_SUCCESS;
247 0 : }
248 :
249 13 : HcclResult TopoInfoExchangeDispather::SendOnce()
250 : {
251 : HcclResult ret;
252 13 : for (auto &it : fdHandleToFdContextMap_) {
253 0 : auto fdCtx = &(it.second);
254 0 : if (fdCtx->txState.Send(fdCtx->socket) != 0) {
255 0 : HCCL_ERROR("[TopoInfoExchangeDispather][SendOnce]Send data to rank[%u] failed.", fdCtx->txState.rankId);
256 0 : stop_ = true;
257 0 : return HCCL_E_INTERNAL;
258 : }
259 :
260 : // 数据未发送完成，添加epoll事件
261 0 : if (!fdCtx->txState.IsOk()) {
262 : // EPOLLOUT_LET_ONESHOT -> EPOLLOUT | EPOLLET | EPOLLONESHOT, 防止多个线程同时操作同一个fd（fd重复触发）
263 0 : ret = hrtRaCtlEventHandle(epollFds_, it.first, EPOLL_CTL_ADD, HcclEpollEvent::HCCL_EPOLLOUT_LET_ONESHOT);
264 0 : if (ret != HCCL_SUCCESS) {
265 0 : HCCL_ERROR("[TopoInfoExchangeDispather][SendOnce]epoll_ctl add fd failed.");
266 0 : stop_ = true;
267 0 : return HCCL_E_INTERNAL;
268 : }
269 : } else {
270 0 : sendDoneCount_++;
271 : }
272 : }
273 13 : return HCCL_SUCCESS;
274 : }
275 :
276 13 : HcclResult TopoInfoExchangeDispather::ProcessSend()
277 : {
278 13 : HcclResult ret = SendOnce(); // 先尝试发送数据
279 13 : CHK_RET(ret);
280 26 : HCCL_INFO("[TopoInfoExchangeDispather][ProcessSend]sendOnce success, start epoll_wait." \
281 : " sendDoneCount[%d], rankNum[%d].", sendDoneCount_.load(), rankNum_);
282 13 : const int sendEvsCount = 20; // epoll_wait 缓冲区大小（单次触发的事件个数）
283 13 : std::vector<SocketEventInfo> eventInfos(sendEvsCount);
284 13 : bool lastEpollWaitFlag = false; // 最后一轮epoll_wait标识位
285 13 : auto startTime = std::chrono::steady_clock::now();
286 13 : auto timeout = std::chrono::seconds(GetExternalInputHcclLinkTimeOut());
287 13 : while (sendDoneCount_ != rankNum_) {
288 0 : if (stop_) {
289 0 : return HCCL_E_INTERNAL;
290 : }
291 0 : if (rankNum_ - sendDoneCount_ < sendEvsCount && !lastEpollWaitFlag) { // 最后一轮epoll_wait
292 0 : lastEpollWaitFlag = true;
293 : }
294 0 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
295 0 : HCCL_ERROR("[TopoInfoExchangeDispather][ProcessSend]epoll_wait timeout!");
296 0 : return HCCL_E_INTERNAL;
297 : }
298 0 : s32 epollTimeout = lastEpollWaitFlag ? LAST_EPOLL_TIMEOUT_MS : EPOLL_TIMEOUT_MS;
299 0 : u32 eventsNum{ 0 };
300 0 : ret = hrtRaWaitEventHandle(epollFds_, eventInfos, epollTimeout, sendEvsCount, eventsNum);
301 0 : if (eventsNum == 0 && ret == HCCL_SUCCESS && sendDoneCount_ == rankNum_) {
302 : // 最后一轮epoll_wait结束, 等待超时，epoll池内无事件
303 0 : HCCL_WARNING("[TopoInfoExchangeDispather][ProcessSend]hrtRaWaitEventHandle is timeout[%d] ms, "\
304 : "eventsNum[%u], ret[%d], sendDoneCount_[%d]", epollTimeout, eventsNum, ret, sendDoneCount_.load());
305 0 : return HCCL_SUCCESS;
306 : }
307 0 : if (eventsNum <= 0 && ret != HCCL_SUCCESS) {
308 0 : HCCL_ERROR("[TopoInfoExchangeDispather][ProcessSend]hrtRaWaitEventHandle failed, eventsNum[%u], "\
309 : "ret[%d]", eventsNum, ret);
310 0 : return HCCL_E_INTERNAL;
311 : }
312 0 : for (u32 i = 0; i < eventsNum; ++i) {
313 0 : std::unique_lock <std::mutex> lck(taskQueueMutex_);
314 0 : taskQueue_.push(std::bind(&TopoInfoExchangeDispather::ProcessOneSendEvent, this, epollFds_,
315 0 : eventInfos[i].fdHandle));
316 0 : lck.unlock();
317 0 : }
318 : // 唤醒处理
319 0 : WakeWoker();
320 : }
321 13 : CHK_RET(CloseEpollFd());
322 13 : return HCCL_SUCCESS;
323 13 : }
324 :
325 0 : HcclResult TopoInfoExchangeDispather::SendState::Send(std::shared_ptr<HcclSocket> socket)
326 : {
327 0 : if (headerSended != headerLen) {
328 0 : header = bodyLen;
329 0 : CHK_RET(SendHeader(socket));
330 : }
331 :
332 0 : if ((headerSended == headerLen) && (bodyLen != bodySended)) {
333 0 : CHK_RET(SendBody(socket));
334 : }
335 :
336 0 : if ((headerSended == headerLen) && (bodyLen == bodySended) &&
337 0 : (identify != UINT_MAX && identifyLen != identifySended)) {
338 0 : CHK_RET(SendIdentify(socket));
339 : }
340 :
341 0 : return HCCL_SUCCESS;
342 : }
343 :
344 0 : HcclResult TopoInfoExchangeDispather::SendState::SendHeader(std::shared_ptr<HcclSocket> socket)
345 : {
346 0 : return SendHelper(socket, reinterpret_cast<char *>(&header), headerLen, headerSended);
347 : }
348 :
349 0 : HcclResult TopoInfoExchangeDispather::SendState::SendBody(std::shared_ptr<HcclSocket> socket)
350 : {
351 0 : return SendHelper(socket, reinterpret_cast<char *>(data), bodyLen, bodySended);
352 : }
353 :
354 0 : HcclResult TopoInfoExchangeDispather::SendState::SendIdentify(std::shared_ptr<HcclSocket> socket)
355 : {
356 0 : return SendHelper(socket, reinterpret_cast<char *>(&identify), identifyLen, identifySended);
357 : }
358 :
359 3 : HcclResult TopoInfoExchangeDispather::SendState::SendHelper(std::shared_ptr<HcclSocket> socket,
360 : char *buf, size_t dataLen, size_t &sendedLen)
361 : {
362 3 : CHK_SMART_PTR_NULL(socket);
363 3 : CHK_PTR_NULL(buf);
364 3 : u64 needSend = dataLen - sendedLen;
365 3 : u64 sentSize = 0;
366 3 : HcclResult ret = socket->ISend(buf + sendedLen, needSend, sentSize);
367 3 : if (ret == HCCL_E_NETWORK) {
368 1 : HCCL_ERROR("[TopoInfoExchangeDispather][SendState][SendHelper]SendHelper fail error[%d].", ret);
369 1 : return HCCL_E_TCP_TRANSFER;
370 : }
371 2 : if (ret != HCCL_SUCCESS && ret != HCCL_E_AGAIN) {
372 1 : HCCL_ERROR("[TopoInfoExchangeDispather][SendState][SendHelper]socket send fail error[%d].", ret);
373 1 : return HCCL_E_INTERNAL;
374 : }
375 1 : if (ret == HCCL_SUCCESS) {
376 1 : sendedLen += sentSize;
377 : }
378 1 : return HCCL_SUCCESS;
379 : }
380 :
381 : } // namespace hccl
|
