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 "log.h"
12 : #include "transport_pub.h"
13 : #include "sal_pub.h"
14 : #include "task_loader.h"
15 :
16 : namespace hccl {
17 0 : TaskLoader::TaskLoader(const s32 deviceLogicId, const HcclDispatcher dispatcher)
18 0 : : deviceLogicId_(deviceLogicId), dispatcher_(dispatcher)
19 0 : {}
20 0 : TaskLoader::~TaskLoader()
21 : {
22 0 : HcclResult ret = Finalize();
23 0 : if (ret != HCCL_SUCCESS) {
24 0 : HCCL_ERROR("[TaskLoader][Destroy]TaskLoader Finalize failed[%d] ", ret);
25 : }
26 0 : }
27 :
28 0 : void TaskLoader::Prepare(Stream *stream, SubCommInfo level0CommInfo)
29 : {
30 : // 参数保存
31 0 : stream_ = stream;
32 0 : HCCL_INFO("[TaskLoader] Prepare stream[%p]", stream_->ptr());
33 0 : commInfo_ = level0CommInfo;
34 0 : executeResult_ = HCCL_SUCCESS;
35 0 : }
36 :
37 0 : HcclResult TaskLoader::Init()
38 : {
39 0 : HCCL_INFO("[TaskLoader] Init");
40 0 : ringThread_.reset(new (std::nothrow) std::thread(&TaskLoader::ThreadExecuteFn, this));
41 0 : CHK_SMART_PTR_NULL(ringThread_);
42 0 : return HCCL_SUCCESS;
43 : }
44 :
45 0 : HcclResult TaskLoader::GetExecuteResult()
46 : {
47 0 : HCCL_INFO("[TaskLoader] ExecuteResult [%d]", executeResult_);
48 0 : return executeResult_;
49 : }
50 :
51 0 : HcclResult TaskLoader::Finalize()
52 : {
53 0 : if (ringThread_) {
54 0 : threadExit = true;
55 0 : NotifyStart();
56 0 : if (ringThread_->joinable()) {
57 0 : ringThread_->join();
58 : }
59 0 : ringThread_ = nullptr;
60 : }
61 0 : HCCL_INFO("[TaskLoader] Finalize");
62 0 : return HCCL_SUCCESS;
63 : }
64 :
65 0 : void TaskLoader::NotifyStart()
66 : {
67 0 : std::unique_lock<std::mutex> lock(startMtx_);
68 0 : startReady = true; // 设置标志位为 true.
69 0 : startCv_.notify_one();
70 0 : workflowMode_ = GetWorkflowMode(); // 每次唤醒前更新下
71 0 : HCCL_INFO("[TaskLoader] NotifyStart");
72 0 : }
73 :
74 0 : void TaskLoader::WaitStart()
75 : {
76 0 : std::unique_lock<std::mutex> lock(startMtx_);
77 0 : while (!startReady) { // 假设标志位不为 true, 则等待...
78 0 : startCv_.wait(lock); // 当前线程被堵塞, 当标志位变为 true 之后,
79 : }
80 0 : startReady = false;
81 :
82 0 : SetWorkflowMode(workflowMode_); // 更新workflowMode
83 0 : }
84 :
85 0 : void TaskLoader::NotifyDone()
86 : {
87 0 : std::unique_lock<std::mutex> lock(doneMtx_);
88 0 : doneReady = true;
89 0 : doneCv_.notify_one();
90 0 : }
91 :
92 0 : void TaskLoader::WaitDone()
93 : {
94 0 : std::unique_lock<std::mutex> lock(doneMtx_);
95 0 : while (!doneReady) {
96 0 : doneCv_.wait(lock);
97 : }
98 0 : doneReady = false;
99 0 : }
100 :
101 0 : HcclResult TaskLoader::ExecuteTransPortTaskInfo(TaskLogicInfo &info)
102 : {
103 0 : u32 index = info.taskLogicCmd.index;
104 :
105 0 : std::shared_ptr<Transport> destTransport = nullptr;
106 0 : if (commInfo_.virtualLinks.size() <= index) {
107 0 : HCCL_ERROR("[ExecuteTransPortTaskInfo]index[%u] is bigger than vlink size[%llu]", index,
108 : commInfo_.virtualLinks.size());
109 0 : } else if (commInfo_.links.size() <= index) {
110 0 : HCCL_ERROR("[ExecuteTransPortTaskInfo]index[%u] is bigger than link size[%llu]", index,
111 : commInfo_.links.size());
112 : } else {
113 0 : destTransport = commInfo_.links[index];
114 : }
115 :
116 0 : CHK_SMART_PTR_NULL(destTransport);
117 :
118 0 : switch (info.taskFuncType) {
119 0 : case TaskLogicFuncType::TRANSPORT_TXACK_TYPE:
120 0 : destTransport->TxAck(*stream_);
121 0 : break;
122 0 : case TaskLogicFuncType::TRANSPORT_RXACK_TYPE:
123 0 : destTransport->RxAck(*stream_);
124 0 : break;
125 0 : case TaskLogicFuncType::TRANSPORT_TXASYNC_TYPE:
126 0 : destTransport->TxAsync(info.txAsync.txMems, *stream_);
127 0 : break;
128 0 : case TaskLogicFuncType::TRANSPORT_RXASYNC_TYPE:
129 0 : destTransport->RxAsync(info.rxAsync.rxMems, *stream_);
130 0 : break;
131 0 : case TaskLogicFuncType::TRANSPORT_TXDATASIGNAL_TYPE:
132 0 : destTransport->TxDataSignal(*stream_);
133 0 : break;
134 0 : case TaskLogicFuncType::TRANSPORT_RXDATASIGNAL_TYPE:
135 0 : destTransport->RxDataSignal(*stream_);
136 0 : break;
137 0 : default:
138 0 : HCCL_ERROR("[TaskLoader][ExecuteTransPortTaskInfo]Invalid taskFuncType[%d]", info.taskFuncType);
139 0 : return HCCL_E_PARA;
140 : }
141 0 : return HCCL_SUCCESS;
142 0 : }
143 :
144 0 : HcclResult TaskLoader::ExecuteDispatcherTaskInfo(TaskLogicInfo &info)
145 : {
146 0 : switch (info.taskFuncType) {
147 0 : case TaskLogicFuncType::DISPATCHER_SIGNALWAIT_TYPE:
148 0 : HcclSignalWait(dispatcher_,
149 : info.taskLogicPara.dispatcherTaskLogicPara.signalWait.signal,
150 0 : *stream_,
151 : info.taskLogicPara.dispatcherTaskLogicPara.signalWait.userRank,
152 : info.taskLogicPara.dispatcherTaskLogicPara.signalWait.remoteRank,
153 : info.taskLogicPara.dispatcherTaskLogicPara.signalWait.stage,
154 : true);
155 0 : break;
156 0 : case TaskLogicFuncType::DISPATCHER_SIGNALRECORD_TYPE:
157 0 : HcclSignalRecord(dispatcher_,
158 : info.taskLogicPara.dispatcherTaskLogicPara.signalRecord.signal,
159 0 : *stream_,
160 : info.taskLogicPara.dispatcherTaskLogicPara.signalRecord.userRank,
161 : info.taskLogicPara.dispatcherTaskLogicPara.signalRecord.offset,
162 : info.taskLogicPara.dispatcherTaskLogicPara.signalRecord.stage,
163 : true, INVALID_U64);
164 0 : break;
165 0 : case TaskLogicFuncType::DISPATCHER_MEMCPYASYNC_TYPE:
166 0 : HcclMemcpyAsync(dispatcher_,
167 : info.taskLogicPara.dispatcherTaskLogicPara.memAsync.dst,
168 : info.taskLogicPara.dispatcherTaskLogicPara.memAsync.destMax,
169 0 : info.taskLogicPara.dispatcherTaskLogicPara.memAsync.src,
170 0 : info.taskLogicPara.dispatcherTaskLogicPara.memAsync.count,
171 : info.taskLogicPara.dispatcherTaskLogicPara.memAsync.kind,
172 0 : *stream_,
173 : INVALID_VALUE_RANKID, LinkType::LINK_ONCHIP);
174 0 : break;
175 0 : default:
176 0 : HCCL_ERROR("[TaskLoader][ExecuteDispatcherTaskInfo]Invalid taskFuncType[%d]", info.taskFuncType);
177 0 : return HCCL_E_PARA;
178 : }
179 0 : return HCCL_SUCCESS;
180 : }
181 :
182 0 : HcclResult TaskLoader::ExecuteTaskLogicPara(TaskLogicInfo &info)
183 : {
184 0 : if (info.taskLogicCmd.taskLogicType == TaskLogicType::TRANSPORT_TYPE) {
185 0 : CHK_RET(ExecuteTransPortTaskInfo(info));
186 0 : } else if (info.taskLogicCmd.taskLogicType == TaskLogicType::DISPATCHER_TYPE) {
187 0 : CHK_RET(ExecuteDispatcherTaskInfo(info));
188 : } else {
189 0 : HCCL_ERROR("[TaskLoader][ExecuteTaskLogicPara]Invalid taskLogicType[%d]", info.taskLogicCmd.taskLogicType);
190 0 : return HCCL_E_PARA;
191 : }
192 0 : return HCCL_SUCCESS;
193 : }
194 :
195 0 : HcclResult TaskLoader::ExecuteService()
196 : {
197 0 : TaskLogicInfo taskLogicInfo;
198 0 : while (stream_->PopTaskLogicInfo(taskLogicInfo) == HCCL_SUCCESS) {
199 0 : CHK_RET(ExecuteTaskLogicPara(taskLogicInfo));
200 : }
201 0 : return HCCL_SUCCESS;
202 0 : }
203 :
204 0 : HcclResult TaskLoader::ThreadExecuteFn()
205 : {
206 : //给当前线程添加名字
207 0 : SetThreadName("Hccl_TaskLoader");
208 :
209 0 : threadId_ = SalGetTid();
210 0 : HCCL_INFO("[TaskLoader][ThreadExecuteFn]deviceLogicId_[%d], threadId_[%u]", deviceLogicId_, threadId_);
211 0 : CHK_RET(hrtSetDevice(deviceLogicId_));
212 :
213 : while (true) {
214 0 : WaitStart(); // 等待线程执行通知
215 0 : if (threadExit) {
216 0 : HCCL_INFO("[TaskLoader][ThreadExecuteFn]threadExit deviceLogicId_[%d]", deviceLogicId_);
217 0 : break;
218 : }
219 0 : HcclResult ret = ExecuteService();
220 0 : if (ret != HCCL_SUCCESS) {
221 0 : HCCL_ERROR("[TaskLoader][ThreadExecuteFn]TaskLoader run ExecuteService fail");
222 0 : executeResult_ = ret;
223 : }
224 0 : NotifyDone(); // 通知主进程本线程执行完成
225 0 : }
226 0 : CHK_RET(hrtResetDevice(deviceLogicId_));
227 :
228 0 : return HCCL_SUCCESS;
229 : }
230 :
231 0 : uint32_t TaskLoader::GetTid()
232 : {
233 0 : if (threadId_ == 0) {
234 0 : threadId_ = SalGetTid();
235 : }
236 0 : HCCL_INFO("[TaskLoader][GetTid]deviceLogicId_[%d], threadId_[%u]", deviceLogicId_, threadId_);
237 0 : return threadId_;
238 : }
239 :
240 0 : HcclResult TaskLoader::ClearTagCommInfo()
241 : {
242 0 : commInfo_ = SubCommInfo{};
243 0 : return HCCL_SUCCESS;
244 : }
245 :
246 : } // namespace hccl
|
