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