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 "profiling_reporter_lite.h"
      12              : 
      13              : namespace Hccl {
      14              : constexpr size_t TASK_INFO_BATCH_RESERVE_SIZE = 8192;
      15           81 : ProfilingReporterLite::ProfilingReporterLite(
      16           81 :     MirrorTaskManagerLite* mirrorTaskMgrLite, ProfilingHandlerLite* profilingHandlerLite, [[maybe_unused]] bool isIndop)
      17           81 :     : mirrorTaskMgrLite_(mirrorTaskMgrLite),
      18           81 :       profilingHandlerLite_(profilingHandlerLite)
      19           81 : {}
      20              : 
      21          161 : ProfilingReporterLite::~ProfilingReporterLite() {}
      22              : 
      23            0 : HcclResult ProfilingReporterLite::Init()
      24              : {
      25            0 :     if (initializedFlag_) {
      26            0 :         return HCCL_SUCCESS;
      27              :     }
      28            0 :     if (UNLIKELY(mirrorTaskMgrLite_ == nullptr || profilingHandlerLite_ == nullptr)) {
      29            0 :         HCCL_ERROR("[ProfilingReporterLite][Init] mirrorTaskMgrLite or profilingHandlerLite is nullptr.");
      30            0 :         return HCCL_E_PTR;
      31              :     }
      32            0 :     mirrorTaskMgrLite_->RegFullyCallBack([this]() {
      33            0 :         ReportAllTasks();
      34            0 :     });
      35            0 :     initializedFlag_ = true;
      36            0 :     return HCCL_SUCCESS;
      37              : }
      38              : 
      39              : /*
      40              :  *  (*currQueue) == Queue<std::unique_ptr<TaskInfo>> = QUEUE
      41              :  *  QUEUE.Begin() =std::shared_ptr<Iterator<unique_ptr<taskInfo>>
      42              :  *  *QUEUE.Begin() = Iterator<unique_ptr<taskInfo>
      43              :  *  *(*QUEUE.Begin()) = unique_ptr<taskInfo>
      44              :  *  *(*(*QUEUE.Begin())) = taskInfo;
      45              :  *  taskInfo.push_back((*(*((*currQueue).Begin())));
      46              :  */
      47              : // 所有的迭代器都是make_shared 永不为空  底层修改之后 需求再看下
      48            1 : void ProfilingReporterLite::ReportAllTasksLog() const
      49              : {
      50            1 :     if (LIKELY(HcclCheckLogLevel(HCCL_LOG_INFO) == 0)) {
      51            0 :         return;
      52              :     }
      53            3 :     for (auto it = mirrorTaskMgrLite_->Begin(); it != mirrorTaskMgrLite_->End(); ++it) {
      54            2 :         u32 streamId = it->first;
      55            2 :         Queue<std::unique_ptr<TaskInfo>>* currQueue = it->second.queue.get();
      56            2 :         if (currQueue == nullptr || currQueue->Begin() == nullptr || currQueue->Tail() == nullptr) {
      57            0 :             continue;
      58              :         }
      59            2 :         bool logAll = (lastPoses_.find(streamId) == lastPoses_.end());
      60            4 :         for (auto logIter = currQueue->Begin(); *logIter != *currQueue->End(); ++(*logIter)) {
      61            2 :             if (*(*logIter) == nullptr) {
      62            0 :                 continue;
      63              :             }
      64            4 :             if (!logAll
      65            2 :                 && *logIter
      66            0 :                        == *lastPoses_.at(
      67            0 :                            streamId)) { //  找到旧 Tail 位置后设为 logAll=true，continue 跳过该位置，后续全打印
      68            0 :                 logAll = true;
      69            0 :                 continue;
      70              :             }
      71            2 :             if (!logAll) {
      72            0 :                 continue;
      73              :             }
      74            2 :             TaskInfo task = (*(*(*logIter)));
      75            6 :             HCCL_INFO("[ProfilingReporterLite][ReportAllTasks] %s", task.Describe().c_str());
      76            4 :         }
      77              :     }
      78              : }
      79              : 
      80            1 : void ProfilingReporterLite::ReportAllTasks()
      81              : {
      82            1 :     ReportAllTasksLog();
      83            1 :     if (ProfilingHandlerLite::GetInstance().GetProfL1State() == false) {
      84            0 :         HCCL_DEBUG("[ProfilingReporterLite][ReportAllTasks] GetProfL1State is false, UpdateAllLastPos and skip report");
      85            0 :         UpdateAllLastPos();
      86            0 :         return;
      87              :     }
      88              : 
      89            1 :     std::vector<TaskInfo*> taskInfo;
      90            1 :     taskInfo.reserve(TASK_INFO_BATCH_RESERVE_SIZE);
      91            3 :     for (auto it = mirrorTaskMgrLite_->Begin(); it != mirrorTaskMgrLite_->End(); ++it) {
      92            2 :         u32 streamId = it->first;
      93            2 :         Queue<std::unique_ptr<TaskInfo>>* currQueue = it->second.queue.get();
      94            2 :         if (currQueue == nullptr || (*(*(currQueue->Begin()))) == nullptr || (*(*(currQueue->Tail()))) == nullptr) {
      95            0 :             HCCL_WARNING("[ProfilingReporterLite][ReportAllTasks] currQueue is nullptr, continue to next task.");
      96            0 :             continue;
      97            0 :         }
      98            2 :         if (lastPoses_.find(streamId) == lastPoses_.end()) {
      99            2 :             taskInfo.emplace_back((*currQueue->Begin())->get());
     100            2 :             lastPoses_[streamId] = currQueue->Begin();
     101              :         }
     102            2 :         auto endPos = currQueue->Tail();
     103            2 :         auto iter = lastPoses_[streamId];
     104            2 :         ++(*iter);
     105            2 :         for (; (*(iter)) != (*(currQueue->End())); ++(*(iter))) {
     106            0 :             taskInfo.emplace_back((*iter)->get());
     107              :         }
     108            2 :         lastPoses_[streamId] = endPos;
     109            2 :     }
     110            1 :     ProfilingHandlerLite::GetInstance().ReportHcclTaskDetails(taskInfo);
     111            1 : }
     112              : 
     113            4 : void ProfilingReporterLite::UpdateProfStat(void) const { ProfilingHandlerLite::GetInstance().UpdateProfSwitch(); }
     114              : 
     115            0 : void ProfilingReporterLite::UpdateAllLastPos()
     116              : {
     117            0 :     for (auto it = mirrorTaskMgrLite_->Begin(); it != mirrorTaskMgrLite_->End(); ++it) {
     118            0 :         u32 streamId = it->first;
     119            0 :         Queue<std::unique_ptr<TaskInfo>>* currQueue = it->second.queue.get(); // 一旦有streamid 必有queue 必不为空
     120              : 
     121            0 :         auto endPos = currQueue->Tail();
     122            0 :         lastPoses_[streamId] = endPos;
     123            0 :     }
     124            0 : }
     125              : 
     126              : } // namespace Hccl