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