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 157 : }
29 : };
30 :
31 39 : inline void VerifySizeIsEqual(const MemoryBuffer &remoteMemBuf, const MemoryBuffer &localMemBuf, const string &desc)
32 : {
33 39 : if (remoteMemBuf.size != localMemBuf.size) {
34 : string msg = StringFormat("Check %s size is error, localMemBufSize[0x%llx], remoteMemBufSize[0x%llx]",
35 7 : desc.c_str(), localMemBuf.size, remoteMemBuf.size);
36 7 : THROW<InvalidParamsException>(msg);
37 7 : }
38 32 : }
39 :
40 : // EXCHANGEABLE状态用于指示UB Connection可与对端交换信息
41 993 : MAKE_ENUM(RmaConnStatus, INIT, READY, SUSPENDED, CLOSE, CONN_INVALID, EXCHANGEABLE)
42 51 : MAKE_ENUM(WqeMode, DB_SEND, DWQE, WRITE_VALUE)
43 307 : 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
56 : {
57 7 : return rmaConnType;
58 : }
59 :
60 0 : virtual std::vector<char> GetUniqueId() const
61 : {
62 0 : MACRO_THROW(NotSupportException, StringFormat("not supported."));
63 : }
64 :
65 : virtual void Connect() = 0;
66 :
67 : virtual void Close();
68 :
69 : virtual RmaConnStatus GetStatus();
70 :
71 : virtual string Describe() const = 0;
72 :
73 : virtual void Bind(RemoteRmaBuffer *remoteRmaBuf, BufferType bufType);
74 :
75 : virtual RemoteRmaBuffer *GetRemoteRmaBuffer(const BufferType &bufType);
76 :
77 : virtual unique_ptr<BaseTask> PrepareRead(const MemoryBuffer &remoteMemBuf, const MemoryBuffer &localMemBuf,
78 : const SqeConfig &config);
79 :
80 : virtual unique_ptr<BaseTask> PrepareReadReduce(const MemoryBuffer &remoteMemBuf, const MemoryBuffer &localMemBuf,
81 : DataType datatype, ReduceOp reduceOp, const SqeConfig &config);
82 :
83 : virtual unique_ptr<BaseTask> PrepareWrite(const MemoryBuffer &remoteMemBuf, const MemoryBuffer &localMemBuf,
84 : const SqeConfig &config);
85 :
86 : virtual unique_ptr<BaseTask> PrepareWriteReduce(const MemoryBuffer &remoteMemBuf, const MemoryBuffer &localMemBuf,
87 : DataType datatype, ReduceOp reduceOp, const SqeConfig &config);
88 :
89 : virtual unique_ptr<BaseTask> PrepareInlineWrite(const MemoryBuffer &remoteMemBuf, u64 data,
90 : const SqeConfig &config);
91 :
92 : virtual unique_ptr<BaseTask> PrepareWriteWithNotify(const MemoryBuffer &remoteMemBuf,
93 : const MemoryBuffer &localMemBuf, u64 data,
94 : const MemoryBuffer &remoteNotifyMemBuf,
95 : const SqeConfig &config);
96 :
97 : virtual unique_ptr<BaseTask> PrepareWriteReduceWithNotify(const MemoryBuffer &remoteMemBuf,
98 : const MemoryBuffer &localMemBuf, DataType datatype,
99 : ReduceOp reduceOp, u64 data,
100 : const MemoryBuffer &remoteNotifyMemBuf,
101 : const SqeConfig &config);
102 :
103 0 : virtual void AddNop(const Stream &stream)
104 : {
105 : (void)stream;
106 0 : }
107 :
108 0 : virtual bool Suspend()
109 : {
110 0 : MACRO_THROW(NotSupportException, StringFormat("Resume is not supported."));
111 : }
112 :
113 0 : virtual unique_ptr<Serializable> GetExchangeDto() // 序列化本地数据
114 : {
115 0 : MACRO_THROW(NotSupportException, StringFormat("not support."));
116 : }
117 :
118 0 : virtual void ParseRmtExchangeDto(const Serializable &rmtDto) // 解析收到得远端序列化数据
119 : {
120 : (void)rmtDto;
121 0 : MACRO_THROW(NotSupportException, StringFormat("not support."));
122 : }
123 :
124 0 : virtual void ImportRmtDto() // 导入远端的数据
125 : {
126 0 : MACRO_THROW(NotSupportException, StringFormat("not support."));
127 : }
128 :
129 0 : virtual HcclResult Describe(std::string &dfxMsg)
130 : {
131 : (void)dfxMsg;
132 0 : HCCL_ERROR("[RmaConnection::%s] not support.", __func__);
133 0 : return HcclResult::HCCL_E_NOT_SUPPORT;
134 : }
135 :
136 : protected:
137 : RmaConnStatus status;
138 : Socket *socket{nullptr};
139 : RmaConnType rmaConnType;
140 :
141 : unordered_map<BufferType, RemoteRmaBuffer *, std::EnumClassHash> remoteBufs;
142 : };
143 :
144 : } // namespace Hccl
145 :
146 : #endif // HCCLV2_RMA_CONNECTION_H
|