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