From 4c8fd6b05bc0fb1250dcb65f1ac89c577c41d770 Mon Sep 17 00:00:00 2001 From: Devendar Bureddy Date: Thu, 4 Sep 2025 02:03:21 +0000 Subject: [PATCH 1/5] net v11 plugin update --- include/net.h | 12 ++- include/net_device.h | 3 +- include/net_v10.h | 17 +--- include/net_v11.h | 193 ++++++++++++++++++++++++++++++++++++++++ include/net_v5.h | 95 -------------------- include/net_v7.h | 1 - include/net_v9.h | 15 ++-- include/ucx_uct_lib.h | 70 +++++++++------ include/utils.h | 2 + src/ib_plugin.c | 136 +++++++++++++++++++++------- src/p2p_plugin.c | 53 ++++++----- src/sharp_plugin.c | 101 +++++++++++++-------- src/ucx_plugin.c | 130 ++++++++++++++++++++------- src/ucx_rma_plugin.c | 130 ++++++++++++++++++++------- src/ucx_uct_lib.c | 45 +++++++++- src/ucx_uct_plugin.c | 21 ++++- src/ucx_uct_rd_plugin.c | 23 ++++- src/utils.c | 20 +++++ 18 files changed, 755 insertions(+), 312 deletions(-) create mode 100644 include/net_v11.h delete mode 100644 include/net_v5.h diff --git a/include/net.h b/include/net.h index d808861..be9858c 100644 --- a/include/net.h +++ b/include/net.h @@ -9,10 +9,12 @@ #include #include +#include "net_device.h" #define NCCL_NET_HANDLE_MAXSIZE 128 //Maximum value NCCL can accept for maxP2pBytes and maxCollBytes net properties #define NCCL_MAX_NET_SIZE_BYTES (1*1024*1024*1024*1024L) +#define MAX_COLLNET_SIZE (512*1024*1024L) #define NCCL_NET_OPTIONAL_RECV_COMPLETION 0x1 @@ -22,6 +24,7 @@ // Maximum number of requests per comm object #define NCCL_NET_MAX_REQUESTS 8 +#define NCCL_NET_MAX_DEVS_PER_NIC 4 typedef enum {NCCL_LOG_NONE=0, NCCL_LOG_VERSION=1, NCCL_LOG_WARN=2, NCCL_LOG_INFO=3, NCCL_LOG_ABORT=4, NCCL_LOG_TRACE=5} ncclDebugLogLevel; typedef enum {NCCL_INIT=1, NCCL_COLL=2, NCCL_P2P=4, NCCL_SHM=8, NCCL_NET=16, NCCL_GRAPH=32, NCCL_TUNING=64, NCCL_ENV=128, NCCL_ALLOC=256, NCCL_CALL=512, NCCL_ALL=~0} ncclDebugLogSubSys; @@ -29,11 +32,18 @@ typedef enum {NCCL_INIT=1, NCCL_COLL=2, NCCL_P2P=4, NCCL_SHM=8, NCCL_NET=16, NCC typedef void (*ncclDebugLogger_t)(ncclDebugLogLevel level, unsigned long flags, const char *file, int line, const char *fmt, ...); typedef ncclResult_t (*ncclProfilerCallback_t)(void** eHandle, int type, void* phandle, int64_t pluginId, void* extData); +#include "net_v11.h" #include "net_v10.h" #include "net_v9.h" #include "net_v8.h" #include "net_v7.h" #include "net_v6.h" -#include "net_v5.h" + +typedef ncclNet_v11_t ncclNet_t; +typedef ncclCollNet_v11_t ncclCollNet_t; +typedef ncclNetProperties_v11_t ncclNetProperties_t; +typedef ncclNetVDeviceProps_v11_t ncclNetVDeviceProps_t; +typedef ncclNetCommConfig_v11_t ncclNetCommConfig_t; +typedef ncclNetAttr_v11_t ncclNetAttr_t; #endif // end include guard diff --git a/include/net_device.h b/include/net_device.h index 61d2319..f37ad5d 100644 --- a/include/net_device.h +++ b/include/net_device.h @@ -29,6 +29,7 @@ typedef struct { typedef ncclNetDeviceHandle_v7_t ncclNetDeviceHandle_v8_t; typedef ncclNetDeviceHandle_v8_t ncclNetDeviceHandle_v9_t; typedef ncclNetDeviceHandle_v9_t ncclNetDeviceHandle_v10_t; -typedef ncclNetDeviceHandle_v10_t ncclNetDeviceHandle_t; +typedef ncclNetDeviceHandle_v10_t ncclNetDeviceHandle_v11_t; +typedef ncclNetDeviceHandle_v11_t ncclNetDeviceHandle_t; #endif diff --git a/include/net_v10.h b/include/net_v10.h index f35ad36..751348f 100644 --- a/include/net_v10.h +++ b/include/net_v10.h @@ -8,20 +8,13 @@ #include "net_device.h" -#define NCCL_NET_MAX_DEVS_PER_NIC_V10 4 -#define NCCL_NET_MAX_DEVS_PER_NIC NCCL_NET_MAX_DEVS_PER_NIC_V10 -typedef struct { - int ndevs; - int devs[NCCL_NET_MAX_DEVS_PER_NIC_V10]; -} ncclNetVDeviceProps_v10_t; -typedef ncclNetVDeviceProps_v10_t ncclNetVDeviceProps_t; +typedef ncclNetVDeviceProps_v11_t ncclNetVDeviceProps_v10_t; #define NCCL_NET_TRAFFIC_CLASS_UNDEF -1 typedef struct { // Plugin-specific TC value int trafficClass; } ncclNetCommConfig_v10_t; -typedef ncclNetCommConfig_v10_t ncclNetCommConfig_t; typedef struct { char* name; // Used mostly for logging. @@ -43,7 +36,6 @@ typedef struct { size_t maxCollBytes; // Max transfer size for collective operations } ncclNetProperties_v10_t; -typedef ncclNetProperties_v10_t ncclNetProperties_t; typedef struct { // Name of the network (mainly for logs) @@ -104,11 +96,7 @@ typedef struct { ncclResult_t (*makeVDevice)(int* d, ncclNetVDeviceProps_v10_t* props); } ncclNet_v10_t; -typedef struct { - void* mhandle; - void* address; - size_t size; -} ncclNetSGE_v10_t; +typedef ncclNetSGE_v11_t ncclNetSGE_v10_t; typedef struct { // Name of the collective network (mainly for logs) @@ -159,6 +147,5 @@ typedef struct { ncclResult_t (*makeVDevice)(int* d, ncclNetVDeviceProps_v10_t* props); } ncclCollNet_v10_t; -typedef ncclCollNet_v10_t ncclCollNet_t; #endif // end include guard diff --git a/include/net_v11.h b/include/net_v11.h new file mode 100644 index 0000000..6b8fd3a --- /dev/null +++ b/include/net_v11.h @@ -0,0 +1,193 @@ +/* + * Copyright (c) 2017-2022, NVIDIA CORPORATION. All rights reserved. + */ + + #ifndef NET_V11_H_ + #define NET_V11_H_ + + #include "net_device.h" + + typedef struct { + int ndevs; + int devs[NCCL_NET_MAX_DEVS_PER_NIC]; + } ncclNetVDeviceProps_v11_t; + + #define NCCL_NET_TRAFFIC_CLASS_UNDEF -1 + + typedef struct { + // Plugin-specific TC value + int trafficClass; + } ncclNetCommConfig_v11_t; + + + typedef struct { + char* name; // Used mostly for logging. + char* pciPath; // Path to the PCI device in /sys. + uint64_t guid; // Unique identifier for the NIC chip. Important for + // cards with multiple PCI functions (Physical or virtual). + int ptrSupport; // [NCCL_PTR_HOST|NCCL_PTR_CUDA|NCCL_PTR_DMABUF] + int regIsGlobal; // regMr is not tied to a particular comm + int forceFlush; // Force a flush on receives + int speed; // Port speed in Mbps. + int port; // Port number. + float latency; // Network latency + int maxComms; // Maximum number of comms we can create + int maxRecvs; // Maximum number of grouped receives. + ncclNetDeviceType netDeviceType; // Network offload type + int netDeviceVersion; // Version number for network offload + ncclNetVDeviceProps_v11_t vProps; + size_t maxP2pBytes; // Max transfer size for point-to-point operations + size_t maxCollBytes; // Max transfer size for collective operations + int maxMultiRequestSize; // Maximum number of requests supported in a single multi-request. + } ncclNetProperties_v11_t; + + #define NCCL_NET_ATTR_UNDEF -1 + + #define NCCL_NET_ATTR_INIT { \ + { NCCL_NET_ATTR_UNDEF, NCCL_NET_ATTR_UNDEF, NCCL_NET_ATTR_UNDEF, NCCL_NET_ATTR_UNDEF }, /* sendCommAttr */ \ + { NCCL_NET_ATTR_UNDEF, NCCL_NET_ATTR_UNDEF, NCCL_NET_ATTR_UNDEF, NCCL_NET_ATTR_UNDEF }, /* recvCommAttr */ \ + (uint32_t)NCCL_NET_ATTR_UNDEF, /* op */ \ + (uint32_t)NCCL_NET_ATTR_UNDEF, /* algo */ \ + (uint32_t)NCCL_NET_ATTR_UNDEF, /* proto */ \ + } + + typedef struct { + int32_t maxConcurrentPeers; + int32_t minConcurrentPeers; + int32_t maxFlowsPerPeer; + int32_t minFlowsPerPeer; + } ncclNetCommAttr_v11_t; + + typedef struct { + ncclNetCommAttr_v11_t sendCommAttr; + ncclNetCommAttr_v11_t recvCommAttr; + uint32_t op; + uint32_t algo; + uint32_t proto; + } ncclNetAttr_v11_t; + + typedef struct { + // Name of the network (mainly for logs) + const char* name; + // Initialize the network. + ncclResult_t (*init)(void** ctx, uint64_t commId, ncclNetCommConfig_v11_t* config, ncclDebugLogger_t logFunction, ncclProfilerCallback_t profFunction); + // Return the number of adapters. + ncclResult_t (*devices)(int* ndev); + // Get various device properties. + ncclResult_t (*getProperties)(int dev, ncclNetProperties_v11_t* props); + // Create a receiving object and provide a handle to connect to it. The + // handle can be up to NCCL_NET_HANDLE_MAXSIZE bytes and will be exchanged + // between ranks to create a connection. + ncclResult_t (*listen)(void* ctx, int dev, void* handle, void** listenComm); + // Connect to a handle and return a sending comm object for that peer. + // This call must not block for the connection to be established, and instead + // should return successfully with sendComm == NULL with the expectation that + // it will be called again until sendComm != NULL. + // If *sendDevComm points to a valid object, then NCCL is requesting device offload for this connection + ncclResult_t (*connect)(void* ctx, int dev, void* handle, void** sendComm, ncclNetDeviceHandle_v11_t** sendDevComm); + // Finalize connection establishment after remote peer has called connect. + // This call must not block for the connection to be established, and instead + // should return successfully with recvComm == NULL with the expectation that + // it will be called again until recvComm != NULL. + // If *recvDevComm points to a valid object, then NCCL is requesting device offload for this connection + ncclResult_t (*accept)(void* listenComm, void** recvComm, ncclNetDeviceHandle_v11_t** recvDevComm); + // Register/Deregister memory. Comm can be either a sendComm or a recvComm. + // Type is either NCCL_PTR_HOST or NCCL_PTR_CUDA. + ncclResult_t (*regMr)(void* comm, void* data, size_t size, int type, void** mhandle); + /* DMA-BUF support */ + ncclResult_t (*regMrDmaBuf)(void* comm, void* data, size_t size, int type, uint64_t offset, int fd, void** mhandle); + ncclResult_t (*deregMr)(void* comm, void* mhandle); + // Asynchronous send to a peer. + // May return request == NULL if the call cannot be performed (or would block) + ncclResult_t (*isend)(void* sendComm, void* data, size_t size, int tag, void* mhandle, void* phandle, void** request); + // Asynchronous recv from a peer. + // May return request == NULL if the call cannot be performed (or would block) + ncclResult_t (*irecv)(void* recvComm, int n, void** data, size_t* sizes, int* tags, void** mhandles, void** phandles, void** request); + // Perform a flush/fence to make sure all data received with NCCL_PTR_CUDA is + // visible to the GPU + ncclResult_t (*iflush)(void* recvComm, int n, void** data, int* sizes, void** mhandles, void** request); + // Test whether a request is complete. If size is not NULL, it returns the + // number of bytes sent/received. + ncclResult_t (*test)(void* request, int* done, int* sizes); + // Close and free send/recv comm objects + ncclResult_t (*closeSend)(void* sendComm); + ncclResult_t (*closeRecv)(void* recvComm); + ncclResult_t (*closeListen)(void* listenComm); + + // Copy the given mhandle to a dptr in a format usable by this plugin's device code + ncclResult_t (*getDeviceMr)(void* comm, void* mhandle, void** dptr_mhandle); + + // Notify the plugin that a recv has completed by the device + ncclResult_t (*irecvConsumed)(void* recvComm, int n, void* request); + + // Virtual NIC APIs. makeVDevice will create a virtual NIC given the specified properties, and tell the caller + // what index this new vNIC exists at + ncclResult_t (*makeVDevice)(int* d, ncclNetVDeviceProps_v11_t* props); + // Finalize the network. + ncclResult_t (*finalize)(void* ctx); + + ncclResult_t (*setNetAttr)(void* ctx, ncclNetAttr_v11_t* netAttr); + } ncclNet_v11_t; + + typedef struct { + void* mhandle; + void* address; + size_t size; + } ncclNetSGE_v11_t; + + typedef struct { + // Name of the collective network (mainly for logs) + const char* name; + // Initialize the collective network. + ncclResult_t (*init)(void** ctx, uint64_t commId, ncclDebugLogger_t logFunction); + // Return the number of adapters capable of doing collective operations. + // If ndev returns 0, all other functions might be set to NULL. + ncclResult_t (*devices)(int* ndev); + // Get various device properties. + ncclResult_t (*getProperties)(int dev, ncclNetProperties_v11_t* props); + // Create a receiving object and provide a handle to connect to it. The + // handle can be up to NCCL_NET_HANDLE_MAXSIZE bytes and will be exchanged + // between ranks to create connections. + ncclResult_t (*listen)(void* ctx, int dev, void* handle, void** listenComm); + // Create a group for collective operations. handles have been created + // using listen() above. rank indicates caller's rank in the collective network. + ncclResult_t (*connect)(void* handles[], int nranks, int rank, void* listenComm, void** collComm); + // Returns whether a reduction operation on a data type is supported. + // 1 for supported, 0 otherwise. + ncclResult_t (*reduceSupport)(ncclDataType_t dataType, ncclRedOp_t redOp, int* supported); + // Register/Deregister memory. Type is either NCCL_PTR_HOST or NCCL_PTR_CUDA. + ncclResult_t (*regMr)(void* collComm, void* data, size_t size, int type, void** mhandle); + /* DMA-BUF support */ + ncclResult_t (*regMrDmaBuf)(void* collComm, void* data, size_t size, int type, uint64_t offset, int fd, void** mhandle); + ncclResult_t (*deregMr)(void* collComm, void* mhandle); + // Performs an asynchronous allreduce operation on the collective group. + // May return request == NULL if the call cannot be performed (or would block). + ncclResult_t (*iallreduce)(void* collComm, void* sendData, void* recvData, size_t count, + ncclDataType_t dataType, ncclRedOp_t redOp, void* sendMhandle, void* recvMhandle, void** request); + ncclResult_t (*iallgather)(void* collComm, void* sendData, int nRecvParts, ncclNetSGE_v11_t* recvParts, + size_t bytesPerRank, size_t windowOffset, size_t windowBytes, + void* sendMhandle, void** request); + ncclResult_t (*ireducescatter)(void* collComm, int nSendParts, ncclNetSGE_v11_t* sendParts, void* recvData, + size_t bytesPerRank, size_t windowOffset, size_t windowBytes, + ncclDataType_t dataType, ncclRedOp_t redOp, + void* recvMhandle, void** request); + // Perform a flush/fence to make sure all data received with NCCL_PTR_CUDA is + // visible to the GPU + ncclResult_t (*iflush)(void* collComm, void* data, int size, void* mhandle, void** request); + // Test whether a request is complete. If size is not NULL, it returns the + // number of bytes sent/received. + ncclResult_t (*test)(void* request, int* done, int* size); + // Close and free collective comm objects + ncclResult_t (*closeColl)(void* collComm); + ncclResult_t (*closeListen)(void* listenComm); + + // Create a virtual NIC given the specified properties, which can be accessed at device index d + ncclResult_t (*makeVDevice)(int* d, ncclNetVDeviceProps_v11_t* props); + // Finalize the collective network. + ncclResult_t (*finalize)(void* ctx); + } ncclCollNet_v11_t; + + typedef ncclCollNet_v11_t ncclCollNet_t; + + #endif // end include guard + \ No newline at end of file diff --git a/include/net_v5.h b/include/net_v5.h deleted file mode 100644 index 27743f2..0000000 --- a/include/net_v5.h +++ /dev/null @@ -1,95 +0,0 @@ -/* - * SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES - * Copyright (c) 2017-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. - * SPDX-License-Identifier: BSD-3-Clause - */ - -#ifndef NCCL_NET_V5_H_ -#define NCCL_NET_V5_H_ - -typedef ncclNetProperties_v6_t ncclNetProperties_v5_t; -typedef struct { - // Name of the network (mainly for logs) - const char* name; - // Initialize the network. - ncclResult_t (*init)(ncclDebugLogger_t logFunction); - // Return the number of adapters. - ncclResult_t (*devices)(int* ndev); - // Get various device properties. - ncclResult_t (*getProperties)(int dev, ncclNetProperties_v5_t* props); - // Create a receiving object and provide a handle to connect to it. The - // handle can be up to NCCL_NET_HANDLE_MAXSIZE bytes and will be exchanged - // between ranks to create a connection. - ncclResult_t (*listen)(int dev, void* handle, void** listenComm); - // Connect to a handle and return a sending comm object for that peer. - // This call must not block for the connection to be established, and instead - // should return successfully with sendComm == NULL with the expectation that - // it will be called again until sendComm != NULL. - ncclResult_t (*connect)(int dev, void* handle, void** sendComm); - // Finalize connection establishment after remote peer has called connect. - // This call must not block for the connection to be established, and instead - // should return successfully with recvComm == NULL with the expectation that - // it will be called again until recvComm != NULL. - ncclResult_t (*accept)(void* listenComm, void** recvComm); - // Register/Deregister memory. Comm can be either a sendComm or a recvComm. - // Type is either NCCL_PTR_HOST or NCCL_PTR_CUDA. - ncclResult_t (*regMr)(void* comm, void* data, int size, int type, void** mhandle); - ncclResult_t (*deregMr)(void* comm, void* mhandle); - // Asynchronous send to a peer. - // May return request == NULL if the call cannot be performed (or would block) - ncclResult_t (*isend)(void* sendComm, void* data, int size, int tag, void* mhandle, void** request); - // Asynchronous recv from a peer. - // May return request == NULL if the call cannot be performed (or would block) - ncclResult_t (*irecv)(void* recvComm, int n, void** data, int* sizes, int* tags, void** mhandles, void** request); - // Perform a flush/fence to make sure all data received with NCCL_PTR_CUDA is - // visible to the GPU - ncclResult_t (*iflush)(void* recvComm, int n, void** data, int* sizes, void** mhandles, void** request); - // Test whether a request is complete. If size is not NULL, it returns the - // number of bytes sent/received. - ncclResult_t (*test)(void* request, int* done, int* sizes); - // Close and free send/recv comm objects - ncclResult_t (*closeSend)(void* sendComm); - ncclResult_t (*closeRecv)(void* recvComm); - ncclResult_t (*closeListen)(void* listenComm); -} ncclNet_v5_t; - - -typedef struct { - // Name of the collective network (mainly for logs) - const char* name; - // Initialize the collective network. - ncclResult_t (*init)(ncclDebugLogger_t logFunction); - // Return the number of adapters capable of doing collective operations. - // If ndev returns 0, all other functions might be set to NULL. - ncclResult_t (*devices)(int* ndev); - // Get various device properties. - ncclResult_t (*getProperties)(int dev, ncclNetProperties_v5_t* props); - // Create a receiving object and provide a handle to connect to it. The - // handle can be up to NCCL_NET_HANDLE_MAXSIZE bytes and will be exchanged - // between ranks to create connections. - ncclResult_t (*listen)(int dev, void* handle, void** listenComm); - // Create a group for collective operations. handles have been created - // using listen() above. rank indicates caller's rank in the collective network. - ncclResult_t (*connect)(void* handles[], int nranks, int rank, void* listenComm, void** collComm); - // Returns whether a reduction operation on a data type is supported. - // 1 for supported, 0 otherwise. - ncclResult_t (*reduceSupport)(ncclDataType_t dataType, ncclRedOp_t redOp, int* supported); - // Register/Deregister memory. Type is either NCCL_PTR_HOST or NCCL_PTR_CUDA. - ncclResult_t (*regMr)(void* collComm, void* data, int size, int type, void** mhandle); - ncclResult_t (*deregMr)(void* collComm, void* mhandle); - // Performs an asynchronous allreduce operation on the collective group. - // May return request == NULL if the call cannot be performed (or would block). - ncclResult_t (*iallreduce)(void* collComm, void* sendData, void* recvData, int count, - ncclDataType_t dataType, ncclRedOp_t redOp, void* sendMhandle, void* recvMhandle, void** request); - // Perform a flush/fence to make sure all data received with NCCL_PTR_CUDA is - // visible to the GPU - ncclResult_t (*iflush)(void* collComm, void* data, int size, void* mhandle, void** request); - // Test whether a request is complete. If size is not NULL, it returns the - // number of bytes sent/received. - ncclResult_t (*test)(void* request, int* done, int* size); - // Close and free collective comm objects - ncclResult_t (*closeColl)(void* collComm); - ncclResult_t (*closeListen)(void* listenComm); -} ncclCollNet_v5_t; - -#endif diff --git a/include/net_v7.h b/include/net_v7.h index 1642335..b8f5902 100644 --- a/include/net_v7.h +++ b/include/net_v7.h @@ -7,7 +7,6 @@ #ifndef NCCL_NET_V7_H_ #define NCCL_NET_V7_H_ -#include "net_device.h" typedef struct { char* name; // Used mostly for logging. diff --git a/include/net_v9.h b/include/net_v9.h index c5ffd48..568a241 100644 --- a/include/net_v9.h +++ b/include/net_v9.h @@ -13,10 +13,7 @@ #define NCCL_NET_MAX_DEVS_PER_NIC_V9 4 -typedef struct { - int ndevs; - int devs[NCCL_NET_MAX_DEVS_PER_NIC_V9]; -} ncclNetVDeviceProps_v9_t; +typedef ncclNetVDeviceProps_v11_t ncclNetVDeviceProps_v9_t; typedef struct { char* name; // Used mostly for logging. @@ -93,9 +90,11 @@ typedef struct { ncclResult_t (*irecvConsumed)(void* recvComm, int n, void* request); // Create a virtual NIC given the specified properties, which can be accessed at device index d - ncclResult_t (*makeVDevice)(int* d, ncclNetVDeviceProps_t* props); + ncclResult_t (*makeVDevice)(int* d, ncclNetVDeviceProps_v9_t* props); } ncclNet_v9_t; +typedef ncclNetSGE_v11_t ncclNetSGE_v9_t; + typedef struct { // Name of the collective network (mainly for logs) const char* name; @@ -125,10 +124,10 @@ typedef struct { // May return request == NULL if the call cannot be performed (or would block). ncclResult_t (*iallreduce)(void* collComm, void* sendData, void* recvData, size_t count, ncclDataType_t dataType, ncclRedOp_t redOp, void* sendMhandle, void* recvMhandle, void** request); - ncclResult_t (*iallgather)(void* collComm, void* sendData, int nRecvParts, ncclNetSGE_v10_t* recvParts, + ncclResult_t (*iallgather)(void* collComm, void* sendData, int nRecvParts, ncclNetSGE_v9_t* recvParts, size_t bytesPerRank, size_t windowOffset, size_t windowBytes, void* sendMhandle, void** request); - ncclResult_t (*ireducescatter)(void* collComm, int nSendParts, ncclNetSGE_v10_t* sendParts, void* recvData, + ncclResult_t (*ireducescatter)(void* collComm, int nSendParts, ncclNetSGE_v9_t* sendParts, void* recvData, size_t bytesPerRank, size_t windowOffset, size_t windowBytes, ncclDataType_t dataType, ncclRedOp_t redOp, void* recvMhandle, void** request); @@ -143,7 +142,7 @@ typedef struct { ncclResult_t (*closeListen)(void* listenComm); // Create a virtual NIC given the specified properties, which can be accessed at device index d - ncclResult_t (*makeVDevice)(int* d, ncclNetVDeviceProps_t* props); + ncclResult_t (*makeVDevice)(int* d, ncclNetVDeviceProps_v9_t* props); } ncclCollNet_v9_t; #endif // end include guard diff --git a/include/ucx_uct_lib.h b/include/ucx_uct_lib.h index 49986c0..bec0d09 100644 --- a/include/ucx_uct_lib.h +++ b/include/ucx_uct_lib.h @@ -219,10 +219,10 @@ ncclResult_t nccl_uct_flush(nccl_uct_comm_t *base_comm, void *data, int size, void nccl_uct_empty_callback(uct_completion_t *comp); /* NCCL common plugin callbacks */ -ncclResult_t nccl_uct_listen(int dev, void *listen_handle, void **listen_comm); +ncclResult_t nccl_uct_listen(void *ctx, int dev, void *listen_handle, void **listen_comm); ncclResult_t nccl_uct_accept(void *listen_comm, void **recv_comm, ncclNetDeviceHandle_v7_t **recvDevComm); -ncclResult_t nccl_uct_connect(int dev, ncclNetCommConfig_t* config, void *listen_handle, void **send_comm, +ncclResult_t nccl_uct_connect(void *ctx, int dev, void *listen_handle, void **send_comm, ncclNetDeviceHandle_t **sendDevComm); ncclResult_t nccl_uct_close_listen(void *listen_comm); ncclResult_t nccl_uct_reg_mr_dmabuf(void *reg_comm, void *data, size_t size, @@ -231,8 +231,15 @@ ncclResult_t nccl_uct_reg_mr_dmabuf(void *reg_comm, void *data, size_t size, ncclResult_t nccl_uct_reg_mr(void *reg_comm, void *data, size_t size, int type, void **mhandle); ncclResult_t nccl_uct_dereg_mr(void *dereg_comm, void *mhandle); +ncclResult_t nccl_uct_finalize(void *ctx); +ncclResult_t nccl_uct_setNetAttr(void *ctx, ncclNetAttr_v11_t *netAttr); /* Compatibility callback */ +ncclResult_t nccl_uct_connect_v10(int dev, ncclNetCommConfig_v10_t* config, void* opaqueHandle, + void** sendComm, ncclNetDeviceHandle_v10_t** sendDevComm); +ncclResult_t nccl_uct_listen_v10(int dev, void* opaqueHandle, void** listenComm); +ncclResult_t nccl_uct_get_properties_v10(int dev, + ncclNetProperties_v10_t *props_v10); ncclResult_t nccl_uct_get_properties_v9(int dev, ncclNetProperties_v9_t *props_v9); ncclResult_t nccl_uct_get_properties_v8(int dev, @@ -250,6 +257,34 @@ ncclResult_t nccl_uct_accept_v6(void *listen_comm, void **recv_comm); ncclResult_t nccl_uct_get_properties(int dev, ncclNetProperties_t *props); +#define NCCL_UCT_PLUGIN_BASE_V11(plugin_name, prefix, init_func, get_properties_func, \ + connect_func, accept_func, reg_mr_func, \ + isend_func, irecv_func) \ +{ \ +.name = plugin_name, \ +.init = prefix##_init, \ +.devices = nccl_uct_devices, \ +.getProperties = get_properties_func, \ +.listen = nccl_uct_listen, \ +.connect = connect_func, \ +.accept = accept_func, \ +.regMr = reg_mr_func, \ +.regMrDmaBuf = nccl_uct_reg_mr_dmabuf, \ +.deregMr = nccl_uct_dereg_mr, \ +.isend = prefix##_##isend_func, \ +.irecv = prefix##_##irecv_func, \ +.iflush = prefix##_iflush, \ +.test = prefix##_test, \ +.closeSend = prefix##_close, \ +.closeRecv = prefix##_close, \ +.closeListen = nccl_uct_close_listen, \ +.getDeviceMr = NULL, \ +.irecvConsumed = NULL, \ +.makeVDevice = NULL, \ +.finalize = prefix##_finalize, \ +.setNetAttr = nccl_uct_setNetAttr \ +} + #define NCCL_UCT_PLUGIN_BASE(plugin_name, prefix, init_func, get_properties_func, \ connect_func, accept_func, reg_mr_func, \ isend_func, irecv_func) \ @@ -258,7 +293,7 @@ ncclResult_t nccl_uct_get_properties(int dev, ncclNetProperties_t *props); .init = prefix##_##init_func, \ .devices = nccl_uct_devices, \ .getProperties = get_properties_func, \ - .listen = nccl_uct_listen, \ + .listen = nccl_uct_listen_v10 , \ .connect = connect_func, \ .accept = accept_func, \ .regMr = reg_mr_func, \ @@ -273,11 +308,16 @@ ncclResult_t nccl_uct_get_properties(int dev, ncclNetProperties_t *props); .closeListen = nccl_uct_close_listen \ } -#define NCCL_UCT_PLUGIN_V10(plugin_name, prefix) \ - NCCL_UCT_PLUGIN_BASE(plugin_name, prefix, init, nccl_uct_get_properties, \ +#define NCCL_UCT_PLUGIN_V11(plugin_name, prefix) \ + NCCL_UCT_PLUGIN_BASE_V11(plugin_name, prefix, init, nccl_uct_get_properties, \ nccl_uct_connect, nccl_uct_accept, nccl_uct_reg_mr, \ isend, irecv) +#define NCCL_UCT_PLUGIN_V10(plugin_name, prefix) \ + NCCL_UCT_PLUGIN_BASE(plugin_name, prefix, init_v10, nccl_uct_get_properties_v10, \ + nccl_uct_connect_v10, nccl_uct_accept, nccl_uct_reg_mr, \ + isend, irecv) + #define NCCL_UCT_PLUGIN_V9(plugin_name, prefix) \ NCCL_UCT_PLUGIN_BASE(plugin_name, prefix, init_v9, nccl_uct_get_properties_v9, \ nccl_uct_connect_v9, nccl_uct_accept, nccl_uct_reg_mr, \ @@ -298,24 +338,4 @@ ncclResult_t nccl_uct_get_properties(int dev, ncclNetProperties_t *props); nccl_uct_connect_v6, nccl_uct_accept_v6, \ nccl_uct_reg_mr_v7, isend_v8, irecv_v8) -#define NCCL_UCT_PLUGIN_V5(plugin_name, prefix) \ - { \ - .name = plugin_name, \ - .init = prefix##_init_v9, \ - .devices = nccl_uct_devices, \ - .getProperties = nccl_uct_get_properties_v6, \ - .listen = nccl_uct_listen, \ - .connect = nccl_uct_connect_v6, \ - .accept = nccl_uct_accept_v6, \ - .regMr = nccl_uct_reg_mr_v7, \ - .deregMr = nccl_uct_dereg_mr, \ - .isend = prefix##_isend_v8, \ - .irecv = prefix##_irecv_v8, \ - .iflush = prefix##_iflush, \ - .test = prefix##_test, \ - .closeSend = prefix##_close, \ - .closeRecv = prefix##_close, \ - .closeListen = nccl_uct_close_listen \ - } - #endif /* NCCL_UCX_UCT_LIB_H_ */ diff --git a/include/utils.h b/include/utils.h index 11259c5..1a37f06 100644 --- a/include/utils.h +++ b/include/utils.h @@ -11,6 +11,7 @@ #include "nccl.h" #include +#include #define NCCL_STATIC_ASSERT(_cond, _msg) \ switch(0) {case 0:case (_cond):;} @@ -29,5 +30,6 @@ struct netIf { int parseStringList(const char* string, struct netIf* ifList, int maxList); int matchIfList(const char* string, int port, struct netIf* ifList, int listSize, int matchExact); const char *get_plugin_lib_path(); +void plugin_get_device_name(const char *input, char *output, size_t output_size); #endif diff --git a/src/ib_plugin.c b/src/ib_plugin.c index f8db0a2..574712e 100644 --- a/src/ib_plugin.c +++ b/src/ib_plugin.c @@ -30,6 +30,8 @@ static char ncclIbIfName[MAX_IF_NAME_SIZE+1]; static union ncclSocketAddress ncclIbIfAddr; +static ncclNetCommConfig_t ibContext; + pthread_mutex_t ncclIbLock = PTHREAD_MUTEX_INITIALIZER; int ncclIbRelaxedOrderingEnabled = 0; @@ -55,6 +57,14 @@ NCCL_PARAM(IbFifoTc, "IB_FIFO_TC", 0); NCCL_PARAM(IbAsyncEvents,"IB_RETURN_ASYNC_EVENTS",1); NCCL_PARAM(IbEceEnable,"IB_ECE_ENABLE",1); +// With ncclNet_v11_t the NCCL core initializes the network plugin per-communicator +// rather than once for all communicators. However, the internal plugin implementation +// still assumes the plugin is initialized only once across all communicators. The ref +// counter makes sure the plugin internally initializes only once. When per communicator +// context support is added to the plugin the ref counter can be removed. +static int ibRefCount = 0; + + static ncclResult_t ncclIbStatsCheckFatalCount(struct ncclIbStats* stat, const char* funcName) { if (ncclParamIbAsyncEvents() && __atomic_load_n(&stat->fatalErrorCount, __ATOMIC_RELAXED)) { WARN("communicator encountered a fatal error (detected in %s)\n", funcName); @@ -337,6 +347,31 @@ ncclResult_t ncclIbGetProperties(int dev, ncclNetProperties_t* props) return nccl_p2p_ib_get_properties(ncclIbDevs, ncclNMergedIbDevs, dev, props); } +ncclResult_t ncclIbGetProperties_v10(int dev, ncclNetProperties_v10_t* props_v10) { + ncclNetProperties_t props; + ncclResult_t ret = nccl_p2p_ib_get_properties(ncclIbDevs, ncclNMergedIbDevs, dev, &props); + if (ret != ncclSuccess) return ret; + props_v10->name = props.name; + props_v10->pciPath = props.pciPath; + props_v10->guid = props.guid; + props_v10->ptrSupport = props.ptrSupport; + props_v10->regIsGlobal = props.regIsGlobal; + props_v10->forceFlush = props.forceFlush; + props_v10->speed = props.speed; + props_v10->port = props.port; + props_v10->latency = props.latency; + props_v10->maxComms = props.maxComms; + props_v10->maxRecvs = props.maxRecvs; + props_v10->netDeviceType = props.netDeviceType; + props_v10->netDeviceVersion = props.netDeviceVersion; + props_v10->maxP2pBytes = props.maxP2pBytes; + props_v10->maxCollBytes = props.maxCollBytes; + props_v10->vProps.ndevs = props.vProps.ndevs; + for(int i=0; ivProps.devs[i] = props.vProps.devs[i]; + } + return ncclSuccess; +} ncclResult_t ncclIbGetProperties_v9(int dev, ncclNetProperties_v9_t* props_v9) { ncclNetProperties_t props; @@ -603,7 +638,14 @@ struct ncclIbRecvComm { int flushEnabled; }; -ncclResult_t ncclIbInit(ncclDebugLogger_t logFunction, ncclProfilerCallback_t profFunction) { +ncclResult_t ncclIbSetNetAttr(void *ctx, ncclNetAttr_t *netAttr) { + (void)ctx; + (void)netAttr; + return ncclSuccess; +} + +ncclResult_t ncclIbInit(void** ctx, uint64_t commId, ncclNetCommConfig_v11_t* config, ncclDebugLogger_t logFunction, ncclProfilerCallback_t profFunction) { + if (ibRefCount++) return ncclSuccess; if (ncclParamIbDisable()) return ncclInternalError; // The SendFifo needs to be 32-byte aligned and each element needs @@ -617,7 +659,11 @@ ncclResult_t ncclIbInit(ncclDebugLogger_t logFunction, ncclProfilerCallback_t pr NCCL_STATIC_ASSERT((offsetof(struct ncclIbRecvComm, remFifo) % 32) == 0, "ncclIbSendComm fifo must be 32-byte aligned"); - return nccl_p2p_ib_init(&ncclNIbDevs, &ncclNMergedIbDevs, ncclIbDevs, ncclIbIfName, &ncclIbIfAddr, &ncclIbAsyncThread, logFunction); + ncclResult_t ret = nccl_p2p_ib_init(&ncclNIbDevs, &ncclNMergedIbDevs, ncclIbDevs, ncclIbIfName, &ncclIbIfAddr, &ncclIbAsyncThread, logFunction); + if (ret != ncclSuccess) return ret; + if (config) ibContext.trafficClass = config->trafficClass; + if (ctx) *ctx = (void*)&ibContext; + return ncclSuccess; } NCCL_PARAM(IbQpsPerConn, "IB_QPS_PER_CONNECTION", 1); @@ -750,7 +796,7 @@ ncclResult_t ncclIbRtsQp(struct ibv_qp* qp) { return ncclSuccess; } -ncclResult_t ncclIbListen(int dev, void* opaqueHandle, void** listenComm) { +ncclResult_t ncclIbListen(void *ctx, int dev, void* opaqueHandle, void** listenComm) { ncclResult_t ret = ncclSuccess; struct ncclIbListenComm* comm; comm = malloc(sizeof(struct ncclIbListenComm)); @@ -772,7 +818,7 @@ ncclResult_t ncclIbListen(int dev, void* opaqueHandle, void** listenComm) { goto exit; } -ncclResult_t ncclIbConnect(int dev, ncclNetCommConfig_t* config, void* opaqueHandle, void** sendComm, ncclNetDeviceHandle_t** sendDevComm) { +ncclResult_t ncclIbConnect(void *ctx, int dev, void* opaqueHandle, void** sendComm, ncclNetDeviceHandle_t** sendDevComm) { ncclResult_t ret = ncclSuccess; struct ncclIbHandle* handle = (struct ncclIbHandle*) opaqueHandle; struct ncclIbCommStage* stage = &handle->stage; @@ -835,6 +881,7 @@ ncclResult_t ncclIbConnect(int dev, ncclNetCommConfig_t* config, void* opaqueHan if (stage->offset != sizeof(ncclNetVDeviceProps_t)) return ncclSuccess; stage->offset = 0; ncclNetVDeviceProps_t remoteVProps; + ncclNetCommConfig_t* config; memcpy(&remoteVProps, stage->buffer, sizeof(ncclNetVDeviceProps_t)); mergedDev = ncclIbMergedDevs + dev; comm->base.vProps = mergedDev->vProps; @@ -930,6 +977,7 @@ ncclResult_t ncclIbConnect(int dev, ncclNetCommConfig_t* config, void* opaqueHan } } + config = (ncclNetCommConfig_t*)ctx; meta.fifoAddr = (uint64_t)comm->fifo; meta.sl = (ncclParamIbSl() != -1) ? ncclParamIbSl() : (config && config->trafficClass != NCCL_NET_TRAFFIC_CLASS_UNDEF) ? config->trafficClass : NCCL_IB_SL_DEFAULT; meta.tc = (ncclParamIbTc() != -1) ? ncclParamIbTc() : (config && config->trafficClass != NCCL_NET_TRAFFIC_CLASS_UNDEF) ? config->trafficClass : NCCL_IB_TC_DEFAULT; @@ -2019,12 +2067,29 @@ ncclResult_t ncclIbMakeVDevice(int* d, ncclNetVDeviceProps_t* props) { return res; } +ncclResult_t ncclIbFinalize(void* ctx) { + ibRefCount--; + return ncclSuccess; +} + +ncclResult_t ncclIbInit_v10(ncclDebugLogger_t logFunction, ncclProfilerCallback_t profilerCallback) { + return ncclIbInit(NULL, 0, NULL, logFunction, NULL); + } + +ncclResult_t ncclIbConnect_v10(int dev, ncclNetCommConfig_v10_t* config, void* opaqueHandle, void** sendComm, ncclNetDeviceHandle_v10_t** sendDevComm) { + return ncclIbConnect(NULL, dev, opaqueHandle, sendComm, sendDevComm); +} + +ncclResult_t ncclIbListen_v10(int dev, void* opaqueHandle, void** listenComm) { + return ncclIbListen(NULL, dev, opaqueHandle, listenComm); +} + ncclResult_t ncclIbConnect_v9(int dev, void* opaqueHandle, void** sendComm, ncclNetDeviceHandle_t** sendDevComm) { - return ncclIbConnect(dev, NULL, opaqueHandle, sendComm, sendDevComm); + return ncclIbConnect_v10(dev, NULL, opaqueHandle, sendComm, sendDevComm); } ncclResult_t ncclIbInit_v9(ncclDebugLogger_t logFunction) { - return ncclIbInit(logFunction, NULL); + return ncclIbInit_v10(logFunction, NULL); } ncclResult_t ncclIbIsend_v9(void* sendComm, void* data, size_t size, int tag, void* mhandle, void** request) { @@ -2061,9 +2126,8 @@ ncclResult_t ncclIbAccept_v6(void* listenComm, void** recvComm) { return ncclIbAccept(listenComm, recvComm, &handle); } - -const ncclNet_v10_t ibPlugin_v10 = { - .name = "IBext_v10", +const ncclNet_v11_t ibPlugin_v11 = { + .name = "IBext_v11", .init = ncclIbInit, .devices = ncclIbDevices, .getProperties = ncclIbGetProperties, @@ -2082,6 +2146,31 @@ const ncclNet_v10_t ibPlugin_v10 = { .closeListen = ncclIbCloseListen, NULL /* getDeviceMr */, NULL /* irecvConsumed */, + ncclIbMakeVDevice, + ncclIbFinalize, + ncclIbSetNetAttr +}; + +const ncclNet_v10_t ibPlugin_v10 = { + .name = "IBext_v10", + .init = ncclIbInit_v10, + .devices = ncclIbDevices, + .getProperties = ncclIbGetProperties_v10, + .listen = ncclIbListen_v10, + .connect = ncclIbConnect_v10, + .accept = ncclIbAccept, + .regMr = ncclIbRegMr, + .regMrDmaBuf = ncclIbRegMrDmaBuf, + .deregMr = ncclIbDeregMr, + .isend = ncclIbIsend, + .irecv = ncclIbIrecv, + .iflush = ncclIbIflush, + .test = ncclIbTest, + .closeSend = ncclIbCloseSend, + .closeRecv = ncclIbCloseRecv, + .closeListen = ncclIbCloseListen, + NULL /* getDeviceMr */, + NULL /* irecvConsumed */, ncclIbMakeVDevice }; @@ -2090,7 +2179,7 @@ const ncclNet_v9_t ibPlugin_v9 = { .init = ncclIbInit_v9, .devices = ncclIbDevices, .getProperties = ncclIbGetProperties_v9, - .listen = ncclIbListen, + .listen = ncclIbListen_v10, .connect = ncclIbConnect_v9, .accept = ncclIbAccept, .regMr = ncclIbRegMr, @@ -2114,7 +2203,7 @@ const ncclNet_v8_t ibPlugin_v8 = { .init = ncclIbInit_v9, .devices = ncclIbDevices, .getProperties = ncclIbGetProperties_v8, - .listen = ncclIbListen, + .listen = ncclIbListen_v10, .connect = ncclIbConnect_v9, .accept = ncclIbAccept, .regMr = ncclIbRegMr, @@ -2136,7 +2225,7 @@ const ncclNet_v7_t ibPlugin_v7 = { .init = ncclIbInit_v9, .devices = ncclIbDevices, .getProperties = ncclIbGetProperties_v7, - .listen = ncclIbListen, + .listen = ncclIbListen_v10, .connect = ncclIbConnect_v9, .accept = ncclIbAccept, .regMr = ncclIbRegMr_v7, @@ -2158,7 +2247,7 @@ const ncclNet_v6_t ibPlugin_v6 = { .init = ncclIbInit_v9, .devices = ncclIbDevices, .getProperties = ncclIbGetProperties_v6, - .listen = ncclIbListen, + .listen = ncclIbListen_v10, .connect = ncclIbConnect_v6, .accept = ncclIbAccept_v6, .regMr = ncclIbRegMr_v7, @@ -2171,23 +2260,4 @@ const ncclNet_v6_t ibPlugin_v6 = { .closeSend = ncclIbCloseSend, .closeRecv = ncclIbCloseRecv, .closeListen = ncclIbCloseListen, -}; - -const ncclNet_v5_t ibPlugin_v5 = { - .name = "IBext_v5", - .init = ncclIbInit_v9, - .devices = ncclIbDevices, - .getProperties = ncclIbGetProperties_v6, - .listen = ncclIbListen, - .connect = ncclIbConnect_v6, - .accept = ncclIbAccept_v6, - .regMr = ncclIbRegMr_v7, - .deregMr = ncclIbDeregMr, - .isend = ncclIbIsend_v8, - .irecv = ncclIbIrecv_v8, - .iflush = ncclIbIflush, - .test = ncclIbTest, - .closeSend = ncclIbCloseSend, - .closeRecv = ncclIbCloseRecv, - .closeListen = ncclIbCloseListen, -}; +}; \ No newline at end of file diff --git a/src/p2p_plugin.c b/src/p2p_plugin.c index 655233a..184ba0e 100644 --- a/src/p2p_plugin.c +++ b/src/p2p_plugin.c @@ -18,41 +18,42 @@ #include "p2p_plugin.h" #ifdef HAVE_UCX_PLUGIN +extern ncclNet_v11_t ucxPlugin_v11; extern ncclNet_v10_t ucxPlugin_v10; extern ncclNet_v9_t ucxPlugin_v9; extern ncclNet_v8_t ucxPlugin_v8; extern ncclNet_v7_t ucxPlugin_v7; extern ncclNet_v6_t ucxPlugin_v6; -extern ncclNet_v5_t ucxPlugin_v5; +extern ncclNet_v11_t ucxRmaPlugin_v11; extern ncclNet_v10_t ucxRmaPlugin_v10; extern ncclNet_v9_t ucxRmaPlugin_v9; extern ncclNet_v8_t ucxRmaPlugin_v8; extern ncclNet_v7_t ucxRmaPlugin_v7; extern ncclNet_v6_t ucxRmaPlugin_v6; -extern ncclNet_v5_t ucxRmaPlugin_v5; +extern ncclNet_v11_t ucxUctPlugin_v11; extern ncclNet_v10_t ucxUctPlugin_v10; extern ncclNet_v9_t ucxUctPlugin_v9; extern ncclNet_v8_t ucxUctPlugin_v8; extern ncclNet_v7_t ucxUctPlugin_v7; extern ncclNet_v6_t ucxUctPlugin_v6; -extern ncclNet_v5_t ucxUctPlugin_v5; +extern ncclNet_v11_t ucxUctRdPlugin_v11; extern ncclNet_v10_t ucxUctRdPlugin_v10; extern ncclNet_v9_t ucxUctRdPlugin_v9; extern ncclNet_v8_t ucxUctRdPlugin_v8; extern ncclNet_v7_t ucxUctRdPlugin_v7; extern ncclNet_v6_t ucxUctRdPlugin_v6; -extern ncclNet_v5_t ucxUctRdPlugin_v5; + #endif +extern ncclNet_v11_t ibPlugin_v11; extern ncclNet_v10_t ibPlugin_v10; extern ncclNet_v9_t ibPlugin_v9; extern ncclNet_v8_t ibPlugin_v8; extern ncclNet_v7_t ibPlugin_v7; extern ncclNet_v6_t ibPlugin_v6; -extern ncclNet_v5_t ibPlugin_v5; pthread_mutex_t nccl_p2p_lock = PTHREAD_MUTEX_INITIALIZER; ncclDebugLogger_t pluginLogFunction; @@ -74,12 +75,18 @@ NCCL_PARAM(SharpMaxComms, "SHARP_MAX_COMMS", 1); NCCL_PARAM(IbAdaptiveRouting, "IB_ADAPTIVE_ROUTING", -2); NCCL_PARAM(IbDataDirect,"IB_DATA_DIRECT", 1); +ncclResult_t pluginInit_v11(void** ctx, uint64_t commId, ncclNetCommConfig_v11_t* config, ncclDebugLogger_t logFunction, ncclProfilerCallback_t profFunction); ncclResult_t pluginInit_v10(ncclDebugLogger_t logFunction, ncclProfilerCallback_t profFunction); ncclResult_t pluginInit_v9(ncclDebugLogger_t logFunction); ncclResult_t pluginInit_v8(ncclDebugLogger_t logFunction); ncclResult_t pluginInit_v7(ncclDebugLogger_t logFunction); ncclResult_t pluginInit_v6(ncclDebugLogger_t logFunction); -ncclResult_t pluginInit_v5(ncclDebugLogger_t logFunction); + + +ncclNet_v11_t ncclNetPlugin_v11 = { + "NCCL RDMA Plugin v11", + pluginInit_v11, +}; ncclNet_v10_t ncclNetPlugin_v10 = { "NCCL RDMA Plugin v10", @@ -107,12 +114,6 @@ ncclNet_v6_t ncclNetPlugin_v6 = { }; -ncclNet_v5_t ncclNetPlugin_v5 = { - "NCCL RDMA Plugin v5", - pluginInit_v5, -}; - - static nccl_p2p_plugin_t p2p_plugin = NCCL_P2P_LAST; static int nccl_p2p_is_uct_plugin(nccl_p2p_plugin_t plugin) { @@ -143,50 +144,59 @@ static void pluginSetup() switch (p2p_plugin) { #ifdef HAVE_UCX_PLUGIN case NCCL_P2P_UCX: + ncclNetPlugin_v11 = ucxPlugin_v11; ncclNetPlugin_v10 = ucxPlugin_v10; ncclNetPlugin_v9 = ucxPlugin_v9; ncclNetPlugin_v8 = ucxPlugin_v8; ncclNetPlugin_v7 = ucxPlugin_v7; ncclNetPlugin_v6 = ucxPlugin_v6; - ncclNetPlugin_v5 = ucxPlugin_v5; + break; case NCCL_P2P_UCX_RMA: + ncclNetPlugin_v11 = ucxRmaPlugin_v11; ncclNetPlugin_v10 = ucxRmaPlugin_v10; ncclNetPlugin_v9 = ucxRmaPlugin_v9; ncclNetPlugin_v8 = ucxRmaPlugin_v8; ncclNetPlugin_v7 = ucxRmaPlugin_v7; ncclNetPlugin_v6 = ucxRmaPlugin_v6; - ncclNetPlugin_v5 = ucxRmaPlugin_v5; + break; case NCCL_P2P_UCX_UCT: + ncclNetPlugin_v11 = ucxUctPlugin_v11; ncclNetPlugin_v10 = ucxUctPlugin_v10; ncclNetPlugin_v9 = ucxUctPlugin_v9; ncclNetPlugin_v8 = ucxUctPlugin_v8; ncclNetPlugin_v7 = ucxUctPlugin_v7; ncclNetPlugin_v6 = ucxUctPlugin_v6; - ncclNetPlugin_v5 = ucxUctPlugin_v5; break; case NCCL_P2P_UCX_UCT_RD: + ncclNetPlugin_v11 = ucxUctRdPlugin_v11; ncclNetPlugin_v10 = ucxUctRdPlugin_v10; ncclNetPlugin_v9 = ucxUctRdPlugin_v9; ncclNetPlugin_v8 = ucxUctRdPlugin_v8; ncclNetPlugin_v7 = ucxUctRdPlugin_v7; ncclNetPlugin_v6 = ucxUctRdPlugin_v6; - ncclNetPlugin_v5 = ucxUctRdPlugin_v5; break; #endif default: + ncclNetPlugin_v11 = ibPlugin_v11; ncclNetPlugin_v10 = ibPlugin_v10; ncclNetPlugin_v9 = ibPlugin_v9; ncclNetPlugin_v8 = ibPlugin_v8; ncclNetPlugin_v7 = ibPlugin_v7; ncclNetPlugin_v6 = ibPlugin_v6; - ncclNetPlugin_v5 = ibPlugin_v5; break; } } +ncclResult_t pluginInit_v11(void** ctx, uint64_t commId, ncclNetCommConfig_v11_t* config, ncclDebugLogger_t logFunction, ncclProfilerCallback_t profFunction) { + pluginLogFunction = logFunction; + pluginSetup(); + INFO(NCCL_INIT|NCCL_NET, "P2P plugin v11 %s", ncclNetPlugin_v11.name); + return ncclNetPlugin_v11.init(ctx, commId, config, logFunction, profFunction); +} + ncclResult_t pluginInit_v10(ncclDebugLogger_t logFunction, ncclProfilerCallback_t profFunction) { pluginLogFunction = logFunction; pluginSetup(); @@ -222,13 +232,6 @@ ncclResult_t pluginInit_v6(ncclDebugLogger_t logFunction) { return ncclNetPlugin_v6.init(logFunction); } -ncclResult_t pluginInit_v5(ncclDebugLogger_t logFunction) { - pluginLogFunction = logFunction; - pluginSetup(); - INFO(NCCL_INIT|NCCL_NET, "P2P plugin %s", ncclNetPlugin_v5.name); - return ncclNetPlugin_v5.init(logFunction); -} - // Detect whether GDR can work on a given NIC with the current CUDA device // Returns : // ncclSuccess : GDR works @@ -332,6 +335,8 @@ ncclResult_t ncclIbGetPhysProperties(int dev, ncclNetProperties_t* props) { props->netDeviceType = NCCL_NET_DEVICE_HOST; props->netDeviceVersion = NCCL_NET_DEVICE_INVALID_VERSION; props->maxP2pBytes = NCCL_MAX_NET_SIZE_BYTES; + props->maxCollBytes = NCCL_MAX_NET_SIZE_BYTES; + props->maxMultiRequestSize = 1; pthread_mutex_unlock(&ibDev->lock); return ncclSuccess; } diff --git a/src/sharp_plugin.c b/src/sharp_plugin.c index 5097215..7ae2f29 100644 --- a/src/sharp_plugin.c +++ b/src/sharp_plugin.c @@ -23,16 +23,25 @@ #include "sharp/api/sharp_coll.h" #include "utils.h" +extern ncclNet_v11_t ncclNetPlugin_v11; extern ncclNet_v10_t ncclNetPlugin_v10; extern ncclNet_v9_t ncclNetPlugin_v9; extern ncclNet_v8_t ncclNetPlugin_v8; extern ncclNet_v7_t ncclNetPlugin_v7; extern ncclNet_v6_t ncclNetPlugin_v6; -extern ncclNet_v5_t ncclNetPlugin_v5; +extern ncclCollNet_v11_t ncclCollNetPlugin_v11; extern ncclCollNet_v10_t ncclCollNetPlugin_v10; extern ncclCollNet_v9_t ncclCollNetPlugin_v9; + +// With ncclNet_v11_t the NCCL core initializes the network plugin per-communicator +// rather than once for all communicators. However, the internal plugin implementation +// still assumes the plugin is initialized only once across all communicators. The ref +// counter makes sure the plugin internally initializes only once. When per communicator +// context support is added to the plugin the ref counter can be removed. +static int sharpRefCount = 0; + int ncclNSharpDevs = -1; struct sharp_coll_caps sharp_caps; static int ncclSharpV3DatatypesSupported = 0; @@ -205,7 +214,8 @@ int ncclSharpOobBcast(void *ctx, void *buf, int size, int root) { return 0; } -ncclResult_t ncclSharpInit(ncclDebugLogger_t logFunction) { +ncclResult_t ncclSharpInit(void **ctx, uint64_t commId, ncclDebugLogger_t logFunction) { + if (sharpRefCount++) return ncclSuccess; struct timeval tval; gettimeofday(&tval, NULL); srand((int) tval.tv_usec); @@ -227,7 +237,7 @@ ncclResult_t ncclSharpInit(ncclDebugLogger_t logFunction) { ncclCollNetPlugin_v9.ireducescatter = NULL; } - return ncclNetPlugin_v10.init(logFunction, NULL); + return ncclNetPlugin_v11.init(ctx, commId, NULL, logFunction, NULL); } ncclResult_t ncclSharpDevices(int* ndev) { @@ -235,6 +245,11 @@ ncclResult_t ncclSharpDevices(int* ndev) { return ncclSuccess; } +ncclResult_t ncclSharpGetProperties_v11(int dev, ncclNetProperties_v11_t* props) { + NCCLCHECK(ncclNetPlugin_v11.getProperties(dev, props)); + props->maxCollBytes = NCCL_MAX_NET_SIZE_BYTES; + return ncclSuccess; +} ncclResult_t ncclSharpGetProperties_v10(int dev, ncclNetProperties_v10_t* props) { NCCLCHECK(ncclNetPlugin_v10.getProperties(dev, props)); @@ -268,11 +283,7 @@ ncclResult_t ncclSharpGetProperties_v6(int dev, ncclNetProperties_v6_t* props) { return ncclNetPlugin_v6.getProperties(dev, props); } -ncclResult_t ncclSharpGetProperties_v5(int dev, ncclNetProperties_v5_t* props) { - return ncclNetPlugin_v5.getProperties(dev, props); -} - -ncclResult_t ncclSharpListen(int dev, void* opaqueHandle, void** listenComm) { +ncclResult_t ncclSharpListen(void *ctx, int dev, void* opaqueHandle, void** listenComm) { struct ncclSharpListenComm *lComm; ncclResult_t status; @@ -803,11 +814,25 @@ ncclResult_t ncclSharpCloseListen(void* listenComm) { return status; } -ncclCollNet_v10_t ncclCollNetPlugin_v10 = { +ncclResult_t ncclSharpFinalize(void* ctx) { + --sharpRefCount; + return ncclSuccess; +} + +ncclResult_t ncclSharpInit_v10(ncclDebugLogger_t logFunction) { + return ncclSharpInit(NULL, 0, logFunction); +} + +ncclResult_t ncclSharpListen_v10(int dev, void* opaqueHandle, void** listenComm) { + return ncclSharpListen(NULL, dev, opaqueHandle, listenComm); +} + + +ncclCollNet_v11_t ncclCollNetPlugin_v11 = { "SHARP", ncclSharpInit, ncclSharpDevices, - ncclSharpGetProperties_v10, + ncclSharpGetProperties_v11, ncclSharpListen, ncclSharpConnect, ncclSharpReduceSupport, @@ -821,14 +846,37 @@ ncclCollNet_v10_t ncclCollNetPlugin_v10 = { ncclSharpTest, ncclSharpCloseColl, ncclSharpCloseListen, + NULL, + ncclSharpFinalize +}; + +ncclCollNet_v10_t ncclCollNetPlugin_v10 = { + "SHARP", + ncclSharpInit_v10, + ncclSharpDevices, + ncclSharpGetProperties_v10, + ncclSharpListen_v10, + ncclSharpConnect, + ncclSharpReduceSupport, + ncclSharpRegMr, + ncclSharpRegMrDmaBuf, + ncclSharpDeregMr, + ncclSharpIallreduce, + ncclSharpIallgather, + ncclSharpIreducescatter, + ncclSharpIflush, + ncclSharpTest, + ncclSharpCloseColl, + ncclSharpCloseListen, NULL }; + ncclCollNet_v9_t ncclCollNetPlugin_v9 = { "SHARP", - ncclSharpInit, + ncclSharpInit_v10, ncclSharpDevices, ncclSharpGetProperties_v9, - ncclSharpListen, + ncclSharpListen_v10, ncclSharpConnect, ncclSharpReduceSupport, ncclSharpRegMr, @@ -846,10 +894,10 @@ ncclCollNet_v9_t ncclCollNetPlugin_v9 = { ncclCollNet_v8_t ncclCollNetPlugin_v8 = { "SHARP", - ncclSharpInit, + ncclSharpInit_v10, ncclSharpDevices, ncclSharpGetProperties_v8, - ncclSharpListen, + ncclSharpListen_v10, ncclSharpConnect, ncclSharpReduceSupport, ncclSharpRegMr, @@ -866,10 +914,10 @@ ncclCollNet_v8_t ncclCollNetPlugin_v8 = { ncclCollNet_v7_t ncclCollNetPlugin_v7 = { "SHARP", - ncclSharpInit, + ncclSharpInit_v10, ncclSharpDevices, ncclSharpGetProperties_v7, - ncclSharpListen, + ncclSharpListen_v10, ncclSharpConnect, ncclSharpReduceSupport, ncclSharpRegMr_v7, @@ -884,10 +932,10 @@ ncclCollNet_v7_t ncclCollNetPlugin_v7 = { ncclCollNet_v6_t ncclCollNetPlugin_v6 = { "SHARP", - ncclSharpInit, + ncclSharpInit_v10, ncclSharpDevices, ncclSharpGetProperties_v6, - ncclSharpListen, + ncclSharpListen_v10, ncclSharpConnect, ncclSharpReduceSupport, ncclSharpRegMr_v7, @@ -899,20 +947,3 @@ ncclCollNet_v6_t ncclCollNetPlugin_v6 = { ncclSharpCloseColl, ncclSharpCloseListen }; - -ncclCollNet_v5_t ncclCollNetPlugin_v5 = { - "SHARP", - ncclSharpInit, - ncclSharpDevices, - ncclSharpGetProperties_v5, - ncclSharpListen, - ncclSharpConnect, - ncclSharpReduceSupport, - ncclSharpRegMr_v7, - ncclSharpDeregMr, - ncclSharpIallreduce_v8, - ncclSharpIflush, - ncclSharpTest, - ncclSharpCloseColl, - ncclSharpCloseListen -}; diff --git a/src/ucx_plugin.c b/src/ucx_plugin.c index 3c6ebea..fc6c585 100644 --- a/src/ucx_plugin.c +++ b/src/ucx_plugin.c @@ -35,6 +35,14 @@ } \ } while(0) +// With ncclNet_v11_t the NCCL core initializes the network plugin per-communicator +// rather than once for all communicators. However, the internal plugin implementation +// still assumes the plugin is initialized only once across all communicators. The ref +// counter makes sure the plugin internally initializes only once. When per communicator +// context support is added to the plugin the ref counter can be removed. +static int ucxRefCount = 0; + + NCCL_PARAM(UCXDisable, "UCX_DISABLE", 0); /* Exclude cuda-related UCX transports */ NCCL_PARAM(UCXCudaDisable, "UCX_CUDA_DISABLE", 1); @@ -75,6 +83,32 @@ ncclResult_t nccl_ucx_get_properties(int dev, ncclNetProperties_t* props) return nccl_p2p_ib_get_properties(ncclIbDevs, ncclNMergedIbDevs, dev, props); } +ncclResult_t nccl_ucx_get_properties_v10(int dev, ncclNetProperties_v10_t* props_v10) { + ncclNetProperties_t props; + ncclResult_t ret = nccl_p2p_ib_get_properties(ncclIbDevs, ncclNMergedIbDevs, dev, &props); + if (ret != ncclSuccess) return ret; + props_v10->name = props.name; + props_v10->pciPath = props.pciPath; + props_v10->guid = props.guid; + props_v10->ptrSupport = props.ptrSupport; + props_v10->regIsGlobal = props.regIsGlobal; + props_v10->forceFlush = props.forceFlush; + props_v10->speed = props.speed; + props_v10->port = props.port; + props_v10->latency = props.latency; + props_v10->maxComms = props.maxComms; + props_v10->maxRecvs = props.maxRecvs; + props_v10->netDeviceType = props.netDeviceType; + props_v10->netDeviceVersion = props.netDeviceVersion; + props_v10->maxP2pBytes = props.maxP2pBytes; + props_v10->maxCollBytes = props.maxCollBytes; + props_v10->vProps.ndevs = props.vProps.ndevs; + for(int i=0; ivProps.devs[i] = props.vProps.devs[i]; + } + return ncclSuccess; +} + ncclResult_t nccl_ucx_get_properties_v9(int dev, ncclNetProperties_v9_t* props_v9) { ncclNetProperties_t props; ncclResult_t ret = nccl_ucx_get_properties(dev, &props); @@ -327,8 +361,10 @@ static ncclResult_t ucx_init_context(ucp_context_h *ctx, int dev) { ncclResult_t result; if (ucp_ctx[dev] == NULL) { - snprintf(ucx_dev_name, PATH_MAX, "%s:%d", ncclIbDevs[dev].devName, + plugin_get_device_name(ncclIbDevs[dev].devName, ucx_dev_name, 64); + snprintf(ucx_dev_name + strlen(ucx_dev_name), PATH_MAX - strlen(ucx_dev_name), ":%d", ncclIbDevs[dev].portNum); + UCXCHECK(ucp_config_read("NCCL", NULL, &config)); UCXCHECK(ucp_config_modify(config, "NET_DEVICES", ucx_dev_name)); @@ -512,7 +548,14 @@ static ncclResult_t nccl_ucx_add_ep(nccl_ucx_worker_t *ucx_worker, return ncclSuccess; } -ncclResult_t nccl_ucx_init(ncclDebugLogger_t logFunction, ncclProfilerCallback_t profFunction) { +ncclResult_t nccl_ucx_setNetAttr(void *ctx, ncclNetAttr_t *netAttr) { + (void)ctx; + (void)netAttr; + return ncclSuccess; +} + +ncclResult_t nccl_ucx_init(void** ctx, uint64_t commId, ncclNetCommConfig_v11_t* config, ncclDebugLogger_t logFunction, ncclProfilerCallback_t profFunction) { + if (ucxRefCount++) return ncclSuccess; if (ncclParamUCXDisable()) return ncclInternalError; for (int i = 0; i < sizeof(worker_tags) / sizeof(*worker_tags); i++) { @@ -523,7 +566,7 @@ ncclResult_t nccl_ucx_init(ncclDebugLogger_t logFunction, ncclProfilerCallback_t &nccl_ucx_if_addr, NULL, logFunction); } -ncclResult_t nccl_ucx_listen(int dev, void *handle, void **listen_comm) { +ncclResult_t nccl_ucx_listen(void *ctx, int dev, void *handle, void **listen_comm) { ucx_listen_handle_t *my_handle = (ucx_listen_handle_t*)handle; ucx_listen_comm_t *comm = (ucx_listen_comm_t*)calloc(1, sizeof(*comm)); @@ -553,7 +596,7 @@ static void ucx_request_init(ucx_comm_t *comm) { } } -ncclResult_t nccl_ucx_connect(int dev, ncclNetCommConfig_t* config, void *handle, void **send_comm, ncclNetDeviceHandle_t** sendDevComm) { +ncclResult_t nccl_ucx_connect(void *ctx, int dev, void *handle, void **send_comm, ncclNetDeviceHandle_t** sendDevComm) { ucx_listen_handle_t *recv_handle = (ucx_listen_handle_t*)handle; struct ncclUCXCommStage *stage = &recv_handle->stage; ucx_comm_t *comm = stage->comm; @@ -582,7 +625,7 @@ ncclResult_t nccl_ucx_connect(int dev, ncclNetCommConfig_t* config, void *handle comm->gpuFlush.enabled = 0; NCCLCHECK(ucx_worker_get_netaddress(comm->ucx_worker->worker, &my_addr, &local_addr_len)); NCCLCHECK(nccl_ucx_add_ep(comm->ucx_worker, &comm->sock)); - INFO(NCCL_NET, "NET/UCX: Worker address length: %zu", local_addr_len); + TRACE(NCCL_NET, "NET/UCX: Worker address length: %zu", local_addr_len); NCCLCHECK(ncclSocketSend(&comm->sock, &local_addr_len, sizeof(size_t))); NCCLCHECK(ncclSocketSend(&comm->sock, my_addr, local_addr_len)); @@ -1093,12 +1136,29 @@ ncclResult_t nccl_ucx_close_listen(void *listen_comm) { return ncclSuccess; } +ncclResult_t nccl_ucx_finalize(void* ctx) { + ucxRefCount--; + return ncclSuccess; +} + +ncclResult_t nccl_ucx_init_v10(ncclDebugLogger_t logFunction, ncclProfilerCallback_t profilerCallback) { + return nccl_ucx_init(NULL, 0, NULL, logFunction, NULL); +} + +ncclResult_t nccl_ucx_connect_v10(int dev, ncclNetCommConfig_v10_t* config, void* opaqueHandle, void** sendComm, ncclNetDeviceHandle_v10_t** sendDevComm) { + return nccl_ucx_connect(NULL, dev, opaqueHandle, sendComm, sendDevComm); +} + +ncclResult_t nccl_ucx_listen_v10(int dev, void* opaqueHandle, void** listenComm) { + return nccl_ucx_listen(NULL, dev, opaqueHandle, listenComm); +} + ncclResult_t nccl_ucx_connect_v9(int dev, void* opaqueHandle, void** sendComm, ncclNetDeviceHandle_t** sendDevComm) { - return nccl_ucx_connect(dev, NULL, opaqueHandle, sendComm, sendDevComm); + return nccl_ucx_connect_v10(dev, NULL, opaqueHandle, sendComm, sendDevComm); } ncclResult_t nccl_ucx_init_v9(ncclDebugLogger_t logFunction) { - return nccl_ucx_init(logFunction, NULL); + return nccl_ucx_init_v10(logFunction, NULL); } ncclResult_t nccl_ucx_isend_v9(void* sendComm, void* data, size_t size, int tag, void* mhandle, void** request) { @@ -1133,8 +1193,7 @@ ncclResult_t nccl_ucx_accept_v6(void *listen_comm, void **recv_comm) { return nccl_ucx_accept(listen_comm, recv_comm, &dev_handle); } - -ncclNet_v10_t ucxPlugin_v10 = { +ncclNet_v11_t ucxPlugin_v11 = { .name = "UCX", .init = nccl_ucx_init, .devices = nccl_ucx_devices, @@ -1154,6 +1213,32 @@ ncclNet_v10_t ucxPlugin_v10 = { .closeListen = nccl_ucx_close_listen, NULL /* getDeviceMr */, NULL /* irecvConsumed */, + NULL, + nccl_ucx_finalize, + nccl_ucx_setNetAttr +}; + + +ncclNet_v10_t ucxPlugin_v10 = { + .name = "UCX", + .init = nccl_ucx_init_v10, + .devices = nccl_ucx_devices, + .getProperties = nccl_ucx_get_properties_v10, + .listen = nccl_ucx_listen_v10, + .connect = nccl_ucx_connect_v10, + .accept = nccl_ucx_accept, + .regMr = nccl_ucx_regmr, + .regMrDmaBuf = nccl_ucx_regmr_dmabuf, + .deregMr = nccl_ucx_deregmr, + .isend = nccl_ucx_isend, + .irecv = nccl_ucx_irecv, + .iflush = nccl_ucx_iflush, + .test = nccl_ucx_test, + .closeSend = nccl_ucx_close_send, + .closeRecv = nccl_ucx_close_recv, + .closeListen = nccl_ucx_close_listen, + NULL /* getDeviceMr */, + NULL /* irecvConsumed */, NULL }; @@ -1162,7 +1247,7 @@ ncclNet_v9_t ucxPlugin_v9 = { .init = nccl_ucx_init_v9, .devices = nccl_ucx_devices, .getProperties = nccl_ucx_get_properties_v9, - .listen = nccl_ucx_listen, + .listen = nccl_ucx_listen_v10, .connect = nccl_ucx_connect_v9, .accept = nccl_ucx_accept, .regMr = nccl_ucx_regmr, @@ -1185,7 +1270,7 @@ ncclNet_v8_t ucxPlugin_v8 = { .init = nccl_ucx_init_v9, .devices = nccl_ucx_devices, .getProperties = nccl_ucx_get_properties_v8, - .listen = nccl_ucx_listen, + .listen = nccl_ucx_listen_v10, .connect = nccl_ucx_connect_v9, .accept = nccl_ucx_accept, .regMr = nccl_ucx_regmr, @@ -1207,7 +1292,7 @@ ncclNet_v7_t ucxPlugin_v7 = { .init = nccl_ucx_init_v9, .devices = nccl_ucx_devices, .getProperties = nccl_ucx_get_properties_v7, - .listen = nccl_ucx_listen, + .listen = nccl_ucx_listen_v10, .connect = nccl_ucx_connect_v9, .accept = nccl_ucx_accept, .regMr = nccl_ucx_regmr_v7, @@ -1229,7 +1314,7 @@ ncclNet_v6_t ucxPlugin_v6 = { .init = nccl_ucx_init_v9, .devices = nccl_ucx_devices, .getProperties = nccl_ucx_get_properties_v6, - .listen = nccl_ucx_listen, + .listen = nccl_ucx_listen_v10, .connect = nccl_ucx_connect_v6, .accept = nccl_ucx_accept_v6, .regMr = nccl_ucx_regmr_v7, @@ -1243,22 +1328,3 @@ ncclNet_v6_t ucxPlugin_v6 = { .closeRecv = nccl_ucx_close_recv, .closeListen = nccl_ucx_close_listen }; - -ncclNet_v5_t ucxPlugin_v5 = { - .name = "UCX", - .init = nccl_ucx_init_v9, - .devices = nccl_ucx_devices, - .getProperties = nccl_ucx_get_properties_v6, - .listen = nccl_ucx_listen, - .connect = nccl_ucx_connect_v6, - .accept = nccl_ucx_accept_v6, - .regMr = nccl_ucx_regmr_v7, - .deregMr = nccl_ucx_deregmr, - .isend = nccl_ucx_isend_v8, - .irecv = nccl_ucx_irecv_v8, - .iflush = nccl_ucx_iflush, - .test = nccl_ucx_test, - .closeSend = nccl_ucx_close_send, - .closeRecv = nccl_ucx_close_recv, - .closeListen = nccl_ucx_close_listen -}; diff --git a/src/ucx_rma_plugin.c b/src/ucx_rma_plugin.c index 83d50ab..823f7f9 100644 --- a/src/ucx_rma_plugin.c +++ b/src/ucx_rma_plugin.c @@ -37,6 +37,12 @@ } while(0) NCCL_PARAM(UCXRMADisable, "UCX_RMA_DISABLE", 0); +// With ncclNet_v11_t the NCCL core initializes the network plugin per-communicator +// rather than once for all communicators. However, the internal plugin implementation +// still assumes the plugin is initialized only once across all communicators. The ref +// counter makes sure the plugin internally initializes only once. When per communicator +// context support is added to the plugin the ref counter can be removed. +static int ucxRmaRefCount = 0; extern ncclDebugLogger_t pluginLogFunction; static char nccl_ucx_rma_tls[32] = ""; @@ -83,6 +89,33 @@ ncclResult_t nccl_ucx_rma_get_properties(int dev, ncclNetProperties_t* props) return nccl_p2p_ib_get_properties(ncclIbDevs, ncclNMergedIbDevs, dev, props); } + +ncclResult_t nccl_ucx_rma_get_properties_v10(int dev, ncclNetProperties_v10_t* props_v10) { + ncclNetProperties_t props; + ncclResult_t ret = nccl_p2p_ib_get_properties(ncclIbDevs, ncclNMergedIbDevs, dev, &props); + if (ret != ncclSuccess) return ret; + props_v10->name = props.name; + props_v10->pciPath = props.pciPath; + props_v10->guid = props.guid; + props_v10->ptrSupport = props.ptrSupport; + props_v10->regIsGlobal = props.regIsGlobal; + props_v10->forceFlush = props.forceFlush; + props_v10->speed = props.speed; + props_v10->port = props.port; + props_v10->latency = props.latency; + props_v10->maxComms = props.maxComms; + props_v10->maxRecvs = props.maxRecvs; + props_v10->netDeviceType = props.netDeviceType; + props_v10->netDeviceVersion = props.netDeviceVersion; + props_v10->maxP2pBytes = props.maxP2pBytes; + props_v10->maxCollBytes = props.maxCollBytes; + props_v10->vProps.ndevs = props.vProps.ndevs; + for(int i=0; ivProps.devs[i] = props.vProps.devs[i]; + } + return ncclSuccess; +} + ncclResult_t nccl_ucx_rma_get_properties_v9(int dev, ncclNetProperties_v9_t* props_v9) { ncclNetProperties_t props; ncclResult_t ret = nccl_ucx_rma_get_properties(dev, &props); @@ -290,8 +323,8 @@ static ncclResult_t nccl_ucx_rma_init_ucp(int dev, ucp_context_h *ctx) ucp_config_t *config; char ucx_dev_name[PATH_MAX]; - snprintf(ucx_dev_name, PATH_MAX, "%s:%d", ncclIbDevs[dev].devName, - ncclIbDevs[dev].portNum); + plugin_get_device_name(ncclIbDevs[dev].devName, ucx_dev_name, 64); + snprintf(ucx_dev_name, PATH_MAX - strlen(ucx_dev_name), ":%d", ncclIbDevs[dev].portNum); UCXCHECK(ucp_config_read("NCCL", NULL, &config)); UCXCHECK(ucp_config_modify(config, "NET_DEVICES", ucx_dev_name)); @@ -431,8 +464,15 @@ static ncclResult_t nccl_ucx_add_ep(ucp_worker_h worker, struct ncclSocket *sock return status; } -ncclResult_t nccl_ucx_rma_init(ncclDebugLogger_t logFunction, ncclProfilerCallback_t profFunction) +ncclResult_t nccl_ucx_rma_setNetAttr(void *ctx, ncclNetAttr_t *netAttr) { + (void)ctx; + (void)netAttr; + return ncclSuccess; +} + +ncclResult_t nccl_ucx_rma_init(void** ctx, uint64_t commId, ncclNetCommConfig_v11_t* config, ncclDebugLogger_t logFunction, ncclProfilerCallback_t profFunction) { + if (ucxRmaRefCount++) return ncclSuccess; char *config_env; if (ncclParamUCXRMADisable()) return ncclInternalError; NCCLCHECK(nccl_p2p_ib_init(&ncclNIbDevs, &ncclNMergedIbDevs, ncclIbDevs, if_name, &nccl_ucx_if_addr, @@ -461,7 +501,7 @@ ncclResult_t nccl_ucx_rma_init(ncclDebugLogger_t logFunction, ncclProfilerCallba return ncclSuccess; } -ncclResult_t nccl_ucx_rma_listen(int dev, void *handle, void **listen_comm) +ncclResult_t nccl_ucx_rma_listen(void *ctx, int dev, void *handle, void **listen_comm) { ucx_rma_listen_handle_t *my_handle = (ucx_rma_listen_handle_t*)handle; nccl_ucx_rma_listen_comm_t *comm; @@ -503,7 +543,7 @@ static ucs_status_t nccl_ucx_rma_am_rkey_cb(void *arg, void *data, size_t length } -ncclResult_t nccl_ucx_rma_connect(int dev, ncclNetCommConfig_t* config, void *handle, void **send_comm, ncclNetDeviceHandle_t** sendDevComm) +ncclResult_t nccl_ucx_rma_connect(void *ctx,int dev, void *handle, void **send_comm, ncclNetDeviceHandle_t** sendDevComm) { ucx_rma_listen_handle_t *recv_handle = (ucx_rma_listen_handle_t*)handle; struct ncclUCXCommStage* stage = &recv_handle->stage; @@ -1211,12 +1251,29 @@ ncclResult_t nccl_ucx_rma_close_listen(void *listen_comm) return ncclSuccess; } +ncclResult_t nccl_ucx_rma_finalize(void* ctx) { + ucxRmaRefCount--; + return ncclSuccess; +} + +ncclResult_t nccl_ucx_rma_init_v10(ncclDebugLogger_t logFunction, ncclProfilerCallback_t profilerCallback) { + return nccl_ucx_rma_init(NULL, 0, NULL, logFunction, NULL); +} + +ncclResult_t nccl_ucx_rma_connect_v10(int dev, ncclNetCommConfig_v10_t* config, void* opaqueHandle, void** sendComm, ncclNetDeviceHandle_v10_t** sendDevComm) { + return nccl_ucx_rma_connect(NULL, dev, opaqueHandle, sendComm, sendDevComm); +} + +ncclResult_t nccl_ucx_rma_listen_v10(int dev, void* opaqueHandle, void** listenComm) { + return nccl_ucx_rma_listen(NULL, dev, opaqueHandle, listenComm); +} + ncclResult_t nccl_ucx_rma_connect_v9(int dev, void* opaqueHandle, void** sendComm, ncclNetDeviceHandle_t** sendDevComm) { - return nccl_ucx_rma_connect(dev, NULL, opaqueHandle, sendComm, sendDevComm); + return nccl_ucx_rma_connect_v10(dev, NULL, opaqueHandle, sendComm, sendDevComm); } ncclResult_t nccl_ucx_rma_init_v9(ncclDebugLogger_t logFunction) { - return nccl_ucx_rma_init(logFunction, NULL); + return nccl_ucx_rma_init_v10(logFunction, NULL); } ncclResult_t nccl_ucx_rma_isend_v9(void* sendComm, void* data, size_t size, int tag, void* mhandle, void** request) { @@ -1255,8 +1312,9 @@ ncclResult_t nccl_ucx_rma_accept_v6(void *listen_comm, void **recv_comm) return nccl_ucx_rma_accept(listen_comm, recv_comm, &dev_handle); } -ncclNet_v10_t ucxRmaPlugin_v10 = { - .name = "UCX-RMA", + +ncclNet_v11_t ucxRmaPlugin_v11 = { + .name = "UCX", .init = nccl_ucx_rma_init, .devices = nccl_ucx_rma_devices, .getProperties = nccl_ucx_rma_get_properties, @@ -1275,6 +1333,31 @@ ncclNet_v10_t ucxRmaPlugin_v10 = { .closeListen = nccl_ucx_rma_close_listen, NULL /* getDeviceMr */, NULL /* irecvConsumed */, + NULL, + nccl_ucx_rma_finalize, + nccl_ucx_rma_setNetAttr +}; + +ncclNet_v10_t ucxRmaPlugin_v10 = { + .name = "UCX-RMA", + .init = nccl_ucx_rma_init_v10, + .devices = nccl_ucx_rma_devices, + .getProperties = nccl_ucx_rma_get_properties_v10, + .listen = nccl_ucx_rma_listen_v10, + .connect = nccl_ucx_rma_connect_v10, + .accept = nccl_ucx_rma_accept, + .regMr = nccl_ucx_rma_regmr, + .regMrDmaBuf = nccl_ucx_rma_regmr_dmabuf, + .deregMr = nccl_ucx_rma_deregmr, + .isend = nccl_ucx_rma_isend, + .irecv = nccl_ucx_rma_irecv, + .iflush = nccl_ucx_rma_iflush, + .test = nccl_ucx_rma_test, + .closeSend = nccl_ucx_rma_close_send, + .closeRecv = nccl_ucx_rma_close_recv, + .closeListen = nccl_ucx_rma_close_listen, + NULL /* getDeviceMr */, + NULL /* irecvConsumed */, NULL }; @@ -1283,7 +1366,7 @@ ncclNet_v9_t ucxRmaPlugin_v9 = { .init = nccl_ucx_rma_init_v9, .devices = nccl_ucx_rma_devices, .getProperties = nccl_ucx_rma_get_properties_v9, - .listen = nccl_ucx_rma_listen, + .listen = nccl_ucx_rma_listen_v10, .connect = nccl_ucx_rma_connect_v9, .accept = nccl_ucx_rma_accept, .regMr = nccl_ucx_rma_regmr, @@ -1306,7 +1389,7 @@ ncclNet_v8_t ucxRmaPlugin_v8 = { .init = nccl_ucx_rma_init_v9, .devices = nccl_ucx_rma_devices, .getProperties = nccl_ucx_rma_get_properties_v8, - .listen = nccl_ucx_rma_listen, + .listen = nccl_ucx_rma_listen_v10, .connect = nccl_ucx_rma_connect_v9, .accept = nccl_ucx_rma_accept, .regMr = nccl_ucx_rma_regmr, @@ -1328,7 +1411,7 @@ ncclNet_v7_t ucxRmaPlugin_v7 = { .init = nccl_ucx_rma_init_v9, .devices = nccl_ucx_rma_devices, .getProperties = nccl_ucx_rma_get_properties_v7, - .listen = nccl_ucx_rma_listen, + .listen = nccl_ucx_rma_listen_v10, .connect = nccl_ucx_rma_connect_v9, .accept = nccl_ucx_rma_accept, .regMr = nccl_ucx_rma_regmr_v7, @@ -1350,7 +1433,7 @@ ncclNet_v6_t ucxRmaPlugin_v6 = { .init = nccl_ucx_rma_init_v9, .devices = nccl_ucx_rma_devices, .getProperties = nccl_ucx_rma_get_properties_v6, - .listen = nccl_ucx_rma_listen, + .listen = nccl_ucx_rma_listen_v10, .connect = nccl_ucx_rma_connect_v6, .accept = nccl_ucx_rma_accept_v6, .regMr = nccl_ucx_rma_regmr_v7, @@ -1363,23 +1446,4 @@ ncclNet_v6_t ucxRmaPlugin_v6 = { .closeSend = nccl_ucx_rma_close_send, .closeRecv = nccl_ucx_rma_close_recv, .closeListen = nccl_ucx_rma_close_listen -}; - -ncclNet_v5_t ucxRmaPlugin_v5 = { - .name = "UCX-RMA", - .init = nccl_ucx_rma_init_v9, - .devices = nccl_ucx_rma_devices, - .getProperties = nccl_ucx_rma_get_properties_v6, - .listen = nccl_ucx_rma_listen, - .connect = nccl_ucx_rma_connect_v6, - .accept = nccl_ucx_rma_accept_v6, - .regMr = nccl_ucx_rma_regmr_v7, - .deregMr = nccl_ucx_rma_deregmr, - .isend = nccl_ucx_rma_isend_v8, - .irecv = nccl_ucx_rma_irecv_v8, - .iflush = nccl_ucx_rma_iflush, - .test = nccl_ucx_rma_test, - .closeSend = nccl_ucx_rma_close_send, - .closeRecv = nccl_ucx_rma_close_recv, - .closeListen = nccl_ucx_rma_close_listen -}; +}; \ No newline at end of file diff --git a/src/ucx_uct_lib.c b/src/ucx_uct_lib.c index 7561744..a7b699e 100644 --- a/src/ucx_uct_lib.c +++ b/src/ucx_uct_lib.c @@ -437,7 +437,7 @@ ncclResult_t nccl_uct_devices(int *ndev) { return ncclSuccess; } -ncclResult_t nccl_uct_listen(int dev, void *listen_handle, void **listen_comm) { +ncclResult_t nccl_uct_listen(void *ctx, int dev, void *listen_handle, void **listen_comm) { nccl_uct_listen_handle_t *handle = listen_handle; nccl_uct_listen_comm_t *l_comm = calloc(1, sizeof(*l_comm)); nccl_uct_comm_t *accept_comm; @@ -532,7 +532,7 @@ static ncclResult_t nccl_uct_comm_gpu_flush_init(nccl_uct_comm_t *comm) { return ncclSystemError; } -ncclResult_t nccl_uct_connect(int dev, ncclNetCommConfig_t* config, void *listen_handle, void **send_comm, +ncclResult_t nccl_uct_connect(void *ctx, int dev, void *listen_handle, void **send_comm, ncclNetDeviceHandle_t **sendDevComm) { int ready = 0; nccl_uct_listen_handle_t *handle = listen_handle; @@ -813,6 +813,32 @@ ncclResult_t nccl_uct_get_properties(int dev, ncclNetProperties_t *props) { return nccl_p2p_ib_get_properties(ncclIbDevs, context.merge_dev_count, dev, props); } +ncclResult_t nccl_uct_get_properties_v10(int dev, ncclNetProperties_v10_t* props_v10) { + ncclNetProperties_t props; + ncclResult_t ret = nccl_uct_get_properties(dev, &props); + if (ret != ncclSuccess) return ret; + props_v10->name = props.name; + props_v10->pciPath = props.pciPath; + props_v10->guid = props.guid; + props_v10->ptrSupport = props.ptrSupport; + props_v10->regIsGlobal = props.regIsGlobal; + props_v10->forceFlush = props.forceFlush; + props_v10->speed = props.speed; + props_v10->port = props.port; + props_v10->latency = props.latency; + props_v10->maxComms = props.maxComms; + props_v10->maxRecvs = props.maxRecvs; + props_v10->netDeviceType = props.netDeviceType; + props_v10->netDeviceVersion = props.netDeviceVersion; + props_v10->maxP2pBytes = props.maxP2pBytes; + props_v10->maxCollBytes = props.maxCollBytes; + props_v10->vProps.ndevs = props.vProps.ndevs; + for(int i=0; ivProps.devs[i] = props.vProps.devs[i]; + } + return ncclSuccess; +} + ncclResult_t nccl_uct_get_properties_v9(int dev, ncclNetProperties_v9_t* props_v9) { ncclNetProperties_t props; ncclResult_t ret = nccl_uct_get_properties(dev, &props); @@ -905,8 +931,17 @@ ncclResult_t nccl_uct_get_properties_v6(int dev, return ncclSuccess; } + +ncclResult_t nccl_uct_listen_v10(int dev, void* opaqueHandle, void** listenComm) { + return nccl_uct_listen(NULL, dev, opaqueHandle, listenComm); +} + +ncclResult_t nccl_uct_connect_v10(int dev, ncclNetCommConfig_v10_t* config, void* opaqueHandle, void** sendComm, ncclNetDeviceHandle_v10_t** sendDevComm) { + return nccl_uct_connect(NULL, dev, opaqueHandle, sendComm, sendDevComm); +} + ncclResult_t nccl_uct_connect_v9(int dev, void* opaqueHandle, void** sendComm, ncclNetDeviceHandle_t** sendDevComm) { - return nccl_uct_connect(dev, NULL, opaqueHandle, sendComm, sendDevComm); + return nccl_uct_connect_v10(dev, NULL, opaqueHandle, sendComm, sendDevComm); } ncclResult_t nccl_uct_reg_mr_v7(void *comm, void *data, int size, int type, @@ -923,3 +958,7 @@ ncclResult_t nccl_uct_accept_v6(void *listen_comm, void **recv_comm) { ncclNetDeviceHandle_v7_t *dev_handle = NULL; return nccl_uct_accept(listen_comm, recv_comm, &dev_handle); } + +ncclResult_t nccl_uct_setNetAttr(void *ctx, ncclNetAttr_v11_t *netAttr) { + return ncclSuccess; +} diff --git a/src/ucx_uct_plugin.c b/src/ucx_uct_plugin.c index 1aab908..6b1d25a 100644 --- a/src/ucx_uct_plugin.c +++ b/src/ucx_uct_plugin.c @@ -8,6 +8,13 @@ #include "ucx_uct_lib.h" +// With ncclNet_v11_t the NCCL core initializes the network plugin per-communicator +// rather than once for all communicators. However, the internal plugin implementation +// still assumes the plugin is initialized only once across all communicators. The ref +// counter makes sure the plugin internally initializes only once. When per communicator +// context support is added to the plugin the ref counter can be removed. +static int ucxUctWrRefCount = 0; + typedef enum { NCCL_UCT_REQ_IRECV = -1, NCCL_UCT_REQ_IFLUSH = -2 @@ -233,7 +240,8 @@ static ncclResult_t nccl_uct_wr_comm_init(nccl_uct_comm_t *base_comm, return nccl_uct_comm_init(&comm->base, context, worker, dev, remote_comm); } -static ncclResult_t nccl_uct_wr_init(ncclDebugLogger_t logFunction, ncclProfilerCallback_t profFunction) { +static ncclResult_t nccl_uct_wr_init(void** ctx, uint64_t commId, ncclNetCommConfig_v11_t* config, ncclDebugLogger_t logFunction, ncclProfilerCallback_t profFunction) { + if (ucxUctWrRefCount++) return ncclSuccess; context.ops.comm_alloc = nccl_uct_wr_comm_alloc; context.ops.comm_init = nccl_uct_wr_comm_init; context.ops.iface_set = nccl_uct_wr_iface_set; @@ -470,9 +478,16 @@ static ncclResult_t nccl_uct_wr_close(void *close_comm) { return ncclSuccess; } +ncclResult_t nccl_uct_wr_finalize(void *ctx) { + ucxUctWrRefCount--; + return ncclSuccess; +} +ncclResult_t nccl_uct_wr_init_v10(ncclDebugLogger_t logFunction, ncclProfilerCallback_t profilerCallback) { + return nccl_uct_wr_init(NULL, 0, NULL, logFunction, NULL); +} static ncclResult_t nccl_uct_wr_init_v9(ncclDebugLogger_t logFunction) { - return nccl_uct_wr_init(logFunction, NULL); + return nccl_uct_wr_init_v10(logFunction, NULL); } static ncclResult_t nccl_uct_wr_isend_v9(void* sendComm, void* data, size_t size, int tag, void* mhandle, void** request) { @@ -496,9 +511,9 @@ static ncclResult_t nccl_uct_wr_irecv_v8(void *recv_comm, int n, void **data, return nccl_uct_wr_irecv_v9(recv_comm, n, data, sizes_sizet, tags, mhandles, request); } +ncclNet_v11_t ucxUctPlugin_v11 = NCCL_UCT_PLUGIN_V11("UCX-UCT", nccl_uct_wr); ncclNet_v10_t ucxUctPlugin_v10 = NCCL_UCT_PLUGIN_V10("UCX-UCT", nccl_uct_wr); ncclNet_v9_t ucxUctPlugin_v9 = NCCL_UCT_PLUGIN_V9("UCX-UCT", nccl_uct_wr); ncclNet_v8_t ucxUctPlugin_v8 = NCCL_UCT_PLUGIN_V8("UCX-UCT", nccl_uct_wr); ncclNet_v7_t ucxUctPlugin_v7 = NCCL_UCT_PLUGIN_V7("UCX-UCT", nccl_uct_wr); ncclNet_v6_t ucxUctPlugin_v6 = NCCL_UCT_PLUGIN_V6("UCX-UCT", nccl_uct_wr); -ncclNet_v5_t ucxUctPlugin_v5 = NCCL_UCT_PLUGIN_V5("UCX-UCT", nccl_uct_wr); diff --git a/src/ucx_uct_rd_plugin.c b/src/ucx_uct_rd_plugin.c index dbcb27f..8dcc47d 100644 --- a/src/ucx_uct_rd_plugin.c +++ b/src/ucx_uct_rd_plugin.c @@ -12,6 +12,13 @@ #define NCCL_UCT_PENDING_SIZE 128 #define NCCL_UCT_PENDING_MASK (NCCL_UCT_PENDING_SIZE - 1) +// With ncclNet_v11_t the NCCL core initializes the network plugin per-communicator +// rather than once for all communicators. However, the internal plugin implementation +// still assumes the plugin is initialized only once across all communicators. The ref +// counter makes sure the plugin internally initializes only once. When per communicator +// context support is added to the plugin the ref counter can be removed. +static int ucxUctRdRefCount = 0; + /* Memory chunk to send or receive */ typedef struct { int tag; @@ -222,7 +229,8 @@ static ncclResult_t nccl_uct_rd_comm_init(nccl_uct_comm_t *base_comm, return nccl_uct_comm_init(&comm->base, context, worker, dev, remote_comm); } -static ncclResult_t nccl_uct_rd_init(ncclDebugLogger_t logFunction, ncclProfilerCallback_t profFunction) { +static ncclResult_t nccl_uct_rd_init(void** ctx, uint64_t commId, ncclNetCommConfig_v11_t* config, ncclDebugLogger_t logFunction, ncclProfilerCallback_t profFunction) { + if (ucxUctRdRefCount++) return ncclSuccess; NCCL_STATIC_ASSERT(NCCL_UCT_RING_SIZE >= 2 * MAX_REQUESTS, "Cannot handle expected/unexpected requests"); NCCL_STATIC_ASSERT(NCCL_UCT_PENDING_SIZE > MAX_REQUESTS, @@ -428,8 +436,17 @@ static ncclResult_t nccl_uct_rd_close(void *close_comm) { return ncclSuccess; } +ncclResult_t nccl_uct_rd_finalize(void *ctx) { + ucxUctRdRefCount--; + return ncclSuccess; +} + +ncclResult_t nccl_uct_rd_init_v10(ncclDebugLogger_t logFunction, ncclProfilerCallback_t profilerCallback) { + return nccl_uct_rd_init(NULL, 0, NULL, logFunction, NULL); +} + static ncclResult_t nccl_uct_rd_init_v9(ncclDebugLogger_t logFunction) { - return nccl_uct_rd_init(logFunction, NULL); + return nccl_uct_rd_init_v10(logFunction, NULL); } static ncclResult_t nccl_uct_rd_isend_v9(void* sendComm, void* data, size_t size, int tag, void* mhandle, void** request) { @@ -453,9 +470,9 @@ static ncclResult_t nccl_uct_rd_irecv_v8(void *recv_comm, int n, void **data, return nccl_uct_rd_irecv_v9(recv_comm, n, data, sizes_sizet, tags, mhandles, request); } +ncclNet_v11_t ucxUctRdPlugin_v11 = NCCL_UCT_PLUGIN_V11("UCX-UCT-RD", nccl_uct_rd); ncclNet_v10_t ucxUctRdPlugin_v10 = NCCL_UCT_PLUGIN_V10("UCX-UCT-RD", nccl_uct_rd); ncclNet_v9_t ucxUctRdPlugin_v9 = NCCL_UCT_PLUGIN_V9("UCX-UCT-RD", nccl_uct_rd); ncclNet_v8_t ucxUctRdPlugin_v8 = NCCL_UCT_PLUGIN_V8("UCX-UCT-RD", nccl_uct_rd); ncclNet_v7_t ucxUctRdPlugin_v7 = NCCL_UCT_PLUGIN_V7("UCX-UCT-RD", nccl_uct_rd); ncclNet_v6_t ucxUctRdPlugin_v6 = NCCL_UCT_PLUGIN_V6("UCX-UCT-RD", nccl_uct_rd); -ncclNet_v5_t ucxUctRdPlugin_v5 = NCCL_UCT_PLUGIN_V5("UCX-UCT-RD", nccl_uct_rd); diff --git a/src/utils.c b/src/utils.c index e153b38..93be349 100644 --- a/src/utils.c +++ b/src/utils.c @@ -138,3 +138,23 @@ void ncclSetThreadName(pthread_t thread, const char *fmt, ...) { pthread_setname_np(thread, threadName); #endif } + +/* for data direct nic, the device name is ends with suffix '_dma`. + * remove this suffix before passing name to libsharp */ + void plugin_get_device_name(const char *input, char *output, size_t output_size) { + const char *suffix = "_dma"; + size_t input_len = strlen(input); + size_t suffix_len = strlen(suffix); + + if (input_len >= suffix_len && strcmp(input + input_len - suffix_len, suffix) == 0) { + size_t new_len = input_len - suffix_len; + if (new_len >= output_size) { + new_len = output_size - 1; + } + memcpy(output, input, new_len); + output[new_len] = '\0'; + } else { + strncpy(output, input, output_size - 1); + output[output_size - 1] = '\0'; + } +} From 62c46a591aac49dfef78cb7ce1096cdf3377df0b Mon Sep 17 00:00:00 2001 From: Devendar Bureddy Date: Tue, 9 Sep 2025 05:40:41 +0300 Subject: [PATCH 2/5] Add string for IB async event IBV_EVENT_GID_CHANGE (18) --- src/p2p_plugin.c | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/p2p_plugin.c b/src/p2p_plugin.c index 184ba0e..f982f59 100644 --- a/src/p2p_plugin.c +++ b/src/p2p_plugin.c @@ -406,6 +406,9 @@ static void* ncclIbAsyncThreadMain(void* args) { // SRQ are not used in NCCL WARN("NET/IB : %s:%d async fatal event on SRQ, unused for now (%p): %s", dev->devName, dev->portNum, srq, str); break; + case IBV_EVENT_GID_CHANGE: + WARN("NET/IB : %s:%d GID table changed", dev->devName, dev->portNum); + break; case IBV_EVENT_PATH_MIG_ERR: case IBV_EVENT_PORT_ERR: case IBV_EVENT_PATH_MIG: From 6bb28a0c5416d310d92e2cfe2007c5b3dfc29b77 Mon Sep 17 00:00:00 2001 From: Devendar Bureddy Date: Tue, 9 Sep 2025 05:42:18 +0300 Subject: [PATCH 3/5] Move libvirt bridge interface to the end of possible interfaces --- src/socket.c | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/socket.c b/src/socket.c index 00a7f4a..fc7b4eb 100755 --- a/src/socket.c +++ b/src/socket.c @@ -382,11 +382,13 @@ ncclResult_t ncclFindInterfaces(char* ifNames, union ncclSocketAddress *ifAddrs, NCCLCHECK(ncclFindInterfaceMatchSubnet(ifNames, ifAddrs, &idAddr, ifNameMaxSize, nIfs)); } } - // Then look for anything else (but not docker or lo) - if (*nIfs == 0) NCCLCHECK(findInterfaces("^docker,lo", ifNames, ifAddrs, sock_family, ifNameMaxSize, maxIfs, nIfs)); - // Finally look for docker, then lo. + // Then look for anything else (but not docker,lo, or virtual) + if (*nIfs == 0) NCCLCHECK(findInterfaces("^docker,lo,virbr", ifNames, ifAddrs, sock_family, ifNameMaxSize, maxIfs, nIfs)); + // Finally look for docker, lo, then virbr. + if (*nIfs == 0) NCCLCHECK(findInterfaces("docker", ifNames, ifAddrs, sock_family, ifNameMaxSize, maxIfs, nIfs)); if (*nIfs == 0) NCCLCHECK(findInterfaces("lo", ifNames, ifAddrs, sock_family, ifNameMaxSize, maxIfs, nIfs)); + if (*nIfs == 0) NCCLCHECK(findInterfaces("virbr", ifNames, ifAddrs, sock_family, ifNameMaxSize, maxIfs, nIfs)); } return ncclSuccess; } From 5d23db839bf46ead9ca5ca3a00f585714949b9ac Mon Sep 17 00:00:00 2001 From: Devendar Bureddy Date: Tue, 9 Sep 2025 18:41:57 +0300 Subject: [PATCH 4/5] Only consider running interfaces for socket traffic - Running means UP and physically attached to a cable - This removes the edge case of a DOWN interface with a manually set IP address --- src/socket.c | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/socket.c b/src/socket.c index fc7b4eb..20166c4 100755 --- a/src/socket.c +++ b/src/socket.c @@ -153,6 +153,9 @@ static ncclResult_t findInterfaces(const char* prefixList, char* names, union nc if (family != AF_INET && family != AF_INET6) continue; + /* Only consider running interfaces, i.e. UP and physically attached. */ + if (!(interface->ifa_flags & IFF_RUNNING)) continue; + TRACE(NCCL_INIT|NCCL_NET,"Found interface %s:%s", interface->ifa_name, ncclSocketToString((union ncclSocketAddress *) interface->ifa_addr, line, 1)); /* Allow the caller to force the socket family type */ From 7d6843aa5add5a62d2db2e7b2543e036c15b7f7f Mon Sep 17 00:00:00 2001 From: Devendar Bureddy Date: Tue, 9 Sep 2025 20:07:34 +0300 Subject: [PATCH 5/5] improve error message of mlx5dv_get_data_direct_sysfs_path --- src/ibvwrap.c | 11 ++-- src/p2p_plugin.c | 165 +++++++++++++++++++++++++---------------------- 2 files changed, 94 insertions(+), 82 deletions(-) diff --git a/src/ibvwrap.c b/src/ibvwrap.c index 8422da2..501b52a 100644 --- a/src/ibvwrap.c +++ b/src/ibvwrap.c @@ -309,12 +309,11 @@ ncclResult_t wrap_mlx5dv_get_data_direct_sysfs_path(struct ibv_context *context, #if HAVE_DECL_MLX5DV_GET_DATA_DIRECT_SYSFS_PATH int ret; ret = mlx5dv_get_data_direct_sysfs_path(context, buf, buf_len); - if (!ret) { - return ncclSuccess; - } else { - INFO(NCCL_NET, "Call to mlx5dv_get_data_direct_sysfs_path failed with error %s errno %d", strerror(ret), ret); - return ncclSystemError; - } + if (ret == 0) return ncclSuccess; + /* ENODEV can happen if the devices is not data-direct but mlx5 is used. It's not an error*/ + if (ret == ENODEV) return ncclInvalidArgument; + INFO(NCCL_NET, "Call to mlx5dv_get_data_direct_sysfs_path failed with error %s errno %d", strerror(ret), ret); + return ncclSystemError; #else INFO(NCCL_NET, "Symbol mlx5dv_get_data_direct_sysfs_path in rdma-core library"); return ncclSystemError; diff --git a/src/p2p_plugin.c b/src/p2p_plugin.c index f982f59..1d8ab5a 100644 --- a/src/p2p_plugin.c +++ b/src/p2p_plugin.c @@ -560,24 +560,15 @@ ncclResult_t nccl_p2p_ib_init(int *nDevs, int *nmDevs, ncclIbDev *ncclIbDevs, ch if (ncclSuccess != wrap_ibv_get_device_list(&devices, &nIbDevs)) { ret = ncclInternalError; goto fail; } for (int d=0; dname); continue; } - enum ncclIbProvider ibProvider = IB_PROVIDER_NONE; - char dataDirectDevicePath[PATH_MAX]; - int dataDirectSupported = 0; - int skipNetDevForDataDirect = 0; - if (wrap_mlx5dv_is_supported(devices[d])) { - ibProvider = IB_PROVIDER_MLX5; - snprintf(dataDirectDevicePath, PATH_MAX, "/sys"); - if((ncclMlx5dvDmaBufCapable(context)) && (wrap_mlx5dv_get_data_direct_sysfs_path(context, dataDirectDevicePath + 4, PATH_MAX - 4) == ncclSuccess)) { - INFO(NCCL_INIT|NCCL_NET, "Data Direct DMA Interface is detected for device:%s", devices[d]->name); - if(ncclParamIbDataDirect() == 1) { dataDirectSupported = 1; skipNetDevForDataDirect = 1; } - if(ncclParamIbDataDirect() == 2) { dataDirectSupported = 1; skipNetDevForDataDirect = 0; } - } - } + char dataDirectDevicePath[PATH_MAX] = "/sys"; + int devCount = /*undefined*/-1, devOffset = 0; + enum ncclIbProvider ibProvider = wrap_mlx5dv_is_supported(devices[d]) ? IB_PROVIDER_MLX5 : IB_PROVIDER_NONE; + int nPorts = 0; struct ibv_device_attr devAttr; if (ncclSuccess != wrap_ibv_query_device(context, &devAttr)) { @@ -586,7 +577,6 @@ ncclResult_t nccl_p2p_ib_init(int *nDevs, int *nmDevs, ncclIbDev *ncclIbDevs, ch continue; } for (int port_num = 1; port_num <= devAttr.phys_port_cnt; port_num++) { - for (int dataDirect = skipNetDevForDataDirect; dataDirect < 1 + dataDirectSupported; ++dataDirect) { struct ibv_port_attr portAttr; uint32_t portSpeed; if (ncclSuccess != wrap_ibv_query_port(context, port_num, &portAttr)) { @@ -594,74 +584,97 @@ ncclResult_t nccl_p2p_ib_init(int *nDevs, int *nmDevs, ncclIbDev *ncclIbDevs, ch continue; } if (portAttr.state != IBV_PORT_ACTIVE) continue; - if (portAttr.link_layer != IBV_LINK_LAYER_INFINIBAND - && portAttr.link_layer != IBV_LINK_LAYER_ETHERNET) continue; + if (portAttr.link_layer != IBV_LINK_LAYER_INFINIBAND && portAttr.link_layer != IBV_LINK_LAYER_ETHERNET) continue; // check against user specified HCAs/ports if (! (matchIfList(devices[d]->name, port_num, userIfs, nUserIfs, searchExact) ^ searchNot)) { continue; } - pthread_mutex_init(&ncclIbDevs[ncclNIbDevs].lock, NULL); - ncclIbDevs[ncclNIbDevs].device = d; - ncclIbDevs[ncclNIbDevs].ibProvider = ibProvider; - ncclIbDevs[ncclNIbDevs].guid = devAttr.sys_image_guid; - ncclIbDevs[ncclNIbDevs].portAttr = portAttr; - ncclIbDevs[ncclNIbDevs].portNum = port_num; - ncclIbDevs[ncclNIbDevs].link = portAttr.link_layer; - #if HAVE_STRUCT_IBV_PORT_ATTR_ACTIVE_SPEED_EX - portSpeed = portAttr.active_speed_ex ? portAttr.active_speed_ex : portAttr.active_speed; - #else - portSpeed = portAttr.active_speed; - #endif - ncclIbDevs[ncclNIbDevs].speed = nccl_p2p_ib_speed(portSpeed) * nccl_p2p_ib_width(portAttr.active_width); - ncclIbDevs[ncclNIbDevs].context = context; - ncclIbDevs[ncclNIbDevs].pdRefs = 0; - ncclIbDevs[ncclNIbDevs].pd = NULL; - if (!dataDirect) { - strncpy(ncclIbDevs[ncclNIbDevs].devName, devices[d]->name, MAXNAMESIZE); - NCCLCHECKGOTO(nccl_p2p_ib_pci_path(ncclIbDevs, ncclNIbDevs, ncclIbDevs[ncclNIbDevs].devName, &ncclIbDevs[ncclNIbDevs].pciPath, &ncclIbDevs[ncclNIbDevs].realPort), ret, fail); - } else { - snprintf(ncclIbDevs[ncclNIbDevs].devName, MAXNAMESIZE, "%s_dma", devices[d]->name); - ncclIbDevs[ncclNIbDevs].pciPath = malloc(PATH_MAX); - strncpy(ncclIbDevs[ncclNIbDevs].pciPath, dataDirectDevicePath, PATH_MAX); - ncclIbDevs[ncclNIbDevs].capsProvider.mlx5.dataDirect = 1; - } - ncclIbDevs[ncclNIbDevs].maxQp = devAttr.max_qp; - ncclIbDevs[ncclNIbDevs].mrCache.capacity = 0; - ncclIbDevs[ncclNIbDevs].mrCache.population = 0; - ncclIbDevs[ncclNIbDevs].mrCache.slots = NULL; - NCCLCHECK(ncclIbStatsInit(&ncclIbDevs[ncclNIbDevs].stats)); - - // Enable ADAPTIVE_ROUTING by default on IB networks - // But allow it to be overloaded by an env parameter - ncclIbDevs[ncclNIbDevs].ar = (portAttr.link_layer == IBV_LINK_LAYER_INFINIBAND) ? 1 : 0; - if (ncclParamIbAdaptiveRouting() != -2) ncclIbDevs[ncclNIbDevs].ar = ncclParamIbAdaptiveRouting(); - - ncclIbDevs[ncclNIbDevs].isSharpDev = 0; - if (portAttr.link_layer == IBV_LINK_LAYER_INFINIBAND) - { - ncclIbDevs[ncclNIbDevs].isSharpDev = 1; - ncclIbDevs[ncclNIbDevs].maxQp = ncclParamSharpMaxComms(); - ncclNSharpDevs++; + // check for mlx5 data direct support only once for a each device + if (devCount == -1) { + devCount = 1; + devOffset = 0; + if (ncclParamIbDataDirect() > 0 && ibProvider == IB_PROVIDER_MLX5 && ncclMlx5dvDmaBufCapable(context)) { + int pathLen = strlen(dataDirectDevicePath); + ncclResult_t res = wrap_mlx5dv_get_data_direct_sysfs_path(context, dataDirectDevicePath + pathLen, sizeof(dataDirectDevicePath) - pathLen); + if (res == ncclSuccess) { + // data direct devices are exposed twice: with the C2C + PCIe link and with the data direct link + devCount = 2; + // by default only expose the data direct NIC (devOffset = 1), unless set to 2 by the user + devOffset = (ncclParamIbDataDirect() == 2) ? 0 : 1; + INFO(NCCL_INIT | NCCL_NET, "NET/IB: Data Direct DMA Interface is detected for device %s", devices[d]->name); + } else if (res == ncclInvalidArgument) { + TRACE(NCCL_NET, "NET/IB: Device %s does not support Data Direct DMA.", devices[d]->name); + } else { + WARN("NET/IB: Error in mlx5dv_get_data_direct_sysfs_path with device %s", devices[d]->name); + ret = res; + goto fail; + } + } } - TRACE(NCCL_NET,"NET/IB: [%d] %s:%s:%d/%s provider=%s speed=%d context=%p pciPath=%s ar=%d", d, devices[d]->name, devices[d]->dev_name, ncclIbDevs[ncclNIbDevs].portNum, - NCCL_IB_LLSTR(portAttr.link_layer), ibProviderName[ncclIbDevs[ncclNIbDevs].ibProvider], ncclIbDevs[ncclNIbDevs].speed, context, ncclIbDevs[ncclNIbDevs].pciPath, ncclIbDevs[ncclNIbDevs].ar); - if (ncclIbAsyncThread != NULL) { - PTHREADCHECKGOTO(pthread_create(ncclIbAsyncThread, NULL, ncclIbAsyncThreadMain, ncclIbDevs + ncclNIbDevs), "pthread_create", ret, fail); - ncclSetThreadName(*ncclIbAsyncThread, "NCCL IbAsync %2d", ncclNIbDevs); - PTHREADCHECKGOTO(pthread_detach(*ncclIbAsyncThread), "pthread_detach", ret, fail); // will not be pthread_join()'d - } - - // Add this plain physical device to the list of virtual devices - int vDev; - ncclNetVDeviceProps_t vProps = {0}; - vProps.ndevs = 1; - vProps.devs[0] = ncclNIbDevs; - NCCLCHECK(ncclIbMakeVDeviceInternal(&vDev, &vProps, ncclNIbDevs, &ncclNMergedIbDevs)); - ncclNIbDevs++; - nPorts++; - } + for (int dev = devOffset; dev < devCount; ++dev) { + pthread_mutex_init(&ncclIbDevs[ncclNIbDevs].lock, NULL); + ncclIbDevs[ncclNIbDevs].device = d; + ncclIbDevs[ncclNIbDevs].ibProvider = ibProvider; + ncclIbDevs[ncclNIbDevs].guid = devAttr.sys_image_guid; + ncclIbDevs[ncclNIbDevs].portAttr = portAttr; + ncclIbDevs[ncclNIbDevs].portNum = port_num; + ncclIbDevs[ncclNIbDevs].link = portAttr.link_layer; + #if HAVE_STRUCT_IBV_PORT_ATTR_ACTIVE_SPEED_EX + portSpeed = portAttr.active_speed_ex ? portAttr.active_speed_ex : portAttr.active_speed; + #else + portSpeed = portAttr.active_speed; + #endif + ncclIbDevs[ncclNIbDevs].speed = nccl_p2p_ib_speed(portSpeed) * nccl_p2p_ib_width(portAttr.active_width); + ncclIbDevs[ncclNIbDevs].context = context; + ncclIbDevs[ncclNIbDevs].pdRefs = 0; + ncclIbDevs[ncclNIbDevs].pd = NULL; + if (dev ==0) { + strncpy(ncclIbDevs[ncclNIbDevs].devName, devices[d]->name, MAXNAMESIZE); + NCCLCHECKGOTO(nccl_p2p_ib_pci_path(ncclIbDevs, ncclNIbDevs, ncclIbDevs[ncclNIbDevs].devName, &ncclIbDevs[ncclNIbDevs].pciPath, &ncclIbDevs[ncclNIbDevs].realPort), ret, fail); + } else { + snprintf(ncclIbDevs[ncclNIbDevs].devName, MAXNAMESIZE, "%s_dma", devices[d]->name); + ncclIbDevs[ncclNIbDevs].pciPath = malloc(PATH_MAX); + strncpy(ncclIbDevs[ncclNIbDevs].pciPath, dataDirectDevicePath, PATH_MAX); + ncclIbDevs[ncclNIbDevs].capsProvider.mlx5.dataDirect = 1; + } + ncclIbDevs[ncclNIbDevs].maxQp = devAttr.max_qp; + ncclIbDevs[ncclNIbDevs].mrCache.capacity = 0; + ncclIbDevs[ncclNIbDevs].mrCache.population = 0; + ncclIbDevs[ncclNIbDevs].mrCache.slots = NULL; + NCCLCHECK(ncclIbStatsInit(&ncclIbDevs[ncclNIbDevs].stats)); + + // Enable ADAPTIVE_ROUTING by default on IB networks + // But allow it to be overloaded by an env parameter + ncclIbDevs[ncclNIbDevs].ar = (portAttr.link_layer == IBV_LINK_LAYER_INFINIBAND) ? 1 : 0; + if (ncclParamIbAdaptiveRouting() != -2) ncclIbDevs[ncclNIbDevs].ar = ncclParamIbAdaptiveRouting(); + + ncclIbDevs[ncclNIbDevs].isSharpDev = 0; + if (portAttr.link_layer == IBV_LINK_LAYER_INFINIBAND) { + ncclIbDevs[ncclNIbDevs].isSharpDev = 1; + ncclIbDevs[ncclNIbDevs].maxQp = ncclParamSharpMaxComms(); + ncclNSharpDevs++; + } + + TRACE(NCCL_NET,"NET/IB: [%d] %s:%s:%d/%s provider=%s speed=%d context=%p pciPath=%s ar=%d", d, devices[d]->name, devices[d]->dev_name, ncclIbDevs[ncclNIbDevs].portNum, + NCCL_IB_LLSTR(portAttr.link_layer), ibProviderName[ncclIbDevs[ncclNIbDevs].ibProvider], ncclIbDevs[ncclNIbDevs].speed, context, ncclIbDevs[ncclNIbDevs].pciPath, ncclIbDevs[ncclNIbDevs].ar); + if (ncclIbAsyncThread != NULL) { + PTHREADCHECKGOTO(pthread_create(ncclIbAsyncThread, NULL, ncclIbAsyncThreadMain, ncclIbDevs + ncclNIbDevs), "pthread_create", ret, fail); + ncclSetThreadName(*ncclIbAsyncThread, "NCCL IbAsync %2d", ncclNIbDevs); + PTHREADCHECKGOTO(pthread_detach(*ncclIbAsyncThread), "pthread_detach", ret, fail); // will not be pthread_join()'d + } + + // Add this plain physical device to the list of virtual devices + int vDev; + ncclNetVDeviceProps_t vProps = {0}; + vProps.ndevs = 1; + vProps.devs[0] = ncclNIbDevs; + NCCLCHECK(ncclIbMakeVDeviceInternal(&vDev, &vProps, ncclNIbDevs, &ncclNMergedIbDevs)); + + ncclNIbDevs++; + nPorts++; + } } if (nPorts == 0 && ncclSuccess != wrap_ibv_close_device(context)) { ret = ncclInternalError; goto fail; } }