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 "plugin_runner.h"
12 : #include "adapter_rts_common.h"
13 : #include "externalinput_pub.h"
14 :
15 : using namespace hccl;
16 936 : PluginRunner::PluginRunner(ProfilerBase* profiler) : profiler_(profiler) {}
17 :
18 2808 : PluginRunner::~PluginRunner() {}
19 :
20 : template <typename T>
21 123 : void PluginRunner::operator()(rtStream_t stream, TaskType taskType, const T& para) const
22 : {
23 : // capture模式下hrtGetStreamId获取的是原来的流对应ID，与实际执行流不是同一个
24 : // capture模式下hrtGetTaskIdAndStreamID获取实际执行的streamID和taskID
25 123 : u32 threadLastTaskID = 0;
26 123 : u32 threadLastStreamID = 0;
27 123 : s32 streamID = 0;
28 123 : bool isOneSideTask = false;
29 123 : bool isCapture = false;
30 123 : CHK_PRT(isStreamCapture(stream, isCapture));
31 :
32 123 : if (profiler_ == nullptr) {
33 0 : return;
34 : }
35 123 : CHK_PRT(hrtGetTaskIdAndStreamID(threadLastTaskID, threadLastStreamID));
36 :
37 123 : std::string tag;
38 123 : CHK_PRT(ProfilerBase::GetTagByStream(threadLastStreamID, tag));
39 123 : if (tag.find("BatchPut_") != std::string::npos || tag.find("BatchGet_") != std::string::npos) {
40 0 : isOneSideTask = true;
41 : }
42 :
43 : HcclResult ret;
44 135 : if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE
45 135 : && !isOneSideTask) {
46 0 : ret = hrtGetStreamId(stream, streamID);
47 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[PluginRunner][Operator]rtGet stream id fail. return[%d]", ret), );
48 :
49 0 : u32 castStreamID = static_cast<u32>(streamID);
50 0 : if (isCapture) {
51 0 : ret = hrtGetTaskIdAndStreamID(threadLastTaskID, threadLastStreamID);
52 0 : CHK_PRT_RET(
53 : ret != HCCL_SUCCESS,
54 : HCCL_ERROR("[PluginRunner][Operator]rtGet task id and stream id fail. return[%d]", ret), );
55 0 : profiler_->Save(castStreamID, threadLastStreamID, threadLastTaskID, taskType, para);
56 : } else {
57 0 : profiler_->Save(castStreamID, threadLastTaskID, taskType, para);
58 : }
59 : } else {
60 123 : ret = hrtGetTaskIdAndStreamID(threadLastTaskID, threadLastStreamID);
61 123 : CHK_PRT_RET(
62 : ret != HCCL_SUCCESS,
63 : HCCL_ERROR("[PluginRunner][Operator]rtGet task id and stream id fail. return[%d]", ret), );
64 :
65 123 : if (isCapture) {
66 0 : ret = hrtGetStreamId(stream, streamID);
67 0 : CHK_PRT_RET(
68 : ret != HCCL_SUCCESS,
69 : HCCL_ERROR("[PluginRunner][Operator]rtGet task id and stream id fail. return[%d]", ret), );
70 0 : u32 castStreamID = static_cast<u32>(streamID);
71 0 : profiler_->Save(castStreamID, threadLastStreamID, threadLastTaskID, taskType, para);
72 : } else {
73 123 : profiler_->Save(threadLastStreamID, threadLastTaskID, taskType, para);
74 : }
75 : }
76 123 : }
77 :
78 : template void PluginRunner::operator()<TaskParaDMA>(rtStream_t, TaskType, const TaskParaDMA&) const;
79 : template void PluginRunner::operator()<TaskParaReduce>(rtStream_t, TaskType, const TaskParaReduce&) const;
80 : template void PluginRunner::operator()<TaskParaNotify>(rtStream_t, TaskType, const TaskParaNotify&) const;
81 :
82 0 : void PluginRunner::operator()(rtStream_t stream, const void* descBuf, size_t descBufLen) const
83 : {
84 0 : u32 threadLastTaskID = 0;
85 0 : u32 threadLastStreamID = 0;
86 0 : s32 streamID = 0;
87 : HcclResult ret;
88 0 : bool isCapture = false;
89 0 : CHK_PRT(isStreamCapture(stream, isCapture));
90 :
91 0 : CHK_PRT_RET(profiler_ == nullptr, HCCL_WARNING("profiler_ is nullptr"), );
92 0 : ret = hrtGetTaskIdAndStreamID(threadLastTaskID, threadLastStreamID);
93 0 : CHK_PRT_RET(
94 : ret != HCCL_SUCCESS, HCCL_ERROR("[PluginRunner][Operator]rtGet task id and stream id fail. return[%d]", ret), );
95 :
96 0 : if (isCapture) {
97 0 : ret = hrtGetStreamId(stream, streamID);
98 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[PluginRunner][Operator]rtGet stream id fail. return[%d]", ret), );
99 0 : u32 castStreamID = static_cast<u32>(streamID);
100 0 : profiler_->Save(castStreamID, threadLastStreamID, threadLastTaskID, descBuf, descBufLen);
101 : } else {
102 0 : profiler_->Save(threadLastStreamID, threadLastTaskID, descBuf, descBufLen);
103 : }
104 : }
105 :
106 0 : void PluginRunner::operator()(const TaskParaHost& paraHost) const
107 : {
108 0 : if (profiler_ != nullptr) {
109 0 : profiler_->SaveToLog(paraHost);
110 : }
111 0 : }
112 :
113 15 : void PluginRunner::operator()(rtStream_t stream, const TaskParaAiv& paraAiv) const
114 : {
115 15 : u32 threadLastTaskID = 0;
116 15 : u32 threadLastStreamID = 0;
117 15 : s32 streamID = 0;
118 : HcclResult result;
119 15 : bool isCapture = false;
120 15 : CHK_PRT(isStreamCapture(stream, isCapture));
121 15 : result = hrtGetTaskIdAndStreamID(threadLastTaskID, threadLastStreamID);
122 15 : CHK_PRT_RET(
123 : result != HCCL_SUCCESS,
124 : HCCL_ERROR("[PluginRunner][Operator]rtGet task id and stream id fail. return[%d]", result), );
125 15 : CHK_PRT_RET(profiler_ == nullptr, HCCL_WARNING("profiler_ is nullptr"), );
126 15 : if (isCapture) {
127 9 : result = hrtGetStreamId(stream, streamID);
128 9 : CHK_PRT_RET(
129 : result != HCCL_SUCCESS, HCCL_ERROR("[PluginRunner][Operator]rtGet stream id fail. return[%d]", result), );
130 9 : u32 castStreamID = static_cast<u32>(streamID);
131 9 : profiler_->Save(castStreamID, threadLastStreamID, threadLastTaskID, paraAiv);
132 : } else {
133 6 : profiler_->Save(threadLastStreamID, threadLastTaskID, paraAiv);
134 : }
135 : }
|
