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_ubg_channel.h"
12 : #include "orion_adpt_utils.h"
13 :
14 : // Orion
15 : #include "topo_common_types.h"
16 : #include "tp_manager.h"
17 :
18 : namespace hcomm {
19 :
20 0 : HcclResult AicpuTsUbgChannel::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 : // UBG 直接从 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("[AicpuTsUbgChannel][%s] locAddr_[%s], rmtAddr_[%s]",
31 : __func__, locAddr_.Describe().c_str(), rmtAddr_.Describe().c_str());
32 :
33 0 : CHK_RET(BuildSocket());
34 0 : CHK_RET(BuildNotify());
35 : /*
36 : HccpRaGetDevBaseAttr
37 : 获取urma read/write 单个wr的最大传输数据大小
38 : 调用前,rdmaHandle_要在ParseInputParam中被赋值好,之后BuildConnection会使用获取的属性
39 : ubg的BuildConnection不再Init里面执行，Init之后会有单独流程建链
40 : */
41 0 : CHK_RET(HccpRaGetDevBaseAttr(rdmaHandle_, &devBaseAttr_));
42 :
43 0 : return HCCL_SUCCESS;
44 : }
45 :
46 1 : HcclResult AicpuTsUbgChannel::BuildConnection()
47 : {
48 1 : Hccl::OpMode opMode = Hccl::OpMode::OPBASE;
49 1 : bool devUsed = true; // aicpu 为 true
50 1 : Hccl::LinkProtocol protocol;
51 1 : CHK_RET(CommProtocolToLinkProtocol(localEp_.protocol, protocol));
52 :
53 : // UBG 的 locAddr_/rmtAddr_ 已经是 EID-based IpAddress，无需额外转换
54 1 : HCCL_INFO("[AicpuTsUbgChannel][%s] LinkProtocol[%s], locAddr_[%s], rmtAddr_[%s]",
55 : __func__, protocol.Describe().c_str(), locAddr_.Describe().c_str(), rmtAddr_.Describe().c_str());
56 :
57 : s32 deviceLogicId;
58 1 : CHK_RET(hrtGetDevice(&deviceLogicId));
59 1 : Hccl::TpManager::GetInstance(deviceLogicId).Init();
60 :
61 : // UBG 使用 DevUbUbgConnection，locAddr_/rmtAddr_ 作为 EID 地址
62 1 : std::unique_ptr<Hccl::DevUbConnection> ubConn = std::make_unique<Hccl::DevUbUbgConnection>(rdmaHandle_,
63 1 : locAddr_, rmtAddr_, opMode, devUsed, Hccl::HrtUbJfcMode::STARS_POLL, locAddr_, rmtAddr_);
64 1 : CHK_SMART_PTR_NULL(ubConn);
65 :
66 1 : if (devBaseAttr_.maxReadSize == 0 || devBaseAttr_.maxWriteSize == 0) {
67 0 : HCCL_ERROR("[AicpuTsUbgChannel][%s] maxReadSize[%u] or maxWriteSize[%u] must not be zero", __func__,
68 : devBaseAttr_.maxReadSize, devBaseAttr_.maxWriteSize);
69 0 : return HCCL_E_PARA;
70 : }
71 1 : ubConn->SetMaxReadSize(devBaseAttr_.maxReadSize);
72 1 : ubConn->SetMaxWriteSize(devBaseAttr_.maxWriteSize);
73 1 : HCCL_INFO("[AicpuTsUbgChannel][%s] maxReadSize[%u], maxWriteSize[%u]", __func__, devBaseAttr_.maxReadSize,
74 : devBaseAttr_.maxWriteSize);
75 :
76 1 : commonRes_.connVec.clear();
77 1 : commonRes_.connVec.emplace_back(ubConn.get());
78 1 : connections_.clear();
79 1 : connections_.push_back(std::move(ubConn));
80 1 : return HCCL_SUCCESS;
81 1 : }
82 :
83 1 : void AicpuTsUbgChannel::SendFinish()
84 : {
85 1 : HCCL_INFO("start send Finish Msg [%s]", UBG_FINISH_MSG);
86 1 : sendFinishMsg_ = std::vector<char>(UBG_FINISH_MSG, UBG_FINISH_MSG + FINISH_MSG_SIZE);
87 1 : socket_->SendAsync(sendFinishMsg_.data(), FINISH_MSG_SIZE);
88 1 : HCCL_INFO("end send Finish Msg [%s]", UBG_FINISH_MSG);
89 1 : }
90 :
91 1 : void AicpuTsUbgChannel::RecvFinish()
92 : {
93 1 : recvFinishMsg_.resize(FINISH_MSG_SIZE);
94 1 : HCCL_INFO("start recv Finish Msg [%s]", UBG_FINISH_MSG);
95 1 : socket_->RecvAsync(reinterpret_cast<u8 *>(recvFinishMsg_.data()), FINISH_MSG_SIZE);
96 1 : HCCL_INFO("end recv Finish Msg [%s]", UBG_FINISH_MSG);
97 1 : }
98 :
99 5 : void AicpuTsUbgChannel::ProcessUbgState()
100 : {
101 9 : auto SetState = [&](UbgStatus next, ChannelStatus ch) { ubgStatus = next; channelStatus = ch; };
102 :
103 5 : switch (ubgStatus) {
104 0 : case UbgStatus::INIT:
105 0 : SetState(UbgStatus::BUILD_CONN, channelStatus);
106 0 : break;
107 0 : case UbgStatus::BUILD_CONN:
108 0 : BuildConn(); SetState(UbgStatus::SEND_SIZE, channelStatus);
109 0 : break;
110 1 : case UbgStatus::SEND_SIZE:
111 1 : if (IsResReady()) { SendDataSize(); SetState(UbgStatus::RECV_SIZE, channelStatus); }
112 1 : break;
113 1 : case UbgStatus::RECV_SIZE:
114 1 : RecvDataSize(); SetState(isRecvFirst_ ? UbgStatus::RECV_DATA : UbgStatus::SEND_DATA, channelStatus);
115 1 : break;
116 1 : case UbgStatus::SEND_DATA:
117 1 : SendExchangeData(); SetState(isRecvFirst_ ? UbgStatus::PROCESS_DATA : UbgStatus::RECV_DATA, channelStatus);
118 1 : break;
119 1 : case UbgStatus::RECV_DATA:
120 1 : RecvExchangeData(); SetState(isRecvFirst_ ? UbgStatus::SEND_DATA : UbgStatus::PROCESS_DATA, channelStatus);
121 1 : break;
122 1 : case UbgStatus::PROCESS_DATA:
123 1 : if (RecvDataProcess()) {
124 0 : ubgStatus = UbgStatus::SEND_FIN;
125 : } else {
126 1 : channelStatus = ChannelStatus::READY;
127 1 : ubgStatus = UbgStatus::READY;
128 : }
129 1 : break;
130 0 : case UbgStatus::SEND_FIN:
131 0 : if (IsConnsReady()) { SendFinish(); SetState(UbgStatus::RECV_FIN, channelStatus); }
132 0 : break;
133 0 : case UbgStatus::RECV_FIN:
134 0 : RecvFinish(); SetState(UbgStatus::SET_READY, channelStatus);
135 0 : break;
136 0 : case UbgStatus::SET_READY:
137 0 : channelStatus = ChannelStatus::READY; SetState(UbgStatus::READY, ChannelStatus::READY);
138 0 : break;
139 0 : default:
140 0 : break;
141 : }
142 5 : }
143 :
144 6 : ChannelStatus AicpuTsUbgChannel::GetStatus()
145 : {
146 6 : if (channelStatus == ChannelStatus::READY) {
147 1 : return channelStatus;
148 : }
149 5 : if (channelStatus == ChannelStatus::INIT) ubgStatus = UbgStatus::INIT;
150 :
151 5 : if (!IsSocketReady()) return channelStatus;
152 :
153 5 : ProcessUbgState();
154 5 : if (channelStatus == ChannelStatus::READY && channelDesc_.socket == nullptr && socket_ != nullptr) {
155 0 : SocketMgr::GetInstance(devicePhyId_).PutSocket(socketConfig_, socket_);
156 0 : socket_ = nullptr;
157 : }
158 :
159 5 : return channelStatus;
160 : }
161 :
162 : } // namespace hcomm
|
