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 5 : CHK_RET(WaitForJettyCreated(*tempConn, 16000)); // 16s 与 jettyTimeOut 对齐
69 :
70 : // 通过适配层提取 jetty 字段（含 sqDepth），不直接调 legacy GetJettyInfo
71 5 : CHK_RET(ExtractJettyInfoFromConn(tempConn.get(), ctx));
72 :
73 : // 分配共享 PI/CI device 内存并清零，供同 endpoint 下后续 channel 复用。
74 : // 失败时释放已分配的指针，避免 device 内存泄漏。
75 0 : auto cleanup = [&ctx]() {
76 0 : if (ctx.sqPiPtr != nullptr) {
77 0 : (void)hrtFree(ctx.sqPiPtr);
78 0 : ctx.sqPiPtr = nullptr;
79 : }
80 0 : if (ctx.sqCiPtr != nullptr) {
81 0 : (void)hrtFree(ctx.sqCiPtr);
82 0 : ctx.sqCiPtr = nullptr;
83 : }
84 0 : if (ctx.cqPiPtr != nullptr) {
85 0 : (void)hrtFree(ctx.cqPiPtr);
86 0 : ctx.cqPiPtr = nullptr;
87 : }
88 0 : if (ctx.cqCiPtr != nullptr) {
89 0 : (void)hrtFree(ctx.cqCiPtr);
90 0 : ctx.cqCiPtr = nullptr;
91 : }
92 5 : };
93 : struct QueueIndexEntry {
94 : void** ptr;
95 : const char* name;
96 : };
97 5 : QueueIndexEntry entries[] = {
98 5 : {&ctx.sqPiPtr, "sqPiPtr"},
99 5 : {&ctx.sqCiPtr, "sqCiPtr"},
100 5 : {&ctx.cqPiPtr, "cqPiPtr"},
101 5 : {&ctx.cqCiPtr, "cqCiPtr"},
102 5 : };
103 25 : for (const auto& entry : entries) {
104 20 : HcclResult allocRet = AllocAndZeroQueueIndex(entry.ptr, sharedQueueIndexMemSize, entry.name);
105 20 : if (allocRet != HCCL_SUCCESS) {
106 0 : cleanup();
107 0 : return allocRet;
108 : }
109 : }
110 5 : ctx.queueIndexMemSize = sharedQueueIndexMemSize;
111 : // 所有可能失败的操作完成后，再转交 jetty 所有权，阻止临时 connection 析构销毁 jetty。
112 : // 若提前 TransferOwnership，后续 PI/CI 分配失败时 jetty 会因所有权已转交而泄漏。
113 : // TransferOwnership 自身失败时也需调用 cleanup()释放已分配的 PI/CI device内存，避免泄漏。
114 5 : HcclResult transferRet = TransferConnJettyOwnership(tempConn.get());
115 5 : if (transferRet != HCCL_SUCCESS) {
116 0 : cleanup();
117 0 : return transferRet;
118 : }
119 : // 临时 connection 析构：ReleaseResource 跳过 DestroyJetty，ReleaseTp 释放 TP
120 5 : return HCCL_SUCCESS;
121 5 : }
122 :
123 8 : HcclResult AcquireSharedJettyForChannel(
124 : Endpoint* endpoint, Hccl::DevUbConnection* connection, const TempConnFactory& tempConnFactory,
125 : Endpoint::SharedJettyCtx& outCtx)
126 : {
127 8 : if (endpoint == nullptr || connection == nullptr) {
128 2 : return HCCL_E_PARA;
129 : }
130 :
131 : // 共享 jetty 下每个 channel 单 conn，PI/CI 每段大小 = connNum(1) * sizeof(void*)。
132 : // 同 endpoint 下多 channel 共用这块 PI/CI device 内存，避免各 channel 各自分配指向同一 SQ
133 : // 导致生产者索引无法协调前进、WQE 互相覆盖、doorbell 不前进、notify 超时。
134 6 : constexpr uint64_t sharedQueueIndexMemSize = sizeof(void*);
135 11 : auto provideCtx = [&tempConnFactory, sharedQueueIndexMemSize](Endpoint::SharedJettyCtx& ctx) -> HcclResult {
136 5 : return ProvideSharedJettyCtx(tempConnFactory, sharedQueueIndexMemSize, ctx);
137 6 : };
138 :
139 6 : Endpoint::SharedJettyCtx ctx{};
140 6 : HcclResult ret = endpoint->AcquireSharedJetty(provideCtx, ctx);
141 6 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s] Acquire shared jetty failed, ret[%d].", __func__, ret), ret);
142 6 : outCtx = ctx;
143 :
144 : // 命中复用或首次创建完成：通过适配层注入 jetty 到主 connection
145 : // releaseCb: connection 销毁时通知 Endpoint 减引用计数
146 0 : auto releaseCb = [](void* tag) {
147 0 : Endpoint* ep = static_cast<Endpoint*>(tag);
148 0 : if (ep != nullptr) {
149 0 : (void)ep->ReleaseSharedJetty();
150 : }
151 0 : };
152 6 : HcclResult injectRet = InjectSharedJettyToConn(connection, ctx, endpoint, std::move(releaseCb));
153 6 : if (injectRet != HCCL_SUCCESS) {
154 0 : HCCL_ERROR("[%s] Inject shared jetty failed, ret[%d], rollback refCount.", __func__, injectRet);
155 0 : (void)endpoint->ReleaseSharedJetty();
156 0 : return injectRet;
157 : }
158 6 : HCCL_INFO(
159 : "[%s] shared jetty acquired and injected, handle[0x%llx], sqPi[%p] sqCi[%p] cqPi[%p] cqCi[%p].", __func__,
160 : static_cast<unsigned long long>(ctx.handle), ctx.sqPiPtr, ctx.sqCiPtr, ctx.cqPiPtr, ctx.cqCiPtr);
161 6 : return HCCL_SUCCESS;
162 : }
163 :
164 : } // namespace hcomm
|
