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 "flush_manager.h"
12 : #include "env_config/env_config_v2.h"
13 :
14 : namespace Hccl {
15 2 : FlushManager::FlushManager() {}
16 :
17 0 : FlushManager& FlushManager::GetInstance()
18 : {
19 0 : static FlushManager flushManager;
20 0 : return flushManager;
21 : }
22 :
23 2 : FlushManager::~FlushManager() { DestroyAll(); }
24 :
25 0 : HcclResult FlushManager::initFlushHandle(IpAddress ip, u32 devPhyId)
26 : {
27 0 : HCCL_INFO("[initFlushHandle]FlushHandle init start.");
28 0 : if (flushHandleMap_.find(ip) != flushHandleMap_.end()) {
29 0 : HCCL_INFO("[initFlushHandle]FlushHandle already exists");
30 0 : return HCCL_SUCCESS;
31 : }
32 :
33 0 : auto flushHandlePtr = std::make_shared<FlushHandle>();
34 0 : HcclResult ret = flushHandlePtr->Init(ip, devPhyId);
35 0 : if (ret != HCCL_SUCCESS) {
36 0 : HCCL_INFO("[initFlushHandle]FlushHandle init fail.");
37 0 : return ret;
38 : }
39 :
40 0 : flushHandleMap_.insert({ip, flushHandlePtr});
41 0 : HCCL_INFO("[initFlushHandle]FlushHandle init success.");
42 0 : return HCCL_SUCCESS;
43 0 : }
44 :
45 2 : HcclResult FlushManager::DestroyAll()
46 : {
47 2 : if (flushHandleMap_.empty()) {
48 6 : HCCL_DEBUG("flushHandleMap_ is empty");
49 2 : return HCCL_SUCCESS;
50 : }
51 0 : for (auto item : flushHandleMap_) {
52 0 : auto flushHandlePtr = item.second;
53 0 : HcclResult ret = flushHandlePtr->Destroy();
54 0 : if (ret != HCCL_SUCCESS) {
55 0 : HCCL_ERROR("[DestroyAll]Failed to destroy flush resources. Error: %d", ret);
56 0 : return ret;
57 : }
58 0 : }
59 0 : flushHandleMap_.clear();
60 0 : HCCL_INFO("[DestroyAll]FlushHandle destroy success.");
61 0 : return HCCL_SUCCESS;
62 : }
63 :
64 1 : HcclResult FlushManager::Flush()
65 : {
66 1 : std::lock_guard<std::mutex> lock(mutex_);
67 3 : HCCL_INFO("[Flush] Start: Entering Flush function.");
68 1 : if (flushHandleMap_.empty()) {
69 0 : HCCL_INFO("[Flush] No FLUSH is needed to be executed.");
70 0 : return HCCL_SUCCESS;
71 : }
72 :
73 1 : for (auto item : flushHandleMap_) {
74 1 : auto flushHandlePtr = item.second;
75 :
76 1 : ibv_qp* loopbackqp0 = static_cast<ibv_qp*>(flushHandlePtr->loopBackQpParam.ibvQp0);
77 1 : CHK_PTR_NULL(loopbackqp0);
78 1 : ibv_cq* cq = loopbackqp0->send_cq;
79 1 : CHK_PTR_NULL(cq);
80 3 : HCCL_DEBUG("[Flush] Successfully retrieved QP and CQ handles: qp=%p, cq=%p", loopbackqp0, cq);
81 :
82 : // 接口数据设置
83 1 : ibv_send_wr swr{};
84 1 : ibv_sge sg_list{};
85 1 : swr.sg_list = &sg_list;
86 1 : HcclResult paramsRet = FlushParamPrepare(flushHandlePtr, &swr);
87 1 : if (paramsRet != HCCL_SUCCESS) {
88 0 : HCCL_INFO("[Flush] Set work request failed.");
89 0 : return paramsRet;
90 : }
91 3 : HCCL_DEBUG("[FlushParamPrepare] Posting RDMA_READ operation... ");
92 :
93 : // 执行读和轮训操作
94 : HcclResult loopQpRet
95 1 : = ExecuteRdmaRead(loopbackqp0, cq, swr, Hccl::EnvConfig::GetInstance().GetRtsConfig().GetExecTimeOut());
96 1 : if (loopQpRet != HCCL_SUCCESS) {
97 3 : HCCL_INFO("[Flush] RDMA_READ operation failed.");
98 1 : return loopQpRet;
99 : }
100 2 : }
101 0 : HCCL_INFO("[Flush] Successfully completed: RDMA_READ operation finished.");
102 0 : return HCCL_SUCCESS;
103 1 : }
104 :
105 1 : HcclResult FlushManager::FlushParamPrepare(std::shared_ptr<FlushHandle> flushHandlePtr, ibv_send_wr* swr) const
106 : {
107 1 : CHK_PTR_NULL(swr);
108 1 : swr->wr_id = 0;
109 1 : CHK_PTR_NULL(swr->sg_list);
110 1 : swr->sg_list->addr = reinterpret_cast<uint64_t>(flushHandlePtr->loopBackQpMrLocalInfo.addr);
111 1 : swr->sg_list->length = flushHandlePtr->loopBackQpMrLocalInfo.size;
112 1 : swr->sg_list->lkey = flushHandlePtr->loopBackQpMrLocalInfo.lkey;
113 1 : swr->next = nullptr;
114 1 : swr->num_sge = 1;
115 1 : swr->opcode = (flushHandlePtr->GetFlushOpcodeSupport()) ? ROCE_WR_FLUSH : IBV_WR_RDMA_READ;
116 1 : swr->send_flags = IBV_SEND_SIGNALED;
117 1 : swr->wr.rdma.remote_addr = reinterpret_cast<uint64_t>(flushHandlePtr->loopBackQpMrRemoteInfo.addr);
118 1 : swr->wr.rdma.rkey = flushHandlePtr->loopBackQpMrRemoteInfo.rkey;
119 1 : return HCCL_SUCCESS;
120 : }
121 :
122 2 : HcclResult FlushManager::ExecuteRdmaRead(ibv_qp* loopbackqp0, ibv_cq* cq, ibv_send_wr& swr, int timeoutSec) const
123 : {
124 2 : ibv_send_wr* send_wr = nullptr;
125 2 : int ret = FlushPostSend(loopbackqp0, &swr, &send_wr);
126 2 : if (ret != 0) {
127 3 : HCCL_ERROR("[ExecuteRdmaRead] ibv_post_send failed: %s", strerror(errno));
128 1 : return HCCL_E_NETWORK;
129 : }
130 :
131 3 : HCCL_DEBUG("[ExecuteRdmaRead] RDMA_READ posted successfully. Starting polling for completion...");
132 1 : ibv_wc wc{};
133 : struct timespec start;
134 : struct timespec current;
135 1 : clock_gettime(CLOCK_MONOTONIC, &start);
136 : while (true) {
137 : // 计算已流逝时间（秒）
138 1 : clock_gettime(CLOCK_MONOTONIC, ¤t);
139 1 : int elapsedSec = (current.tv_sec - start.tv_sec) + (current.tv_nsec - start.tv_nsec) / 1000000000;
140 :
141 : // 超时判断
142 1 : if (elapsedSec >= timeoutSec) {
143 3 : HCCL_ERROR(
144 : "[ExecuteRdmaRead] Failed: Wait for completion queue timeout (elapsed=%d s, max=%d s)", elapsedSec,
145 : timeoutSec);
146 1 : return HCCL_E_TIMEOUT;
147 : }
148 :
149 : // 轮询 CQ
150 0 : int numCqes = FlushPollCq(cq, 1, &wc);
151 0 : if (numCqes < 0) {
152 0 : HCCL_ERROR("[ExecuteRdmaRead] ibv_poll_cq returned error: %s", strerror(errno));
153 0 : return HCCL_E_NETWORK;
154 : }
155 : // 成功收到完成事件
156 0 : if (numCqes > 0) {
157 0 : if (wc.status == IBV_WC_SUCCESS) {
158 0 : HCCL_DEBUG(
159 : "[ExecuteRdmaRead] RDMA_READ completed successfully. "
160 : "wr_id=%llu, status=%d",
161 : wc.wr_id, wc.status);
162 0 : return HCCL_SUCCESS;
163 : } else {
164 0 : HCCL_ERROR("[ExecuteRdmaRead] RDMA_READ operation failed: status=%d, wr_id=%llu", wc.status, wc.wr_id);
165 0 : return HCCL_E_NETWORK;
166 : }
167 : }
168 0 : }
169 : }
170 :
171 : } // namespace Hccl
|
