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), reductionOp_(reductionOp), reduceAttribute_(reduceAttribute)
      16              : {
      17            4 : }
      18              : 
      19            4 : Sender::~Sender()
      20              : {
      21            4 : }
      22              : 
      23            0 : HcclResult Sender::run(const std::shared_ptr<Transport> &link, const u64 dstOffset, DeviceMem &src,
      24              :                        Stream &stream, 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() && (link->GetLinkType() == LinkType::LINK_STANDARD_ROCE ||
      32              :         isSpRdmaReduce)) {
      33              :         // 数据发送端执行Write With Reduce操作
      34            0 :         CHK_RET(link->TxWithReduce(dstMemType, dstOffset, src.ptr(), src.size(), dataType_,
      35              :             reductionOp_, stream));
      36            0 :     } else if (isSpInlineReduce && (INLINE_REDUCE_BITMASK & reduceAttribute_)) {
      37              :         // link支持inline reduce 并且 reduceAttribute_ 也支持
      38              :         // notify 下一个rank做 inline reduce
      39            0 :         CHK_RET(link->TxDataSignal(stream));
      40            0 :     } else {
      41              :         // 向下一个节点发送数据
      42            0 :         CHK_RET(link->TxAsync(UserMemType::OUTPUT_MEM, dstOffset, src.ptr(), src.size(), stream));
      43              :     }
      44              : 
      45            0 :     return HCCL_SUCCESS;
      46              : }
      47              : 
      48            0 : HcclResult Sender::run(const std::shared_ptr<Transport> &link, const std::vector<SenderMemoryInfo> &senderMems,
      49              :     Stream &stream) const
      50              : {
      51            0 :     LinkType linkType = link->GetLinkType();
      52            0 :     bool isSpInlineReduce = link->IsSpInlineReduce();
      53            0 :     bool isSpRdmaReduce = RDMA_REDUCE_BITMASK & reduceAttribute_;
      54            0 :     bool isSpTransportWithReduce = link->IsSupportTransportWithReduce();
      55              : 
      56            0 :     std::vector<TxMemoryInfo> txMems;
      57            0 :     for (const SenderMemoryInfo& senderMem : senderMems) {
      58            0 :         txMems.emplace_back(TxMemoryInfo{UserMemType::INPUT_MEM, senderMem.dstOffset,
      59            0 :             senderMem.src.ptr(), senderMem.src.size()});
      60              :     }
      61              : 
      62            0 :     if (isSpTransportWithReduce && (linkType == LinkType::LINK_STANDARD_ROCE || isSpRdmaReduce)) {
      63            0 :         CHK_RET(link->TxWithReduce(txMems, dataType_, reductionOp_, stream));
      64            0 :     } else if (isSpInlineReduce && (INLINE_REDUCE_BITMASK & reduceAttribute_)) {
      65              :         // link支持inline reduce 并且 reduceAttribute_ 也支持
      66              :         // notify 下一个rank做 inline reduce
      67            0 :         CHK_RET(link->TxDataSignal(stream));
      68            0 :     } else {
      69            0 :         for (TxMemoryInfo& txMem : txMems) {
      70            0 :             txMem.dstMemType = UserMemType::OUTPUT_MEM;
      71              :         }
      72            0 :         CHK_RET(link->TxAsync(txMems, stream));
      73              :     }
      74              : 
      75            0 :     return HCCL_SUCCESS;
      76            0 : }
      77              : 
      78            0 : HcclResult Sender::run(const std::shared_ptr<Transport> &link, const std::vector<SenderMemoryInfo> &senderMems,
      79              :     u32 notifyIdx, 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(TxMemoryInfo{UserMemType::INPUT_MEM, senderMem.dstOffset,
      87            0 :             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