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