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 "p2p_enable_manager.h"
12 : #include "log.h"
13 : #include "env_config.h"
14 :
15 : namespace Hccl {
16 :
17 886 : P2PEnableManager &P2PEnableManager::GetInstance()
18 : {
19 886 : static P2PEnableManager p2pEnableManager;
20 886 : return p2pEnableManager;
21 : }
22 :
23 39 : HcclResult P2PEnableManager::EnableP2P(std::vector<uint32_t> remoteDevices)
24 : {
25 39 : auto localDeviceLogicID = HrtGetDevice();
26 :
27 42 : for (auto remoteDevicePhysicID : remoteDevices) {
28 3 : CHK_RET(EnableP2P(localDeviceLogicID, remoteDevicePhysicID));
29 : }
30 39 : return HCCL_SUCCESS;
31 : }
32 :
33 3 : HcclResult P2PEnableManager::EnableP2P(uint32_t localDeviceLogicID, uint32_t remoteDevicePhysicID)
34 : {
35 3 : std::unique_lock<std::mutex> lock(connectionsLock_[localDeviceLogicID]);
36 3 : auto &iterLocalDevice = connectionsInfo_[localDeviceLogicID];
37 3 : auto iterRemoteDevice = iterLocalDevice.find(remoteDevicePhysicID);
38 3 : if ((iterRemoteDevice == iterLocalDevice.end()) || (iterRemoteDevice->second.reference == 0)) {
39 3 : auto localDevicePhysicID = HrtGetDevicePhyIdByIndex(localDeviceLogicID);
40 3 : CHK_RET(HrtEnableP2P(localDeviceLogicID, remoteDevicePhysicID));
41 9 : HCCL_INFO("[EnableP2P]enable p2p: local logic id:%u, local physic id:%u, remote physic id:%u.",
42 : localDeviceLogicID, localDevicePhysicID, remoteDevicePhysicID);
43 3 : iterLocalDevice[remoteDevicePhysicID].status = P2PStatus::P2P_STATUS_ENABLING;
44 3 : iterLocalDevice[remoteDevicePhysicID].reference++;
45 3 : return HCCL_SUCCESS;
46 : } else {
47 : // 使已执行过 enable，且未执行过 disable，不重复执行 enable p2p。
48 0 : iterLocalDevice[remoteDevicePhysicID].reference++;
49 : }
50 0 : return HCCL_SUCCESS;
51 3 : }
52 :
53 0 : HcclResult P2PEnableManager::WaitP2PEnabled(std::vector<uint32_t> remoteDevices)
54 : {
55 0 : auto localDeviceLogicID = HrtGetDevice();
56 :
57 0 : for (auto &remoteDevicePhysicID : remoteDevices) {
58 0 : CHK_RET(WaitP2PEnabled(localDeviceLogicID, remoteDevicePhysicID));
59 : }
60 0 : return HCCL_SUCCESS;
61 : }
62 :
63 0 : HcclResult P2PEnableManager::WaitP2PEnabled(uint32_t localDeviceLogicID, uint32_t remoteDevicePhysicID)
64 : {
65 0 : std::unique_lock<std::mutex> lock(connectionsLock_[localDeviceLogicID]);
66 :
67 0 : auto &iterLocalDevice = connectionsInfo_[localDeviceLogicID];
68 0 : auto iterRemoteDevice = iterLocalDevice.find(remoteDevicePhysicID);
69 0 : bool bErr = (iterRemoteDevice == iterLocalDevice.end()) || (iterRemoteDevice->second.reference == 0) ||
70 0 : (iterRemoteDevice->second.status == P2PStatus::P2P_STATUS_DISABLED);
71 0 : CHK_PRT_RET(bErr, HCCL_ERROR("[Wait][P2PEnabled]wait p2p enabled failed. enable operation has not been executed, "\
72 : "ret[%u]. device info: local logic id:%u, remote physic id:%u.", HCCL_E_INTERNAL, localDeviceLogicID,
73 : remoteDevicePhysicID), HCCL_E_INTERNAL);
74 :
75 0 : if (iterRemoteDevice->second.status == P2PStatus::P2P_STATUS_ENABLED) {
76 0 : return HCCL_SUCCESS;
77 : } else {
78 0 : auto localDevicePhysicID = HrtGetDevicePhyIdByIndex(localDeviceLogicID);
79 :
80 0 : CHK_RET(WaitP2PConnected(localDeviceLogicID, remoteDevicePhysicID));
81 0 : HCCL_INFO("[Wait]enable p2p: local logic id:%u, local physic id:%u, remote physic id:%u.",
82 : localDeviceLogicID, localDevicePhysicID, remoteDevicePhysicID);
83 0 : iterLocalDevice[remoteDevicePhysicID].status = P2PStatus::P2P_STATUS_ENABLED;
84 : }
85 0 : return HCCL_SUCCESS;
86 0 : }
87 :
88 0 : HcclResult P2PEnableManager::WaitP2PConnected(uint32_t localDeviceLogicID, uint32_t remoteDevicePhysicID)
89 : {
90 : // 读取P2P状态超时时间
91 0 : const std::chrono::seconds timeout(EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
92 0 : const std::chrono::milliseconds checkP2PTimeInterval(1); // 轮询P2P状态时间 1ms
93 0 : const auto start = TIME_NOW();
94 :
95 : while (true) {
96 0 : uint32_t status = DRV_P2P_STATUS_DISABLE;
97 0 : CHK_RET(HrtGetP2PStatus(localDeviceLogicID, remoteDevicePhysicID, &status));
98 :
99 0 : if (status == DRV_P2P_STATUS_ENABLE) {
100 0 : HCCL_INFO("connected p2p success, take time [%lld]us. device info: local logic id:%u, remote physic id:%u.",
101 : DURATION_US(TIME_NOW() - start), localDeviceLogicID, remoteDevicePhysicID);
102 0 : return HCCL_SUCCESS;
103 : }
104 0 : std::this_thread::sleep_for(checkP2PTimeInterval);
105 : /* 获取当前时间，如果耗时超过timeout，则返回错误 */
106 : const auto elapsed =
107 0 : std::chrono::duration_cast<std::chrono::seconds>(TIME_NOW() - start);
108 0 : if (elapsed > timeout) {
109 0 : HCCL_ERROR("[Wait][P2PConnected]connected p2p timeout, timeout:%d s. local logicDevid:%u, "\
110 : "remote physic id:%u.", EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut(),
111 : localDeviceLogicID, remoteDevicePhysicID);
112 0 : return HCCL_E_DRV;
113 : }
114 0 : }
115 : return HCCL_SUCCESS;
116 : }
117 :
118 846 : HcclResult P2PEnableManager::DisableP2P(uint32_t localDeviceLogicID, std::vector<uint32_t> remoteDevices)
119 : {
120 : try {
121 852 : for (auto &remoteDevicePhysicID : remoteDevices) {
122 6 : CHK_RET(DisableP2P(localDeviceLogicID, remoteDevicePhysicID));
123 : }
124 0 : } catch (HcclException &e) {
125 0 : HCCL_ERROR("%s", e.what());
126 0 : return e.GetErrorCode();
127 0 : } catch (...) {
128 0 : HCCL_ERROR("Unknown error occurs!");
129 0 : return HcclResult::HCCL_E_INTERNAL;
130 0 : }
131 846 : return HCCL_SUCCESS;
132 : }
133 :
134 6 : HcclResult P2PEnableManager::DisableP2P(uint32_t localDeviceLogicID, uint32_t remoteDevicePhysicID)
135 : {
136 6 : std::unique_lock<std::mutex> lock(connectionsLock_[localDeviceLogicID]);
137 6 : auto &iterLocalDevice = connectionsInfo_[localDeviceLogicID];
138 6 : auto iterRemoteDevice = iterLocalDevice.find(remoteDevicePhysicID);
139 6 : if ((iterRemoteDevice == iterLocalDevice.end()) || (iterRemoteDevice->second.reference == 0)) {
140 18 : HCCL_WARNING("there is no p2p connections, no need to disable p2p. "\
141 : "device info: local logic id:%u, remote physic id:%u.", localDeviceLogicID, remoteDevicePhysicID);
142 6 : return HCCL_SUCCESS;
143 : }
144 :
145 0 : iterRemoteDevice->second.reference--;
146 0 : if (iterRemoteDevice->second.reference == 0) {
147 0 : auto localDevicePhysicID = HrtGetDevicePhyIdByIndex(localDeviceLogicID);
148 0 : HCCL_INFO("disable p2p: local logic id:%u, local physic id:%u, remote physic id:%u.", localDeviceLogicID,
149 : localDevicePhysicID, remoteDevicePhysicID);
150 0 : CHK_RET(HrtDisableP2P(localDeviceLogicID, remoteDevicePhysicID));
151 0 : iterLocalDevice[remoteDevicePhysicID].status = P2PStatus::P2P_STATUS_DISABLED;
152 : }
153 0 : return HCCL_SUCCESS;
154 6 : }
155 :
156 1 : P2PEnableManager::~P2PEnableManager()
157 : {
158 1 : }
159 :
160 : } // namespace Hccl
|
