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