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