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