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              : #ifndef HCCLV2_RMA_CONNECTION_H
      12              : #define HCCLV2_RMA_CONNECTION_H
      13              : #include "task.h"
      14              : #include "virtual_topo.h"
      15              : #include "buffer_type.h"
      16              : #include "remote_rma_buffer.h"
      17              : #include "stream.h"
      18              : #include "socket.h"
      19              : #include "serializable.h"
      20              : 
      21              : namespace Hccl {
      22              : 
      23              : struct MemoryBuffer {
      24              :     u64 addr{0};
      25              :     u64 size{0};
      26              :     u64 memHandle{0};
      27          157 :     MemoryBuffer(u64 addr, u64 size, u64 memHandle) : addr(addr), size(size), memHandle(memHandle) {}
      28              : };
      29              : 
      30           39 : inline void VerifySizeIsEqual(const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, const string& desc)
      31              : {
      32           39 :     if (remoteMemBuf.size != localMemBuf.size) {
      33              :         string msg = StringFormat(
      34            7 :             "Check %s size is error, localMemBufSize[0x%llx], remoteMemBufSize[0x%llx]", desc.c_str(), localMemBuf.size,
      35            7 :             remoteMemBuf.size);
      36            7 :         THROW<InvalidParamsException>(msg);
      37            7 :     }
      38           32 : }
      39              : 
      40              : // EXCHANGEABLE状态用于指示UB Connection可与对端交换信息
      41         1062 : MAKE_ENUM(RmaConnStatus, INIT, READY, SUSPENDED, CLOSE, CONN_INVALID, EXCHANGEABLE)
      42           51 : MAKE_ENUM(WqeMode, DB_SEND, DWQE, WRITE_VALUE)
      43          338 : MAKE_ENUM(RmaConnType, P2P, RDMA, UB, CCU)
      44              : 
      45              : class SqeConfig {
      46              : public:
      47              :     WqeMode wqeMode{WqeMode::DB_SEND};
      48              : };
      49              : 
      50              : class RmaConnection {
      51              : public:
      52              :     explicit RmaConnection(Socket* socket, const RmaConnType rmaConnType);
      53              :     virtual ~RmaConnection();
      54              : 
      55            7 :     RmaConnType GetRmaConnType() const { return rmaConnType; }
      56              : 
      57            0 :     virtual std::vector<char> GetUniqueId() const { MACRO_THROW(NotSupportException, StringFormat("not supported.")); }
      58              : 
      59              :     virtual void Connect() = 0;
      60              : 
      61              :     virtual void Close();
      62              : 
      63              :     virtual RmaConnStatus GetStatus();
      64              : 
      65              :     virtual string Describe() const = 0;
      66              : 
      67              :     virtual void Bind(RemoteRmaBuffer* remoteRmaBuf, BufferType bufType);
      68              : 
      69              :     virtual RemoteRmaBuffer* GetRemoteRmaBuffer(const BufferType& bufType);
      70              : 
      71              :     virtual unique_ptr<BaseTask>
      72              :     PrepareRead(const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, const SqeConfig& config);
      73              : 
      74              :     virtual unique_ptr<BaseTask> PrepareReadReduce(
      75              :         const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, DataType datatype, ReduceOp reduceOp,
      76              :         const SqeConfig& config);
      77              : 
      78              :     virtual unique_ptr<BaseTask>
      79              :     PrepareWrite(const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, const SqeConfig& config);
      80              : 
      81              :     virtual unique_ptr<BaseTask> PrepareWriteReduce(
      82              :         const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, DataType datatype, ReduceOp reduceOp,
      83              :         const SqeConfig& config);
      84              : 
      85              :     virtual unique_ptr<BaseTask>
      86              :     PrepareInlineWrite(const MemoryBuffer& remoteMemBuf, u64 data, const SqeConfig& config);
      87              : 
      88              :     virtual unique_ptr<BaseTask> PrepareWriteWithNotify(
      89              :         const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, u64 data,
      90              :         const MemoryBuffer& remoteNotifyMemBuf, const SqeConfig& config);
      91              : 
      92              :     virtual unique_ptr<BaseTask> PrepareWriteReduceWithNotify(
      93              :         const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, DataType datatype, ReduceOp reduceOp,
      94              :         u64 data, const MemoryBuffer& remoteNotifyMemBuf, const SqeConfig& config);
      95              : 
      96            0 :     virtual void AddNop(const Stream& stream) { (void)stream; }
      97              : 
      98            0 :     virtual bool Suspend() { MACRO_THROW(NotSupportException, StringFormat("Resume is not supported.")); }
      99              : 
     100            0 :     virtual unique_ptr<Serializable> GetExchangeDto() // 序列化本地数据
     101              :     {
     102            0 :         MACRO_THROW(NotSupportException, StringFormat("not support."));
     103              :     }
     104              : 
     105            0 :     virtual void ParseRmtExchangeDto(const Serializable& rmtDto) // 解析收到得远端序列化数据
     106              :     {
     107              :         (void)rmtDto;
     108            0 :         MACRO_THROW(NotSupportException, StringFormat("not support."));
     109              :     }
     110              : 
     111            0 :     virtual void ImportRmtDto() // 导入远端的数据
     112              :     {
     113            0 :         MACRO_THROW(NotSupportException, StringFormat("not support."));
     114              :     }
     115              : 
     116            0 :     virtual HcclResult Describe(std::string& dfxMsg)
     117              :     {
     118              :         (void)dfxMsg;
     119            0 :         HCCL_ERROR("[RmaConnection::%s] not support.", __func__);
     120            0 :         return HcclResult::HCCL_E_NOT_SUPPORT;
     121              :     }
     122              : 
     123              : protected:
     124              :     RmaConnStatus status;
     125              :     Socket* socket{nullptr};
     126              :     RmaConnType rmaConnType;
     127              : 
     128              :     unordered_map<BufferType, RemoteRmaBuffer*, std::EnumClassHash> remoteBufs;
     129              : };
     130              : 
     131              : } // namespace Hccl
     132              : 
     133              : #endif // HCCLV2_RMA_CONNECTION_H