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 "p2p_connection.h"
12 : #include "socket.h"
13 : #include "log.h"
14 : #include "exception_util.h"
15 : #include "rma_conn_exception.h"
16 :
17 : namespace Hccl {
18 :
19 5 : P2PConnection::P2PConnection(Socket* socket, const std::string& tag) : 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(
59 : const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, [[maybe_unused]] 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(
69 : const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, DataType datatype, ReduceOp reduceOp,
70 : [[maybe_unused]] const SqeConfig& config)
71 : {
72 6 : VerifySizeIsEqual(remoteMemBuf, localMemBuf, "P2PConnection PrepareReadReduce");
73 4 : if (localMemBuf.size == 0) {
74 1 : return nullptr;
75 : }
76 3 : return make_unique<TaskSdmaReduce>(localMemBuf.addr, remoteMemBuf.addr, localMemBuf.size, datatype, reduceOp);
77 : }
78 :
79 1 : unique_ptr<BaseTask> P2PConnection::PrepareWrite(
80 : const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, [[maybe_unused]] const SqeConfig& config)
81 : {
82 1 : VerifySizeIsEqual(remoteMemBuf, localMemBuf, "P2PConnection PrepareWrite");
83 1 : if (localMemBuf.size == 0) {
84 0 : return nullptr;
85 : }
86 1 : return make_unique<TaskP2pMemcpy>(remoteMemBuf.addr, localMemBuf.addr, localMemBuf.size, MemcpyKind::D2D);
87 : }
88 :
89 1 : unique_ptr<BaseTask> P2PConnection::PrepareWriteReduce(
90 : const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, DataType datatype, ReduceOp reduceOp,
91 : [[maybe_unused]] const SqeConfig& config)
92 : {
93 1 : VerifySizeIsEqual(remoteMemBuf, localMemBuf, "P2PConnection PrepareWriteReduce");
94 1 : if (localMemBuf.size == 0) {
95 0 : return nullptr;
96 : }
97 1 : return make_unique<TaskSdmaReduce>(remoteMemBuf.addr, localMemBuf.addr, localMemBuf.size, datatype, reduceOp);
98 : }
99 :
100 3 : string P2PConnection::Describe() const { return StringFormat("P2PConnection[status=%s]", status.Describe().c_str()); }
101 :
102 : } // namespace Hccl
|