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 "alg_template_base_pub.h"
13 : #include "hccl_impl_pub.h"
14 : #include "reduce_scatter_ring_pub.h"
15 : #include "reduce_scatter_ring_concurrent_direct_pub.h"
16 : #include "all_gather_ring_pub.h"
17 : #include "all_gather_ring_concurrent_direct_pub.h"
18 : #include "coll_executor_base.h"
19 : #include "sal_pub.h"
20 : #include "profiler_base_pub.h"
21 : #include "threadManage.h"
22 :
23 : namespace hccl {
24 70 : ThreadManage::ThreadManage(s32 deviceLogicId, u32 userRank, const HcclDispatcher dispatcher)
25 70 : : deviceLogicId_(deviceLogicId),
26 70 : userRank_(userRank),
27 70 : dispatcher_(dispatcher),
28 70 : context_(nullptr)
29 70 : {}
30 70 : ThreadManage::~ThreadManage()
31 : {
32 70 : HcclResult ret = Finalize();
33 70 : if (ret != HCCL_SUCCESS) {
34 0 : HCCL_ERROR("[ThreadManage][Destroy]threadManage Finalize failed[%d] ", ret);
35 : }
36 70 : }
37 :
38 70 : HcclResult ThreadManage::Init()
39 : {
40 70 : HCCL_INFO("ThreadManage::Init");
41 70 : CHK_RET(hrtCtxGetCurrent(&context_));
42 :
43 70 : ringThread_.reset(new (std::nothrow) std::thread(&ThreadManage::ThreadExecuteFn, this));
44 70 : CHK_SMART_PTR_NULL(ringThread_);
45 70 : return HCCL_SUCCESS;
46 : }
47 :
48 70 : HcclResult ThreadManage::Finalize()
49 : {
50 70 : if (ringThread_) {
51 70 : threadExit = true;
52 70 : NotifyStart();
53 70 : if (ringThread_->joinable()) {
54 70 : ringThread_->join();
55 : }
56 70 : ringThread_ = nullptr;
57 : }
58 70 : return HCCL_SUCCESS;
59 : }
60 :
61 70 : void ThreadManage::NotifyStart()
62 : {
63 70 : std::unique_lock<std::mutex> lock(startMtx_);
64 70 : startReady = true; // 设置标志位为 true.
65 70 : startCv_.notify_one();
66 70 : workflowMode_ = GetWorkflowMode();
67 70 : }
68 :
69 70 : void ThreadManage::WaitStart()
70 : {
71 70 : std::unique_lock<std::mutex> lock(startMtx_);
72 140 : while (!startReady) { // 假设标志位不为 true, 则等待...
73 70 : startCv_.wait(lock); // 当前线程被堵塞, 当标志位变为 true 之后,
74 : }
75 70 : startReady = false;
76 :
77 70 : SetWorkflowMode(workflowMode_);
78 70 : }
79 :
80 0 : void ThreadManage::NotifyDone()
81 : {
82 0 : std::unique_lock<std::mutex> lock(doneMtx_);
83 0 : doneReady = true;
84 0 : doneCv_.notify_one();
85 0 : }
86 :
87 0 : void ThreadManage::WaitDone()
88 : {
89 0 : std::unique_lock<std::mutex> lock(doneMtx_);
90 0 : while (!doneReady) {
91 0 : doneCv_.wait(lock);
92 : }
93 0 : doneReady = false;
94 0 : }
95 :
96 0 : HcclResult ThreadManage::ExecuteService()
97 : {
98 0 : HcclResult ret = HCCL_SUCCESS;
99 :
100 0 : std::unique_ptr<AlgTemplateBase> tempAlg;
101 0 : if (executorType_ == ExecutorType::REDUCE_SCATTER_RING) {
102 : tempAlg
103 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_RING, dispatcher_);
104 0 : CHK_SMART_PTR_NULL(tempAlg);
105 0 : CHK_RET(tempAlg->Prepare(reduceAttr_));
106 0 : } else if (executorType_ == ExecutorType::ALLGATHER_RING) {
107 0 : tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
108 0 : } else if (executorType_ == ExecutorType::REDUCE_SCATTER_RING_DIRECT) {
109 0 : tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
110 0 : TemplateType::TEMPLATE_REDUCESCATTER_RING_DIRECT, dispatcher_);
111 0 : CHK_SMART_PTR_NULL(tempAlg);
112 0 : CHK_RET(tempAlg->Prepare(
113 : reduceAttr_, opInfo_, userRank_, subStreamsInOneRing_, mainSignalsInOneRing_, subSignalsInOneRing_,
114 : ringsOrder_, userMemInputSlices_));
115 0 : } else if (executorType_ == ExecutorType::REDUCE_SCATTER_RING_DIRECT_RDMA) {
116 0 : tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
117 0 : TemplateType::TEMPLATE_REDUCESCATTER_RING_DIRECT, dispatcher_);
118 0 : CHK_SMART_PTR_NULL(tempAlg);
119 0 : CHK_RET(tempAlg->Prepare(
120 : reduceAttr_, opInfo_, userRank_, subStreamsInOneRing_, mainSignalsInOneRing_, subSignalsInOneRing_,
121 : ringsOrder_, userMemInputSlices_, false));
122 0 : } else if (executorType_ == ExecutorType::ALLGATHER_RING_DIRECT) {
123 0 : tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
124 0 : TemplateType::TEMPLATE_ALL_GATHER_RING_CONCURRENT_DIRECT, dispatcher_);
125 0 : CHK_SMART_PTR_NULL(tempAlg);
126 0 : CHK_RET(tempAlg->Prepare(
127 : const_cast<HcomCollOpInfo*>(opInfo_), userRank_, subStreamsInOneRing_, mainSignalsInOneRing_,
128 : subSignalsInOneRing_, ringsOrder_, userMemInputSlices_));
129 0 : } else if (executorType_ == ExecutorType::ALLGATHER_RING_DIRECT_RDMA) {
130 0 : tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
131 0 : TemplateType::TEMPLATE_ALL_GATHER_RING_CONCURRENT_DIRECT, dispatcher_);
132 0 : CHK_SMART_PTR_NULL(tempAlg);
133 0 : CHK_RET(tempAlg->Prepare(
134 : const_cast<HcomCollOpInfo*>(opInfo_), userRank_, subStreamsInOneRing_, mainSignalsInOneRing_,
135 : subSignalsInOneRing_, ringsOrder_, userMemInputSlices_, false));
136 : }
137 0 : CHK_SMART_PTR_NULL(tempAlg);
138 :
139 : /* 从环等待启动 */
140 0 : ret = LocalNotify::Wait(stream_, dispatcher_, signalAux_, profStage_);
141 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Execute][Service]stream[%u] wait failed", ringIndex_), ret);
142 :
143 0 : ret = tempAlg->Prepare(
144 0 : inputMem_, outputMem_, scratchMem_, count_, dataType_, stream_, reductionOp_, LEVEL0_BRIDGE_RANK_ID, slices_,
145 0 : baseOffset_, nicRankList_);
146 0 : CHK_PRT_RET(
147 : ret != HCCL_SUCCESS, HCCL_ERROR("[Execute][Service]stream[%u],prepare failed,return[%d]", ringIndex_, ret),
148 : ret);
149 :
150 0 : ret = tempAlg->RegisterProfiler(
151 0 : ((ringIndex_ + 1) << PROF_RINGINDEX_OFFSET_OF_PLANEID)
152 0 : + (ringSubCommInfo_.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + ringSubCommInfo_.localRank,
153 0 : profStage_, HCCL_EXEC_STEP_NOT_SET, stream_);
154 0 : CHK_PRT_RET(
155 : ret != HCCL_SUCCESS,
156 : HCCL_ERROR("[Execute][Service]stream[%u],register Profiler failed,return[%d]", ringIndex_, ret), ret);
157 :
158 0 : ret = CollExecutorBase::RunTemplate(tempAlg, ringSubCommInfo_);
159 :
160 0 : CHK_PRT_RET(
161 : ret != HCCL_SUCCESS, HCCL_ERROR("[Execute][Service]stream[%u],run failed,return[%d]", ringIndex_, ret), ret);
162 : /* 从环record通知主环结束 */
163 0 : ret = LocalNotify::Post(stream_, dispatcher_, signalMain_, profStage_);
164 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Execute][Service]stream[%u] record failed", ringIndex_), ret);
165 0 : return HCCL_SUCCESS;
166 0 : }
167 :
168 70 : HcclResult ThreadManage::ThreadExecuteFn()
169 : {
170 : // 给当前线程添加名字
171 70 : SetThreadName("Hccl_ThrdManage");
172 :
173 70 : threadId_ = SalGetTid();
174 70 : HCCL_INFO("[ThreadManage][ThreadExecuteFn]deviceLogicId_[%d], threadId_[%u]", deviceLogicId_, threadId_);
175 :
176 70 : CHK_RET(hrtSetDevice(deviceLogicId_));
177 70 : if (context_ != nullptr) {
178 68 : CHK_RET(hrtCtxSetCurrent(context_));
179 : }
180 : while (true) {
181 70 : WaitStart(); // 等待线程执行通知
182 70 : if (threadExit) {
183 70 : HCCL_INFO("threadExit deviceLogicId_[%d] ringIndex_[%u]", deviceLogicId_, ringIndex_);
184 70 : break;
185 : }
186 0 : HcclResult ret = ExecuteService();
187 0 : if (ret != HCCL_SUCCESS) {
188 0 : HCCL_ERROR("[ThreadManage][ThreadExecuteFn]ThreadManage run ExecuteService fail");
189 : }
190 0 : NotifyDone(); // 通知主进程本线程执行完成
191 0 : }
192 70 : CHK_RET(hrtResetDevice(deviceLogicId_));
193 :
194 70 : return HCCL_SUCCESS;
195 : }
196 :
197 0 : HcclResult ThreadManage::Prepare(
198 : DeviceMem& inputMem, DeviceMem& outputMem, DeviceMem& scratchMem, const u64 count, const HcclDataType dataType,
199 : const Stream& stream, const HcclReduceOp reductionOp, const u32 root, const std::vector<Slice>& slices,
200 : const u64 baseOffset, std::vector<u32> nicRankList, const std::string& tag, s32 profStage,
201 : const SubCommInfo& ringSubCommInfo, std::shared_ptr<LocalNotify>& signalAux,
202 : std::shared_ptr<LocalNotify>& signalMain, u32 ringIndex, ExecutorType type, u64 reduceAttr,
203 : const HcomCollOpInfo* opInfo, std::vector<Stream> subStreamsInOneRing,
204 : std::vector<std::shared_ptr<LocalNotify>> mainSignalsInOneRing,
205 : std::vector<std::shared_ptr<LocalNotify>> subSignalsInOneRing, std::vector<u32> ringsOrder,
206 : std::vector<Slice> userMemInputSlices)
207 : {
208 : /* * 参数保存 */
209 0 : inputMem_ = inputMem;
210 0 : outputMem_ = outputMem;
211 0 : scratchMem_ = scratchMem;
212 0 : count_ = count;
213 0 : dataType_ = dataType;
214 0 : stream_ = stream;
215 0 : reductionOp_ = reductionOp;
216 0 : root_ = root;
217 0 : baseOffset_ = baseOffset;
218 0 : profStage_ = profStage;
219 0 : ringSubCommInfo_ = ringSubCommInfo;
220 0 : signalAux_ = signalAux;
221 0 : signalMain_ = signalMain;
222 0 : ringIndex_ = ringIndex;
223 0 : reduceAttr_ = reduceAttr;
224 0 : opInfo_ = opInfo;
225 0 : subStreamsInOneRing_ = subStreamsInOneRing;
226 0 : mainSignalsInOneRing_ = mainSignalsInOneRing;
227 0 : subSignalsInOneRing_ = subSignalsInOneRing;
228 0 : ringsOrder_ = ringsOrder;
229 0 : userMemInputSlices_ = userMemInputSlices;
230 0 : executorType_ = type;
231 :
232 0 : tag_.assign(tag.begin(), tag.end());
233 0 : slices_.assign(slices.begin(), slices.end());
234 0 : nicRankList_.assign(nicRankList.begin(), nicRankList.end());
235 :
236 0 : HCCL_PROFILER_ADD_STREAM_BY_STREAMID(stream.id(), tag_, 0, AlgType::Reserved());
237 0 : return HCCL_SUCCESS;
238 : }
239 :
240 0 : uint32_t ThreadManage::GetTid()
241 : {
242 0 : if (threadId_ == 0) {
243 0 : threadId_ = SalGetTid();
244 : }
245 0 : HCCL_INFO("[ThreadManage][GetTid]deviceLogicId_[%d], threadId_[%u]", deviceLogicId_, threadId_);
246 0 : return threadId_;
247 : }
248 :
249 : } // namespace hccl
|
