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