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 "opbase_stream_manager.h"
      12              : #include "log.h"
      13              : #include "exception_util.h"
      14              : #include "communicator_impl.h"
      15              : #include "invalid_params_exception.h"
      16              : 
      17              : namespace Hccl {
      18              : 
      19          465 : OpbaseStreamManager::~OpbaseStreamManager()
      20              : {
      21          465 :     DECTOR_TRY_CATCH("OpbaseStreamManager", Clear());
      22          465 : }
      23              : 
      24           17 : void OpbaseStreamManager::ReplaceMaster(std::unique_ptr<Stream> stream)
      25              : {
      26           17 :     master = std::move(stream);
      27           17 :     master->SetStmMode(HrtStreamGetMode(master->GetPtr()));
      28           17 : }
      29              : 
      30           23 : void OpbaseStreamManager::RegisterMaster(std::unique_ptr<Stream> stream)
      31              : {
      32           69 :     HCCL_INFO("[OpbaseStreamManager::%s] start.", __func__);
      33              : 
      34           23 :     if (master == nullptr) {
      35           16 :         ReplaceMaster(std::move(stream));
      36           48 :         HCCL_INFO("[OpbaseStreamManager::%s] end, master is nullptr.", __func__);
      37           16 :         return;
      38              :     }
      39            7 :     if (*master == *stream) {
      40           18 :         HCCL_INFO("[OpbaseStreamManager::%s] end, master is same as stream.", __func__);
      41            6 :         return;
      42              :     }
      43            1 :     ReplaceMaster(std::move(stream));
      44              : 
      45            3 :     HCCL_INFO("[OpbaseStreamManager::%s] end.", __func__);
      46              : }
      47              : 
      48          466 : void OpbaseStreamManager::Clear()
      49              : {
      50          466 :     slaves.clear();
      51          466 : }
      52              : 
      53           16 : Stream *OpbaseStreamManager::GetOrCreateSlave()
      54              : {
      55           48 :     HCCL_INFO("[OpbaseStreamManager::%s] start.", __func__);
      56              : 
      57              :     // 如果从流不够用，就申请一个新的从流
      58           16 :     u32 slavesSize = slaves.size();
      59           48 :     HCCL_INFO("[OpbaseStreamManager::%s] slavesSize[%u] slaveIndex[%u]", __func__, slavesSize, slaveIndex);
      60           16 :     if (slaveIndex >= slavesSize) {
      61           16 :         slaves.emplace_back(std::make_unique<Stream>(comm->GetOpAiCpuTSFeatureFlag(), false)); // 算子粒度
      62           16 :         if (master != nullptr && !comm->GetOpAiCpuTSFeatureFlag()) {  // 算子粒度
      63           15 :             slaves[slaveIndex]->SetStmMode(master->GetMode());
      64              :         }
      65              :     }
      66              : 
      67           48 :     HCCL_INFO("[OpbaseStreamManager::%s] end, get index[%u] slave stream", __func__, slaveIndex);
      68           16 :     return slaves[slaveIndex++].get();
      69              : }
      70              : 
      71           15 : Stream *OpbaseStreamManager::GetSlave(u32 index) const
      72              : {
      73           15 :     if (index >= slaves.size()) {
      74            0 :         THROW<InvalidParamsException>(StringFormat("[OpbaseStreamManager::%s] index[%u] is invalid.", __func__, index));
      75              :     }
      76           15 :     return slaves[index].get();
      77              : }
      78              : 
      79              : } // namespace Hccl