diff --git a/include/net.h b/include/net.h index be9858c..465d0c0 100644 --- a/include/net.h +++ b/include/net.h @@ -9,6 +9,7 @@ #include #include +#include #include "net_device.h" #define NCCL_NET_HANDLE_MAXSIZE 128 @@ -17,11 +18,14 @@ #define MAX_COLLNET_SIZE (512*1024*1024L) #define NCCL_NET_OPTIONAL_RECV_COMPLETION 0x1 - #define NCCL_PTR_HOST 0x1 #define NCCL_PTR_CUDA 0x2 #define NCCL_PTR_DMABUF 0x4 +#define NCCL_NET_MR_FLAG_FORCE_SO (1 << 0) +#define NCCL_NET_SIGNAL_OP_INC 0x1 +#define NCCL_NET_SIGNAL_OP_ADD 0x2 + // Maximum number of requests per comm object #define NCCL_NET_MAX_REQUESTS 8 #define NCCL_NET_MAX_DEVS_PER_NIC 4 diff --git a/include/net_v11.h b/include/net_v11.h index 6b8fd3a..0930dd3 100644 --- a/include/net_v11.h +++ b/include/net_v11.h @@ -2,192 +2,237 @@ * 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 +#ifndef NET_V11_H_ +#define NET_V11_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; + +typedef struct { + // Name of the GIN support (mainly for logs) + const char* name; + // Initialize the GIN support. + ncclResult_t (*init)(void** ctx, uint64_t commId, ncclDebugLogger_t logFunction); + // Return the number of adapters capable of doing GIN 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 GIN operations. handles have been created + // using listen() above. rank indicates caller's rank in the collective network. + ncclResult_t (*connect)(void* ctx, void* handles[], int nranks, int rank, void* listenComm, void** collComm); + // Create device-side GIN context. devHandle will be passed to device code. + // This function is not used in GIN_PROXY mode. + ncclResult_t (*createContext)(void* collComm, int nSignals, int nCounters, void** ginCtx, ncclNetDeviceHandle_v11_t** devHandle); + // Collective memory registration + ncclResult_t (*regMrSym)(void* collComm, void* data, size_t size, int type, uint64_t mrFlags, void** mhandle, void **ginHandle); + ncclResult_t (*regMrSymDmaBuf)(void* collComm, void* data, size_t size, int type, uint64_t offset, int fd, uint64_t mrFlags, void** mhandle, void **ginHandle); + ncclResult_t (*deregMrSym)(void* collComm, void* mhandle); + // Close and free collective comm objects + ncclResult_t (*destroyContext)(void* ginCtx); + ncclResult_t (*closeColl)(void* collComm); + ncclResult_t (*closeListen)(void* listenComm); + + // Put operations + ncclResult_t (*iput)(void* collComm, uint64_t srcOff, void* srcMhandle, size_t size, + uint64_t dstOff, void* dstMhandle, uint32_t rank, void** request); + ncclResult_t (*iputSignal)(void* collComm, uint64_t srcOff, void* srcMhandle, + size_t size, uint64_t dstOff, void* dstMhandle, + uint32_t rank, uint64_t signalOff, void *signalMhandle, + uint64_t signalValue, uint32_t signalOp, void** request); + + // Test whether a request is complete. + ncclResult_t (*test)(void* collComm, void* request, int* done); + + // Progress function. Will be called if non-NULL in GIN_PROXY mode, or if devHandle.needsProxyProgress=1. + ncclResult_t (*ginProgress)(void* collComm); + + // Query the last error for the GIN support. Particularly important when ginProgress is not used, to report errors. + ncclResult_t (*queryLastError)(void* ginCtx, bool *hasError); + + // Finalize the GIN support + ncclResult_t (*finalize)(void* ctx); +} ncclGin_v11_t; +#endif // end include guard diff --git a/src/ib_plugin.c b/src/ib_plugin.c index 574712e..5d7f68c 100644 --- a/src/ib_plugin.c +++ b/src/ib_plugin.c @@ -30,8 +30,6 @@ 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; @@ -73,6 +71,50 @@ static ncclResult_t ncclIbStatsCheckFatalCount(struct ncclIbStats* stat, const c return ncclSuccess; } +// Helper function to convert IB work completion status to string +static const char* ibvWcStatusStr(enum ibv_wc_status status) { + switch (status) { + case IBV_WC_SUCCESS: return "IBV_WC_SUCCESS"; + case IBV_WC_LOC_LEN_ERR: return "IBV_WC_LOC_LEN_ERR"; + case IBV_WC_LOC_QP_OP_ERR: return "IBV_WC_LOC_QP_OP_ERR"; + case IBV_WC_LOC_EEC_OP_ERR: return "IBV_WC_LOC_EEC_OP_ERR"; + case IBV_WC_LOC_PROT_ERR: return "IBV_WC_LOC_PROT_ERR"; + case IBV_WC_WR_FLUSH_ERR: return "IBV_WC_WR_FLUSH_ERR"; + case IBV_WC_MW_BIND_ERR: return "IBV_WC_MW_BIND_ERR"; + case IBV_WC_BAD_RESP_ERR: return "IBV_WC_BAD_RESP_ERR"; + case IBV_WC_LOC_ACCESS_ERR: return "IBV_WC_LOC_ACCESS_ERR"; + case IBV_WC_REM_INV_REQ_ERR: return "IBV_WC_REM_INV_REQ_ERR"; + case IBV_WC_REM_ACCESS_ERR: return "IBV_WC_REM_ACCESS_ERR"; + case IBV_WC_REM_OP_ERR: return "IBV_WC_REM_OP_ERR"; + case IBV_WC_RETRY_EXC_ERR: return "IBV_WC_RETRY_EXC_ERR"; + case IBV_WC_RNR_RETRY_EXC_ERR: return "IBV_WC_RNR_RETRY_EXC_ERR"; + case IBV_WC_LOC_RDD_VIOL_ERR: return "IBV_WC_LOC_RDD_VIOL_ERR"; + case IBV_WC_REM_INV_RD_REQ_ERR: return "IBV_WC_REM_INV_RD_REQ_ERR"; + case IBV_WC_REM_ABORT_ERR: return "IBV_WC_REM_ABORT_ERR"; + case IBV_WC_INV_EECN_ERR: return "IBV_WC_INV_EECN_ERR"; + case IBV_WC_INV_EEC_STATE_ERR: return "IBV_WC_INV_EEC_STATE_ERR"; + case IBV_WC_FATAL_ERR: return "IBV_WC_FATAL_ERR"; + case IBV_WC_RESP_TIMEOUT_ERR: return "IBV_WC_RESP_TIMEOUT_ERR"; + case IBV_WC_GENERAL_ERR: return "IBV_WC_GENERAL_ERR"; + default: return "UNKNOWN_STATUS"; + } +} + +// Helper function to convert IB work completion opcode to string +static const char* ibvWcOpcodeStr(enum ibv_wc_opcode opcode) { + switch (opcode) { + case IBV_WC_SEND: return "IBV_WC_SEND"; + case IBV_WC_RDMA_WRITE: return "IBV_WC_RDMA_WRITE"; + case IBV_WC_RDMA_READ: return "IBV_WC_RDMA_READ"; + case IBV_WC_COMP_SWAP: return "IBV_WC_COMP_SWAP"; + case IBV_WC_FETCH_ADD: return "IBV_WC_FETCH_ADD"; + case IBV_WC_BIND_MW: return "IBV_WC_BIND_MW"; + case IBV_WC_RECV: return "IBV_WC_RECV"; + case IBV_WC_RECV_RDMA_WITH_IMM: return "IBV_WC_RECV_RDMA_WITH_IMM"; + default: return "UNKNOWN_OPCODE"; + } +} + static pthread_t ncclIbAsyncThread; // Determine whether RELAXED_ORDERING is enabled and possible @@ -644,7 +686,12 @@ ncclResult_t ncclIbSetNetAttr(void *ctx, ncclNetAttr_t *netAttr) { return ncclSuccess; } -ncclResult_t ncclIbInit(void** ctx, uint64_t commId, ncclNetCommConfig_v11_t* config, ncclDebugLogger_t logFunction, ncclProfilerCallback_t profFunction) { +static ncclResult_t ncclIbFinalizeDevices(void) { + ibRefCount--; + return ncclSuccess; +} + +static ncclResult_t ncclIbInitDevices(ncclDebugLogger_t logFunction, ncclProfilerCallback_t profFunction) { if (ibRefCount++) return ncclSuccess; if (ncclParamIbDisable()) return ncclInternalError; @@ -661,8 +708,20 @@ ncclResult_t ncclIbInit(void** ctx, uint64_t commId, ncclNetCommConfig_v11_t* co 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; +} + +ncclResult_t ncclIbInit(void** ctx, uint64_t commId, ncclNetCommConfig_t* config, ncclDebugLogger_t logFunction, ncclProfilerCallback_t profFunction) { + ncclNetCommConfig_t* netCommConfig = NULL; + NCCLCHECK(ncclIbInitDevices(logFunction, profFunction)); + + if (ctx) { + netCommConfig = malloc(sizeof(ncclNetCommConfig_t)); + if (!netCommConfig) return ncclInternalError; + memset(netCommConfig, 0, sizeof(ncclNetCommConfig_t)); + if (config) netCommConfig->trafficClass = config->trafficClass; + *ctx = (void *)netCommConfig; + } return ncclSuccess; } @@ -956,7 +1015,7 @@ ncclResult_t ncclIbConnect(void *ctx, int dev, void* opaqueHandle, void** sendC INFO(NCCL_NET,"NET/IB: %s %d IbDev %d Port %d qpn %d mtu %d LID %d subnet-prefix %lu FLID %d fifoRkey=0x%x fifoLkey=0x%x", comm->base.vProps.ndevs > 2 ? "NCCL MergedDev" : "NCCL Dev", dev, commDev->base.ibDevN, ibDev->portNum, meta.qpInfo[q].qpn, devInfo->mtu, devInfo->lid, - devInfo->gid.global.subnet_prefix, ncclIbExtractFlid(&devInfo->gid), devInfo->fifoRkey, commDev->fifoMr->lkey); + (uint64_t)devInfo->gid.global.subnet_prefix, ncclIbExtractFlid(&devInfo->gid), devInfo->fifoRkey, commDev->fifoMr->lkey); } } else { // RoCE for (int q = 0; q < comm->base.nqps; q++) { @@ -965,7 +1024,7 @@ ncclResult_t ncclIbConnect(void *ctx, int dev, void* opaqueHandle, void** sendC INFO(NCCL_NET,"NET/IB: %s %d IbDev %d Port %d qpn %d mtu %d query_ece={supported=%d, vendor_id=0x%x, options=0x%x, comp_mask=0x%x} GID %ld (%lX/%lX) fifoRkey=0x%x fifoLkey=0x%x", comm->base.vProps.ndevs > 2 ? "NCCL MergedDev" : "NCCL Dev", dev, commDev->base.ibDevN, ibDev->portNum, meta.qpInfo[q].qpn, devInfo->mtu, meta.qpInfo[q].ece_supported, meta.qpInfo[q].ece.vendor_id, meta.qpInfo[q].ece.options, meta.qpInfo[q].ece.comp_mask, (int64_t)commDev->base.gidInfo.localGidIndex, - devInfo->gid.global.subnet_prefix, devInfo->gid.global.interface_id, devInfo->fifoRkey, commDev->fifoMr->lkey); + (uint64_t)devInfo->gid.global.subnet_prefix, devInfo->gid.global.interface_id, devInfo->fifoRkey, commDev->fifoMr->lkey); } } if (link_layer == IBV_LINK_LAYER_UNSPECIFIED) link_layer = devInfo->link_layer; @@ -1420,7 +1479,7 @@ ncclResult_t ncclIbFreeRequest(struct ncclIbRequest* r) { ncclResult_t ncclIbTest(void* request, int* done, int* size); -ncclResult_t ncclIbRegMrDmaBufInternal(ncclIbNetCommDevBase* base, void* data, size_t size, int type, uint64_t offset, int fd, struct ibv_mr** mhandle) { +ncclResult_t ncclIbRegMrDmaBufInternal2(ncclIbNetCommDevBase* base, void* data, size_t size, int type, uint64_t offset, int fd, uint64_t mrFlags, struct ibv_mr** mhandle) { static __thread uintptr_t pageSize = 0; if (pageSize == 0) pageSize = sysconf(_SC_PAGESIZE); @@ -1437,8 +1496,9 @@ ncclResult_t ncclIbRegMrDmaBufInternal(ncclIbNetCommDevBase* base, void* data, s } // Deregister / register struct ibv_mr* mr; - unsigned int flags = IBV_ACCESS_LOCAL_WRITE|IBV_ACCESS_REMOTE_WRITE|IBV_ACCESS_REMOTE_READ; - if (ncclIbRelaxedOrderingEnabled) flags |= IBV_ACCESS_RELAXED_ORDERING; + unsigned int flags = IBV_ACCESS_LOCAL_WRITE|IBV_ACCESS_REMOTE_WRITE|IBV_ACCESS_REMOTE_READ|IBV_ACCESS_REMOTE_ATOMIC; + bool relaxedOrdering = ncclIbRelaxedOrderingEnabled && (mrFlags & NCCL_NET_MR_FLAG_FORCE_SO) == 0; + if (relaxedOrdering) flags |= IBV_ACCESS_RELAXED_ORDERING; if (fd != -1) { /* DMA-BUF support */ if (!ncclIbDevs[base->ibDevN].capsProvider.mlx5.dataDirect) { @@ -1447,7 +1507,7 @@ ncclResult_t ncclIbRegMrDmaBufInternal(ncclIbNetCommDevBase* base, void* data, s NCCLCHECKGOTO(wrap_mlx5dv_reg_dmabuf_mr(&mr, base->pd, offset, pages*pageSize, addr, fd, flags, 1), res, returning); } } else { - if (ncclIbRelaxedOrderingEnabled) { + if (relaxedOrdering) { // Use IBVERBS_1.8 API - needed for IBV_ACCESS_RELAXED_ORDERING support NCCLCHECKGOTO(wrap_ibv_reg_mr_iova2(&mr, base->pd, (void*)addr, pages*pageSize, addr, flags), res, returning); } @@ -1489,7 +1549,7 @@ struct ncclIbNetCommDevBase* ncclIbGetNetCommDevBase(ncclIbNetCommBase* base, in } /* DMA-BUF support */ -ncclResult_t ncclIbRegMrDmaBuf(void* comm, void* data, size_t size, int type, uint64_t offset, int fd, void** mhandle) { +ncclResult_t ncclIbRegMrDmaBufInternal(void* comm, void* data, size_t size, int type, uint64_t offset, int fd, uint64_t mrFlags, void** mhandle) { ncclResult_t ret = ncclSuccess; assert(size > 0); struct ncclIbNetCommBase* base = (struct ncclIbNetCommBase*) comm; @@ -1497,7 +1557,7 @@ ncclResult_t ncclIbRegMrDmaBuf(void* comm, void* data, size_t size, int type, ui for (int i = 0; i < base->vProps.ndevs; i++) { // Each ncclIbNetCommDevBase is at different offset in send and recv netComms struct ncclIbNetCommDevBase* devComm = ncclIbGetNetCommDevBase(base, i); - NCCLCHECKGOTO(ncclIbRegMrDmaBufInternal(devComm, data, size, type, offset, fd, mhandleWrapper->mrs + i), ret, fail); + NCCLCHECKGOTO(ncclIbRegMrDmaBufInternal2(devComm, data, size, type, offset, fd, mrFlags, mhandleWrapper->mrs + i), ret, fail); } *mhandle = (void*) mhandleWrapper; exit: @@ -1507,8 +1567,12 @@ ncclResult_t ncclIbRegMrDmaBuf(void* comm, void* data, size_t size, int type, ui goto exit; } +ncclResult_t ncclIbRegMrDmaBuf(void* comm, void* data, size_t size, int type, uint64_t offset, int fd, void** mhandle) { + return ncclIbRegMrDmaBufInternal(comm, data, size, type, offset, fd, 0ULL, mhandle); +} + ncclResult_t ncclIbRegMr(void* comm, void* data, size_t size, int type, void** mhandle) { - return ncclIbRegMrDmaBuf(comm, data, size, type, 0ULL, -1, mhandle); + return ncclIbRegMrDmaBufInternal(comm, data, size, type, 0ULL, -1, 0, mhandle); } ncclResult_t ncclIbDeregMrInternal(ncclIbNetCommDevBase* base, struct ibv_mr* mhandle) { @@ -1960,8 +2024,15 @@ ncclResult_t ncclIbTest(void* request, int* done, int* sizes) { char line[SOCKET_NAME_MAXLEN+1]; char *hcaName = r->devBases[i]->pd->context->device->name; - WARN("NET/IB: Got completion from peer %s with status=%d opcode=%d len=%u vendor err %u (%s)%s%s%s%s hca %s", - ncclSocketToString(&addr, line, 1), wc->status, wc->opcode, wc->byte_len, wc->vendor_err, reqTypeStr[r->type], + int reqSize = wc->byte_len; + struct ncclIbRequest* req = r->base->reqs+(wc->wr_id & 0xff); + if (req && req->type == NCCL_NET_IB_REQ_SEND) { + // For Send use the request size as WC byte_len is not reliable + reqSize = req->send.size; + } + WARN("NET/IB: Got completion from peer %s with status=%s(%d) opcode=%s(%d) reqSize=%d vendor_err=%u req_type=%s%s%s%s%s hca %s", + ncclSocketToString(&addr, line, 1), ibvWcStatusStr(wc->status), wc->status, + ibvWcOpcodeStr(wc->opcode), wc->opcode, reqSize, wc->vendor_err, reqTypeStr[r->type], localGidStr ? " localGid ":"", localGidString, remoteGidStr ? " remoteGids":"", remoteGidString, hcaName); return ncclRemoteError; } @@ -1972,8 +2043,9 @@ ncclResult_t ncclIbTest(void* request, int* done, int* sizes) { #ifdef ENABLE_TRACE char line[SOCKET_NAME_MAXLEN+1]; - TTRACE(NCCL_NET, "Got completion from peer %s with status=%d opcode=%d len=%u wr_id=%lu r=%p type=%d events={%d,%d}, i=%d", - ncclSocketToString(&addr, line, 1), wc->status, wc->opcode,wc->byte_len, wc->wr_id, req, req->type, req->events[0], req->events[1], i); + TRACE(NCCL_NET, "Got completion from peer %s with status=%s(%d) opcode=%s(%d) len=%u wr_id=%lu r=%p type=%d events={%d,%d}, i=%d", + ncclSocketToString(&addr, line, 1), ibvWcStatusStr(wc->status), wc->status, ibvWcOpcodeStr(wc->opcode), wc->opcode,wc->byte_len, + wc->wr_id, req, req->type, req->events[0], req->events[1], i); #endif if (req && req->type == NCCL_NET_IB_REQ_SEND) { for (int j = 0; j < req->nreqs; j++) { @@ -2068,8 +2140,8 @@ ncclResult_t ncclIbMakeVDevice(int* d, ncclNetVDeviceProps_t* props) { } ncclResult_t ncclIbFinalize(void* ctx) { - ibRefCount--; - return ncclSuccess; + if (ctx) free(ctx); + return ncclIbFinalizeDevices(); } ncclResult_t ncclIbInit_v10(ncclDebugLogger_t logFunction, ncclProfilerCallback_t profilerCallback) { @@ -2260,4 +2332,4 @@ const ncclNet_v6_t ibPlugin_v6 = { .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 1d8ab5a..ad54c4d 100644 --- a/src/p2p_plugin.c +++ b/src/p2p_plugin.c @@ -421,7 +421,7 @@ static void* ncclIbAsyncThreadMain(void* args) { case IBV_EVENT_CLIENT_REREGISTER: case IBV_EVENT_SRQ_LIMIT_REACHED: // the above are non-fatal - WARN("NET/IB : %s:%d Got async error event: %s", dev->devName, dev->portNum, str); + WARN("NET/IB : %s:%d Got non-fatal async event: %s(%d)", dev->devName, dev->portNum, str, event.event_type); break; case IBV_EVENT_COMM_EST: break;