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 "hccl_primitive_local.h"
12 : #include "log.h"
13 : #include "mem_device_pub.h"
14 : #include "hccl_dispatcher_ctx.h"
15 : #include "dispatcher_task_types.h"
16 : #include "dispatcher_aicpu_pub.h"
17 : #include "local_notify.h"
18 : #include "dispatcher_ctx.h"
19 :
20 7 : HcclResult GetPubDispatcher(hccl::DispatcherPub** dispatcherPtr)
21 : {
22 7 : DispatcherCtxPtr ctx = nullptr;
23 7 : CHK_RET(AcquireDispatcherCtx(&ctx));
24 7 : CHK_PTR_NULL(ctx);
25 7 : hccl::DispatcherCtx* ctx_temp = reinterpret_cast<hccl::DispatcherCtx*>(ctx);
26 7 : CHK_PTR_NULL(ctx_temp->GetDispatcher());
27 7 : *dispatcherPtr = reinterpret_cast<hccl::DispatcherPub*>(ctx_temp->GetDispatcher());
28 7 : CHK_PTR_NULL(*dispatcherPtr);
29 7 : return HCCL_SUCCESS;
30 : }
31 :
32 5 : HcclResult HcclLocalCopy(StreamHandle streamHandle, HcclBuf* dst, HcclBuf* src)
33 : {
34 5 : CHK_PTR_NULL(src);
35 5 : CHK_PTR_NULL(dst);
36 5 : CHK_PTR_NULL(streamHandle);
37 5 : hccl::DeviceMem srcDevMem(src->addr, src->len);
38 5 : hccl::DeviceMem dstDevMem(dst->addr, dst->len);
39 5 : HCCL_INFO(
40 : "[hcclLocalCopy] dst addr[%p], size[%llu], src addr[%p], size[%llu]", dst->addr, dst->len, src->addr, src->len);
41 :
42 5 : hccl::Stream* stream = reinterpret_cast<hccl::Stream*>(streamHandle);
43 :
44 5 : hccl::DispatcherPub* dispatcherPtr = nullptr;
45 5 : CHK_RET(GetPubDispatcher(&dispatcherPtr));
46 5 : HCCL_INFO("[%s] dispatcherPtr[%p]", __func__, (void*)dispatcherPtr);
47 5 : return dispatcherPtr->MemcpyAsync(dstDevMem, srcDevMem, *stream);
48 5 : }
49 :
50 2 : HcclResult HcclLocalCopyReduce(StreamHandle streamHandle, HcclBuf* dst, HcclBuf* src, HcclReduceInfo reduceInfo)
51 : {
52 2 : CHK_PTR_NULL(src);
53 2 : CHK_PTR_NULL(dst);
54 2 : CHK_PTR_NULL(streamHandle);
55 :
56 1 : HCCL_INFO(
57 : "[HcclLocalCopyReduce] dst ptr[%p], size[%llu], src ptr[%p], size[%llu], datatype[%d], reduceOp[%d]", dst->addr,
58 : dst->len, src->addr, src->len, static_cast<int>(reduceInfo.dataType), static_cast<int>(reduceInfo.reduceOp));
59 :
60 1 : hccl::DispatcherPub* dispatcherPtr = nullptr;
61 1 : CHK_RET(GetPubDispatcher(&dispatcherPtr));
62 :
63 1 : hccl::Stream* stream = reinterpret_cast<hccl::Stream*>(streamHandle);
64 :
65 2 : return dispatcherPtr->InlineReduceAsync(
66 1 : src->addr, src->len / SIZE_TABLE[reduceInfo.dataType], reduceInfo.dataType, reduceInfo.reduceOp, *stream,
67 1 : dst->addr, INVALID_VALUE_RANKID, hccl::LinkType::LINK_ONCHIP);
68 : }
69 :
70 2 : HcclResult HcclLocalLaunchTaskExtend(aclrtStream& stream, std::vector<aclrtStream>& subStreams)
71 : {
72 2 : CHK_PTR_NULL(stream);
73 2 : hccl::Stream stream_temp(stream, false);
74 :
75 2 : hccl::DispatcherPub* dispatcherPtr = nullptr;
76 2 : hccl::DispatcherCtx* ctx_temp = reinterpret_cast<hccl::DispatcherCtx*>(GetDispatcherCtx());
77 2 : CHK_PTR_NULL(ctx_temp);
78 1 : CHK_PTR_NULL(ctx_temp->GetDispatcher());
79 1 : dispatcherPtr = reinterpret_cast<hccl::DispatcherPub*>(ctx_temp->GetDispatcher());
80 :
81 1 : if (ctx_temp->GetLaunchTaskCallback() != nullptr) {
82 0 : CHK_RET(ctx_temp->GetLaunchTaskCallback()(dispatcherPtr, stream_temp));
83 : }
84 :
85 1 : std::vector<hccl::Stream> subStreams_temp;
86 1 : for (auto& s : subStreams) {
87 0 : CHK_PTR_NULL(s);
88 0 : subStreams_temp.push_back(*(reinterpret_cast<hccl::Stream*>(s)));
89 : }
90 :
91 1 : return dispatcherPtr->LaunchTasksEx(stream_temp, subStreams_temp);
92 2 : }
93 :
94 : HcclResult
95 2 : HcclLocalInitTask(aclrtStream stream, const bool enableCache, const std::string& key, bool useGraphConstructorV2)
96 : {
97 2 : CHK_PTR_NULL(stream);
98 :
99 1 : hccl::DispatcherPub* dispatcherPtr = nullptr;
100 1 : hccl::DispatcherCtx* ctx_temp = reinterpret_cast<hccl::DispatcherCtx*>(GetDispatcherCtx());
101 1 : CHK_PTR_NULL(ctx_temp);
102 1 : CHK_PTR_NULL(ctx_temp->GetDispatcher());
103 1 : dispatcherPtr = reinterpret_cast<hccl::DispatcherPub*>(ctx_temp->GetDispatcher());
104 :
105 1 : HCCL_INFO(
106 : "InitTask enableCache[%d], key[%s], useGraphConstructorV2[%d]", enableCache, key.c_str(),
107 : useGraphConstructorV2);
108 :
109 1 : CHK_RET(dispatcherPtr->ResetGraphCtx(enableCache, key, useGraphConstructorV2));
110 :
111 1 : hccl::Stream stream_temp(stream, false);
112 :
113 1 : if (ctx_temp->GetInitTaskCallback() != nullptr) {
114 0 : CHK_RET(ctx_temp->GetInitTaskCallback()(dispatcherPtr, stream_temp));
115 : }
116 1 : return HCCL_SUCCESS;
117 1 : }
118 :
119 2 : HcclResult HcclLocalNotifyRecord(StreamHandle streamHandle, aclrtNotify notify)
120 : {
121 2 : CHK_PTR_NULL(streamHandle);
122 1 : CHK_PTR_NULL(notify);
123 1 : hccl::DispatcherPub* dispatcherPtr = nullptr;
124 1 : CHK_RET(GetPubDispatcher(&dispatcherPtr));
125 :
126 1 : hccl::Stream* stream = reinterpret_cast<hccl::Stream*>(streamHandle);
127 1 : hccl::LocalNotify* localNotify = reinterpret_cast<hccl::LocalNotify*>(notify);
128 :
129 1 : return dispatcherPtr->SignalRecord(
130 : localNotify->ptr(), *stream, INVALID_VALUE_RANKID, INVALID_U64, INVALID_VALUE_STAGE, true, INVALID_U64,
131 1 : localNotify->notifyId_);
132 : }
133 :
134 1 : HcclResult HcclLocalNotifyWait(StreamHandle streamHandle, aclrtNotify notify, const uint32_t timeOut)
135 : {
136 1 : CHK_PTR_NULL(streamHandle);
137 0 : CHK_PTR_NULL(notify);
138 :
139 0 : hccl::DispatcherPub* dispatcherPtr = nullptr;
140 0 : CHK_RET(GetPubDispatcher(&dispatcherPtr));
141 :
142 0 : hccl::LocalNotify* localNotify = reinterpret_cast<hccl::LocalNotify*>(notify);
143 0 : hccl::Stream* stream = reinterpret_cast<hccl::Stream*>(streamHandle);
144 0 : return dispatcherPtr->SignalWait(
145 : localNotify->ptr(), *stream, INVALID_VALUE_RANKID, INVALID_VALUE_RANKID, INVALID_VALUE_STAGE, true,
146 0 : localNotify->notifyId_, timeOut);
147 : }
148 :
149 0 : HcclResult HcclTaskPrepare(char* key, uint32_t keyLen) // host ffts+使用
150 : {
151 0 : bool enableCache = false;
152 0 : std::string keyStr = "temp_key";
153 0 : if (key != nullptr && keyLen != 0) {
154 0 : enableCache = true;
155 0 : keyStr = std::string(key, keyLen);
156 0 : HCCL_DEBUG("[HcclTaskPrepare]key[%s], keyLen[%u]", key, keyLen);
157 : } else {
158 0 : HCCL_DEBUG("[HcclTaskPrepare]disable cache, key[%p], keyLen[%u]", key, keyLen);
159 : }
160 :
161 0 : hccl::DispatcherPub* dispatcherPtr = nullptr;
162 0 : CHK_RET(GetPubDispatcher(&dispatcherPtr));
163 :
164 0 : return dispatcherPtr->ResetGraphCtx(enableCache, keyStr, true);
165 0 : }
166 :
167 0 : HcclResult HcclTaskLaunch(hccl::Stream* streams, uint32_t streamNum) // host ffts+或aicpu stars使用"
168 : {
169 0 : CHK_PTR_NULL(streams);
170 0 : CHK_PRT_RET(streamNum < 1, HCCL_ERROR("[HcclTaskLaunch]threadNum is less than 1"), HCCL_E_PARA);
171 0 : hccl::Stream mainStream = streams[0];
172 0 : std::vector<hccl::Stream> subStreams;
173 0 : for (uint32_t i = 1; i < streamNum; i++) {
174 0 : subStreams.push_back(streams[i]);
175 : }
176 :
177 0 : hccl::DispatcherPub* dispatcherPtr = nullptr;
178 0 : CHK_RET(GetPubDispatcher(&dispatcherPtr));
179 :
180 0 : return dispatcherPtr->LaunchTasksEx(mainStream, subStreams);
181 0 : }
182 :
183 0 : HcclResult HcclLocalBareNotifyRecord(StreamHandle streamHandle, uint64_t dstNotifyId)
184 : {
185 0 : CHK_PTR_NULL(streamHandle);
186 0 : hccl::DispatcherPub* dispatcherPtr = nullptr;
187 0 : CHK_RET(GetPubDispatcher(&dispatcherPtr));
188 :
189 0 : hccl::Stream* stream = reinterpret_cast<hccl::Stream*>(streamHandle);
190 :
191 0 : return dispatcherPtr->SignalRecord(*stream, dstNotifyId);
192 : }
193 :
194 0 : HcclResult HcclLocalBareNotifyWait(StreamHandle streamHandle, uint64_t notifyId, uint32_t timeOut)
195 : {
196 0 : CHK_PTR_NULL(streamHandle);
197 :
198 0 : hccl::DispatcherPub* dispatcherPtr = nullptr;
199 0 : CHK_RET(GetPubDispatcher(&dispatcherPtr));
200 :
201 0 : hccl::Stream* stream = reinterpret_cast<hccl::Stream*>(streamHandle);
202 0 : HCCL_INFO("%s notifyId[%llu]", __func__, notifyId);
203 0 : return dispatcherPtr->SignalWait(*stream, notifyId, timeOut);
204 : }
205 :
206 0 : HcclResult HcclTaskClear(std::string key) // host ffts+使用
207 : {
208 0 : hccl::DispatcherPub* dispatcherPtr = nullptr;
209 0 : CHK_RET(GetPubDispatcher(&dispatcherPtr));
210 0 : return dispatcherPtr->ResetGraphCtx(false, key, true);
211 : }
|
