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