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