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