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