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 "global_mirror_tasks.h"
11 : #include <stdexcept>
12 :
13 : namespace Hccl {
14 :
15 : GlobalMirrorTasks GlobalMirrorTasks::ins_;
16 :
17 1 : GlobalMirrorTasks::GlobalMirrorTasks()
18 : {
19 3 : HCCL_INFO("[GlobalMirrorTasks][GlobalMirrorTasks]GlobalMirrorTasks Contruct");
20 1 : }
21 :
22 1 : GlobalMirrorTasks::~GlobalMirrorTasks()
23 : {
24 3 : HCCL_INFO("[GlobalMirrorTasks][~GlobalMirrorTasks]GlobalMirrorTasks Destroy");
25 1 : }
26 :
27 416 : GlobalMirrorTasks &GlobalMirrorTasks::Instance()
28 : {
29 416 : return ins_;
30 : }
31 :
32 1 : u32 GlobalMirrorTasks::DevSize() const
33 : {
34 1 : return DEVICE_MAX_NUM;
35 : }
36 :
37 17 : TaskInfoQueue *GlobalMirrorTasks::GetQueue(u32 devId, u32 streamId) const
38 : {
39 17 : if (devId >= DEVICE_MAX_NUM) {
40 6 : HCCL_ERROR("GlobalMirrorTasks::GetQueue devId[%u] out of range", devId);
41 2 : THROW<InternalException>(
42 6 : StringFormat("GlobalMirrorTasks::GetQueue devId[%u] out of range", devId));
43 : }
44 :
45 15 : auto &devMap = taskMaps_[devId];
46 15 : auto streamIterator = devMap.find(streamId);
47 15 : if (streamIterator == devMap.end()) {
48 3 : HCCL_ERROR("GlobalMirrorTasks::GetQueue devId[%u], streamId(sqId)[%u] not found", devId, streamId);
49 1 : THROW<InternalException>(
50 3 : StringFormat("GlobalMirrorTasks::GetQueue devId[%u], streamId(sqId)[%u] not found", devId, streamId));
51 : }
52 :
53 42 : HCCL_INFO("[GlobalMirrorTasks][GetQueue]find devId[%u], streamId(sqId)[%u]", devId, streamId);
54 :
55 28 : return streamIterator->second.get();
56 : }
57 :
58 40 : TaskInfoQueue &GlobalMirrorTasks::CreateQueue(u32 devId, u32 streamId, QueueType type)
59 : {
60 40 : if (devId >= DEVICE_MAX_NUM) {
61 1 : THROW<InternalException>(
62 3 : StringFormat("GlobalMirrorTasks::CreateQueue devId[%u] out of range, streamId(sqId)[%u] ", devId, streamId));
63 : }
64 :
65 39 : auto &devMap = taskMaps_[devId];
66 39 : auto streamIterator = devMap.find(streamId);
67 39 : if (streamIterator != devMap.end()) {
68 14 : return *(streamIterator->second.get());
69 : }
70 :
71 25 : std::unique_ptr<TaskInfoQueue> newQueue;
72 25 : if (type == QueueType::Circular_Queue) {
73 23 : newQueue = std::make_unique<CircularQueue<std::unique_ptr<TaskInfo>>>(MAX_CIRCULAR_QUEUE_LENGTH);
74 69 : HCCL_INFO("[GlobalMirrorTasks][CreateQueue]Create circular queue, devId[%u] streamId(sqId)[%u]", devId, streamId);
75 : } else {
76 2 : newQueue = std::make_unique<VectorQueue<std::unique_ptr<TaskInfo>>>();
77 6 : HCCL_INFO("[GlobalMirrorTasks][CreateQueue]Create vector queue, devId[%u] streamId(sqId)[%u]", devId, streamId);
78 : }
79 :
80 25 : devMap[streamId] = std::move(newQueue);
81 :
82 25 : return *devMap[streamId].get();
83 25 : }
84 :
85 10 : void GlobalMirrorTasks::DestroyQueue(u32 devId, u32 streamId)
86 : {
87 10 : if (devId >= DEVICE_MAX_NUM) {
88 1 : THROW<InternalException>(
89 2 : StringFormat("GlobalMirrorTasks::DestroyQueue devId[%u] out of range, streamId(sqId)[%u]", devId, streamId));
90 : return;
91 : }
92 9 : taskMaps_[devId].erase(streamId);
93 : }
94 :
95 6 : TaskInfo* GlobalMirrorTasks::GetTaskInfo(u32 devId, u32 streamId, u32 taskId) const
96 : {
97 6 : TaskInfoQueue *queue = nullptr;
98 : try {
99 6 : queue = GetQueue(devId, streamId);
100 1 : }catch(HcclException &e){
101 1 : return nullptr;
102 1 : }
103 :
104 66 : auto FindTask = [taskId](const std::unique_ptr<TaskInfo> &taskInfo) {
105 66 : return taskInfo->taskId_ == taskId;
106 5 : };
107 :
108 5 : auto task = queue->Find(FindTask);
109 5 : if (*task == *queue->End()) {
110 1 : return nullptr;
111 : };
112 :
113 12 : HCCL_INFO("[GlobalMirrorTasks][GetTaskInfo]find devId[%u], streamId(sqId)[%u] taskId(sqeId)[%u]", devId, streamId, taskId);
114 :
115 4 : return (*(*task)).get();
116 5 : }
117 :
118 34 : TaskInfoQueueMap::iterator GlobalMirrorTasks::Begin(u32 devId)
119 : {
120 34 : if (devId >= DEVICE_MAX_NUM) {
121 1 : THROW<InternalException>(StringFormat("GlobalMirrorTasks::Begin devId[%u] out of range", devId));
122 : }
123 33 : auto &devMap = taskMaps_[devId];
124 33 : return devMap.begin();
125 : }
126 :
127 34 : TaskInfoQueueMap::iterator GlobalMirrorTasks::End(u32 devId)
128 : {
129 34 : if (devId >= DEVICE_MAX_NUM) {
130 1 : THROW<InternalException>(StringFormat("GlobalMirrorTasks::End devId[%u] out of range", devId));
131 : }
132 33 : auto &devMap = taskMaps_[devId];
133 33 : return devMap.end();
134 : }
135 :
136 6 : HcclResult GlobalMirrorTasks::FindTaskInfo(u32 devId, u32 streamId, u32 taskId, TaskInfo*& curTask) const
137 : {
138 18 : HCCL_INFO("[%s]start, devId[%u] streamId(sqId)[%u] taskId(sqeId)[%u].", __func__, devId, streamId, taskId);
139 9 : CHK_PRT_RET(devId >= DEVICE_MAX_NUM, HCCL_ERROR("[%s]fail, devId[%u] out of range.", __func__, devId), HCCL_E_PARA);
140 :
141 5 : const TaskInfoQueueMap &devMap = taskMaps_[devId];
142 5 : auto streamIterator = devMap.find(streamId);
143 5 : if (streamIterator == devMap.end()) { // rts回调时不会判断异常task是否HCCL task，索引不到可能是其他组件task，此处不打印ERROR日志
144 3 : HCCL_RUN_INFO("[%s]devId[%u] streamId(sqId)[%u] not hccl task.", __func__, devId, streamId);
145 1 : return HCCL_E_NOT_FOUND;
146 : }
147 :
148 4 : TaskInfoQueue* queue = streamIterator->second.get();
149 4 : CHK_PTR_NULL(queue);
150 :
151 4 : auto FindTask = [taskId](const std::unique_ptr<TaskInfo> &taskInfo) {
152 4 : return taskInfo->taskId_ == taskId;
153 4 : };
154 :
155 4 : auto task = queue->Find(FindTask);
156 4 : if (*task == *queue->End() || *(*task) == nullptr) {
157 6 : HCCL_RUN_INFO("[%s]devId[%u] streamId(sqId)[%u] taskId(sqeId)[%u] not hccl task.",
158 : __func__, devId, streamId, taskId);
159 2 : return HCCL_E_NOT_FOUND;
160 : };
161 :
162 2 : curTask = (*(*task)).get();
163 6 : HCCL_INFO("[%s]success, devId[%u] streamId(sqId)[%u] taskId(sqeId)[%u].", __func__, devId, streamId, taskId);
164 2 : return HCCL_SUCCESS;
165 4 : }
166 :
167 : } // namespace Hccl
|
