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