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