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