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 "shared_jetty_channel_helper.h"
12 : #include "log.h"
13 : #include "adapter_rts_common.h"
14 : #include <chrono>
15 : #include <thread>
16 :
17 : namespace hcomm {
18 :
19 : // 轮询临时 connection 状态机到 EXCHANGEABLE（jetty 已创建），或超时失败
20 5 : static HcclResult WaitForJettyCreated(Hccl::DevUbConnection& conn, uint32_t timeoutMs)
21 : {
22 5 : auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(timeoutMs);
23 : while (true) {
24 5 : Hccl::RmaConnStatus st = conn.GetStatus();
25 5 : if (st == Hccl::RmaConnStatus::EXCHANGEABLE || st == Hccl::RmaConnStatus::READY) {
26 5 : return HCCL_SUCCESS;
27 : }
28 0 : if (st == Hccl::RmaConnStatus::CONN_INVALID) {
29 0 : HCCL_ERROR("[%s] temp connection became CONN_INVALID.", __func__);
30 0 : return HCCL_E_INTERNAL;
31 : }
32 0 : if (st == Hccl::RmaConnStatus::CLOSE) {
33 0 : HCCL_ERROR("[%s] temp connection CLOSED.", __func__);
34 0 : return HCCL_E_INTERNAL;
35 : }
36 0 : if (std::chrono::steady_clock::now() >= deadline) {
37 0 : HCCL_ERROR("[%s] wait jetty create timeout[%ums].", __func__, timeoutMs);
38 0 : return HCCL_E_TIMEOUT;
39 : }
40 0 : std::this_thread::sleep_for(std::chrono::milliseconds(2));
41 0 : }
42 : }
43 :
44 : // 分配 device 内存并清零：PI/CI 必须初值为 0，否则生产者索引非 0 会导致首条 WQE 越界
45 20 : static HcclResult AllocAndZeroQueueIndex(void** ptr, uint64_t size, const char* name)
46 : {
47 20 : HcclResult allocRet = hrtMalloc(ptr, size);
48 20 : if (allocRet != HCCL_SUCCESS || *ptr == nullptr) {
49 0 : HCCL_ERROR("[%s] hrtMalloc %s failed, ret[%d].", __func__, name, allocRet);
50 0 : return HCCL_E_MEMORY;
51 : }
52 20 : aclError memsetRet = aclrtMemset(*ptr, size, 0, size);
53 20 : if (memsetRet != ACL_SUCCESS) {
54 0 : HCCL_ERROR("[%s] aclrtMemset %s failed, ret[%d].", __func__, name, memsetRet);
55 0 : (void)hrtFree(*ptr);
56 0 : *ptr = nullptr;
57 0 : return HCCL_E_MEMORY;
58 : }
59 20 : return HCCL_SUCCESS;
60 : }
61 :
62 : // 首次创建共享 jetty：用临时 connection 走完整 jetty 创建流程，分配共享 PI/CI device 内存
63 5 : static HcclResult ProvideSharedJettyCtx(
64 : const TempConnFactory& tempConnFactory, uint64_t sharedQueueIndexMemSize, Endpoint::SharedJettyCtx& ctx)
65 : {
66 5 : std::unique_ptr<Hccl::DevUbConnection> tempConn = tempConnFactory();
67 5 : CHK_SMART_PTR_NULL(tempConn);
68 : // 16s 覆盖 TP 建链(~8s) + jetty 异步创建(~8s) 的串行最坏时长, 与 jettyTimeOut hwValue=16(=8s) 配套
69 5 : CHK_RET(WaitForJettyCreated(*tempConn, 16000));
70 :
71 : // 通过适配层提取 jetty 字段（含 sqDepth），不直接调 legacy GetJettyInfo
72 5 : CHK_RET(ExtractJettyInfoFromConn(tempConn.get(), ctx));
73 :
74 : // 分配共享 PI/CI device 内存并清零，供同 endpoint 下后续 channel 复用。
75 : // 失败时释放已分配的指针，避免 device 内存泄漏。
76 0 : auto cleanup = [&ctx]() {
77 0 : if (ctx.sqPiPtr != nullptr) {
78 0 : (void)hrtFree(ctx.sqPiPtr);
79 0 : ctx.sqPiPtr = nullptr;
80 : }
81 0 : if (ctx.sqCiPtr != nullptr) {
82 0 : (void)hrtFree(ctx.sqCiPtr);
83 0 : ctx.sqCiPtr = nullptr;
84 : }
85 0 : if (ctx.cqPiPtr != nullptr) {
86 0 : (void)hrtFree(ctx.cqPiPtr);
87 0 : ctx.cqPiPtr = nullptr;
88 : }
89 0 : if (ctx.cqCiPtr != nullptr) {
90 0 : (void)hrtFree(ctx.cqCiPtr);
91 0 : ctx.cqCiPtr = nullptr;
92 : }
93 5 : };
94 : struct QueueIndexEntry {
95 : void** ptr;
96 : const char* name;
97 : };
98 5 : QueueIndexEntry entries[] = {
99 5 : {&ctx.sqPiPtr, "sqPiPtr"},
100 5 : {&ctx.sqCiPtr, "sqCiPtr"},
101 5 : {&ctx.cqPiPtr, "cqPiPtr"},
102 5 : {&ctx.cqCiPtr, "cqCiPtr"},
103 5 : };
104 25 : for (const auto& entry : entries) {
105 20 : HcclResult allocRet = AllocAndZeroQueueIndex(entry.ptr, sharedQueueIndexMemSize, entry.name);
106 20 : if (allocRet != HCCL_SUCCESS) {
107 0 : cleanup();
108 0 : return allocRet;
109 : }
110 : }
111 5 : ctx.queueIndexMemSize = sharedQueueIndexMemSize;
112 : // 所有可能失败的操作完成后，再分离 jetty 所有权，阻止临时 connection 析构销毁 jetty。
113 : // 若提前 DetachJetty，后续 PI/CI 分配失败时 jetty 会因所有权已分离而泄漏。
114 : // DetachJetty 自身失败时也需调用 cleanup() 释放已分配的 PI/CI device 内存，避免泄漏。
115 5 : HcclResult transferRet = DetachConnJetty(tempConn.get());
116 5 : if (transferRet != HCCL_SUCCESS) {
117 0 : cleanup();
118 0 : return transferRet;
119 : }
120 : // 临时 connection 析构：ReleaseResource 跳过 DestroyJetty，ReleaseTp 释放 TP
121 5 : return HCCL_SUCCESS;
122 5 : }
123 :
124 8 : HcclResult AcquireSharedJettyForChannel(
125 : Endpoint* endpoint, Hccl::DevUbConnection* connection, const TempConnFactory& tempConnFactory,
126 : Endpoint::SharedJettyCtx& outCtx)
127 : {
128 8 : if (endpoint == nullptr || connection == nullptr) {
129 2 : return HCCL_E_PARA;
130 : }
131 :
132 : // 共享 jetty 下每个 channel 单 conn，PI/CI 每段大小 = connNum(1) * sizeof(void*)。
133 : // 同 endpoint 下多 channel 共用这块 PI/CI device 内存，避免各 channel 各自分配指向同一 SQ
134 : // 导致生产者索引无法协调前进、WQE 互相覆盖、doorbell 不前进、notify 超时。
135 6 : constexpr uint64_t sharedQueueIndexMemSize = sizeof(void*);
136 11 : auto provideCtx = [&tempConnFactory, sharedQueueIndexMemSize](Endpoint::SharedJettyCtx& ctx) -> HcclResult {
137 5 : return ProvideSharedJettyCtx(tempConnFactory, sharedQueueIndexMemSize, ctx);
138 6 : };
139 :
140 6 : Endpoint::SharedJettyCtx ctx{};
141 6 : HcclResult ret = endpoint->AcquireSharedJetty(provideCtx, ctx);
142 6 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] Acquire shared jetty failed, ret[%d].", __func__, ret), ret);
143 6 : outCtx = ctx;
144 :
145 : // 命中复用或首次创建完成：通过适配层填充 jetty 字段到主 connection
146 : // releaseCb: connection 销毁时通知 Endpoint 减引用计数
147 0 : auto releaseCb = [](void* tag) {
148 0 : Endpoint* ep = static_cast<Endpoint*>(tag);
149 0 : if (ep != nullptr) {
150 0 : (void)ep->ReleaseSharedJetty();
151 : }
152 0 : };
153 6 : HcclResult injectRet = SetSharedJettyFieldsToConn(connection, ctx, endpoint, std::move(releaseCb));
154 6 : if (injectRet != HCCL_SUCCESS) {
155 0 : HCCL_ERROR("[%s] SetSharedJettyFields failed, ret[%d], rollback refCount.", __func__, injectRet);
156 0 : (void)endpoint->ReleaseSharedJetty();
157 0 : return injectRet;
158 : }
159 6 : HCCL_INFO(
160 : "[%s] shared jetty acquired and fields set, handle[0x%llx], sqPi[%p] sqCi[%p] cqPi[%p] cqCi[%p].", __func__,
161 : static_cast<unsigned long long>(ctx.handle), ctx.sqPiPtr, ctx.sqCiPtr, ctx.cqPiPtr, ctx.cqCiPtr);
162 6 : return HCCL_SUCCESS;
163 : }
164 :
165 : } // namespace hcomm
|
