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 : #include "p2p_connection.h"
11 : #include "socket.h"
12 : #include "log.h"
13 : #include "exception_util.h"
14 : #include "rma_conn_exception.h"
15 :
16 : namespace Hccl {
17 :
18 5 : P2PConnection::P2PConnection(Socket *socket, const std::string &tag)
19 5 : : RmaConnection(socket, RmaConnType::P2P)
20 : {
21 15 : HCCL_INFO("P2PConnection::P2PConnection tag = [%s]", tag.c_str());
22 5 : }
23 :
24 0 : void P2PConnection::Connect()
25 : {
26 0 : EnableP2p();
27 0 : GetStatus();
28 0 : }
29 :
30 0 : RmaConnStatus P2PConnection::GetStatus()
31 : {
32 0 : switch (status) {
33 0 : case RmaConnStatus::READY:
34 0 : break;
35 0 : case RmaConnStatus::INIT:
36 0 : if (socket->GetStatus() == SocketStatus::OK) {
37 0 : status = RmaConnStatus::READY;
38 0 : } else if (socket->GetStatus() == SocketStatus::TIMEOUT) {
39 0 : status = RmaConnStatus::CONN_INVALID;
40 : }
41 0 : break;
42 0 : case RmaConnStatus::CLOSE:
43 0 : break;
44 0 : default:
45 0 : auto msg = StringFormat("Invalid status of %s", status.Describe().c_str());
46 0 : THROW<RmaConnException>(msg);
47 : break;
48 0 : }
49 :
50 0 : return status;
51 : }
52 :
53 0 : void P2PConnection::EnableP2p() const
54 : {
55 : // SDMA P2pEnable
56 0 : }
57 :
58 5 : unique_ptr<BaseTask> P2PConnection::PrepareRead(const MemoryBuffer &remoteMemBuf, const MemoryBuffer &localMemBuf,
59 : const SqeConfig &config)
60 : {
61 6 : VerifySizeIsEqual(remoteMemBuf, localMemBuf, "P2PConnection PrepareRead");
62 4 : if (localMemBuf.size == 0) {
63 1 : return nullptr;
64 : }
65 3 : return make_unique<TaskP2pMemcpy>(localMemBuf.addr, remoteMemBuf.addr, localMemBuf.size, MemcpyKind::D2D);
66 : }
67 :
68 5 : unique_ptr<BaseTask> P2PConnection::PrepareReadReduce(const MemoryBuffer &remoteMemBuf, const MemoryBuffer &localMemBuf,
69 : DataType datatype, ReduceOp reduceOp, const SqeConfig &config)
70 : {
71 6 : VerifySizeIsEqual(remoteMemBuf, localMemBuf, "P2PConnection PrepareReadReduce");
72 4 : if (localMemBuf.size == 0) {
73 1 : return nullptr;
74 : }
75 3 : return make_unique<TaskSdmaReduce>(localMemBuf.addr, remoteMemBuf.addr, localMemBuf.size, datatype, reduceOp);
76 : }
77 :
78 1 : unique_ptr<BaseTask> P2PConnection::PrepareWrite(const MemoryBuffer &remoteMemBuf, const MemoryBuffer &localMemBuf,
79 : const SqeConfig &config)
80 : {
81 1 : VerifySizeIsEqual(remoteMemBuf, localMemBuf, "P2PConnection PrepareWrite");
82 1 : if (localMemBuf.size == 0) {
83 0 : return nullptr;
84 : }
85 1 : return make_unique<TaskP2pMemcpy>(remoteMemBuf.addr, localMemBuf.addr, localMemBuf.size, MemcpyKind::D2D);
86 : }
87 :
88 2 : unique_ptr<BaseTask> P2PConnection::PrepareWriteReduce(const MemoryBuffer &remoteMemBuf,
89 : const MemoryBuffer &localMemBuf, DataType datatype,
90 : ReduceOp reduceOp, const SqeConfig &config)
91 : {
92 1 : VerifySizeIsEqual(remoteMemBuf, localMemBuf, "P2PConnection PrepareWriteReduce");
93 1 : if (localMemBuf.size == 0) {
94 0 : return nullptr;
95 : }
96 1 : return make_unique<TaskSdmaReduce>(remoteMemBuf.addr, localMemBuf.addr, localMemBuf.size, datatype, reduceOp);
97 : }
98 :
99 3 : string P2PConnection::Describe() const
100 : {
101 3 : return StringFormat("P2PConnection[status=%s]", status.Describe().c_str());
102 : }
103 :
104 : } // namespace Hccl
|