Line data Source code
1 : /**
2 : * Copyright (c) 2026 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 "aicpu_ts_ub_rtp_channel.h"
12 : #include "orion_adpt_utils.h"
13 : #include "hcomm_res_mgr.h"
14 : #include "endpoint.h"
15 :
16 : // Orion
17 : #include "topo_common_types.h"
18 :
19 : namespace hcomm {
20 :
21 0 : HcclResult AicpuTsUbRtpChannel::Init()
22 : {
23 : s32 devLogicId;
24 0 : CHK_RET(hrtGetDevice(&devLogicId));
25 0 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<u32>(devLogicId), devicePhyId_));
26 0 : CHK_RET(ParseInputParam());
27 :
28 : // UB_RTP 直接从 EID type CommAddr 获取地址，不做 IP→EID 转换
29 0 : CHK_RET(CommAddrToIpAddress(localEp_.commAddr, locAddr_));
30 0 : CHK_RET(CommAddrToIpAddress(remoteEp_.commAddr, rmtAddr_));
31 0 : HCCL_INFO(
32 : "[AicpuTsUbRtpChannel][%s] locAddr_[%s], rmtAddr_[%s]", __func__, locAddr_.Describe().c_str(),
33 : rmtAddr_.Describe().c_str());
34 :
35 0 : CHK_RET(BuildSocket());
36 0 : CHK_RET(BuildNotify());
37 : /*
38 : HccpRaGetDevBaseAttr
39 : 获取urma read/write 单个wr的最大传输数据大小
40 : 调用前,rdmaHandle_要在ParseInputParam中被赋值好,之后BuildConnection会使用获取的属性
41 : ub_rtp的BuildConnection不再Init里面执行，Init之后会有单独流程建链
42 : */
43 0 : CHK_RET(HccpRaGetDevBaseAttr(rdmaHandle_, &devBaseAttr_));
44 :
45 0 : return HCCL_SUCCESS;
46 : }
47 :
48 1 : HcclResult AicpuTsUbRtpChannel::BuildConnection()
49 : {
50 1 : UbConnBuildContext ctx;
51 1 : CHK_RET(PrepareUbConnBuildContext(localEp_, remoteEp_, channelDesc_, ctx));
52 :
53 1 : Hccl::OpMode opMode = Hccl::OpMode::OPBASE;
54 1 : bool devUsed = true; // aicpu 为 true
55 : // UB_RTP 的 locAddr_/rmtAddr_ 已经是 EID-based IpAddress，无需额外转换
56 1 : HCCL_INFO(
57 : "[AicpuTsUbRtpChannel][%s] LinkProtocol[%s], locAddr_[%s], rmtAddr_[%s], qos[%u]", __func__,
58 : ctx.protocol.Describe().c_str(), locAddr_.Describe().c_str(), rmtAddr_.Describe().c_str(),
59 : static_cast<unsigned int>(ctx.qosPre));
60 :
61 : // UB_RTP 使用 DevUbRtpConnection，locAddr_/rmtAddr_ 作为 EID 地址；qos 与 UBOE 一致来自 channelDesc_
62 : // UB_RTP 协议对应 HCOMM_TA_RTP_UB_TIMEOUT，由 base_comm 从环境变量获取后传入
63 1 : u8 taTimeOut = 0;
64 1 : uint32_t taTimeOutValue = 0;
65 1 : CHK_RET(hcomm::HcommResMgr::GetInstance().GetConfigMgr().GetRdmaConfig().GetTaRtpUbTimeOut(taTimeOutValue));
66 1 : taTimeOut = static_cast<u8>(taTimeOutValue);
67 1 : std::unique_ptr<Hccl::DevUbConnection> ubConn = std::make_unique<Hccl::DevUbRtpConnection>(
68 1 : rdmaHandle_, locAddr_, rmtAddr_, opMode, devUsed, Hccl::HrtUbJfcMode::STARS_POLL, locAddr_, rmtAddr_,
69 1 : ctx.qosPre, taTimeOut);
70 1 : CHK_SMART_PTR_NULL(ubConn);
71 :
72 1 : if (devBaseAttr_.maxReadSize == 0 || devBaseAttr_.maxWriteSize == 0) {
73 0 : HCCL_ERROR(
74 : "[AicpuTsUbRtpChannel][%s] maxReadSize[%u] or maxWriteSize[%u] must not be zero", __func__,
75 : devBaseAttr_.maxReadSize, devBaseAttr_.maxWriteSize);
76 0 : return HCCL_E_PARA;
77 : }
78 1 : ubConn->SetMaxReadSize(devBaseAttr_.maxReadSize);
79 1 : ubConn->SetMaxWriteSize(devBaseAttr_.maxWriteSize);
80 1 : HCCL_INFO(
81 : "[AicpuTsUbRtpChannel][%s] maxReadSize[%u], maxWriteSize[%u]", __func__, devBaseAttr_.maxReadSize,
82 : devBaseAttr_.maxWriteSize);
83 :
84 1 : commonRes_.connVec.clear();
85 1 : commonRes_.connVec.emplace_back(ubConn.get());
86 1 : connections_.clear();
87 1 : connections_.push_back(std::move(ubConn));
88 1 : return HCCL_SUCCESS;
89 1 : }
90 :
91 1 : void AicpuTsUbRtpChannel::SendFinish()
92 : {
93 1 : HCCL_INFO("start send Finish Msg [%s]", UB_RTP_FINISH_MSG);
94 1 : sendFinishMsg_ = std::vector<char>(UB_RTP_FINISH_MSG, UB_RTP_FINISH_MSG + FINISH_MSG_SIZE);
95 1 : socket_->SendAsync(sendFinishMsg_.data(), FINISH_MSG_SIZE);
96 1 : HCCL_INFO("end send Finish Msg [%s]", UB_RTP_FINISH_MSG);
97 1 : }
98 :
99 1 : void AicpuTsUbRtpChannel::RecvFinish()
100 : {
101 1 : recvFinishMsg_.resize(FINISH_MSG_SIZE);
102 1 : HCCL_INFO("start recv Finish Msg [%s]", UB_RTP_FINISH_MSG);
103 1 : socket_->RecvAsync(reinterpret_cast<u8*>(recvFinishMsg_.data()), FINISH_MSG_SIZE);
104 1 : HCCL_INFO("end recv Finish Msg [%s]", UB_RTP_FINISH_MSG);
105 1 : }
106 :
107 5 : void AicpuTsUbRtpChannel::ProcessUbRtpState()
108 : {
109 1 : auto SetState = [&](UbRtpStatus next, ChannelStatus ch) {
110 1 : ubRtpStatus = next;
111 1 : channelStatus = ch;
112 6 : };
113 :
114 5 : switch (ubRtpStatus) {
115 0 : case UbRtpStatus::INIT:
116 0 : SetState(UbRtpStatus::BUILD_CONN, channelStatus);
117 0 : break;
118 0 : case UbRtpStatus::BUILD_CONN:
119 0 : BuildConn();
120 0 : SetState(UbRtpStatus::SEND_SIZE, channelStatus);
121 0 : break;
122 1 : case UbRtpStatus::SEND_SIZE:
123 1 : if (IsResReady()) {
124 1 : SendDataSize();
125 1 : SetState(UbRtpStatus::RECV_SIZE, channelStatus);
126 : }
127 1 : break;
128 4 : case UbRtpStatus::RECV_SIZE:
129 : case UbRtpStatus::SEND_DATA:
130 : case UbRtpStatus::RECV_DATA:
131 : case UbRtpStatus::PROCESS_DATA:
132 4 : ProcessUbRtpDataState();
133 4 : break;
134 0 : case UbRtpStatus::SEND_FIN:
135 0 : if (IsConnsReady()) {
136 0 : SendFinish();
137 0 : SetState(UbRtpStatus::RECV_FIN, channelStatus);
138 : }
139 0 : break;
140 0 : case UbRtpStatus::RECV_FIN:
141 0 : RecvFinish();
142 0 : SetState(UbRtpStatus::SET_READY, channelStatus);
143 0 : break;
144 0 : case UbRtpStatus::SET_READY:
145 0 : channelStatus = ChannelStatus::READY;
146 0 : SetState(UbRtpStatus::READY, ChannelStatus::READY);
147 0 : break;
148 0 : default:
149 0 : break;
150 : }
151 5 : }
152 :
153 4 : void AicpuTsUbRtpChannel::ProcessUbRtpDataState()
154 : {
155 4 : switch (ubRtpStatus) {
156 1 : case UbRtpStatus::RECV_SIZE:
157 1 : RecvDataSize();
158 1 : ubRtpStatus = isRecvFirst_ ? UbRtpStatus::RECV_DATA : UbRtpStatus::SEND_DATA;
159 1 : break;
160 1 : case UbRtpStatus::SEND_DATA:
161 1 : SendExchangeData();
162 1 : ubRtpStatus = isRecvFirst_ ? UbRtpStatus::PROCESS_DATA : UbRtpStatus::RECV_DATA;
163 1 : break;
164 1 : case UbRtpStatus::RECV_DATA:
165 1 : RecvExchangeData();
166 1 : ubRtpStatus = isRecvFirst_ ? UbRtpStatus::SEND_DATA : UbRtpStatus::PROCESS_DATA;
167 1 : break;
168 1 : case UbRtpStatus::PROCESS_DATA:
169 1 : if (RecvDataProcess()) {
170 0 : ubRtpStatus = UbRtpStatus::SEND_FIN;
171 : } else {
172 1 : channelStatus = ChannelStatus::READY;
173 1 : ubRtpStatus = UbRtpStatus::READY;
174 : }
175 1 : break;
176 0 : default:
177 0 : break;
178 : }
179 4 : }
180 :
181 6 : ChannelStatus AicpuTsUbRtpChannel::GetStatus()
182 : {
183 6 : if (channelStatus == ChannelStatus::READY) {
184 1 : return channelStatus;
185 : }
186 5 : if (channelStatus == ChannelStatus::INIT)
187 0 : ubRtpStatus = UbRtpStatus::INIT;
188 :
189 5 : if (!IsSocketReady())
190 0 : return channelStatus;
191 :
192 5 : ProcessUbRtpState();
193 :
194 5 : return channelStatus;
195 : }
196 :
197 1 : HcclResult AicpuTsUbRtpChannel::Clean()
198 : {
199 1 : commonRes_.connVec.clear();
200 1 : connections_.clear();
201 :
202 1 : rmtNotifyVec_.clear();
203 1 : locBufferVec_.clear();
204 :
205 1 : recvData_.clear();
206 1 : recvFinishMsg_.clear();
207 1 : sendData_.clear();
208 1 : sendFinishMsg_.clear();
209 :
210 1 : bufferNum_ = 0;
211 1 : connNum_ = 0;
212 1 : recvDataSize_ = 0;
213 :
214 : {
215 1 : std::lock_guard<std::mutex> lock(remoteMemsMutex_);
216 1 : rmtBufferVec_.clear();
217 1 : cacheValid_ = false;
218 1 : remoteUserMems_.clear();
219 1 : memInfoCopies_.clear();
220 1 : memInfoPointers_.clear();
221 1 : }
222 :
223 1 : channelStatus = ChannelStatus::INIT;
224 1 : ubRtpStatus = UbRtpStatus::INIT;
225 :
226 1 : return HCCL_SUCCESS;
227 : }
228 :
229 1 : HcclResult AicpuTsUbRtpChannel::Resume()
230 : {
231 1 : channelStatus = ChannelStatus::INIT;
232 1 : ubRtpStatus = UbRtpStatus::INIT;
233 1 : return HCCL_SUCCESS;
234 : }
235 :
236 : } // namespace hcomm
|
