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 "op_base_stream_manager.h"
12 :
13 : namespace hccl {
14 1053 : OpBaseStreamManager::OpBaseStreamManager() : master_()
15 : {
16 1053 : slaves_.reserve(MAX_SUBSTREAM_NUM);
17 1053 : slaveDevices_.reserve(MAX_SUBSTREAM_NUM);
18 1053 : HCCL_DEBUG("[OpBaseStreamManager]reserve slaves[%u]", MAX_SUBSTREAM_NUM);
19 1053 : }
20 :
21 1047 : OpBaseStreamManager::~OpBaseStreamManager() = default;
22 :
23 48 : HcclResult OpBaseStreamManager::RegisterMaster(Stream stream)
24 : {
25 48 : std::unique_lock<std::mutex> lock(masterMutex_);
26 46 : master_ = stream;
27 43 : HCCL_DEBUG("[OpBaseStreamManager][RegisterMaster]register master stream[%p] success.", master_.ptr());
28 48 : lock.unlock();
29 46 : return HCCL_SUCCESS;
30 46 : }
31 :
32 0 : HcclResult OpBaseStreamManager::RegisterMaster(rtStream_t rtStream)
33 : {
34 0 : std::unique_lock<std::mutex> lock(masterMutex_);
35 0 : if (rtStream == master_.ptr()) {
36 0 : return HCCL_SUCCESS;
37 : }
38 0 : master_ = Stream(rtStream);
39 0 : if (!master_.ptr()) {
40 0 : HCCL_ERROR("[OpBaseStreamManager][RegisterMaster]register master stream by rtStream[%p] failed.", rtStream);
41 0 : return HCCL_E_INTERNAL;
42 : }
43 0 : HCCL_DEBUG("[OpBaseStreamManager][RegisterMaster]register master stream by rtStream[%p] success.", rtStream);
44 0 : lock.unlock();
45 0 : return HCCL_SUCCESS;
46 0 : }
47 :
48 1 : HcclResult OpBaseStreamManager::AllocMaster(const StreamType streamType)
49 : {
50 1 : std::unique_lock<std::mutex> lock(masterMutex_);
51 1 : Stream stream(streamType);
52 1 : if (!stream.ptr()) {
53 0 : HCCL_ERROR("[OpBaseStreamManager][AllocMaster]alloc master stream of type[%d] failed.", streamType);
54 0 : return HCCL_E_INTERNAL;
55 : }
56 1 : master_ = stream;
57 1 : HCCL_INFO("[OpBaseStreamManager][AllocMaster]alloc master stream[%p] success.", master_.ptr());
58 1 : lock.unlock();
59 1 : return HCCL_SUCCESS;
60 1 : }
61 43 : std::vector<Stream> OpBaseStreamManager::AllocSlaves(const StreamType streamType, u32 num)
62 : {
63 43 : HCCL_INFO("[OpBaseStreamManager][AllocSlaves]requesting for [%u] slave streams.", num);
64 46 : std::unique_lock<std::mutex> masterLock(masterMutex_);
65 46 : if (!master_ || !master_.ptr()) {
66 0 : HCCL_ERROR("[OpBaseStreamManager][AllocSlaves]master not found, alloc slave stream failed.");
67 0 : return std::vector<Stream>();
68 : }
69 : std::vector<Stream>* slavesPtr;
70 46 : slavesPtr = (streamType == StreamType::STREAM_TYPE_ONLINE) ? &slaves_ : &slaveDevices_;
71 46 : if (slavesPtr->capacity() < num) {
72 1 : HCCL_ERROR(
73 : "[OpBaseStreamManager][AllocSlaves]request number[%u] exceed max substream num[%u], alloc failed.", num,
74 : slavesPtr->capacity());
75 1 : return std::vector<Stream>();
76 : }
77 45 : std::unique_lock<std::mutex> slaveLock(slavesMutex_);
78 45 : if (slavesPtr->size() < num) {
79 43 : HCCL_INFO(
80 : "[OpBaseStreamManager][AllocSlaves]expanding slave streams, original size[%u], target size[%u].",
81 : slavesPtr->size(), num);
82 302 : for (u32 i = slavesPtr->size(); i < num; i++) {
83 260 : slavesPtr->emplace_back(Stream(streamType));
84 260 : if (!(*slavesPtr)[i].ptr()) {
85 : // 创建足够数量的slave stream失败，直接返回空vector
86 1 : HCCL_ERROR("[OpBaseStreamManager][AllocSlaves]alloc slave stream[%u] failed.", i);
87 1 : return std::vector<Stream>();
88 : }
89 259 : if (streamType != StreamType::STREAM_TYPE_DEVICE) {
90 99 : HcclResult ret = SetSlaveMode((*slavesPtr)[i]);
91 99 : if (ret != HCCL_SUCCESS) {
92 0 : HCCL_ERROR("[OpBaseStreamManager][AllocSlaves]set mode to slave stream[%u] failed.", i);
93 0 : return std::vector<Stream>();
94 : }
95 : }
96 : }
97 : }
98 44 : HCCL_INFO("[OpBaseStreamManager][AllocSlaves]find enough slave streams, return size[%u].", num);
99 44 : return std::vector<Stream>(slavesPtr->begin(), slavesPtr->begin() + num);
100 46 : }
101 :
102 99 : HcclResult OpBaseStreamManager::SetSlaveMode(Stream& slave)
103 : {
104 99 : if (master_) {
105 99 : uint64_t streamMode = 0;
106 99 : HcclResult ret = HCCL_SUCCESS;
107 99 : ret = master_.GetMode(&streamMode);
108 99 : if (ret != HCCL_SUCCESS) {
109 0 : HCCL_ERROR(
110 : "[OpBaseStreamManager][SetSlaveMode]errNo[0x%016llx], get master stream mode failed.",
111 : HCCL_ERROR_CODE(ret));
112 0 : return ret;
113 : }
114 99 : ret = slave.SetMode(streamMode);
115 99 : if (ret != HCCL_SUCCESS) {
116 0 : HCCL_ERROR(
117 : "[OpBaseStreamManager][SetSlaveMode]errNo[0x%016llx], set slave stream mode failed.",
118 : HCCL_ERROR_CODE(ret));
119 0 : return ret;
120 : }
121 99 : return HCCL_SUCCESS;
122 : }
123 0 : HCCL_ERROR("[OpBaseStreamManager][SetSlaveMode]master not registered, slave mode will not be set.");
124 0 : return HCCL_E_INTERNAL;
125 : }
126 :
127 1 : Stream OpBaseStreamManager::GetMaster()
128 : {
129 1 : std::unique_lock<std::mutex> lock(masterMutex_);
130 1 : HCCL_DEBUG("[OpBaseStreamManager][GetMaster]get master stream[%p].", master_.ptr());
131 2 : return master_;
132 1 : }
133 :
134 5 : HcclResult OpBaseStreamManager::ClearSlaves()
135 : {
136 5 : std::unique_lock<std::mutex> lock(slavesMutex_);
137 5 : slaves_.clear();
138 5 : slaveDevices_.clear();
139 5 : HCCL_DEBUG("[OpBaseStreamManager][GetMaster]clear slave streams success.");
140 5 : return HCCL_SUCCESS;
141 5 : }
142 :
143 : } // namespace hccl
|
