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 "aiv_ins_preprocessor.h"
      12              : #include "aiv_ins.h"
      13              : #include "env_config_v2.h"
      14              : 
      15              : namespace Hccl {
      16              : 
      17            3 : void AivInsPreprocessor::SetProtocol(uint8_t protocol) { protocol_ = protocol; }
      18              : 
      19            4 : uint8_t AivInsPreprocessor::GetProtocol() const { return protocol_; }
      20              : 
      21            6 : void AivInsPreprocessor::Preprocess(std::shared_ptr<InsQueue>& insQueue) const
      22              : {
      23           18 :     HCCL_INFO("[AivInsPreprocessor::%s] insQueue Preprocess start.", __func__);
      24              : 
      25              :     // 对每主queue中每个ins进行预处理
      26           10 :     for (auto ins = insQueue->Iter(); ins.HasNext(); ++ins) {
      27            4 :         if (ins->GetType() != InstructionType::AIV_INS) {
      28            1 :             continue;
      29              :         }
      30            3 :         InsPreprocess(ins);
      31            6 :     }
      32              : 
      33           18 :     HCCL_INFO("[AivInsPreprocessor::%s] insQueue Preprocess end.", __func__);
      34            6 : }
      35              : 
      36            3 : void AivInsPreprocessor::InsPreprocess(InsIterator& insIter) const
      37              : {
      38            9 :     HCCL_INFO("[AivInsPreprocessor::%s] start.", __func__);
      39              : 
      40            3 :     const AivInstruction& aivIns = dynamic_cast<const AivInstruction&>(*insIter);
      41              : 
      42            3 :     auto links = aivIns.GetLinks();
      43              : 
      44            3 :     if (protocol_ == 0) { // ubmemory
      45            3 :         BatchBuildTransports(links);
      46            0 :     } else if (protocol_ == 1) { // urma
      47            0 :         BatchBuildUrmaTransports(links);
      48              :     } else {
      49            0 :         THROW<InvalidParamsException>(StringFormat("protocol[%u] not supported", protocol_));
      50              :     }
      51              : 
      52            9 :     HCCL_INFO("[AivInsPreprocessor::%s] end.", __func__);
      53            3 : }
      54              : 
      55            3 : void AivInsPreprocessor::BatchBuildTransports(const vector<LinkData>& links) const
      56              : {
      57            9 :     HCCL_INFO("[AivInsPreprocessor::%s] start.", __func__);
      58              : 
      59              :     // 创建MemTransport并进行异步建链、交换
      60            3 :     comm->GetUbMemoryTransportMgr()->BatchCreateTransport(links);
      61              : 
      62            3 :     comm->GetUbMemoryTransportMgr()->TransportsConnect();
      63              : 
      64            9 :     HCCL_INFO("[AivInsPreprocessor::%s] end.", __func__);
      65            3 : }
      66              : 
      67            0 : void AivInsPreprocessor::BatchBuildUrmaTransports(const vector<LinkData>& links) const
      68              : {
      69            0 :     HCCL_RUN_INFO("[AivInsPreprocessor::%s] start.", __func__);
      70              : 
      71            0 :     std::string opTag = comm->GetCurrentCollOperator()->opTag;
      72              : 
      73              :     // 创建RmaConnectiuon
      74            0 :     RmaConnManager& connManager = comm->GetRmaConnManager();
      75            0 :     for (auto& link : links) {
      76            0 :         auto conn = connManager.Create(opTag, link, HrtUbJfcMode::USER_CTL);
      77            0 :         CHECK_NULLPTR(conn, "[AivInsPreprocessor::BatchBuildUrmaTransports] conn is nullptr!");
      78              :     }
      79            0 :     HCCL_INFO("[AivInsPreprocessor::%s] end creating rma connection", __func__);
      80              : 
      81              :     // 创建UrmaDirectTransport并进行异步建链、交换
      82            0 :     auto transportMgr = comm->GetMemTransportManager();
      83            0 :     CHECK_NULLPTR(transportMgr, "[AivInsPreprocessor::BatchBuildUrmaTransports] transportMgr is nullptr!");
      84            0 :     transportMgr->BatchBuildUrmaDirectTransports(links);
      85              : 
      86            0 :     auto timeout = std::chrono::seconds(EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
      87              : 
      88            0 :     HcclUs startTime = std::chrono::steady_clock::now();
      89            0 :     bool isReady = false;
      90            0 :     while (!isReady) {
      91            0 :         isReady = transportMgr->IsAllTransportReady();
      92            0 :         if (isReady) {
      93            0 :             break;
      94              :         }
      95            0 :         if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
      96            0 :             transportMgr->DumpNotReadyTransportsUrma();
      97            0 :             RPT_INPUT_ERR(
      98              :                 true, "EI0006", std::vector<std::string>({"reason"}),
      99              :                 std::vector<std::string>({"Aiv urma wait transports ready timeout."}));
     100            0 :             THROW<InternalException>("Aiv WaitTransportReady timeout, commId[%s].", comm->GetId().c_str());
     101              :             break;
     102              :         }
     103              :     }
     104              : 
     105            0 :     HCCL_INFO("[AivInsPreprocessor::%s] end.", __func__);
     106            0 : }
     107              : 
     108            0 : std::vector<HcclAiRMAWQ> AivInsPreprocessor::GetWqs() const
     109              : {
     110            0 :     HCCL_INFO("[AivInsPreprocessor::%s] start.", __func__);
     111            0 :     if (protocol_ != 1) {
     112            0 :         THROW<InvalidParamsException>(StringFormat("can not get wq info when protocol is [%u]", protocol_));
     113              :     }
     114            0 :     auto memTransportMgr = comm->GetMemTransportManager();
     115            0 :     CHECK_NULLPTR(memTransportMgr, "[AivInsPreprocessor::GetWqs] memTransportMgr is nullptr!");
     116            0 :     return memTransportMgr->GetUrmaWqs();
     117              : }
     118              : 
     119            0 : std::vector<HcclAiRMACQ> AivInsPreprocessor::GetCqs() const
     120              : {
     121            0 :     HCCL_INFO("[AivInsPreprocessor::%s] start.", __func__);
     122            0 :     if (protocol_ != 1) {
     123            0 :         THROW<InvalidParamsException>(StringFormat("can not get cq info when protocol is [%u]", protocol_));
     124              :     }
     125            0 :     auto memTransportMgr = comm->GetMemTransportManager();
     126            0 :     CHECK_NULLPTR(memTransportMgr, "[AivInsPreprocessor::GetCqs] memTransportMgr is nullptr!");
     127            0 :     return memTransportMgr->GetUrmaCqs();
     128              : }
     129              : 
     130          238 : AivInsPreprocessor::~AivInsPreprocessor() {}
     131              : 
     132              : } // namespace Hccl