fix: C流程从内存聚合op_statistic，不再依赖op_counter.db-MindStudio-Profiler-AtomGit

Manalysis/csrc/application/database/db_constant.h+5-0

	@@ -68,7 +68,12 @@ const std::string PROCESSOR_NAME_COMM_STATISTIC = "COMMUNICATION_STATISTIC";
68	const std::string TABLE_NAME_COMMUNICATION_TASK_INFO = "COMMUNICATION_TASK_INFO";	68	const std::string TABLE_NAME_COMMUNICATION_TASK_INFO = "COMMUNICATION_TASK_INFO";
69	const std::string TABLE_NAME_COMMUNICATION_OP = "COMMUNICATION_OP";	69	const std::string TABLE_NAME_COMMUNICATION_OP = "COMMUNICATION_OP";
70		70
		71	+const std::string PROCESSOR_NAME_TASK_ASSOCIATION = "TASK_ASSOCIATION";
71	const std::string PROCESSOR_NAME_OP_STATISTIC = "OP_STATISTIC";	72	const std::string PROCESSOR_NAME_OP_STATISTIC = "OP_STATISTIC";
		73	+const std::string TASK_TYPE_WRITE_BACK = "WRITE_BACK";
		74	+const std::string TASK_TYPE_INVALID = "INVALID";
		75	+const std::string TASK_TYPE_HCCL_AI_CPU = "HCCL_AI_CPU";
		76	+const std::string TASK_TYPE_COMMUNICATION = "COMMUNICATION";
72		77
73	const std::string PROCESSOR_NAME_API = "API";	78	const std::string PROCESSOR_NAME_API = "API";
74	const std::string TABLE_NAME_CANN_API = "CANN_API";	79	const std::string TABLE_NAME_CANN_API = "CANN_API";

Manalysis/csrc/application/summary/op_summary_assembler.cpp+14-35

	@@ -62,9 +62,6 @@ const std::string AIV_TOTAL_TIME = "aiv_total_time(us)";
62	const std::string DUR_IDX = "durIdx";	62	const std::string DUR_IDX = "durIdx";
63	const std::string USAGE_IDX = "usageIdx";	63	const std::string USAGE_IDX = "usageIdx";
64		64
65	-// WRITE_BACK与INVALID类型不需要处理，针对helper场景, 去除运行在AI_CPU的HCCL小算子不生成op_summary,
66	-// 运行在AI_CORE上的HCCL小算子也不呈现在op_summary,因此这四个类型都不生成数据即可，直接排除掉
67	-const std::vector<std::string> INVALID_TASK_TYPE{"WRITE_BACK", "INVALID", "HCCL_AI_CPU", "COMMUNICATION"};
68	const std::vector<std::string> BASE_HEADER{DEVICE_ID, MODEL_ID, TASK_ID, STREAM_ID, OP_NAME, OP_TYPE,	65	const std::vector<std::string> BASE_HEADER{DEVICE_ID, MODEL_ID, TASK_ID, STREAM_ID, OP_NAME, OP_TYPE,
69	OP_STATE, TASK_TYPE, TASK_START_TIME, TASK_DURATION, TASK_WAIT_TIME};	66	OP_STATE, TASK_TYPE, TASK_START_TIME, TASK_DURATION, TASK_WAIT_TIME};
70	const std::vector<std::string> ADDITIONAL_TENSOR_HEADER{	67	const std::vector<std::string> ADDITIONAL_TENSOR_HEADER{
	@@ -123,15 +120,6 @@ void OpSummaryAssembler::GenerateHcclBody(std::vector<CommunicationOpData> &opDa
123	}	120	}
124	}	121	}
125		122
126	-void OpSummaryAssembler::SplitDataByTaskId(std::vector<TaskInfoData> &taskInfo)
127	-{
128	- for (auto &data : taskInfo)
129	- {
130	- TaskId id{data.streamId, data.batchId, data.taskId, data.contextId, data.deviceId};
131	- computeTask_[id] = &data;
132	- }
133	-}
134	-
135	std::vector<std::string> OpSummaryAssembler::GenerateOneTaskRow(const TaskInfoData &computeTask,	123	std::vector<std::string> OpSummaryAssembler::GenerateOneTaskRow(const TaskInfoData &computeTask,
136	const AscendTaskData &task)	124	const AscendTaskData &task)
137	{	125	{
	@@ -195,7 +183,8 @@ void OpSummaryAssembler::MergeTaskAndPmu(std::shared_ptr<MetricSummary> &pmu, st
195	}	183	}
196	}	184	}
197		185
198	-void OpSummaryAssembler::GenerateOpBody(std::vector<AscendTaskData> &taskData, std::shared_ptr<MetricSummary> &pmu)	186	+void OpSummaryAssembler::GenerateOpBody(const AssociatedTaskCollection &associatedTasks,
		187	+ std::shared_ptr<MetricSummary> &pmu)
199	{	188	{
200	if (pmu != nullptr)	189	if (pmu != nullptr)
201	{	190	{
	@@ -203,24 +192,17 @@ void OpSummaryAssembler::GenerateOpBody(std::vector<AscendTaskData> &taskData, s
203	}	192	}
204	std::unordered_map<std::string, int> indexTable{{DUR_IDX, GetIndexForVec(headers_, TASK_DURATION)},	193	std::unordered_map<std::string, int> indexTable{{DUR_IDX, GetIndexForVec(headers_, TASK_DURATION)},
205	{USAGE_IDX, GetIndexForVec(headers_, CUBE_UTILIZATION)}};	194	{USAGE_IDX, GetIndexForVec(headers_, CUBE_UTILIZATION)}};
206	- for (const auto &task : taskData)	195	+ for (const auto &item : associatedTasks.records)
207	{	196	{
208	- TaskId id{task.streamId, task.batchId, task.taskId, task.contextId, task.deviceId};	197	+ if (!item.opSummaryRequired)
209	- auto it = computeTask_.find(id);
210	- if (it != computeTask_.end())
211	{	198	{
212	- const std::string &taskType = it->second->taskType;	199	+ continue;
213	- const std::string &opName = it->second->opName;
214	- bool isInvalidType =
215	- std::find(INVALID_TASK_TYPE.begin(), INVALID_TASK_TYPE.end(), taskType) != INVALID_TASK_TYPE.end();
216	- bool isCommWithValidSuffix = (taskType == "COMMUNICATION" && EndsWith(opName, AIV_KERNEL));
217	- if (!isInvalidType \|\| isCommWithValidSuffix)
218	- {
219	- auto row = GenerateOneTaskRow(*it->second, task);
220	- MergeTaskAndPmu(pmu, row, id, indexTable);
221	- res_.emplace_back(row);
222	- }
223	}	200	}
		201	+ auto row = GenerateOneTaskRow(item.taskInfo, item.ascendTask);
		202	+ TaskId taskId{item.ascendTask->streamId, item.ascendTask->batchId, item.ascendTask->taskId,
		203	+ item.ascendTask->contextId, item.ascendTask->deviceId};
		204	+ MergeTaskAndPmu(pmu, row, taskId, indexTable);
		205	+ res_.emplace_back(row);
224	}	206	}
225	}	207	}
226		208
	@@ -319,22 +301,19 @@ void OpSummaryAssembler::WriteToFile(const std::string &fileName, const std::set
319		301
320	uint8_t OpSummaryAssembler::AssembleData(DataInventory &dataInventory)	302	uint8_t OpSummaryAssembler::AssembleData(DataInventory &dataInventory)
321	{	303	{
322	- auto taskInfoData = dataInventory.GetPtr<std::vector<TaskInfoData>>();	304	+ auto associatedTasks = dataInventory.GetPtr<AssociatedTaskCollection>();
323	- auto ascendTaskData = dataInventory.GetPtr<std::vector<AscendTaskData>>();
324	auto hcclOpData = dataInventory.GetPtr<std::vector<CommunicationOpData>>();	305	auto hcclOpData = dataInventory.GetPtr<std::vector<CommunicationOpData>>();
325	auto metricData = dataInventory.GetPtr<MetricSummary>();	306	auto metricData = dataInventory.GetPtr<MetricSummary>();
326	- if ((taskInfoData == nullptr \|\| ascendTaskData == nullptr) && hcclOpData == nullptr)	307	+ if (associatedTasks == nullptr && hcclOpData == nullptr)
327	{	308	{
328	WARN("No data to export op summary");	309	WARN("No data to export op summary");
329	return DATA_NOT_EXIST;	310	return DATA_NOT_EXIST;
330	}	311	}
331	headers_ = BASE_HEADER;	312	headers_ = BASE_HEADER;
332	- // 当没有ascendTask或者没有taskInfo数据时，只生成hccl数据	313	+ if (associatedTasks != nullptr)
333	- if (taskInfoData != nullptr && ascendTaskData != nullptr)
334	{	314	{
335	headers_.insert(headers_.end(), ADDITIONAL_TENSOR_HEADER.begin(), ADDITIONAL_TENSOR_HEADER.end());	315	headers_.insert(headers_.end(), ADDITIONAL_TENSOR_HEADER.begin(), ADDITIONAL_TENSOR_HEADER.end());
336	- SplitDataByTaskId(*taskInfoData);	316	+ GenerateOpBody(*associatedTasks, metricData);
337	- GenerateOpBody(*ascendTaskData, metricData);
338	}	317	}
339	else	318	else
340	{	319	{

Manalysis/csrc/application/summary/op_summary_assembler.h+21-17

	@@ -18,38 +18,42 @@
18	#define ANALYSIS_APPLICATION_OP_SUMMARY_ASSEMBLE_H	18	#define ANALYSIS_APPLICATION_OP_SUMMARY_ASSEMBLE_H
19		19
20	#include <cstdint>	20	#include <cstdint>
21	-#include <map>	21	+#include <unordered_map>
		22	+
22	#include "analysis/csrc/application/summary/summary_assembler.h"	23	#include "analysis/csrc/application/summary/summary_assembler.h"
23	-#include "analysis/csrc/infrastructure/data_inventory/include/data_inventory.h"	24	+#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/associated_task_data.h"
24	-#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/task_info_data.h"
25	-#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/ascend_task_data.h"
26	#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/communication_info_data.h"	25	#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/communication_info_data.h"
27	#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/metric_summary.h"	26	#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/metric_summary.h"
		27	+#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/task_info_data.h"
		28	+#include "analysis/csrc/domain/valueobject/include/task_id.h"
		29	+#include "analysis/csrc/infrastructure/data_inventory/include/data_inventory.h"
28		30
29	-namespace Analysis {	31	+namespace Analysis
30	-namespace Application {	32	+{
		33	+namespace Application
		34	+{
31	using namespace Analysis::Infra;	35	using namespace Analysis::Infra;
32	using namespace Analysis::Domain;	36	using namespace Analysis::Domain;
33	-class OpSummaryAssembler : public SummaryAssembler {	37	+class OpSummaryAssembler : public SummaryAssembler
34	-public:	38	+{
		39	+ public:
35	OpSummaryAssembler() = default;	40	OpSummaryAssembler() = default;
36	OpSummaryAssembler(const std::string &name, const std::string &profPath);	41	OpSummaryAssembler(const std::string &name, const std::string &profPath);
37	-protected:	42	+
		43	+ protected:
38	void WriteToFile(const std::string &fileName, const std::set<int> &maskCols);	44	void WriteToFile(const std::string &fileName, const std::set<int> &maskCols);
39	-private:	45	+
		46	+ private:
40	uint8_t AssembleData(DataInventory &dataInventory);	47	uint8_t AssembleData(DataInventory &dataInventory);
41	- void GenerateOpBody(std::vector<AscendTaskData> &taskData, std::shared_ptr<MetricSummary> &pmu);	48	+ void GenerateOpBody(const AssociatedTaskCollection &associatedTasks, std::shared_ptr<MetricSummary> &pmu);
42	void GenerateHcclBody(std::vector<CommunicationOpData> &opData);	49	void GenerateHcclBody(std::vector<CommunicationOpData> &opData);
43	- void SplitDataByTaskId(std::vector<TaskInfoData> &taskInfo);
44	std::vector<std::string> GenerateOneTaskRow(const TaskInfoData &computeTask, const AscendTaskData &task);	50	std::vector<std::string> GenerateOneTaskRow(const TaskInfoData &computeTask, const AscendTaskData &task);
45	void MergeTaskAndPmu(std::shared_ptr<MetricSummary> &pmu, std::vector<std::string> &row, const TaskId &id,	51	void MergeTaskAndPmu(std::shared_ptr<MetricSummary> &pmu, std::vector<std::string> &row, const TaskId &id,
46	std::unordered_map<std::string, int> &indexTable);	52	std::unordered_map<std::string, int> &indexTable);
47	void AddCubeUsage(std::vector<std::string> &data, std::unordered_map<std::string, int> &indexTable);	53	void AddCubeUsage(std::vector<std::string> &data, std::unordered_map<std::string, int> &indexTable);
48	void CalculateWaitTime();	54	void CalculateWaitTime();
49	std::set<int> GetMaskCols();	55	std::set<int> GetMaskCols();
50	-private:
51	- std::map<TaskId, TaskInfoData*> computeTask_;
52	};	56	};
53	-}	57	+} // namespace Application
54	-}	58	+} // namespace Analysis
55	-#endif // ANALYSIS_APPLICATION_OP_SUMMARY_ASSEMBLE_H	59	+#endif // ANALYSIS_APPLICATION_OP_SUMMARY_ASSEMBLE_H

Manalysis/csrc/application/summary/summary_topology_register.cpp+4-4

Manalysis/csrc/domain/data_process/ai_task/op_statistic_processor.cpp+146-74

	@@ -14,95 +14,167 @@
14	* See the Mulan PSL v2 for more details.	14	* See the Mulan PSL v2 for more details.
15	* -------------------------------------------------------------------------*/	15	* -------------------------------------------------------------------------*/
16		16
17	-
18	#include "analysis/csrc/domain/data_process/ai_task/op_statistic_processor.h"	17	#include "analysis/csrc/domain/data_process/ai_task/op_statistic_processor.h"
19	-#include "analysis/csrc/domain/services/environment/context.h"
20		18
21	-namespace Analysis {	19	+#include <algorithm>
22	-namespace Domain {	20	+#include <cstdint>
23	-using namespace Analysis::Domain::Environment;	21	+#include <limits>
		22	+#include <map>
		23	+#include <string>
		24	+#include <unordered_map>
		25	+
		26	+#include "analysis/csrc/application/database/db_constant.h"
		27	+
		28	+namespace Analysis
		29	+{
		30	+namespace Domain
		31	+{
		32	+using namespace Analysis::Application;
24	using namespace Analysis::Utils;	33	using namespace Analysis::Utils;
25		34
26	-OpStatisticProcessor::OpStatisticProcessor(const std::string& profPaths) : DataProcessor(profPaths)	35	+namespace
27	-{}
28	-OriOpCountDataFormat OpStatisticProcessor::LoadData(const DBInfo& dbInfo, const std::string& dbPath)
29	{	36	{
30	- OriOpCountDataFormat oriData;	37	+struct OpStatisticAggKey
31	- if (dbInfo.dbRunner == nullptr) {	38	+{
32	- ERROR("Create % connection failed.", dbPath);	39	+ uint16_t deviceId;
33	- return oriData;	40	+ std::string opType;
		41	+ std::string taskType;
		42	+
		43	+ bool operator<(const OpStatisticAggKey &other) const
		44	+ {
		45	+ if (deviceId != other.deviceId)
		46	+ {
		47	+ return deviceId < other.deviceId;
		48	+ }
		49	+ if (opType != other.opType)
		50	+ {
		51	+ return opType < other.opType;
		52	+ }
		53	+ return taskType < other.taskType;
34	}	54	}
35	- std::string sql{	55	+};
36	- "SELECT op_type, core_type, occurrences, round(total_time/1000.0, 3), round(min/1000.0, 3), "	56	+
37	- "round(avg/1000.0, 3), round(max/1000.0, 3), round(ratio, 3) FROM " + dbInfo.tableName +	57	+struct OpStatisticAgg
38	- " WHERE op_type != 'N/A' and core_type not in ('WRITE_BACK', 'INVALID')"	58	+{
39	- };	59	+ uint64_t count = 0;
40	- if (!dbInfo.dbRunner->QueryData(sql, oriData)) {	60	+ double totalTimeNs = 0.0;
41	- ERROR("Failed to obtain data from the % table.", dbInfo.tableName);	61	+ double minNs = std::numeric_limits<double>::infinity();
42	- }	62	+ double maxNs = 0.0;
43	- return oriData;	63	+};
		64	+
		65	+using OpStatisticAggMap = std::map<OpStatisticAggKey, OpStatisticAgg>;
		66	+using DeviceTotalNsMap = std::unordered_map<uint16_t, double>;
		67	+
		68	+// Match the Python pipeline: op_report excludes communication tasks before aggregation, while N/A,
		69	+// WRITE_BACK and INVALID remain in the ratio denominator and are filtered only from op_statistic.csv.
		70	+bool isOpReportRequired(const AssociatedTaskData &associatedTask)
		71	+{
		72	+ return associatedTask.taskInfo->taskType != TASK_TYPE_COMMUNICATION &&
		73	+ associatedTask.taskInfo->taskType != TASK_TYPE_HCCL_AI_CPU;
44	}	74	}
45		75
46	-std::vector<OpStatisticData> OpStatisticProcessor::FormatData(const OriOpCountDataFormat& oriData,	76	+bool isOpStatisticRequired(const OpStatisticAggKey &key)
47	- const uint16_t deviceId)
48	{	77	{
49	- std::vector<OpStatisticData> processedData;	78	+ return key.opType != NA && key.taskType != TASK_TYPE_WRITE_BACK && key.taskType != TASK_TYPE_INVALID;
50	- OpStatisticData data;
51	- if (!Reserve(processedData, oriData.size())) {
52	- ERROR("Reserve for Op Statistic data failed.");
53	- return processedData;
54	- }
55	- data.deviceId = deviceId;
56	- for (auto& row : oriData) {
57	- std::tie(data.opType, data.coreType, data.count, data.totalTime, data.min, data.avg, data.max,
58	- data.ratio) = row;
59	- processedData.push_back(data);
60	- }
61	- return processedData;
62	}	79	}
63		80
64	-bool OpStatisticProcessor::Process(Analysis::Infra::DataInventory& dataInventory)	81	+double getTaskDurationNs(const AscendTaskData &ascendTask)
65	{	82	{
66	- bool flag = true;	83	+ return static_cast<double>(ascendTask.end - ascendTask.timestamp);
67	- std::vector<OpStatisticData> res;	84	+}
68	- auto deviceList = File::GetFilesWithPrefix(profPath_, DEVICE_PREFIX);	85	+
69	- for (const auto& devicePath : deviceList) {	86	+void aggregate(const AssociatedTaskCollection &associatedTasks, OpStatisticAggMap &aggregatedData,
70	- DBInfo opCounterDB("op_counter.db", "op_report");	87	+ DeviceTotalNsMap &deviceTotalNs)
71	- std::string dbPath = File::PathJoin({devicePath, SQLITE, opCounterDB.dbName});	88	+{
72	- auto deviceId = Utils::GetDeviceIdByDevicePath(devicePath);	89	+ for (const auto &associatedTask : associatedTasks.records)
73	- if (deviceId == INVALID_DEVICE_ID) {	90	+ {
74	- ERROR("the invalid deviceId cannot to be identified, profPath is %", profPath_);	91	+ if (!isOpReportRequired(associatedTask))
75	- return false;	92	+ {
76	- }
77	- if (!opCounterDB.ConstructDBRunner(dbPath)) {
78	- flag = false;
79	continue;	93	continue;
80	}	94	}
81	- auto status = CheckPathAndTable(dbPath, opCounterDB, false);	95	+ const auto &taskInfo = *associatedTask.taskInfo;
82	- if (status != CHECK_SUCCESS) {	96	+ const auto &ascendTask = *associatedTask.ascendTask;
83	- if (status == CHECK_FAILED) {	97	+ const double durationNs = getTaskDurationNs(ascendTask);
84	- flag = false;	98	+ OpStatisticAggKey key{ascendTask.deviceId, taskInfo.opType, taskInfo.taskType};
		C C chenhao_12097 天前【review】【正确性】聚合 key 字段名 coreType 与实际取值 taskType 不符问题描述：OpStatisticAggKey 构造处将 taskInfo.taskType 赋给名为 coreType 的字段（struct 定义第 38 行），聚合 key 层面字段名与语义错位问题原因：该命名沿袭了存量 OpStatisticData.coreType（对应 op_counter.db 的 core_type 列，其值即 task_type），但对新维护者而言，key 层直接叫 coreType 容易误以为取自其它 coreType 语义字段，排查口径时需要跨两层追溯修改建议：聚合 key 字段改名为 taskType 以准确表达取值来源，输出侧（formatData）再映射到存量 OpStatisticData.coreType，保持 CSV 列名不变 C chenqm 4 天前评论： 4 天前评论：
85	- }	99	+ OpStatisticAgg &aggregated = aggregatedData[key];
86	- continue;	100	+ ++aggregated.count;
87	- }	101	+ aggregated.totalTimeNs += durationNs;
88	- auto oriData = LoadData(opCounterDB, dbPath);	102	+ aggregated.minNs = std::min(aggregated.minNs, durationNs);
89	- if (oriData.empty()) {	103	+ aggregated.maxNs = std::max(aggregated.maxNs, durationNs);
90	- WARN("Op Statistics original data has no valid type data. DBPath is %", dbPath);	104	+ deviceTotalNs[ascendTask.deviceId] += durationNs;
91	- continue;
92	- }
93	- auto formatData = FormatData(oriData, deviceId);
94	- if (formatData.empty()) {
95	- ERROR("Op Statistics data format failed, DBPath is %", dbPath);
96	- flag = false;
97	- continue;
98	- }
99	- res.insert(res.end(), formatData.begin(), formatData.end());
100	}	105	}
101	- if (!SaveToDataInventory<OpStatisticData>(std::move(res), dataInventory, PROCESSOR_NAME_OP_STATISTIC)) {	106	+}
		107	+
		108	+double getDeviceTotalNs(const DeviceTotalNsMap &deviceTotalNs, uint16_t deviceId)
		109	+{
		110	+ auto total = deviceTotalNs.find(deviceId);
		111	+ return total == deviceTotalNs.end() ? 0.0 : total->second;
		112	+}
		113	+
		114	+bool formatData(const OpStatisticAggMap &aggregatedData, const DeviceTotalNsMap &deviceTotalNs,
		115	+ std::vector<OpStatisticData> &result)
		116	+{
		117	+ if (!Reserve(result, aggregatedData.size()))
		118	+ {
		119	+ ERROR("Reserve for op statistic data failed.");
		120	+ return false;
		121	+ }
		122	+ const double nsToUs = static_cast<double>(NS_TO_US);
		123	+ for (const auto &item : aggregatedData)
		124	+ {
		125	+ const OpStatisticAggKey &key = item.first;
		126	+ const OpStatisticAgg &aggregated = item.second;
		127	+ if (!isOpStatisticRequired(key))
		128	+ {
		129	+ continue;
		130	+ }
		131	+
		132	+ const double totalNs = getDeviceTotalNs(deviceTotalNs, key.deviceId);
		133	+ const double ratio = totalNs > 0.0 ? aggregated.totalTimeNs * static_cast<double>(PERCENTAGE) / totalNs : 0.0;
		134	+
		135	+ OpStatisticData data;
		136	+ data.deviceId = key.deviceId;
		137	+ data.opType = key.opType;
		138	+ data.coreType = key.taskType;
		139	+ data.count = std::to_string(aggregated.count);
		140	+ data.totalTime = RoundToDecimalPlaces(aggregated.totalTimeNs / nsToUs);
		141	+ data.min = RoundToDecimalPlaces(aggregated.minNs / nsToUs);
		142	+ data.avg = aggregated.count == 0
		143	+ ? 0.0
		144	+ : RoundToDecimalPlaces(aggregated.totalTimeNs / static_cast<double>(aggregated.count) / nsToUs);
		145	+ data.max = RoundToDecimalPlaces(aggregated.maxNs / nsToUs);
		146	+ data.ratio = RoundToDecimalPlaces(ratio);
		147	+ result.push_back(std::move(data));
		148	+ }
		149	+ return true;
		150	+}
		151	+} // namespace
		152	+
		153	+OpStatisticProcessor::OpStatisticProcessor(const std::string &profPaths) : DataProcessor(profPaths) {}
		154	+
		155	+bool OpStatisticProcessor::Process(DataInventory &dataInventory)
		156	+{
		157	+ auto associatedTasks = dataInventory.GetPtr<AssociatedTaskCollection>();
		158	+ if (associatedTasks == nullptr)
		159	+ {
		160	+ WARN("Op Statistic source data not exist.");
		161	+ return true;
		162	+ }
		163	+
		164	+ OpStatisticAggMap aggregatedData;
		165	+ DeviceTotalNsMap deviceTotalNs;
		166	+ aggregate(*associatedTasks, aggregatedData, deviceTotalNs);
		167	+ std::vector<OpStatisticData> result;
		168	+ if (!formatData(aggregatedData, deviceTotalNs, result))
		169	+ {
		170	+ return false;
		171	+ }
		172	+ if (!SaveToDataInventory<OpStatisticData>(std::move(result), dataInventory, PROCESSOR_NAME_OP_STATISTIC))
		173	+ {
102	ERROR("Save data failed, %.", PROCESSOR_NAME_OP_STATISTIC);	174	ERROR("Save data failed, %.", PROCESSOR_NAME_OP_STATISTIC);
103	- flag = false;	175	+ return false;
104	}	176	}
105	- return flag;	177	+ return true;
106	}	178	}
107	-}	179	+} // namespace Domain
108	-}	180	+} // namespace Analysis

Manalysis/csrc/domain/data_process/ai_task/op_statistic_processor.h+16-18

	@@ -18,25 +18,23 @@
18	#define ANALYSIS_DOMAIN_OP_STATISTIC_PROCESSOR_H	18	#define ANALYSIS_DOMAIN_OP_STATISTIC_PROCESSOR_H
19		19
20	#include "analysis/csrc/domain/data_process/data_processor.h"	20	#include "analysis/csrc/domain/data_process/data_processor.h"
		21	+#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/associated_task_data.h"
21	#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/op_statistic_data.h"	22	#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/op_statistic_data.h"
22		23
23	-	24	+namespace Analysis
24	-namespace Analysis {	25	+{
25	-namespace Domain {	26	+namespace Domain
26	-// op_type, core_type, occurrences, total_time, min, avg, max, ratio	27	+{
27	-using OriOpCountDataFormat = std::vector<std::tuple<std::string, std::string, std::string,	28	+class OpStatisticProcessor : public DataProcessor
28	- double, double, double, double, double>>;	29	+{
29	-	30	+ public:
30	-class OpStatisticProcessor : public DataProcessor {
31	-public:
32	OpStatisticProcessor() = default;	31	OpStatisticProcessor() = default;
33	- explicit OpStatisticProcessor(const std::string& profPaths);	32	+ explicit OpStatisticProcessor(const std::string &profPaths);
34	-private:
35	- bool Process(DataInventory& dataInventory) override;
36	- OriOpCountDataFormat LoadData(const DBInfo& dbInfo, const std::string& dbPath);
37	- std::vector<OpStatisticData> FormatData(const OriOpCountDataFormat& oriData, const uint16_t deviceId);
38	-};
39	-} // namespace Domain
40	-} // namespace Analysis
41		33
42	-#endif // ANALYSIS_DOMAIN_OP_STATISTIC_PROCESSOR_H	34	+ private:
		35	+ bool Process(DataInventory &dataInventory) override;
		36	+};
		37	+} // namespace Domain
		38	+} // namespace Analysis
		39	+
		40	+#endif // ANALYSIS_DOMAIN_OP_STATISTIC_PROCESSOR_H

Aanalysis/csrc/domain/data_process/ai_task/task_association_processor.cpp+119-0

@@ -0,0 +1,119 @@
	1	+/* -------------------------------------------------------------------------
	2	+ * Copyright (c) 2026 Huawei Technologies Co., Ltd.
	3	+ * This file is part of the MindStudio project.
	4	+ *
	5	+ * MindStudio is licensed under Mulan PSL v2.
	6	+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
	7	+ * You may obtain a copy of Mulan PSL v2 at:
	8	+ *
	9	+ * http://license.coscl.org.cn/MulanPSL2
	10	+ *
	11	+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
	12	+ * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
	13	+ * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
	14	+ * See the Mulan PSL v2 for more details.
	15	+ * -------------------------------------------------------------------------*/
	16	+
	17	+#include "analysis/csrc/domain/data_process/ai_task/task_association_processor.h"
	18	+
	19	+#include <unordered_map>
	20	+
	21	+#include "analysis/csrc/application/database/db_constant.h"
	22	+#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/ascend_task_data.h"
	23	+#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/associated_task_data.h"
	24	+#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/task_info_data.h"
	25	+#include "analysis/csrc/domain/valueobject/include/task_id.h"
	26	+#include "analysis/csrc/infrastructure/utils/utils.h"
	27	+
	28	+namespace Analysis
	29	+{
	30	+namespace Domain
	31	+{
	32	+using namespace Analysis::Application;
	33	+using namespace Analysis::Utils;
	34	+
	35	+namespace
	36	+{
	37	+template <typename TaskData>
	38	+TaskId makeTaskId(const TaskData &taskData)
	39	+{
	40	+ return TaskId{taskData.streamId, taskData.batchId, taskData.taskId, taskData.contextId, taskData.deviceId};
	41	+}
	42	+
	43	+bool isOpSummaryRequired(const TaskInfoData &taskInfo)
	44	+{
	45	+ const bool isExcludedTaskType =
	46	+ taskInfo.taskType == TASK_TYPE_WRITE_BACK \|\| taskInfo.taskType == TASK_TYPE_INVALID \|\|
	47	+ taskInfo.taskType == TASK_TYPE_HCCL_AI_CPU \|\| taskInfo.taskType == TASK_TYPE_COMMUNICATION;
	48	+ const bool isCommunicationAivKernel =
	49	+ taskInfo.taskType == TASK_TYPE_COMMUNICATION && EndsWith(taskInfo.opName, AIV_KERNEL);
	50	+ return !isExcludedTaskType \|\| isCommunicationAivKernel;
	51	+}
	52	+
	53	+bool associateTasks(const std::vector<TaskInfoData> &taskInfoData, const std::vector<AscendTaskData> &ascendTaskData,
	54	+ std::vector<AssociatedTaskData> &associatedTasks)
	55	+{
	56	+ std::unordered_map<TaskId, const TaskInfoData *, IDHasher> taskInfoIndex;
	57	+ for (const auto &taskInfo : taskInfoData)
	58	+ {
	59	+ taskInfoIndex[makeTaskId(taskInfo)] = &taskInfo;
	60	+ }
	61	+
	62	+ if (!Reserve(associatedTasks, ascendTaskData.size()))
	63	+ {
	64	+ ERROR("Reserve for associated task data failed.");
	65	+ return false;
	66	+ }
	67	+ size_t invalidTimeRangeCount = 0;
	68	+ for (const auto &ascendTask : ascendTaskData)
	69	+ {
	70	+ auto taskInfo = taskInfoIndex.find(makeTaskId(ascendTask));
	71	+ if (taskInfo != taskInfoIndex.end())
	72	+ {
	73	+ // Validate once before sharing to prevent unsigned duration subtraction in every consumer.
	74	+ if (ascendTask.end < ascendTask.timestamp)
	75	+ {
	76	+ ++invalidTimeRangeCount;
	77	+ continue;
	78	+ }
	79	+ associatedTasks.push_back(
	C C chenhao_12097 天前【review】【性能】AssociatedTaskData 全量值拷贝放大内存峰值问题描述：关联结果通过值拷贝构造（拷贝整个 TaskInfoData + AscendTaskData），TaskInfoData 含 16 个 std::string（其中 6 个 shape/format 串在真实模型数据中可达数百字节）问题原因：百万级任务场景下，关联结果与源数据并存导致内存峰值约放大 2-3 倍；而 taskInfoIndex 已持有源数据指针、DataInventory 持有 shared_ptr 保证生命周期，拷贝并非必要修改建议：AssociatedTaskData 改存 const TaskInfoData* / const AscendTaskData*（或索引对），消费侧字段访问同步调整；若评估内存峰值可接受，可保持现状但在注释中说明取舍 C chenqm 4 天前评论： 4 天前评论：
	80	+ AssociatedTaskData{taskInfo->second, &ascendTask, isOpSummaryRequired(*taskInfo->second)});
	81	+ }
	82	+ }
	83	+ if (invalidTimeRangeCount > 0)
	84	+ {
	85	+ WARN("Task association skipped % matched records with reversed time range.", invalidTimeRangeCount);
	86	+ }
	87	+ return true;
	88	+}
	89	+} // namespace
	90	+
	91	+TaskAssociationProcessor::TaskAssociationProcessor(const std::string &profPaths) : DataProcessor(profPaths) {}
	92	+
	93	+bool TaskAssociationProcessor::Process(DataInventory &dataInventory)
	94	+{
	95	+ auto taskInfoData = dataInventory.GetPtr<std::vector<TaskInfoData>>();
	96	+ auto ascendTaskData = dataInventory.GetPtr<std::vector<AscendTaskData>>();
	97	+ if (taskInfoData == nullptr \|\| ascendTaskData == nullptr)
	98	+ {
	99	+ WARN("Task association source data not exist.");
	100	+ return true;
	101	+ }
	102	+ std::shared_ptr<AssociatedTaskCollection> sharedData;
	103	+ MAKE_SHARED_RETURN_VALUE(sharedData, AssociatedTaskCollection, false);
	104	+ sharedData->taskInfoData = taskInfoData;
	105	+ sharedData->ascendTaskData = ascendTaskData;
	106	+ if (!associateTasks(sharedData->taskInfoData, sharedData->ascendTaskData, sharedData->records))
	107	+ {
	108	+ return false;
	109	+ }
	110	+ INFO("Task association matched % records.", sharedData->records.size());
	111	+ if (!dataInventory.Inject(sharedData))
	112	+ {
	113	+ ERROR("Save data failed, %.", PROCESSOR_NAME_TASK_ASSOCIATION);
	114	+ return false;
	115	+ }
	116	+ return true;
	117	+}
	118	+} // namespace Domain
	119	+} // namespace Analysis

Aanalysis/csrc/domain/data_process/ai_task/task_association_processor.h+38-0

@@ -0,0 +1,38 @@
	1	+/* -------------------------------------------------------------------------
	2	+ * Copyright (c) 2026 Huawei Technologies Co., Ltd.
	3	+ * This file is part of the MindStudio project.
	4	+ *
	5	+ * MindStudio is licensed under Mulan PSL v2.
	6	+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
	7	+ * You may obtain a copy of Mulan PSL v2 at:
	8	+ *
	9	+ * http://license.coscl.org.cn/MulanPSL2
	10	+ *
	11	+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
	12	+ * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
	13	+ * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
	14	+ * See the Mulan PSL v2 for more details.
	15	+ * -------------------------------------------------------------------------*/
	16	+
	17	+#ifndef ANALYSIS_DOMAIN_TASK_ASSOCIATION_PROCESSOR_H
	18	+#define ANALYSIS_DOMAIN_TASK_ASSOCIATION_PROCESSOR_H
	19	+
	20	+#include "analysis/csrc/domain/data_process/data_processor.h"
	21	+
	22	+namespace Analysis
	23	+{
	24	+namespace Domain
	25	+{
	26	+class TaskAssociationProcessor : public DataProcessor
	27	+{
	28	+ public:
	29	+ TaskAssociationProcessor() = default;
	30	+ explicit TaskAssociationProcessor(const std::string &profPaths);
	31	+
	32	+ private:
	33	+ bool Process(DataInventory &dataInventory) override;
	34	+};
	35	+} // namespace Domain
	36	+} // namespace Analysis
	37	+
	38	+#endif // ANALYSIS_DOMAIN_TASK_ASSOCIATION_PROCESSOR_H

Manalysis/csrc/domain/data_process/data_processor_topology_register.cpp+8-1

	@@ -35,6 +35,7 @@
35	#include "analysis/csrc/domain/data_process/ai_task/op_statistic_processor.h"	35	#include "analysis/csrc/domain/data_process/ai_task/op_statistic_processor.h"
36	#include "analysis/csrc/domain/data_process/ai_task/overlap_analysis_processor.h"	36	#include "analysis/csrc/domain/data_process/ai_task/overlap_analysis_processor.h"
37	#include "analysis/csrc/domain/data_process/ai_task/step_trace_processor.h"	37	#include "analysis/csrc/domain/data_process/ai_task/step_trace_processor.h"
		38	+#include "analysis/csrc/domain/data_process/ai_task/task_association_processor.h"
38	#include "analysis/csrc/domain/data_process/ai_task/task_processor.h"	39	#include "analysis/csrc/domain/data_process/ai_task/task_processor.h"
39	#include "analysis/csrc/domain/data_process/ai_task/unified_pmu_processor.h"	40	#include "analysis/csrc/domain/data_process/ai_task/unified_pmu_processor.h"
40	#include "analysis/csrc/domain/data_process/include/data_processor_factory.h"	41	#include "analysis/csrc/domain/data_process/include/data_processor_factory.h"
	@@ -126,7 +127,13 @@ REGISTER_PROCESSOR(FusionOpProcessor, PROCESSOR_NAME_FUSION_OP, TOPO_DEPS());
126	REGISTER_PROCESSOR(FusionTaskProcessor, PROCESSOR_NAME_FUSION_TASK, TOPO_DEPS());	127	REGISTER_PROCESSOR(FusionTaskProcessor, PROCESSOR_NAME_FUSION_TASK, TOPO_DEPS());
127	REGISTER_PROCESSOR(ModelNameProcessor, PROCESSOR_NAME_MODEL_NAME, TOPO_DEPS());	128	REGISTER_PROCESSOR(ModelNameProcessor, PROCESSOR_NAME_MODEL_NAME, TOPO_DEPS());
128	REGISTER_PROCESSOR(HcclStatisticProcessor, PROCESSOR_NAME_COMM_STATISTIC, TOPO_DEPS());	129	REGISTER_PROCESSOR(HcclStatisticProcessor, PROCESSOR_NAME_COMM_STATISTIC, TOPO_DEPS());
129	-REGISTER_PROCESSOR(OpStatisticProcessor, PROCESSOR_NAME_OP_STATISTIC, TOPO_DEPS());	130	+REGISTER_PROCESSOR_WITH_DATA(TaskAssociationProcessor, PROCESSOR_NAME_TASK_ASSOCIATION,
		W W wangzixuan7 天前 [review] 没有看到op_summary相关类的依赖修改，请确认相关逻辑是否同步适配。当前已新增聚合类，op_summary侧的数据依赖可迁移至当前新增类，避免数据的冗余依赖和数据处理。 C chenqm 4 天前评论： 4 天前评论：
		131	+ TOPO_DEPS(TOPO_NODE(DATA_PROCESSING, PROCESSOR_NAME_COMPUTE_TASK_INFO),
		132	+ TOPO_NODE(DATA_PROCESSING, PROCESSOR_NAME_TASK)),
		133	+ std::vector<TaskInfoData>, std::vector<AscendTaskData>);
		134	+REGISTER_PROCESSOR_WITH_DATA(OpStatisticProcessor, PROCESSOR_NAME_OP_STATISTIC,
		135	+ TOPO_DEPS(TOPO_NODE(DATA_PROCESSING, PROCESSOR_NAME_TASK_ASSOCIATION)),
		136	+ AssociatedTaskCollection);
130	REGISTER_PROCESSOR(HostTaskProcessor, PROCESSOR_HOST_TASK, TOPO_DEPS());	137	REGISTER_PROCESSOR(HostTaskProcessor, PROCESSOR_HOST_TASK, TOPO_DEPS());
131	REGISTER_PROCESSOR_WITH_DATA(OverlapAnalysisProcessor, PROCESSOR_NAME_OVERLAP_ANALYSIS,	138	REGISTER_PROCESSOR_WITH_DATA(OverlapAnalysisProcessor, PROCESSOR_NAME_OVERLAP_ANALYSIS,
132	TOPO_DEPS(TOPO_NODE(DATA_PROCESSING, PROCESSOR_NAME_TASK),	139	TOPO_DEPS(TOPO_NODE(DATA_PROCESSING, PROCESSOR_NAME_TASK),

Aanalysis/csrc/domain/entities/viewer_data/ai_task/include/associated_task_data.h+55-0

@@ -0,0 +1,55 @@
	1	+/* -------------------------------------------------------------------------
	2	+ * Copyright (c) 2026 Huawei Technologies Co., Ltd.
	3	+ * This file is part of the MindStudio project.
	4	+ *
	5	+ * MindStudio is licensed under Mulan PSL v2.
	6	+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
	7	+ * You may obtain a copy of Mulan PSL v2 at:
	8	+ *
	9	+ * http://license.coscl.org.cn/MulanPSL2
	10	+ *
	11	+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
	12	+ * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
	13	+ * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
	14	+ * See the Mulan PSL v2 for more details.
	15	+ * -------------------------------------------------------------------------*/
	16	+
	17	+#ifndef ANALYSIS_DOMAIN_ASSOCIATED_TASK_DATA_H
	18	+#define ANALYSIS_DOMAIN_ASSOCIATED_TASK_DATA_H
	19	+
	20	+#include <memory>
	21	+#include <vector>
	22	+
	23	+#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/ascend_task_data.h"
	24	+#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/task_info_data.h"
	25	+
	26	+namespace Analysis
	27	+{
	28	+namespace Domain
	29	+{
	30	+struct AssociatedTaskData
	31	+{
	32	+ AssociatedTaskData() = default;
	33	+ AssociatedTaskData(const TaskInfoData taskInfoData, const AscendTaskData ascendTaskData,
	34	+ bool isOpSummaryRequired = true)
	35	+ : taskInfo(taskInfoData), ascendTask(ascendTaskData), opSummaryRequired(isOpSummaryRequired)
	36	+ {
	37	+ }
	38	+
	39	+ const TaskInfoData *taskInfo = nullptr;
	40	+ const AscendTaskData *ascendTask = nullptr;
	41	+ // This flag controls only the op_summary view; other consumers still receive every matched task.
	42	+ bool opSummaryRequired = true;
	43	+};
	44	+
	45	+struct AssociatedTaskCollection
	46	+{
	47	+ // Keep source vectors alive after DataInventory releases their standalone entries.
	48	+ std::shared_ptr<const std::vector<TaskInfoData>> taskInfoData;
	49	+ std::shared_ptr<const std::vector<AscendTaskData>> ascendTaskData;
	50	+ std::vector<AssociatedTaskData> records;
	51	+};
	52	+} // namespace Domain
	53	+} // namespace Analysis
	54	+
	55	+#endif // ANALYSIS_DOMAIN_ASSOCIATED_TASK_DATA_H

Mtest/msprof_cpp/analysis_ut/application/assemble/test/op_summary_assembler_utest.cpp+27-16

	@@ -20,6 +20,7 @@
20	#include "analysis/csrc/application/summary/op_summary_assembler.h"	20	#include "analysis/csrc/application/summary/op_summary_assembler.h"
21	#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/ascend_task_data.h"	21	#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/ascend_task_data.h"
22	#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/communication_info_data.h"	22	#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/communication_info_data.h"
		23	+#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/associated_task_data.h"
23	#include "analysis/csrc/application/database/db_constant.h"	24	#include "analysis/csrc/application/database/db_constant.h"
24	#include "analysis/csrc/application/summary/summary_constant.h"	25	#include "analysis/csrc/application/summary/summary_constant.h"
25	#include "analysis/csrc/domain/services/environment/context.h"	26	#include "analysis/csrc/domain/services/environment/context.h"
	@@ -143,6 +144,24 @@ static std::vector<CommunicationOpData> GenerateOpData()
143	return res;	144	return res;
144	}	145	}
145		146
		147	+static AssociatedTaskCollection generateAssociatedTaskData()
		148	+{
		149	+ std::shared_ptr<std::vector<TaskInfoData>> taskInfoData;
		150	+ std::shared_ptr<std::vector<AscendTaskData>> ascendTaskData;
		151	+ MAKE_SHARED_NO_OPERATION(taskInfoData, std::vector<TaskInfoData>, GenerateTaskInfoData());
		152	+ MAKE_SHARED_NO_OPERATION(ascendTaskData, std::vector<AscendTaskData>, GenerateTaskData());
		153	+
		154	+ AssociatedTaskCollection associatedTasks;
		155	+ associatedTasks.taskInfoData = taskInfoData;
		156	+ associatedTasks.ascendTaskData = ascendTaskData;
		157	+ associatedTasks.records = {{&associatedTasks.taskInfoData->at(0), &associatedTasks.ascendTaskData->at(0), true},
		158	+ {&associatedTasks.taskInfoData->at(0), &associatedTasks.ascendTaskData->at(1), true},
		159	+ {&associatedTasks.taskInfoData->at(1), &associatedTasks.ascendTaskData->at(2), true},
		160	+ {&associatedTasks.taskInfoData->at(2), &associatedTasks.ascendTaskData->at(3), true},
		161	+ {&associatedTasks.taskInfoData->at(0), &associatedTasks.ascendTaskData->at(0), false}};
		162	+ return associatedTasks;
		163	+}
		164	+
146	static MetricSummary GenerateMetricSummary()	165	static MetricSummary GenerateMetricSummary()
147	{	166	{
148	MetricSummary summary;	167	MetricSummary summary;
	@@ -182,17 +201,13 @@ TEST_F(OpSummaryAssemblerUTest, ShouldReturnTrueWhenDataNotExist)
182	TEST_F(OpSummaryAssemblerUTest, ShouldReturnTrueWhenTaskAndHcclExistWithNoStars)	201	TEST_F(OpSummaryAssemblerUTest, ShouldReturnTrueWhenTaskAndHcclExistWithNoStars)
183	{	202	{
184	DataInventory dataInventory;	203	DataInventory dataInventory;
185	- std::shared_ptr<std::vector<AscendTaskData>> taskS;	204	+ std::shared_ptr<AssociatedTaskCollection> associatedTasksPtr;
186	- std::shared_ptr<std::vector<TaskInfoData>> infoS;
187	std::shared_ptr<std::vector<CommunicationOpData>> opDataS;	205	std::shared_ptr<std::vector<CommunicationOpData>> opDataS;
188	- auto task = GenerateTaskData();	206	+ auto associatedTasks = generateAssociatedTaskData();
189	- auto info = GenerateTaskInfoData();
190	auto opData = GenerateOpData();	207	auto opData = GenerateOpData();
191	- MAKE_SHARED_NO_OPERATION(taskS, std::vector<AscendTaskData>, task);	208	+ MAKE_SHARED_NO_OPERATION(associatedTasksPtr, AssociatedTaskCollection, associatedTasks);
192	- MAKE_SHARED_NO_OPERATION(infoS, std::vector<TaskInfoData>, info);
193	MAKE_SHARED_NO_OPERATION(opDataS, std::vector<CommunicationOpData>, opData);	209	MAKE_SHARED_NO_OPERATION(opDataS, std::vector<CommunicationOpData>, opData);
194	- dataInventory.Inject(taskS);	210	+ dataInventory.Inject(associatedTasksPtr);
195	- dataInventory.Inject(infoS);
196	dataInventory.Inject(opDataS);	211	dataInventory.Inject(opDataS);
197	OpSummaryAssembler assembler(PROCESSOR_OP_SUMMARY, PROF_PATH);	212	OpSummaryAssembler assembler(PROCESSOR_OP_SUMMARY, PROF_PATH);
198	EXPECT_TRUE(assembler.Run(dataInventory));	213	EXPECT_TRUE(assembler.Run(dataInventory));
	@@ -235,20 +250,16 @@ TEST_F(OpSummaryAssemblerUTest, ShouldReturnTrueWhenOnlyHcclWithStars)
235	TEST_F(OpSummaryAssemblerUTest, ShouldReturnTrueWhenTaskAndHcclAndPmuExistWithStars)	250	TEST_F(OpSummaryAssemblerUTest, ShouldReturnTrueWhenTaskAndHcclAndPmuExistWithStars)
236	{	251	{
237	DataInventory dataInventory;	252	DataInventory dataInventory;
238	- std::shared_ptr<std::vector<AscendTaskData>> taskS;	253	+ std::shared_ptr<AssociatedTaskCollection> associatedTasksPtr;
239	- std::shared_ptr<std::vector<TaskInfoData>> infoS;
240	std::shared_ptr<std::vector<CommunicationOpData>> opDataS;	254	std::shared_ptr<std::vector<CommunicationOpData>> opDataS;
241	std::shared_ptr<MetricSummary> metricDataS;	255	std::shared_ptr<MetricSummary> metricDataS;
242	- auto task = GenerateTaskData();	256	+ auto associatedTasks = generateAssociatedTaskData();
243	- auto info = GenerateTaskInfoData();
244	auto opData = GenerateOpData();	257	auto opData = GenerateOpData();
245	auto metricSummary = GenerateMetricSummary();	258	auto metricSummary = GenerateMetricSummary();
246	- MAKE_SHARED_NO_OPERATION(taskS, std::vector<AscendTaskData>, task);	259	+ MAKE_SHARED_NO_OPERATION(associatedTasksPtr, AssociatedTaskCollection, associatedTasks);
247	- MAKE_SHARED_NO_OPERATION(infoS, std::vector<TaskInfoData>, info);
248	MAKE_SHARED_NO_OPERATION(opDataS, std::vector<CommunicationOpData>, opData);	260	MAKE_SHARED_NO_OPERATION(opDataS, std::vector<CommunicationOpData>, opData);
249	MAKE_SHARED_NO_OPERATION(metricDataS, MetricSummary, metricSummary);	261	MAKE_SHARED_NO_OPERATION(metricDataS, MetricSummary, metricSummary);
250	- dataInventory.Inject(taskS);	262	+ dataInventory.Inject(associatedTasksPtr);
251	- dataInventory.Inject(infoS);
252	dataInventory.Inject(opDataS);	263	dataInventory.Inject(opDataS);
253	dataInventory.Inject(metricDataS);	264	dataInventory.Inject(metricDataS);
254	OpSummaryAssembler assembler(PROCESSOR_OP_SUMMARY, PROF_PATH);	265	OpSummaryAssembler assembler(PROCESSOR_OP_SUMMARY, PROF_PATH);

Mtest/msprof_cpp/analysis_ut/domain/data_process/test/op_statistic_processor_utest.cpp+192-118

	@@ -1,4 +1,4 @@
1	-/* -------------------------------------------------------------------------	1	+/* -------------------------------------------------------------------------
2	* Copyright (c) 2025 Huawei Technologies Co., Ltd.	2	* Copyright (c) 2025 Huawei Technologies Co., Ltd.
3	* This file is part of the MindStudio project.	3	* This file is part of the MindStudio project.
4	*	4	*
	@@ -15,149 +15,223 @@
15	* -------------------------------------------------------------------------*/	15	* -------------------------------------------------------------------------*/
16		16
17	#include "gtest/gtest.h"	17	#include "gtest/gtest.h"
18	-#include "mockcpp/mockcpp.hpp"	18	+
19	-#include "analysis/csrc/domain//data_process/ai_task/op_statistic_processor.h"
20	-#include "analysis/csrc/infrastructure/dfx/error_code.h"
21	-#include "analysis/csrc/domain/services/environment/context.h"
22	#include "analysis/csrc/application/database/db_constant.h"	19	#include "analysis/csrc/application/database/db_constant.h"
23	-#include "reserve_mock_utils.h"	20	+#include "analysis/csrc/domain/data_process/ai_task/op_statistic_processor.h"
		21	+#include "analysis/csrc/domain/data_process/ai_task/task_association_processor.h"
		22	+#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/associated_task_data.h"
		23	+#include "analysis/csrc/infrastructure/utils/utils.h"
24		24
25	-using namespace Analysis::Domain;
26	-using namespace Domain::Environment;
27	-using namespace Analysis::Utils;
28	using namespace Analysis::Application;	25	using namespace Analysis::Application;
29	-using namespace Analysis::Test;	26	+using namespace Analysis::Domain;
30	-using ProcessedFormat = std::vector<OpStatisticData>;	27	+using namespace Analysis::Infra;
		28	+using namespace Analysis::Utils;
31		29
32	-const std::string BASE_PATH = "./op_statistic";	30	+namespace
33	-const std::string DEVICE_SUFFIX = "device_0";	31	+{
34	-const std::string SQLITE_SUFFIX = "sqlite";	32	+const std::string PROF_PATH = "./op_statistic_processor_test/PROF_0";
35	-const std::string PROF_PATH_A = File::PathJoin({BASE_PATH, "PROF_0"});
36	-const std::string PROF_PATH_B = File::PathJoin({BASE_PATH, "PROF_1"});
37	-const std::string DB_SUFFIX = "op_counter.db";
38	-const std::string TABLE_NAME = "op_report";
39	-const std::set<std::string> PROF_PATHS = {PROF_PATH_A, PROF_PATH_B};
40		33
41	-const OriOpCountDataFormat OP_DATA = {	34	+TaskInfoData makeTaskInfo(uint16_t deviceId, uint32_t streamId, uint32_t taskId, const std::string &opType,
42	- {"RmsNormTactic", "AI_CORE", "1610", 20218924.375, 3900.125, 12558.338121, 93541.875, 4.833029},	35	+ const std::string &taskType, uint32_t batchId = 0, uint32_t contextId = 0,
43	- {"PagedAttentionMaskNdKernel", "MIX_AIC", "720", 18795054.5, 23860.5, 26104.242361, 32020.75, 4.492674},	36	+ const std::string &opName = "")
44	- {"AddBF16Tactic", "AI_CORE", "1600", 11507650.75, 2260, 7192.281719, 51141, 2.75073},	37	+{
45	- {"UnpadFlashAttentionBF16NdKernel", "MIX_AIC", "80", 11179324, 137662.75, 139741.55, 147122.875, 2.672249}	38	+ TaskInfoData data;
		39	+ data.deviceId = deviceId;
		40	+ data.streamId = streamId;
		41	+ data.taskId = taskId;
		42	+ data.batchId = batchId;
		43	+ data.contextId = contextId;
		44	+ data.opType = opType;
		45	+ data.taskType = taskType;
		46	+ data.opName = opName;
		47	+ return data;
		48	+}
		49	+
		50	+AscendTaskData makeAscendTask(uint16_t deviceId, uint32_t streamId, uint32_t taskId, uint64_t timestamp,
		51	+ uint64_t end, uint32_t batchId = 0, uint32_t contextId = 0)
		52	+{
		53	+ AscendTaskData data;
		54	+ data.deviceId = deviceId;
		55	+ data.streamId = streamId;
		56	+ data.taskId = taskId;
		57	+ data.batchId = batchId;
		58	+ data.contextId = contextId;
		59	+ data.timestamp = timestamp;
		60	+ data.end = end;
		61	+ data.duration = 1.0;
		62	+ return data;
		63	+}
		64	+
		65	+struct AssociatedTaskInput
		66	+{
		67	+ AssociatedTaskInput(TaskInfoData taskInfoData, AscendTaskData ascendTaskData, bool isOpSummaryRequired = true)
		68	+ : taskInfo(std::move(taskInfoData)), ascendTask(std::move(ascendTaskData)),
		69	+ opSummaryRequired(isOpSummaryRequired)
		70	+ {
		71	+ }
		72	+
		73	+ TaskInfoData taskInfo;
		74	+ AscendTaskData ascendTask;
		75	+ bool opSummaryRequired;
46	};	76	};
47		77
48	-class OpStatisticProcessorUTest : public testing::Test {	78	+AssociatedTaskInput makeAssociatedTask(uint16_t deviceId, uint32_t streamId, uint32_t taskId,
49	-protected:	79	+ const std::string &opType, const std::string &taskType, uint64_t durationNs)
50	- virtual void SetUp()	80	+{
		81	+ return {makeTaskInfo(deviceId, streamId, taskId, opType, taskType),
		82	+ makeAscendTask(deviceId, streamId, taskId, 100, 100 + durationNs), true};
		83	+}
		84	+
		85	+void injectAssociatedTasks(DataInventory &dataInventory, std::vector<AssociatedTaskInput> associatedTasks)
		86	+{
		87	+ std::vector<TaskInfoData> taskInfoData;
		88	+ std::vector<AscendTaskData> ascendTaskData;
		89	+ taskInfoData.reserve(associatedTasks.size());
		90	+ ascendTaskData.reserve(associatedTasks.size());
		91	+ for (auto &associatedTask : associatedTasks)
51	{	92	{
52	- if (File::Check(BASE_PATH)) {	93	+ taskInfoData.emplace_back(std::move(associatedTask.taskInfo));
53	- File::RemoveDir(BASE_PATH, 0);	94	+ ascendTaskData.emplace_back(std::move(associatedTask.ascendTask));
		95	+ }
		96	+
		97	+ std::shared_ptr<std::vector<TaskInfoData>> taskInfoDataPtr;
		98	+ std::shared_ptr<std::vector<AscendTaskData>> ascendTaskDataPtr;
		99	+ std::shared_ptr<AssociatedTaskCollection> associatedTaskCollection;
		100	+ MAKE_SHARED_NO_OPERATION(taskInfoDataPtr, std::vector<TaskInfoData>, std::move(taskInfoData));
		101	+ MAKE_SHARED_NO_OPERATION(ascendTaskDataPtr, std::vector<AscendTaskData>, std::move(ascendTaskData));
		102	+ MAKE_SHARED_NO_OPERATION(associatedTaskCollection, AssociatedTaskCollection);
		103	+ associatedTaskCollection->taskInfoData = taskInfoDataPtr;
		104	+ associatedTaskCollection->ascendTaskData = ascendTaskDataPtr;
		105	+ for (size_t index = 0; index < associatedTasks.size(); ++index)
		106	+ {
		107	+ associatedTaskCollection->records.push_back(AssociatedTaskData{
		108	+ &associatedTaskCollection->taskInfoData->at(index), &associatedTaskCollection->ascendTaskData->at(index),
		109	+ associatedTasks.at(index).opSummaryRequired});
		110	+ }
		111	+ dataInventory.Inject(associatedTaskCollection);
		112	+}
		113	+
		114	+void injectSourceData(DataInventory &dataInventory, std::vector<TaskInfoData> taskInfo,
		115	+ std::vector<AscendTaskData> ascendTask)
		116	+{
		117	+ std::shared_ptr<std::vector<TaskInfoData>> taskInfoPtr;
		118	+ MAKE_SHARED_NO_OPERATION(taskInfoPtr, std::vector<TaskInfoData>, std::move(taskInfo));
		119	+ dataInventory.Inject(taskInfoPtr);
		120	+ std::shared_ptr<std::vector<AscendTaskData>> ascendTaskPtr;
		121	+ MAKE_SHARED_NO_OPERATION(ascendTaskPtr, std::vector<AscendTaskData>, std::move(ascendTask));
		122	+ dataInventory.Inject(ascendTaskPtr);
		123	+}
		124	+
		125	+const OpStatisticData *findStatistic(const std::vector<OpStatisticData> &data, uint16_t deviceId,
		126	+ const std::string &opType, const std::string &coreType)
		127	+{
		128	+ for (const auto &item : data)
		129	+ {
		130	+ if (item.deviceId == deviceId && item.opType == opType && item.coreType == coreType)
		131	+ {
		132	+ return &item;
54	}	133	}
55	- EXPECT_TRUE(File::CreateDir(BASE_PATH));
56	- EXPECT_TRUE(File::CreateDir(PROF_PATH_A));
57	- EXPECT_TRUE(File::CreateDir(PROF_PATH_B));
58	- EXPECT_TRUE(File::CreateDir(File::PathJoin({PROF_PATH_A, DEVICE_SUFFIX})));
59	- EXPECT_TRUE(File::CreateDir(File::PathJoin({PROF_PATH_B, DEVICE_SUFFIX})));
60	- EXPECT_TRUE(File::CreateDir(File::PathJoin({PROF_PATH_A, DEVICE_SUFFIX, SQLITE_SUFFIX})));
61	- EXPECT_TRUE(File::CreateDir(File::PathJoin({PROF_PATH_B, DEVICE_SUFFIX, SQLITE_SUFFIX})));
62	- CreateOpMetricData(File::PathJoin({PROF_PATH_A, DEVICE_SUFFIX, SQLITE_SUFFIX, DB_SUFFIX}), OP_DATA);
63	- CreateOpMetricData(File::PathJoin({PROF_PATH_B, DEVICE_SUFFIX, SQLITE_SUFFIX, DB_SUFFIX}), OP_DATA);
64	}	134	}
65	- virtual void TearDown()	135	+ return nullptr;
66	- {	136	+}
67	- EXPECT_TRUE(File::RemoveDir(BASE_PATH, 0));	137	+} // namespace
68	- }
69	- static void CreateOpMetricData(const std::string& dbPath, OriOpCountDataFormat data)
70	- {
71	- std::shared_ptr<OpCounterDB> database;
72	- MAKE_SHARED0_RETURN_VOID(database, OpCounterDB);
73	- std::shared_ptr<DBRunner> dbRunner;
74	- MAKE_SHARED_RETURN_VOID(dbRunner, DBRunner, dbPath);
75	- auto cols = database->GetTableCols(TABLE_NAME);
76	- dbRunner->CreateTable(TABLE_NAME, cols);
77	- dbRunner->InsertData(TABLE_NAME, data);
78	- }
79	-};
80		138
81	-	139	+TEST(OpStatisticProcessorUTest, TestRunShouldReturnTrueWhenSourceDataNotExist)
82	-TEST_F(OpStatisticProcessorUTest, TestRunShouldReturnTrueWhenProcessorRunSuccess)
83	{	140	{
84	- for (auto path: PROF_PATHS) {	141	+ DataInventory dataInventory;
85	- auto processor = OpStatisticProcessor(path);	142	+ EXPECT_TRUE(OpStatisticProcessor(PROF_PATH).Run(dataInventory, PROCESSOR_NAME_OP_STATISTIC));
86	- auto dataInventory = DataInventory();	143	+ EXPECT_EQ(nullptr, dataInventory.GetPtr<std::vector<OpStatisticData>>());
87	- EXPECT_TRUE(processor.Run(dataInventory, PROCESSOR_NAME_OP_STATISTIC));
88	- }
89	}	144	}
90		145
91	-TEST_F(OpStatisticProcessorUTest, TestRunShouldReturnTrueWhenSourceTableNotExist)	146	+TEST(OpStatisticProcessorUTest, TestRunShouldReturnTrueWhenAssociatedTaskDataEmpty)
92	{	147	{
93	- auto dbPath = File::PathJoin({PROF_PATH_A, DEVICE_SUFFIX, SQLITE_SUFFIX, DB_SUFFIX});	148	+ DataInventory dataInventory;
94	- std::shared_ptr<DBRunner> dbRunner;	149	+ injectAssociatedTasks(dataInventory, {});
95	- MAKE_SHARED0_NO_OPERATION(dbRunner, DBRunner, dbPath);	150	+ EXPECT_TRUE(OpStatisticProcessor(PROF_PATH).Run(dataInventory, PROCESSOR_NAME_OP_STATISTIC));
96	- dbRunner->DropTable(TABLE_NAME);	151	+ EXPECT_EQ(nullptr, dataInventory.GetPtr<std::vector<OpStatisticData>>());
97	- dbPath = File::PathJoin({PROF_PATH_B, DEVICE_SUFFIX, SQLITE_SUFFIX, DB_SUFFIX});
98	- MAKE_SHARED0_NO_OPERATION(dbRunner, DBRunner, dbPath);
99	- dbRunner->DropTable(TABLE_NAME);
100	- for (auto path: PROF_PATHS) {
101	- auto processor = OpStatisticProcessor(path);
102	- auto dataInventory = DataInventory();
103	- EXPECT_TRUE(processor.Run(dataInventory, PROCESSOR_NAME_OP_STATISTIC));
104	- }
105	}	152	}
106		153
107	-	154	+TEST(OpStatisticProcessorUTest, TestRunShouldAggregateAssociatedTasks)
108	-TEST_F(OpStatisticProcessorUTest, TestRunShouldReturnFalseWhenCheckPathFailed)
109	{	155	{
110	- MOCKER_CPP(&Analysis::Utils::File::Check).stubs().will(returnValue(false));	156	+ DataInventory dataInventory;
111	- for (auto path: PROF_PATHS) {	157	+ injectAssociatedTasks(dataInventory,
112	- auto processor = OpStatisticProcessor(path);	158	+ {makeAssociatedTask(0, 1, 1, "MatMul", "AI_CORE", 2000),
113	- auto dataInventory = DataInventory();	159	+ makeAssociatedTask(0, 1, 2, "MatMul", "AI_CORE", 4000),
114	- EXPECT_FALSE(processor.Run(dataInventory, PROCESSOR_NAME_OP_STATISTIC));	160	+ makeAssociatedTask(0, 2, 1, "Add", "MIX_AIC", 4000)});
115	- }	161	+
116	- MOCKER_CPP(&Analysis::Utils::File::Check).reset();	162	+ EXPECT_TRUE(OpStatisticProcessor(PROF_PATH).Run(dataInventory, PROCESSOR_NAME_OP_STATISTIC));
		163	+ auto result = dataInventory.GetPtr<std::vector<OpStatisticData>>();
		164	+ ASSERT_NE(nullptr, result);
		165	+ ASSERT_EQ(2ul, result->size());
		166	+
		167	+ auto matmul = findStatistic(*result, 0, "MatMul", "AI_CORE");
		168	+ ASSERT_NE(nullptr, matmul);
		169	+ EXPECT_EQ("2", matmul->count);
		170	+ EXPECT_DOUBLE_EQ(6.0, matmul->totalTime);
		171	+ EXPECT_DOUBLE_EQ(2.0, matmul->min);
		172	+ EXPECT_DOUBLE_EQ(3.0, matmul->avg);
		173	+ EXPECT_DOUBLE_EQ(4.0, matmul->max);
		174	+ EXPECT_DOUBLE_EQ(60.0, matmul->ratio);
		175	+
		176	+ auto add = findStatistic(*result, 0, "Add", "MIX_AIC");
		177	+ ASSERT_NE(nullptr, add);
		178	+ EXPECT_EQ("1", add->count);
		179	+ EXPECT_DOUBLE_EQ(4.0, add->totalTime);
		180	+ EXPECT_DOUBLE_EQ(40.0, add->ratio);
117	}	181	}
118		182
119	-TEST_F(OpStatisticProcessorUTest, TestRunShouldReturnTrueWhenNoDb)	183	+TEST(OpStatisticProcessorUTest, TestRunShouldComputeRatioPerDevice)
120	{	184	{
121	- std::vector<std::string> deviceList = {File::PathJoin({BASE_PATH, "test", "device_1"})};	185	+ DataInventory dataInventory;
122	- MOCKER_CPP(&Utils::File::GetFilesWithPrefix).stubs().will(returnValue(deviceList));	186	+ injectAssociatedTasks(dataInventory,
123	- auto processor = OpStatisticProcessor({File::PathJoin({BASE_PATH, "test"})});	187	+ {makeAssociatedTask(0, 1, 1, "MatMul", "AI_CORE", 1000),
124	- auto dataInventory = DataInventory();	188	+ makeAssociatedTask(1, 1, 1, "Add", "AI_CORE", 2000)});
125	- EXPECT_TRUE(processor.Run(dataInventory, PROCESSOR_NAME_OP_STATISTIC));	189	+
126	- MOCKER_CPP(&Utils::File::GetFilesWithPrefix).reset();	190	+ EXPECT_TRUE(OpStatisticProcessor(PROF_PATH).Run(dataInventory, PROCESSOR_NAME_OP_STATISTIC));
		191	+ auto result = dataInventory.GetPtr<std::vector<OpStatisticData>>();
		192	+ ASSERT_NE(nullptr, result);
		193	+ ASSERT_EQ(2ul, result->size());
		194	+ EXPECT_DOUBLE_EQ(100.0, findStatistic(*result, 0, "MatMul", "AI_CORE")->ratio);
		195	+ EXPECT_DOUBLE_EQ(100.0, findStatistic(*result, 1, "Add", "AI_CORE")->ratio);
127	}	196	}
128		197
129	-TEST_F(OpStatisticProcessorUTest, TestRunShouldReturnFalseWhenReserveFailed)	198	+TEST(OpStatisticProcessorUTest, TestRunShouldUseCompleteOpReportForRatioAndFilterCsvRows)
130	{	199	{
131	- StubReserveFailureForVector<ProcessedFormat>();	200	+ DataInventory dataInventory;
132	- for (auto path: PROF_PATHS) {	201	+ auto communication = makeAssociatedTask(0, 1, 2, "AllReduce", "COMMUNICATION", 3000);
133	- auto processor = OpStatisticProcessor(path);	202	+ communication.taskInfo.opName = "opAivKernel";
134	- auto dataInventory = DataInventory();	203	+ injectAssociatedTasks(
135	- EXPECT_FALSE(processor.Run(dataInventory, PROCESSOR_NAME_OP_STATISTIC));	204	+ dataInventory,
136	- }	205	+ {makeAssociatedTask(0, 1, 1, "N/A", "AI_CORE", 1000), communication,
137	- ResetReserveFailureForVector<ProcessedFormat>();	206	+ makeAssociatedTask(0, 1, 3, "MatMul", "AI_CORE", 2000),
		207	+ makeAssociatedTask(0, 1, 4, "Write", "WRITE_BACK", 3000),
		208	+ makeAssociatedTask(0, 1, 5, "Invalid", "INVALID", 4000),
		209	+ makeAssociatedTask(0, 1, 6, "CpuTask", "HCCL_AI_CPU", 5000)});
		210	+
		211	+ EXPECT_TRUE(OpStatisticProcessor(PROF_PATH).Run(dataInventory, PROCESSOR_NAME_OP_STATISTIC));
		212	+ auto result = dataInventory.GetPtr<std::vector<OpStatisticData>>();
		213	+ ASSERT_NE(nullptr, result);
		214	+ ASSERT_EQ(1ul, result->size());
		215	+ auto matmul = findStatistic(*result, 0, "MatMul", "AI_CORE");
		216	+ ASSERT_NE(nullptr, matmul);
		217	+ EXPECT_DOUBLE_EQ(20.0, matmul->ratio);
138	}	218	}
139		219
140	-TEST_F(OpStatisticProcessorUTest, TestRunShouldReturnFalseWhenConstructDBRunnerFailed)	220	+TEST(OpStatisticProcessorUTest, TestRunShouldConsumeAssociatedTasksFromPredecessor)
141	{	221	{
142	- MOCKER_CPP(&DBInfo::ConstructDBRunner).stubs().will(returnValue(false));	222	+ DataInventory dataInventory;
143	- for (auto path: PROF_PATHS) {	223	+ injectSourceData(dataInventory,
144	- auto processor = OpStatisticProcessor(path);	224	+ {makeTaskInfo(0, 1, 1, "MatMul", "AI_CORE"),
145	- auto dataInventory = DataInventory();	225	+ makeTaskInfo(0, 1, 2, "Write", "WRITE_BACK")},
146	- EXPECT_FALSE(processor.Run(dataInventory, PROCESSOR_NAME_OP_STATISTIC));	226	+ {makeAscendTask(0, 1, 1, 0, 1000), makeAscendTask(0, 1, 2, 0, 3000)});
147	- }
148	- MOCKER_CPP(&DBInfo::ConstructDBRunner).reset();
149	-}
150		227
151	-TEST_F(OpStatisticProcessorUTest, TestRunShouldReturnTureWhenProcessRunSuccessAndCheckData)	228	+ EXPECT_TRUE(TaskAssociationProcessor(PROF_PATH).Run(dataInventory, PROCESSOR_NAME_TASK_ASSOCIATION));
152	-{	229	+ dataInventory.RemoveRestData({typeid(AssociatedTaskCollection)});
153	- auto dataInventory = DataInventory();	230	+ EXPECT_TRUE(OpStatisticProcessor(PROF_PATH).Run(dataInventory, PROCESSOR_NAME_OP_STATISTIC));
154	- for (auto path: PROF_PATHS) {	231	+ auto result = dataInventory.GetPtr<std::vector<OpStatisticData>>();
155	- auto processor = OpStatisticProcessor(path);	232	+ ASSERT_NE(nullptr, result);
156	- EXPECT_TRUE(processor.Run(dataInventory, PROCESSOR_NAME_OP_STATISTIC));	233	+ ASSERT_EQ(1ul, result->size());
157	- }	234	+ auto matmul = findStatistic(*result, 0, "MatMul", "AI_CORE");
158	- auto res = dataInventory.GetPtr<std::vector<OpStatisticData>>();	235	+ ASSERT_NE(nullptr, matmul);
159	- EXPECT_EQ(4ul, res->size());	236	+ EXPECT_DOUBLE_EQ(25.0, matmul->ratio);
160	- auto data = res->at(0);
161	- EXPECT_EQ("RmsNormTactic", data.opType);
162	- EXPECT_EQ("1610", data.count);
163	}	237	}

Atest/msprof_cpp/analysis_ut/domain/data_process/test/task_association_processor_utest.cpp+180-0

@@ -0,0 +1,180 @@
	1	+/* -------------------------------------------------------------------------
	2	+ * Copyright (c) 2026 Huawei Technologies Co., Ltd.
	3	+ * This file is part of the MindStudio project.
	4	+ *
	5	+ * MindStudio is licensed under Mulan PSL v2.
	6	+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
	7	+ * You may obtain a copy of Mulan PSL v2 at:
	8	+ *
	9	+ * http://license.coscl.org.cn/MulanPSL2
	10	+ *
	11	+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
	12	+ * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
	13	+ * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
	14	+ * See the Mulan PSL v2 for more details.
	15	+ * -------------------------------------------------------------------------*/
	16	+
	17	+#include "gtest/gtest.h"
	18	+
	19	+#include "analysis/csrc/application/database/db_constant.h"
	20	+#include "analysis/csrc/domain/data_process/ai_task/task_association_processor.h"
	21	+#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/associated_task_data.h"
	22	+#include "analysis/csrc/infrastructure/utils/utils.h"
	23	+
	24	+using namespace Analysis::Domain;
	25	+using namespace Analysis::Utils;
	26	+using namespace Analysis::Application;
	27	+
	28	+namespace
	29	+{
	30	+TaskInfoData makeTaskInfo(uint16_t deviceId, uint32_t streamId, uint32_t taskId, const std::string &opType,
	31	+ const std::string &taskType, uint32_t batchId = 0, uint32_t contextId = 0,
	32	+ const std::string &opName = "")
	33	+{
	34	+ TaskInfoData data;
	35	+ data.deviceId = deviceId;
	36	+ data.streamId = streamId;
	37	+ data.taskId = taskId;
	38	+ data.batchId = batchId;
	39	+ data.contextId = contextId;
	40	+ data.opType = opType;
	41	+ data.taskType = taskType;
	42	+ data.opName = opName;
	43	+ return data;
	44	+}
	45	+
	46	+AscendTaskData makeAscendTask(uint16_t deviceId, uint32_t streamId, uint32_t taskId, double durationNs,
	47	+ uint32_t batchId = 0, uint32_t contextId = 0)
	48	+{
	49	+ AscendTaskData data;
	50	+ data.deviceId = deviceId;
	51	+ data.streamId = streamId;
	52	+ data.taskId = taskId;
	53	+ data.batchId = batchId;
	54	+ data.contextId = contextId;
	55	+ data.timestamp = 0;
	56	+ data.end = static_cast<uint64_t>(durationNs);
	57	+ data.duration = durationNs;
	58	+ return data;
	59	+}
	60	+
	61	+void injectSourceData(DataInventory &dataInventory, std::vector<TaskInfoData> taskInfo,
	62	+ std::vector<AscendTaskData> ascendTask)
	63	+{
	64	+ std::shared_ptr<std::vector<TaskInfoData>> taskInfoPtr;
	65	+ MAKE_SHARED_NO_OPERATION(taskInfoPtr, std::vector<TaskInfoData>, std::move(taskInfo));
	66	+ dataInventory.Inject(taskInfoPtr);
	67	+ std::shared_ptr<std::vector<AscendTaskData>> ascendTaskPtr;
	68	+ MAKE_SHARED_NO_OPERATION(ascendTaskPtr, std::vector<AscendTaskData>, std::move(ascendTask));
	69	+ dataInventory.Inject(ascendTaskPtr);
	70	+}
	71	+} // namespace
	72	+
	73	+class TaskAssociationProcessorUTest : public testing::Test
	74	+{
	75	+};
	76	+
	77	+TEST_F(TaskAssociationProcessorUTest, TestRunShouldReturnTrueWhenSourceDataNotExist)
	78	+{
	79	+ auto processor = TaskAssociationProcessor("./task_association");
	80	+ auto dataInventory = DataInventory();
	81	+ EXPECT_TRUE(processor.Run(dataInventory, PROCESSOR_NAME_TASK_ASSOCIATION));
	82	+ EXPECT_EQ(nullptr, dataInventory.GetPtr<AssociatedTaskCollection>());
	83	+}
	84	+
	85	+TEST_F(TaskAssociationProcessorUTest, TestRunShouldReturnTrueWhenOnlyTaskInfoExist)
	86	+{
	87	+ auto dataInventory = DataInventory();
	88	+ std::vector<TaskInfoData> taskInfo{makeTaskInfo(0, 1, 1, "MatMul", "AI_CORE")};
	89	+ std::shared_ptr<std::vector<TaskInfoData>> taskInfoPtr;
	90	+ MAKE_SHARED_NO_OPERATION(taskInfoPtr, std::vector<TaskInfoData>, std::move(taskInfo));
	91	+ dataInventory.Inject(taskInfoPtr);
	92	+ auto processor = TaskAssociationProcessor("./task_association");
	93	+ EXPECT_TRUE(processor.Run(dataInventory, PROCESSOR_NAME_TASK_ASSOCIATION));
	94	+ EXPECT_EQ(nullptr, dataInventory.GetPtr<AssociatedTaskCollection>());
	95	+}
	96	+
	97	+TEST_F(TaskAssociationProcessorUTest, TestRunShouldKeepMatchedTaskExcludedFromOpSummary)
	98	+{
	99	+ auto dataInventory = DataInventory();
	100	+ injectSourceData(dataInventory, {makeTaskInfo(0, 1, 1, "MatMul", "COMMUNICATION")},
	101	+ {makeAscendTask(0, 1, 1, 1000.0)});
	102	+ auto processor = TaskAssociationProcessor("./task_association");
	103	+ EXPECT_TRUE(processor.Run(dataInventory, PROCESSOR_NAME_TASK_ASSOCIATION));
	104	+ auto res = dataInventory.GetPtr<AssociatedTaskCollection>();
	105	+ ASSERT_NE(nullptr, res);
	106	+ ASSERT_EQ(1ul, res->records.size());
	107	+ EXPECT_FALSE(res->records.front().opSummaryRequired);
	108	+}
	109	+
	110	+TEST_F(TaskAssociationProcessorUTest, TestRunShouldKeepAllMatchedTasksAndMarkOpSummaryRecords)
	111	+{
	112	+ auto dataInventory = DataInventory();
	113	+ injectSourceData(dataInventory,
	114	+ {makeTaskInfo(0, 1, 1, "MatMul", "AI_CORE", 0, 0, "mm"),
	115	+ makeTaskInfo(0, 1, 2, "AllReduce", "COMMUNICATION", 0, 0, "opAivKernel"),
	116	+ makeTaskInfo(0, 1, 3, "AllReduce", "COMMUNICATION", 0, 0, "opNormal"),
	117	+ makeTaskInfo(0, 1, 4, "Write", "WRITE_BACK"),
	118	+ makeTaskInfo(0, 1, 5, "CpuTask", "HCCL_AI_CPU")},
	119	+ {makeAscendTask(0, 1, 1, 2000.0), makeAscendTask(0, 1, 2, 3000.0),
	120	+ makeAscendTask(0, 1, 3, 7000.0), makeAscendTask(0, 1, 4, 1000.0),
	121	+ makeAscendTask(0, 1, 5, 5000.0), makeAscendTask(0, 9, 9, 4000.0)});
	122	+ auto processor = TaskAssociationProcessor("./task_association");
	123	+ EXPECT_TRUE(processor.Run(dataInventory, PROCESSOR_NAME_TASK_ASSOCIATION));
	124	+ auto res = dataInventory.GetPtr<AssociatedTaskCollection>();
	125	+ ASSERT_NE(nullptr, res);
	126	+ ASSERT_EQ(5ul, res->records.size());
	127	+ EXPECT_EQ("MatMul", res->records.at(0).taskInfo->opType);
	128	+ EXPECT_EQ("AI_CORE", res->records.at(0).taskInfo->taskType);
	129	+ EXPECT_EQ(2000ul, res->records.at(0).ascendTask->end);
	130	+ EXPECT_TRUE(res->records.at(0).opSummaryRequired);
	131	+ EXPECT_EQ("AllReduce", res->records.at(1).taskInfo->opType);
	132	+ EXPECT_EQ("COMMUNICATION", res->records.at(1).taskInfo->taskType);
	133	+ EXPECT_EQ("opAivKernel", res->records.at(1).taskInfo->opName);
	134	+ EXPECT_TRUE(res->records.at(1).opSummaryRequired);
	135	+ EXPECT_FALSE(res->records.at(2).opSummaryRequired);
	136	+ EXPECT_FALSE(res->records.at(3).opSummaryRequired);
	137	+ EXPECT_FALSE(res->records.at(4).opSummaryRequired);
	138	+}
	139	+
	140	+TEST_F(TaskAssociationProcessorUTest, TestRunShouldSkipMatchedTaskWithReversedTimeRange)
	141	+{
	142	+ auto invalidTask = makeAscendTask(0, 1, 1, 1000.0);
	143	+ invalidTask.timestamp = 2000;
	144	+ invalidTask.end = 1000;
	145	+ auto dataInventory = DataInventory();
	146	+ injectSourceData(dataInventory,
	147	+ {makeTaskInfo(0, 1, 1, "InvalidTime", "AI_CORE"),
	148	+ makeTaskInfo(0, 1, 2, "MatMul", "AI_CORE")},
	149	+ {invalidTask, makeAscendTask(0, 1, 2, 3000.0)});
	150	+
	151	+ auto processor = TaskAssociationProcessor("./task_association");
	152	+ EXPECT_TRUE(processor.Run(dataInventory, PROCESSOR_NAME_TASK_ASSOCIATION));
	153	+ auto res = dataInventory.GetPtr<AssociatedTaskCollection>();
	154	+ ASSERT_NE(nullptr, res);
	155	+ ASSERT_EQ(1ul, res->records.size());
	156	+ EXPECT_EQ(2ul, res->records.front().ascendTask->taskId);
	157	+ EXPECT_EQ("MatMul", res->records.front().taskInfo->opType);
	158	+}
	159	+
	160	+TEST_F(TaskAssociationProcessorUTest, TestRunShouldMatchOnlyCompleteTaskId)
	161	+{
	162	+ auto dataInventory = DataInventory();
	163	+ injectSourceData(dataInventory,
	164	+ {makeTaskInfo(0, 1, 7, "DeviceZero", "AI_CORE", 0, 0),
	165	+ makeTaskInfo(1, 1, 7, "DeviceOne", "AI_CORE", 0, 0),
	166	+ makeTaskInfo(0, 1, 7, "BatchOne", "AI_CORE", 1, 0),
	167	+ makeTaskInfo(0, 1, 7, "ContextOne", "AI_CORE", 0, 1)},
	168	+ {makeAscendTask(0, 1, 7, 1000.0, 0, 0), makeAscendTask(1, 1, 7, 1000.0, 0, 0),
	169	+ makeAscendTask(0, 1, 7, 1000.0, 1, 0), makeAscendTask(0, 1, 7, 1000.0, 0, 1)});
	170	+
	171	+ auto processor = TaskAssociationProcessor("./task_association");
	172	+ EXPECT_TRUE(processor.Run(dataInventory, PROCESSOR_NAME_TASK_ASSOCIATION));
	173	+ auto res = dataInventory.GetPtr<AssociatedTaskCollection>();
	174	+ ASSERT_NE(nullptr, res);
	175	+ ASSERT_EQ(4ul, res->records.size());
	176	+ EXPECT_EQ("DeviceZero", res->records.at(0).taskInfo->opType);
	177	+ EXPECT_EQ("DeviceOne", res->records.at(1).taskInfo->opType);
	178	+ EXPECT_EQ("BatchOne", res->records.at(2).taskInfo->opType);
	179	+ EXPECT_EQ("ContextOne", res->records.at(3).taskInfo->opType);
	180	+}

	@@ -21,6 +21,7 @@
21	#include "analysis/csrc/domain/entities/hal/include/ascend_obj.h"	21	#include "analysis/csrc/domain/entities/hal/include/ascend_obj.h"
22	#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/api_data.h"	22	#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/api_data.h"
23	#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/ascend_task_data.h"	23	#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/ascend_task_data.h"
		24	+#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/associated_task_data.h"
24	#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/communication_info_data.h"	25	#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/communication_info_data.h"
25	#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/fusion_op_data.h"	26	#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/fusion_op_data.h"
26	#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/hccl_statistic_data.h"	27	#include "analysis/csrc/domain/entities/viewer_data/ai_task/include/hccl_statistic_data.h"
	@@ -46,13 +47,12 @@ using StringMap = std::unordered_map<std::string, std::string>;
46		47
47	REGISTER_TOPO_NODE_SEQUENCE(typeid(void), TOPO_NODE(SUMMARY_GENERATION, PROCESSOR_OP_SUMMARY), true,	48	REGISTER_TOPO_NODE_SEQUENCE(typeid(void), TOPO_NODE(SUMMARY_GENERATION, PROCESSOR_OP_SUMMARY), true,
48	SummaryManager::CreateSummaryAssembler(PROCESSOR_OP_SUMMARY),	49	SummaryManager::CreateSummaryAssembler(PROCESSOR_OP_SUMMARY),
49	- TOPO_DEPS(TOPO_NODE(DATA_PROCESSING, PROCESSOR_NAME_COMPUTE_TASK_INFO),	50	+ TOPO_DEPS(TOPO_NODE(DATA_PROCESSING, PROCESSOR_NAME_TASK_ASSOCIATION),
50	- TOPO_NODE(DATA_PROCESSING, PROCESSOR_NAME_TASK),
51	TOPO_NODE(DATA_PROCESSING, PROCESSOR_NAME_COMMUNICATION),	51	TOPO_NODE(DATA_PROCESSING, PROCESSOR_NAME_COMMUNICATION),
52	TOPO_NODE(DATA_PROCESSING, PROCESSOR_PMU)),	52	TOPO_NODE(DATA_PROCESSING, PROCESSOR_PMU)),
53	nullptr);	53	nullptr);
54	-REGISTER_TOPO_NODE_DEPENDENT_DATA(TOPO_NODE(SUMMARY_GENERATION, PROCESSOR_OP_SUMMARY), std::vector<TaskInfoData>,	54	+REGISTER_TOPO_NODE_DEPENDENT_DATA(TOPO_NODE(SUMMARY_GENERATION, PROCESSOR_OP_SUMMARY), AssociatedTaskCollection,
55	- std::vector<AscendTaskData>, std::vector<CommunicationOpData>, MetricSummary);	55	+ std::vector<CommunicationOpData>, MetricSummary);
56		56
57	#define REGISTER_SIMPLE_SUMMARY_NODE(Name, DataType, Dependency) \	57	#define REGISTER_SIMPLE_SUMMARY_NODE(Name, DataType, Dependency) \
58	REGISTER_TOPO_NODE_SEQUENCE(typeid(void), TOPO_NODE(SUMMARY_GENERATION, Name), true, \	58	REGISTER_TOPO_NODE_SEQUENCE(typeid(void), TOPO_NODE(SUMMARY_GENERATION, Name), true, \