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 "sender.h"
      12              : 
      13              : namespace hccl {
      14            4 : Sender::Sender(const HcclDataType dataType, const HcclReduceOp reductionOp, const u64 reduceAttribute)
      15            4 :     : dataType_(dataType),
      16            4 :       reductionOp_(reductionOp),
      17            4 :       reduceAttribute_(reduceAttribute)
      18            4 : {}
      19              : 
      20            4 : Sender::~Sender() {}
      21              : 
      22            0 : HcclResult Sender::run(
      23              :     const std::shared_ptr<Transport>& link, const u64 dstOffset, DeviceMem& src, Stream& stream,
      24              :     const UserMemType dstMemType) const
      25              : {
      26              :     // server 内通信并且 reduceAttribute_ 也支持，走该分支
      27            0 :     bool isSpInlineReduce = link->IsSpInlineReduce();
      28              :     // 溢出检测为：Warning && INF/NAN 模式时, 支持Write With Reduce
      29            0 :     bool isSpRdmaReduce = RDMA_REDUCE_BITMASK & reduceAttribute_;
      30              : 
      31            0 :     if (link->IsSupportTransportWithReduce()
      32            0 :         && (link->GetLinkType() == LinkType::LINK_STANDARD_ROCE || isSpRdmaReduce)) {
      33              :         // 数据发送端执行Write With Reduce操作
      34            0 :         CHK_RET(link->TxWithReduce(dstMemType, dstOffset, src.ptr(), src.size(), dataType_, reductionOp_, stream));
      35            0 :     } else if (isSpInlineReduce && (INLINE_REDUCE_BITMASK & reduceAttribute_)) {
      36              :         // link支持inline reduce 并且 reduceAttribute_ 也支持
      37              :         // notify 下一个rank做 inline reduce
      38            0 :         CHK_RET(link->TxDataSignal(stream));
      39            0 :     } else {
      40              :         // 向下一个节点发送数据
      41            0 :         CHK_RET(link->TxAsync(UserMemType::OUTPUT_MEM, dstOffset, src.ptr(), src.size(), stream));
      42              :     }
      43              : 
      44            0 :     return HCCL_SUCCESS;
      45              : }
      46              : 
      47            0 : HcclResult Sender::run(
      48              :     const std::shared_ptr<Transport>& link, const std::vector<SenderMemoryInfo>& senderMems, Stream& stream) const
      49              : {
      50            0 :     LinkType linkType = link->GetLinkType();
      51            0 :     bool isSpInlineReduce = link->IsSpInlineReduce();
      52            0 :     bool isSpRdmaReduce = RDMA_REDUCE_BITMASK & reduceAttribute_;
      53            0 :     bool isSpTransportWithReduce = link->IsSupportTransportWithReduce();
      54              : 
      55            0 :     std::vector<TxMemoryInfo> txMems;
      56            0 :     for (const SenderMemoryInfo& senderMem : senderMems) {
      57            0 :         txMems.emplace_back(
      58            0 :             TxMemoryInfo{UserMemType::INPUT_MEM, senderMem.dstOffset, senderMem.src.ptr(), senderMem.src.size()});
      59              :     }
      60              : 
      61            0 :     if (isSpTransportWithReduce && (linkType == LinkType::LINK_STANDARD_ROCE || isSpRdmaReduce)) {
      62            0 :         CHK_RET(link->TxWithReduce(txMems, dataType_, reductionOp_, stream));
      63            0 :     } else if (isSpInlineReduce && (INLINE_REDUCE_BITMASK & reduceAttribute_)) {
      64              :         // link支持inline reduce 并且 reduceAttribute_ 也支持
      65              :         // notify 下一个rank做 inline reduce
      66            0 :         CHK_RET(link->TxDataSignal(stream));
      67            0 :     } else {
      68            0 :         for (TxMemoryInfo& txMem : txMems) {
      69            0 :             txMem.dstMemType = UserMemType::OUTPUT_MEM;
      70              :         }
      71            0 :         CHK_RET(link->TxAsync(txMems, stream));
      72              :     }
      73              : 
      74            0 :     return HCCL_SUCCESS;
      75            0 : }
      76              : 
      77            0 : HcclResult Sender::run(
      78              :     const std::shared_ptr<Transport>& link, const std::vector<SenderMemoryInfo>& senderMems, u32 notifyIdx,
      79              :     Stream& stream) const
      80              : {
      81            0 :     CHK_SMART_PTR_NULL(link);
      82            0 :     bool isSpInlineReduce = link->IsSpInlineReduce();
      83              : 
      84            0 :     std::vector<TxMemoryInfo> txMems;
      85            0 :     for (const SenderMemoryInfo& senderMem : senderMems) {
      86            0 :         txMems.emplace_back(
      87            0 :             TxMemoryInfo{UserMemType::INPUT_MEM, senderMem.dstOffset, senderMem.src.ptr(), senderMem.src.size()});
      88              :     }
      89              : 
      90            0 :     if (isSpInlineReduce && static_cast<bool>((INLINE_REDUCE_BITMASK & reduceAttribute_))) {
      91              :         // link支持inline reduce 并且 reduceAttribute_ 也支持
      92              :         // notify 下一个rank做 inline reduce
      93            0 :         CHK_RET(link->Post(notifyIdx, stream));
      94            0 :     } else {
      95            0 :         for (TxMemoryInfo& txMem : txMems) {
      96            0 :             txMem.dstMemType = UserMemType::OUTPUT_MEM;
      97              :         }
      98            0 :         CHK_RET(link->Post(notifyIdx, stream));
      99              :     }
     100            0 :     return HCCL_SUCCESS;
     101            0 : }
     102              : } // namespace hccl