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