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