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 "aiv_ins_preprocessor.h"
12 : #include "aiv_ins.h"
13 : #include "env_config_v2.h"
14 : #include "local_rma_buf_manager.h"
15 :
16 : namespace Hccl {
17 :
18 4 : void AivInsPreprocessor::SetProtocol(uint8_t protocol) { protocol_ = protocol; }
19 :
20 5 : uint8_t AivInsPreprocessor::GetProtocol() const { return protocol_; }
21 :
22 6 : void AivInsPreprocessor::Preprocess(std::shared_ptr<InsQueue>& insQueue) const
23 : {
24 18 : HCCL_INFO("[AivInsPreprocessor::%s] insQueue Preprocess start.", __func__);
25 :
26 : // 对每主queue中每个ins进行预处理
27 10 : for (auto ins = insQueue->Iter(); ins.HasNext(); ++ins) {
28 4 : if (ins->GetType() != InstructionType::AIV_INS) {
29 1 : continue;
30 : }
31 3 : InsPreprocess(ins);
32 6 : }
33 :
34 18 : HCCL_INFO("[AivInsPreprocessor::%s] insQueue Preprocess end.", __func__);
35 6 : }
36 :
37 3 : void AivInsPreprocessor::InsPreprocess(InsIterator& insIter) const
38 : {
39 9 : HCCL_INFO("[AivInsPreprocessor::%s] start.", __func__);
40 :
41 3 : const AivInstruction& aivIns = dynamic_cast<const AivInstruction&>(*insIter);
42 :
43 3 : auto links = aivIns.GetLinks();
44 :
45 3 : if (protocol_ == 0) { // ubmemory
46 3 : BatchBuildTransports(links);
47 0 : } else if (protocol_ == 1) { // urma
48 0 : BatchBuildUrmaTransports(links);
49 : } else {
50 0 : THROW<InvalidParamsException>(StringFormat("protocol[%u] not supported", protocol_));
51 : }
52 :
53 9 : HCCL_INFO("[AivInsPreprocessor::%s] end.", __func__);
54 3 : }
55 :
56 3 : void AivInsPreprocessor::BatchBuildTransports(const vector<LinkData>& links) const
57 : {
58 9 : HCCL_INFO("[AivInsPreprocessor::%s] start.", __func__);
59 :
60 : // 创建MemTransport并进行异步建链、交换
61 3 : comm->GetUbMemoryTransportMgr()->BatchCreateTransport(links);
62 :
63 3 : comm->GetUbMemoryTransportMgr()->TransportsConnect();
64 :
65 9 : HCCL_INFO("[AivInsPreprocessor::%s] end.", __func__);
66 3 : }
67 :
68 1 : void AivInsPreprocessor::BatchBuildUrmaTransports(const vector<LinkData>& links) const
69 : {
70 3 : HCCL_RUN_INFO("[AivInsPreprocessor::%s] start.", __func__);
71 :
72 1 : std::string opTag = comm->GetCurrentCollOperator()->opTag;
73 :
74 1 : auto& localRmaBufManager = comm->GetLocalRmaBufManager();
75 1 : const string aivUrmaBufferTag = GetAivUrmaBufferTag(comm->GetId());
76 1 : for (size_t i = 0; i < links.size(); ++i) {
77 0 : const auto& link = links[i];
78 0 : auto localBuffer = localRmaBufManager.Reg(
79 0 : aivUrmaBufferTag, BufferType::SCRATCH, comm->GetCclBuffer(), link.GetLocalPort(), link.GetLinkProtocol());
80 0 : CHECK_NULLPTR(
81 0 : localBuffer, StringFormat(
82 : "[AivInsPreprocessor::%s] fake local buffer is nullptr, linkIdx[%zu], port[%u]", __func__,
83 0 : i, link.GetLocalPort().GetId()));
84 : }
85 :
86 : // 创建RmaConnectiuon
87 1 : RmaConnManager& connManager = comm->GetRmaConnManager();
88 1 : for (auto& link : links) {
89 0 : auto conn = connManager.Create(opTag, link, HrtUbJfcMode::USER_CTL);
90 0 : CHECK_NULLPTR(conn, "[AivInsPreprocessor::BatchBuildUrmaTransports] conn is nullptr!");
91 : }
92 3 : HCCL_INFO("[AivInsPreprocessor::%s] end creating rma connection", __func__);
93 :
94 : // 创建UrmaDirectTransport并进行异步建链、交换
95 1 : auto transportMgr = comm->GetMemTransportManager();
96 2 : CHECK_NULLPTR(transportMgr, "[AivInsPreprocessor::BatchBuildUrmaTransports] transportMgr is nullptr!");
97 0 : transportMgr->BatchBuildUrmaDirectTransports(links);
98 :
99 0 : auto timeout = std::chrono::seconds(EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
100 :
101 0 : HcclUs startTime = std::chrono::steady_clock::now();
102 0 : bool isReady = false;
103 0 : while (!isReady) {
104 0 : isReady = transportMgr->IsAllTransportReady();
105 0 : if (isReady) {
106 0 : break;
107 : }
108 0 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
109 0 : transportMgr->DumpNotReadyTransportsUrma();
110 0 : RPT_INPUT_ERR(
111 : true, "EI0006", std::vector<std::string>({"reason"}),
112 : std::vector<std::string>({"Aiv urma wait transports ready timeout."}));
113 0 : THROW<InternalException>("Aiv WaitTransportReady timeout, commId[%s].", comm->GetId().c_str());
114 : break;
115 : }
116 : }
117 :
118 0 : HCCL_INFO("[AivInsPreprocessor::%s] end.", __func__);
119 2 : }
120 :
121 0 : std::vector<HcclAiRMAWQ> AivInsPreprocessor::GetWqs() const
122 : {
123 0 : HCCL_INFO("[AivInsPreprocessor::%s] start.", __func__);
124 0 : if (protocol_ != 1) {
125 0 : THROW<InvalidParamsException>(StringFormat("can not get wq info when protocol is [%u]", protocol_));
126 : }
127 0 : auto memTransportMgr = comm->GetMemTransportManager();
128 0 : CHECK_NULLPTR(memTransportMgr, "[AivInsPreprocessor::GetWqs] memTransportMgr is nullptr!");
129 0 : return memTransportMgr->GetUrmaWqs();
130 : }
131 :
132 0 : std::vector<HcclAiRMACQ> AivInsPreprocessor::GetCqs() const
133 : {
134 0 : HCCL_INFO("[AivInsPreprocessor::%s] start.", __func__);
135 0 : if (protocol_ != 1) {
136 0 : THROW<InvalidParamsException>(StringFormat("can not get cq info when protocol is [%u]", protocol_));
137 : }
138 0 : auto memTransportMgr = comm->GetMemTransportManager();
139 0 : CHECK_NULLPTR(memTransportMgr, "[AivInsPreprocessor::GetCqs] memTransportMgr is nullptr!");
140 0 : return memTransportMgr->GetUrmaCqs();
141 : }
142 :
143 241 : AivInsPreprocessor::~AivInsPreprocessor() {}
144 :
145 : } // namespace Hccl
|
