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 : #include "stream_manager.h"
11 : #include "log.h"
12 : #include "exception_util.h"
13 : #include "communicator_impl.h"
14 : #include "stream_utils.h"
15 :
16 : namespace Hccl {
17 :
18 282 : StreamManager::StreamManager(CommunicatorImpl *comm) : comm(comm)
19 : {
20 282 : opbase = std::make_unique<OpbaseStreamManager>(comm);
21 282 : offload = std::make_unique<OffloadStreamManager>();
22 282 : }
23 :
24 5 : Stream *StreamManager::GetSlave() const
25 : {
26 15 : HCCL_INFO("[StreamManager::%s] start.", __func__);
27 :
28 5 : Stream *stream = nullptr;
29 5 : auto op = comm->GetCurrentCollOperator();
30 5 : OpMode opMode = op->opMode;
31 5 : if (opMode == OpMode::OPBASE) {
32 5 : stream = comm->GetStreamManager().opbase->GetOrCreateSlave();
33 0 : } else if (opMode == OpMode::OFFLOAD) {
34 0 : stream = comm->GetStreamManager().offload->GetSlave(op->opTag);
35 : } else {
36 0 : THROW<NotSupportException>(StringFormat("Unsupported OpMode: %s", opMode.Describe().c_str()));
37 : }
38 :
39 15 : HCCL_INFO("[StreamManager::%s] end, opMode[%s], slave stream[%u].",
40 : __func__, opMode.Describe().c_str(), stream->GetId());
41 5 : return stream;
42 : }
43 :
44 3 : Stream *StreamManager::GetSlaveByIndex(u32 index) const
45 : {
46 9 : HCCL_INFO("[StreamManager::%s] start.", __func__);
47 :
48 3 : Stream *stream = nullptr;
49 3 : auto op = comm->GetCurrentCollOperator();
50 3 : OpMode opMode = op->opMode;
51 3 : if (opMode == OpMode::OPBASE) {
52 3 : stream = comm->GetStreamManager().opbase->GetSlave(index);
53 0 : } else if (opMode == OpMode::OFFLOAD) {
54 0 : stream = comm->GetStreamManager().offload->GetSlave(op->opTag, index);
55 : } else {
56 0 : THROW<NotSupportException>(StringFormat("Unsupported OpMode: %s", opMode.Describe().c_str()));
57 : }
58 :
59 9 : HCCL_INFO("[StreamManager::%s] end, opMode[%s], slave stream[%u].",
60 : __func__, opMode.Describe().c_str(), stream->GetId());
61 3 : return stream;
62 : }
63 :
64 22 : Stream *StreamManager::GetMaster() const
65 : {
66 66 : HCCL_INFO("[StreamManager::%s] start.", __func__);
67 :
68 22 : Stream *stream = nullptr;
69 22 : auto op = comm->GetCurrentCollOperator();
70 22 : OpMode opMode = op->opMode;
71 22 : if (opMode == OpMode::OPBASE) {
72 21 : stream = comm->GetStreamManager().opbase->GetMaster();
73 1 : } else if (opMode == OpMode::OFFLOAD) {
74 1 : stream = comm->GetStreamManager().offload->GetMaster(op->opTag);
75 : } else {
76 0 : THROW<NotSupportException>(StringFormat("Unsupported OpMode: %s", opMode.Describe().c_str()));
77 : }
78 :
79 66 : HCCL_INFO("[StreamManager::%s] end, opMode[%s], master stream[%u].",
80 : __func__, opMode.Describe().c_str(), stream->GetId());
81 22 : return stream;
82 : }
83 :
84 16 : void StreamManager::CaptureSlaveStream(const Stream *masterStream, const Stream *slaveStream) const
85 : {
86 48 : HCCL_RUN_INFO("[StreamManager][%s] masterStream[%u] slaveStream[%u]", __func__,
87 : masterStream->GetId(), slaveStream->GetId());
88 16 : rtModel_t rtModel = nullptr;
89 16 : bool isCapture = false;
90 16 : u32 modelId = 0;
91 16 : auto op = comm->GetCurrentCollOperator();
92 16 : OpMode opMode = op->opMode;
93 16 : if (opMode == OpMode::OPBASE) {
94 10 : auto ret = GetStreamCaptureInfo(masterStream->GetPtr(), rtModel, isCapture);
95 10 : if (ret != HCCL_SUCCESS) {
96 2 : THROW<InternalException>(StringFormat("[StreamManager::%s] Failed to obtain masterStream capture status, "
97 : "ret[%d]", __func__, ret));
98 : }
99 :
100 9 : if (isCapture) {
101 4 : if (rtModel == nullptr) {
102 2 : THROW<NullPtrException>(StringFormat("[StreamManager::%s] rtModel is NULL.", __func__));
103 : }
104 :
105 3 : ret = GetModelId(rtModel, modelId);
106 3 : if (ret != HCCL_SUCCESS) {
107 2 : THROW<InternalException>(StringFormat("[StreamManager::%s] Failed to obtain the modelId corresponding "
108 : "to the masterStream rtModel, ret[%d]", __func__, ret));
109 : }
110 :
111 2 : ret = AddStreamToModel(slaveStream->GetPtr(), rtModel);
112 2 : if (ret != HCCL_SUCCESS) {
113 2 : THROW<InternalException>(StringFormat("[StreamManager::%s] Adding the salveStream to the masterStream "
114 : "failed, ret[%d]", __func__, ret));
115 : }
116 3 : HCCL_RUN_INFO("[StreamManager::%s] Add slaveStream[%u] to model[%u] success, masterStream[%u]",
117 : __func__, slaveStream->GetId(), modelId, masterStream->GetId());
118 : }
119 : }
120 12 : }
121 :
122 5 : u32 StreamManager::GetSlaveIndex() const
123 : {
124 15 : HCCL_INFO("[StreamManager::%s] start.", __func__);
125 :
126 5 : u32 res = 0;
127 5 : auto op = comm->GetCurrentCollOperator();
128 5 : OpMode opMode = op->opMode;
129 5 : if (opMode == OpMode::OPBASE) {
130 5 : res = comm->GetStreamManager().opbase->GetSlaveIndex();
131 0 : } else if (opMode == OpMode::OFFLOAD) {
132 0 : res = comm->GetStreamManager().offload->GetSlaveIndex(op->opTag);
133 : } else {
134 0 : THROW<NotSupportException>(StringFormat("Unsupported OpMode: %s", opMode.Describe().c_str()));
135 : }
136 :
137 15 : HCCL_INFO("[StreamManager::%s] end, opMode[%s].", __func__, opMode.Describe().c_str());
138 5 : return res;
139 : }
140 :
141 17 : void StreamManager::ResetSlaveIndex(u32 index) const
142 : {
143 51 : HCCL_INFO("[StreamManager::%s] start.", __func__);
144 :
145 17 : auto op = comm->GetCurrentCollOperator();
146 17 : OpMode opMode = op->opMode;
147 17 : if (opMode == OpMode::OPBASE) {
148 16 : comm->GetStreamManager().opbase->ResetIndex(index);
149 1 : } else if (opMode == OpMode::OFFLOAD) {
150 1 : comm->GetStreamManager().offload->ResetIndex(op->opTag, index);
151 : } else {
152 0 : THROW<NotSupportException>(StringFormat("Unsupported OpMode: %s", opMode.Describe().c_str()));
153 : }
154 :
155 51 : HCCL_INFO("[StreamManager::%s] end, opMode[%s].", __func__, opMode.Describe().c_str());
156 17 : }
157 :
158 13 : void StreamManager::RecordStreamIdToIndex(u32 streamId, u32 streamIndex)
159 : {
160 13 : streamIdToIndexMap_[streamId] = streamIndex;
161 13 : }
162 :
163 1 : u32 StreamManager::GetStreamIndex(u32 streamId)
164 : {
165 1 : return streamIdToIndexMap_[streamId];
166 : }
167 :
168 13 : void StreamManager::InitBucket(u32 bucket)
169 : {
170 13 : streamBucket_[bucket] = std::vector<u32>{};
171 13 : }
172 :
173 4 : void StreamManager::RegisterBucket(u32 bucket, u32 subStreamIndex)
174 : {
175 4 : streamBucket_[bucket].emplace_back(subStreamIndex);
176 4 : }
177 :
178 1 : std::vector<u32>& StreamManager::GetSubSlaveIndexes(u32 slaveIndex)
179 : {
180 1 : return streamBucket_[slaveIndex];
181 : }
182 :
183 12 : void StreamManager::DestroyRecords()
184 : {
185 12 : streamIdToIndexMap_.clear();
186 12 : streamBucket_.clear();
187 12 : }
188 :
189 : } // namespace Hccl
|
