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 "rank_info_dispatcher.h"
12 :
13 : #include <algorithm>
14 : #include <cerrno>
15 : #include <cmath>
16 : #include <cstring>
17 : #include <sys/socket.h>
18 : #include <sys/epoll.h>
19 : #include <unistd.h>
20 : #include <ctime>
21 : #include "sal.h"
22 : #include "hccp.h"
23 : #include "env_config/env_config_v2.h"
24 : #include "hccp_common.h"
25 : #include "network_api_exception.h"
26 : #include "adapter_error_manager_pub.h"
27 :
28 : namespace Hccl {
29 :
30 20 : RankInfoDispather::~RankInfoDispather()
31 : {
32 20 : DECTOR_TRY_CATCH("RankInfoDispather", CleanResource());
33 20 : DECTOR_TRY_CATCH("RankInfoDispather", CloseEpollFd());
34 20 : }
35 :
36 1 : void RankInfoDispather::BroadcastRankTable(
37 : const std::unordered_map<std::string, std::shared_ptr<Socket>>& connectSockets, const RankTableInfo& clusterInfo,
38 : const std::string& failedAgentIdList, u32 step)
39 : {
40 1 : PrepareResource(connectSockets, clusterInfo, failedAgentIdList, step);
41 1 : ProcessSend();
42 1 : HCCL_INFO(
43 : "[RankInfoDispather::%s] broadcast topoinfo success, rankNum[%u], threadNum[%u]", __func__, rankNum_,
44 : threadNum_);
45 1 : }
46 :
47 1 : void RankInfoDispather::InitWorkerThread()
48 : {
49 1 : threadNum_ = std::max(1, std::min(s32(rankNum_ / RANK_CAPACITY_PER_THREAD), s32(MAX_THREAD_NUM)));
50 2 : for (u32 i = 0; i < threadNum_; ++i) {
51 1 : auto th = std::thread(&RankInfoDispather::RunWorkerThread, this, i);
52 1 : workerThreads_.emplace_back(std::move(th));
53 1 : }
54 1 : HCCL_INFO("[RankInfoDispather::%s]calculate threadNum[%u], rankNum[%u]", __func__, threadNum_, rankNum_);
55 1 : }
56 :
57 1 : void RankInfoDispather::WorkerWait(s32 workId)
58 : {
59 1 : HCCL_DEBUG("[RankInfoDispather::%s]start wait! workId[%d]", __func__, workId);
60 1 : std::unique_lock<std::mutex> lck(wakeMutex_);
61 1 : while (!ready_ && !stop_) {
62 0 : wakeManager_.wait(lck);
63 : }
64 1 : HCCL_DEBUG("[RankInfoDispather::%s]finish wait! workId[%d]", __func__, workId);
65 1 : }
66 :
67 1 : bool RankInfoDispather::GetTask(WorkerTask& workTask)
68 : {
69 1 : auto& taskQueue = taskQueue_;
70 1 : std::unique_lock<std::mutex> lckForGetTask(taskQueueMutex_);
71 1 : if (taskQueue.empty()) {
72 1 : ready_ = false;
73 1 : return false;
74 : }
75 0 : workTask = taskQueue.front();
76 0 : taskQueue.pop();
77 0 : return true;
78 1 : }
79 :
80 1 : void RankInfoDispather::RunWorkerThread(s32 workId)
81 : {
82 : // 给当前线程添加名字
83 1 : SetThreadName("Hccl_RunWorker");
84 :
85 1 : while (!stop_) {
86 0 : WorkerWait(workId);
87 : while (true) {
88 0 : WorkerTask task;
89 0 : if (GetTask(task)) {
90 0 : task();
91 : } else {
92 0 : break;
93 : }
94 0 : }
95 : }
96 1 : HCCL_DEBUG("[RankInfoDispather::%s]finish thread! workId[%d]", __func__, workId);
97 1 : }
98 :
99 2 : void RankInfoDispather::PrepareResource(
100 : const std::unordered_map<std::string, std::shared_ptr<Socket>> connectSockets, const RankTableInfo& clusterInfo,
101 : const std::string& failedAgentIdList, u32 step)
102 : {
103 2 : rankNum_ = connectSockets.size();
104 2 : InitWorkerThread();
105 :
106 2 : s32 res = RaCreateEventHandle(&epollFds_);
107 2 : CHK_PRT_THROW(
108 : res != 0, HCCL_ERROR("[RankInfoDispather::%s] create epoll event failed, res[%d].", __func__, res),
109 : NetworkApiException, "create epoll event error.");
110 2 : epollCreate_ = true;
111 :
112 2 : BinaryStream binaryStream;
113 2 : clusterInfo.GetBinStream(true, binaryStream);
114 2 : binaryStream << step;
115 2 : binaryStream << failedAgentIdList;
116 :
117 2 : binaryStream.Dump(rankTableMsg_);
118 :
119 3 : for (auto& it : connectSockets) {
120 2 : FdContext fdcontext;
121 2 : fdcontext.socket = it.second;
122 2 : fdcontext.txState.bodyLen = rankTableMsg_.size();
123 2 : fdcontext.txState.data = rankTableMsg_.data();
124 3 : CHK_RET_THROW(
125 : InvalidParamsException,
126 : StringFormat("[RankInfoDispather::%s] ranid[%s] strToULong fail.", __func__, it.first.c_str()),
127 : SalStrToULong(it.first, HCCL_BASE_DECIMAL, fdcontext.txState.rankId));
128 1 : HCCL_DEBUG(
129 : "[RankInfoDispather::%s]rankId:%u, bodyLen:%u", __func__, fdcontext.txState.rankId,
130 : fdcontext.txState.bodyLen);
131 1 : fdHandleToFdContextMap_.emplace(it.second->GetFdHandle(), fdcontext);
132 2 : }
133 :
134 1 : HCCL_INFO("[RankInfoDispather::%s]fdHandleToFdContextMap_ size[%d]", __func__, fdHandleToFdContextMap_.size());
135 2 : }
136 :
137 494577 : void RankInfoDispather::WakeWoker()
138 : {
139 494577 : std::unique_lock<std::mutex> lck(wakeMutex_);
140 494577 : ready_ = true;
141 494577 : wakeManager_.notify_all();
142 494577 : }
143 :
144 21 : void RankInfoDispather::CleanResource()
145 : {
146 : // 主线程广播结束，结束从线程（不确定是否存在出于wait状态的线程，统一全部唤醒）
147 21 : stop_ = true;
148 21 : WakeWoker();
149 21 : HCCL_INFO("[RankInfoDispather::%s]wake all workers.", __func__);
150 22 : for (auto& th : workerThreads_) {
151 1 : if (th.joinable()) {
152 1 : th.join();
153 : }
154 : }
155 21 : fdHandleToFdContextMap_.clear();
156 21 : workerThreads_.clear();
157 21 : }
158 :
159 4 : void RankInfoDispather::ProcessOneSendEvent([[maybe_unused]] s32 epollFd, FdHandle& fdHanlde)
160 : {
161 4 : std::unique_lock<std::mutex> lckForMap(fdHandleMapMutex_);
162 4 : bool noFdHandle = (fdHandleToFdContextMap_.find(fdHanlde) == fdHandleToFdContextMap_.end());
163 4 : if (noFdHandle) {
164 1 : stop_ = true;
165 : }
166 4 : CHK_PRT_RET(noFdHandle, HCCL_ERROR("[RankInfoDispather::%s]no fdhandle[%p]", __func__, fdHanlde), );
167 3 : auto ctx = &(fdHandleToFdContextMap_.at(fdHanlde));
168 3 : bool sendFailed = !ctx->txState.Send(ctx->socket);
169 3 : if (sendFailed) {
170 1 : stop_ = true;
171 : }
172 3 : CHK_PRT_RET(
173 : sendFailed,
174 : HCCL_ERROR("[RankInfoDispather::%s]send data to rank[%u] failed.", __func__, ctx->txState.rankId), );
175 :
176 2 : s32 ctlType = EPOLL_CTL_DEL;
177 2 : if (ctx->txState.IsOk()) {
178 1 : sendDoneCount_++;
179 : } else {
180 1 : ctlType = EPOLL_CTL_MOD;
181 : }
182 : // EPOLLOUT_LET_ONESHOT -> EPOLLOUT | EPOLLET | EPOLLONESHOT, 防止多个线程同时操作同一个fd（fd重复触发）
183 2 : s32 ret = RaCtlEventHandle(epollFds_, fdHanlde, ctlType, RaEpollEvent::RA_EPOLLOUT_LET_ONESHOT);
184 2 : bool ctlFailed = (ret != 0);
185 2 : if (ctlFailed) {
186 1 : stop_ = true;
187 : }
188 2 : CHK_PRT_RET(ctlFailed, HCCL_ERROR("[RankInfoDispather::%s]epoll_ctl failed, ctlType[%d]", __func__, ctlType), );
189 4 : }
190 :
191 3 : void RankInfoDispather::SendOnce()
192 : {
193 5 : for (auto& it : fdHandleToFdContextMap_) {
194 3 : auto fdCtx = &(it.second);
195 5 : CHK_PRT_THROW(
196 : !fdCtx->txState.Send(fdCtx->socket),
197 : HCCL_ERROR("[RankInfoDispather::%s]Send data to rank[%u] failed.", __func__, fdCtx->txState.rankId),
198 : InvalidParamsException, "send data error.");
199 :
200 : // 数据未发送完成，添加epoll事件
201 2 : if (!fdCtx->txState.IsOk()) {
202 : // EPOLLOUT_LET_ONESHOT -> EPOLLOUT | EPOLLET | EPOLLONESHOT, 防止多个线程同时操作同一个fd（fd重复触发）
203 1 : s32 ret = RaCtlEventHandle(epollFds_, it.first, EPOLL_CTL_ADD, RaEpollEvent::RA_EPOLLOUT_LET_ONESHOT);
204 1 : CHK_PRT_THROW(
205 : ret != 0, HCCL_ERROR("[RankInfoDispather::%s]epoll_ctl add fd failed.", __func__),
206 : InvalidParamsException, "send data error.");
207 : } else {
208 1 : sendDoneCount_++;
209 : }
210 : }
211 2 : }
212 :
213 2 : void RankInfoDispather::ProcessSend()
214 : {
215 2 : SendOnce(); // 先尝试发送数据
216 4 : HCCL_INFO(
217 : "[RankInfoDispather::%s]sendOnce success, start epoll_wait. sendDoneCount[%d], rankNum[%u].", __func__,
218 : sendDoneCount_.load(), rankNum_);
219 2 : const s32 sendEvsCount = 20; // epoll_wait 缓冲区大小（单次触发的事件个数）
220 2 : std::vector<SocketEventInfo> eventInfos(sendEvsCount);
221 2 : bool lastEpollWaitFlag = false; // 最后一轮epoll_wait标识位
222 2 : auto timeout = std::chrono::seconds(EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
223 2 : auto startTime = std::chrono::steady_clock::now();
224 : HcclResult ret;
225 494560 : while (sendDoneCount_ != rankNum_) {
226 494560 : CHK_PRT_THROW(
227 : stop_, HCCL_ERROR("[RankInfoDispather::%s] process stop.", __func__), InvalidParamsException,
228 : "process stop.");
229 :
230 494560 : if (rankNum_ - sendDoneCount_ < sendEvsCount && !lastEpollWaitFlag) { // 最后一轮epoll_wait
231 2 : lastEpollWaitFlag = true;
232 : }
233 :
234 : // 循环超时
235 494560 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
236 1 : HCCL_ERROR(
237 : "[RankInfoDispather::%s] epoll_wait timeout, timeout[%lld s].", __func__,
238 : static_cast<long long>(timeout.count()));
239 7 : RPT_INPUT_ERR(
240 : true, "EI0015", std::vector<std::string>({"error_reason"}),
241 : std::vector<std::string>({StringFormat(
242 : "Receiving message from the root node timed out "
243 : "Timeout was set to %lld seconds. Expected to send to %u nodes, completed %u nodes.",
244 : static_cast<long long>(timeout.count()), rankNum_, sendDoneCount_.load())}));
245 1 : THROW<TimeoutException>("epoll_wait timeout");
246 : }
247 :
248 : // 等待epoll事件
249 494559 : s32 epollTimeout = lastEpollWaitFlag ? LAST_EPOLL_TIMEOUT_MS : EPOLL_TIMEOUT_MS;
250 494559 : u32 eventsNum{0};
251 494559 : ret = HrtRaWaitEventHandle(epollFds_, eventInfos, epollTimeout, sendEvsCount, eventsNum);
252 :
253 : // 最后一轮epoll_wait结束, 等待超时，epoll池内无事件
254 494559 : CHK_PRT_RET(
255 : (eventsNum == 0 && ret == HCCL_SUCCESS && sendDoneCount_ == rankNum_),
256 : HCCL_WARNING(
257 : "[RankInfoDispather::%s]hrtRaWaitEventHandle is timeout[%d] ms, eventsNum[%u], "
258 : "sendDoneCount_[%d]",
259 : __func__, epollTimeout, eventsNum, sendDoneCount_.load()), );
260 :
261 : // epoll wait事件失败
262 : // 可能出现ret==HCCL_SUCCESS但eventsNum==0的情况，属于正常情况，不报错退出，需要继续循环
263 494561 : CHK_PRT_THROW(
264 : ret != HCCL_SUCCESS,
265 : HCCL_ERROR(
266 : "[RankInfoDispather::%s] HrtRaWaitEventHandle failed ret[%d], eventsNum[%u].", __func__, ret,
267 : eventsNum),
268 : InvalidParamsException, "epoll_wait fail");
269 989116 : for (u32 i = 0; i < eventsNum; ++i) {
270 494558 : std::unique_lock<std::mutex> lck(taskQueueMutex_);
271 494558 : taskQueue_.push(std::bind(
272 494558 : &RankInfoDispather::ProcessOneSendEvent, this, epollFds_, static_cast<void*>(eventInfos[i].fdHandle)));
273 494558 : lck.unlock();
274 494558 : }
275 : // 唤醒处理
276 494558 : WakeWoker();
277 : }
278 :
279 0 : CloseEpollFd();
280 0 : HCCL_INFO(
281 : "[RankInfoDispather::%s]ProcessSend success, sendDoneCount[%d], rankNum[%d].", __func__, sendDoneCount_.load(),
282 : rankNum_);
283 4 : }
284 :
285 20 : void RankInfoDispather::CloseEpollFd()
286 : {
287 20 : if (epollCreate_) {
288 2 : s32 ret = RaDestroyEventHandle(&epollFds_);
289 2 : CHK_PRT_THROW(
290 : ret != 0, HCCL_ERROR("[RankInfoDispather::%s] destroy epoll event failed, res[%d].", __func__, ret),
291 : NetworkApiException, "destroy epoll event error.");
292 2 : epollCreate_ = false;
293 : }
294 20 : }
295 :
296 3 : bool RankInfoDispather::SendState::Send(std::shared_ptr<Socket> socket)
297 : {
298 3 : if (headerSended != headerLen) {
299 2 : header = bodyLen;
300 2 : CHK_PRT_RET(!SendHeader(socket), HCCL_ERROR("SendHeader error"), false);
301 : }
302 :
303 2 : if ((headerSended == headerLen) && (bodyLen != bodySended)) {
304 1 : CHK_PRT_RET(!SendBody(socket), HCCL_ERROR("SendBody error"), false);
305 : }
306 :
307 2 : return true;
308 : }
309 :
310 2 : bool RankInfoDispather::SendState::SendHeader(std::shared_ptr<Socket> socket)
311 : {
312 2 : return SendHelper(socket, &header, headerLen, headerSended);
313 : }
314 :
315 1 : bool RankInfoDispather::SendState::SendBody(std::shared_ptr<Socket> socket)
316 : {
317 1 : return SendHelper(socket, data, bodyLen, bodySended);
318 : }
319 :
320 5 : bool RankInfoDispather::SendState::SendHelper(
321 : std::shared_ptr<Socket> socket, void* buf, size_t dataLen, size_t& sendedLen)
322 : {
323 5 : u64 needSend = dataLen - sendedLen;
324 5 : u64 sentSize = 0;
325 5 : u8* sendData = static_cast<u8*>(buf) + sendedLen;
326 5 : CHK_PRT_RET(!socket->ISend(sendData, needSend, sentSize), HCCL_ERROR("ISend fail"), false);
327 3 : sendedLen += sentSize;
328 3 : return true;
329 : }
330 :
331 : } // namespace Hccl
|
