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 "connections_builder.h"
12 : #include "stl_util.h"
13 : #include "exception_util.h"
14 : #include "rma_conn_exception.h"
15 : #include "communicator_impl.h"
16 :
17 : #include <list>
18 :
19 : namespace Hccl {
20 1 : vector<LinkData> ConnectionsBuilder::GetAvailableLinksVec() const
21 : {
22 2 : return vector<LinkData>(availableLinks.begin(), availableLinks.end());
23 : }
24 :
25 8 : void ConnectionsBuilder::BatchBuild(const std::string& opTag, const vector<LinkData>& links)
26 : {
27 8 : vector<LinkData> pendingLinks;
28 8 : for (auto& link : links) {
29 0 : if (Contain(availableLinks, link)) {
30 0 : continue;
31 : }
32 0 : pendingLinks.emplace_back(link);
33 : }
34 24 : HCCL_INFO(
35 : "[%s]opTag[%s] links size[%u], pendingLinks size[%u]", __func__, opTag.c_str(), links.size(),
36 : pendingLinks.size());
37 :
38 8 : if (pendingLinks.empty()) {
39 8 : return;
40 : }
41 :
42 : // rmaConnections创建
43 0 : HCCL_INFO("create rma connection start");
44 0 : CreateRmaConnections(opTag, pendingLinks);
45 0 : HCCL_INFO("create rma connection end");
46 :
47 0 : availableLinks.insert(pendingLinks.begin(), pendingLinks.end());
48 8 : }
49 :
50 1 : void ConnectionsBuilder::CreateRmaConnections(const std::string& opTag, const vector<LinkData>& links) const
51 : {
52 1 : list<RmaConnection*> connTasks;
53 2 : for (auto& link : links) {
54 1 : connTasks.emplace_back(connManager->Create(opTag, link));
55 : }
56 1 : CHECK_NULLPTR(comm, "[ConnectionsBuilder::CreateRmaConnections] comm is nullptr!");
57 1 : if (!comm->GetOpCcuFeatureFlag()) { // 非CCU展开,则在transport中交换 connection // 直接return,下面删除?
58 1 : return;
59 : }
60 : // CCU使用的connection,则走到下面来搞,待CCU transport搞定后,再删除下面这段
61 0 : while (!connTasks.empty()) {
62 0 : for (auto connIter = connTasks.begin(); connIter != connTasks.end();) {
63 0 : CHECK_NULLPTR((*connIter), "[ConnectionsBuilder::CreateRmaConnections] (*connIter) is nullptr!");
64 0 : auto status = (*connIter)->GetStatus();
65 0 : if (status == RmaConnStatus::READY) {
66 0 : connIter = connTasks.erase(connIter);
67 0 : } else if (status == RmaConnStatus::CONN_INVALID || status == RmaConnStatus::CLOSE) {
68 0 : THROW<RmaConnException>(StringFormat(
69 0 : "Invalid status occurs when creating RMA connection %s!", (*connIter)->Describe().c_str()));
70 : } else {
71 0 : ++connIter;
72 : }
73 : }
74 : }
75 1 : }
76 :
77 18 : ConnectionsBuilder::ConnectionsBuilder(CommunicatorImpl& communicator)
78 : {
79 18 : this->comm = &communicator;
80 18 : connManager = &(comm->GetRmaConnManager());
81 18 : }
82 :
83 : } // namespace Hccl
|