diff --git a/src/include/socket.h b/src/include/socket.h index a3246eabb8..44d152d208 100644 --- a/src/include/socket.h +++ b/src/include/socket.h @@ -333,7 +333,8 @@ static ncclResult_t createListenSocket(int *fd, union socketAddress *localAddr) } #if defined(RCCL_IB_TEST) - localAddr->sin.sin_port = htons(23456); + static int port = 23456; + localAddr->sin.sin_port = htons(port++); #endif if (socketToPort(&localAddr->sa)) { diff --git a/tools/ib-test/ib_test.cpp b/tools/ib-test/ib_test.cpp index 53556f9451..f60efc37eb 100755 --- a/tools/ib-test/ib_test.cpp +++ b/tools/ib-test/ib_test.cpp @@ -28,6 +28,8 @@ #include #include #include +#include +#include ncclResult_t initNet(); @@ -49,50 +51,10 @@ bool cmdOptionExists(char** begin, char** end, const std::string& option) { #define DEFAULT_CYCLES 4000 #define VEGA_GPU_RTC_FREQUENCY 2.5E7 #define ENABLE_VALIDATION -#define USE_MEMALIGN #define USE_UNROLL 8 typedef ulong2 Pack128; -struct sockaddr_in netConnectAddr; -void* netSendComm; -int netSendDev; -char *sendDevBuffer; -char *sendHostBuffer, *d_sendHostBuffer; -void *sendDevHandle; -void *sendHostHandle; -int sendBuffSize; -uint64_t *sendHead, *sendTail, *sourceCycle, *sourceBytes; -struct timeval send_tvs; -uint64_t send_sizes; -int send_active_req; -float send_bw_cumulative; -int send_bw_count; - -struct sockaddr_in netListenAddr; -void* netListenComm; -void* netRecvComm; -int netRecvDev; -char *recvDevBuffer; -char *recvHostBuffer, *d_recvHostBuffer; -void *recvDevHandle; -void *recvHostHandle; -int recvBuffSize; -uint64_t *recvHead, *recvTail, *recvErrorCount, *sinkCycle, *sinkBytes; -struct timeval recv_tvs; -uint64_t recv_sizes; -int recv_active_req; -float recv_bw_cumulative; -int recv_bw_count; - -bool use_gdr_read = false, use_gdr_write = true; -bool runSend = false, runRecv = false; - -uint64_t send_byte; -uint64_t recv_byte; - -uint64_t iterations = 1; - __device__ inline __attribute((always_inline)) long long int __rtc64() { @@ -176,50 +138,6 @@ __global__ void DataSinkKernel(const uint64_t end, Pack128* data, uint64_t* recv if (tid == 0) STORE(mismatch, error); } -ncclResult_t netRecvProxy(struct ncclProxyArgs* args) { - char* localBuff = use_gdr_write ? recvDevBuffer : recvHostBuffer; - void* mhandle = use_gdr_write ? recvDevHandle : recvHostHandle; - int stepSize = recvBuffSize / NCCL_STEPS; - if (args->head < args->end) { - if ((args->tail < args->head + NCCL_STEPS) && (args->tail < LOAD(recvTail) + NCCL_STEPS) && (args->tail < args->end)) { - int buffSlot = args->tail%NCCL_STEPS; - int sliceSize = stepSize * args->sliceSteps; - NCCLCHECK(ncclNetIrecv(netRecvComm, localBuff+buffSlot*stepSize, sliceSize, mhandle, args->requests+buffSlot)); - if (args->requests[buffSlot] != NULL) { - if (recv_active_req == 0) { - gettimeofday(&recv_tvs, NULL); - recv_sizes = 0; - } - recv_active_req ++; - args->tail += args->sliceSteps; - args->idle = 0; - } - } - if (args->tail > args->head) { - int buffSlot = args->head%NCCL_STEPS; - int done, size; - NCCLCHECK(ncclNetTest(args->requests[buffSlot], &done, &size)); - if (done) { - recv_active_req --; - recv_sizes += size; - if (recv_active_req == 0) { - struct timeval tv; - gettimeofday(&tv, NULL); - recv_bw_cumulative += (float)recv_sizes/((tv.tv_sec - recv_tvs.tv_sec)*1000*1000 + tv.tv_usec - recv_tvs.tv_usec)/1000.0; - recv_bw_count ++; - } - args->head += args->sliceSteps; - recv_byte += size; - NCCLCHECK(ncclNetFlush(netRecvComm, localBuff+buffSlot*stepSize, size, mhandle)); - STORE(recvHead, args->head); - args->idle = 0; - } - } - } else { - runRecv = false; - } - return ncclSuccess; -} __global__ void DataSourceKernel(const uint64_t end, Pack128* data, uint64_t* send_head, uint64_t* send_tail, uint64_t *source_cycle, uint64_t *source_bytes) { const int N = DEFAULT_BUFFSIZE*SLICE_STEPS/NCCL_STEPS/sizeof(Pack128); @@ -247,58 +165,358 @@ __global__ void DataSourceKernel(const uint64_t end, Pack128* data, uint64_t* se } while (head < end); } -ncclResult_t netSendProxy(struct ncclProxyArgs* args) { - char* localBuff = use_gdr_read ? sendDevBuffer : sendHostBuffer; - void* mhandle = use_gdr_read ? sendDevHandle : sendHostHandle; - int stepSize = sendBuffSize / NCCL_STEPS; - int sliceSize = stepSize * args->sliceSteps; - if (args->head < args->end) { - if (args->tail < args->end && args->tail < args->head + NCCL_STEPS) { - if (args->tail < LOAD(sendHead)) { - int buffSlot = args->tail%NCCL_STEPS; - NCCLCHECK(ncclNetIsend(netSendComm, localBuff+buffSlot*stepSize, sliceSize, mhandle, args->requests+buffSlot)); - if (args->requests[buffSlot] != NULL) { - if (send_active_req == 0) { - gettimeofday(&send_tvs, NULL); - send_sizes = 0; - } - send_active_req ++; - send_sizes += sliceSize; - send_byte += sliceSize; - __sync_synchronize(); - args->tail += args->sliceSteps; - args->idle = 0; - } - } - } - if (args->head < args->tail) { - int done; - int buffSlot = args->head%NCCL_STEPS; - NCCLCHECK(ncclNetTest(args->requests[buffSlot], &done, NULL)); - if (done) { - send_active_req --; - if (send_active_req == 0) { - struct timeval tv; - gettimeofday(&tv, NULL); - send_bw_cumulative += (float)send_sizes/((tv.tv_sec - send_tvs.tv_sec)*1000*1000 + tv.tv_usec - send_tvs.tv_usec)/1000.0; - send_bw_count ++; - } - args->head += args->sliceSteps; - STORE(sendTail, args->head); - args->idle = 0; - } - } +class sendrecvChannel { +protected: + int id; + hipStream_t stream; + +public: + sendrecvChannel(int id) : id(id) { + hipStreamCreate(&stream); } - else - runSend = false; - return ncclSuccess; -} + + virtual ~sendrecvChannel() { + hipStreamDestroy(stream); + } + + virtual void launchKernel(uint64_t end) = 0; + virtual void printProgress(uint64_t total_time) = 0; + virtual bool netProxy() = 0; +}; + +class sendChannel : public sendrecvChannel { +private: + struct sockaddr_in netConnectAddr; + void* netSendComm; + int netSendDev; + char *sendDevBuffer; + char *sendHostBuffer, *d_sendHostBuffer; + void *sendDevHandle; + void *sendHostHandle; + int sendBuffSize; + uint64_t *sendHead, *sendTail, *sourceCycle, *sourceBytes; + struct timeval send_tvs; + uint64_t send_sizes; + int send_active_req; + float send_bw_cumulative; + int send_bw_count; + uint64_t send_byte; + bool runSend; + bool use_gdr_read; + int sliceSteps; + struct ncclProxyArgs args; + + ncclResult_t connect(char* ip, uint16_t port) { + inet_pton(AF_INET, ip, &netConnectAddr.sin_addr); + if (port) + netConnectAddr.sin_port = htons(port); + else + netConnectAddr.sin_port = htons(23456); + + netConnectAddr.sin_family = AF_INET; + printf("Connecting to %s:%d\n", ip, ntohs(netConnectAddr.sin_port)); + + printf("GDR Read %s\n", use_gdr_read ? "enabled" : "disabled"); + + if (use_gdr_read) { + NCCLCHECK(ncclCudaCalloc(&sendDevBuffer, sendBuffSize, 1)); + printf("Allocated sendDevBuffer %p of %d bytes, sliceSteps %d\n", + sendDevBuffer, sendBuffSize, sliceSteps); + } + else { + NCCLCHECK(ncclCudaHostCalloc(&sendHostBuffer, sendBuffSize)); + d_sendHostBuffer = sendHostBuffer; + int status[1] = {-1}; + if (!move_pages(0, 1, (void **)&sendHostBuffer, NULL, status, 0)) + printf("Allocated sendHostBuffer %p of %d bytes on node %d, sliceSteps %d\n", + sendHostBuffer, sendBuffSize, status[0], sliceSteps); + } + + NCCLCHECK(ncclCudaHostCalloc(&sendHead, 1)); + NCCLCHECK(ncclCudaHostCalloc(&sendTail, 1)); + NCCLCHECK(ncclCudaHostCalloc(&sourceCycle, 1)); + NCCLCHECK(ncclCudaHostCalloc(&sourceBytes, 1)); + netSendDev = 0; + NCCLCHECK(ncclNetConnect(netSendDev, &netConnectAddr, &netSendComm)); + + if (use_gdr_read) { + NCCLCHECK(ncclNetRegMr(netSendComm, sendDevBuffer, sendBuffSize, NCCL_PTR_CUDA, &sendDevHandle)); + } else { + NCCLCHECK(ncclNetRegMr(netSendComm, sendHostBuffer, sendBuffSize, NCCL_PTR_HOST, &sendHostHandle)); + } + + return ncclSuccess; + } + + ncclResult_t teardown() { + NCCLCHECK(ncclCudaHostFree(sourceCycle)); + NCCLCHECK(ncclCudaHostFree(sourceBytes)); + NCCLCHECK(ncclCudaHostFree(sendHead)); + NCCLCHECK(ncclCudaHostFree(sendTail)); + if (use_gdr_read) { + NCCLCHECK(ncclNetDeregMr(netSendComm, sendDevHandle)); + CUDACHECK(hipFree(sendDevBuffer)); + } else { + NCCLCHECK(ncclNetDeregMr(netSendComm, sendHostHandle)); + NCCLCHECK(ncclCudaHostFree(sendHostBuffer)); + } + NCCLCHECK(ncclNetCloseSend(netSendComm)); + return ncclSuccess; + } + +public: + sendChannel(int id, bool gdr_read, char* ip, uint16_t port) : sendrecvChannel(id), use_gdr_read(gdr_read) { + sendBuffSize = DEFAULT_BUFFSIZE; + sliceSteps = SLICE_STEPS; + connect(ip, port); + } + + ~sendChannel() { + teardown(); + } + + bool netProxy() { + char* localBuff = use_gdr_read ? sendDevBuffer : sendHostBuffer; + void* mhandle = use_gdr_read ? sendDevHandle : sendHostHandle; + int stepSize = sendBuffSize / NCCL_STEPS; + int sliceSize = stepSize * args.sliceSteps; + if (!runSend) return runSend; + if (args.head < args.end) { + if (args.tail < args.end && args.tail < args.head + NCCL_STEPS) { + if (args.tail < LOAD(sendHead)) { + int buffSlot = args.tail%NCCL_STEPS; + NCCLCHECK(ncclNetIsend(netSendComm, localBuff+buffSlot*stepSize, sliceSize, mhandle, args.requests+buffSlot)); + if (args.requests[buffSlot] != NULL) { + if (send_active_req == 0) { + gettimeofday(&send_tvs, NULL); + send_sizes = 0; + } + send_active_req ++; + send_sizes += sliceSize; + send_byte += sliceSize; + __sync_synchronize(); + args.tail += args.sliceSteps; + args.idle = 0; + } + } + } + if (args.head < args.tail) { + int done; + int buffSlot = args.head%NCCL_STEPS; + NCCLCHECK(ncclNetTest(args.requests[buffSlot], &done, NULL)); + if (done) { + send_active_req --; + if (send_active_req == 0) { + struct timeval tv; + gettimeofday(&tv, NULL); + send_bw_cumulative += (float)send_sizes/((tv.tv_sec - send_tvs.tv_sec)*1000*1000 + tv.tv_usec - send_tvs.tv_usec)/1000.0; + send_bw_count ++; + } + args.head += args.sliceSteps; + STORE(sendTail, args.head); + args.idle = 0; + } + } + } + else + runSend = false; + return runSend; + } + + void launchKernel(uint64_t end) { + *sendHead = 0; *sendTail = 0; *sourceCycle = 0; *sourceBytes = 0; + send_sizes = 0; send_bw_cumulative = 0; send_bw_count =0; send_byte = 0; + memset(&args, 0, sizeof(struct ncclProxyArgs)); + args.head = 0; + args.tail = 0; + args.end = end; + args.sliceSteps = SLICE_STEPS; + hipLaunchKernelGGL(DataSourceKernel, dim3(1, 1, 1), dim3(256, 1, 1), 0, stream, + end, (Pack128 *)(use_gdr_read ? sendDevBuffer : d_sendHostBuffer), sendHead, sendTail, sourceCycle, sourceBytes); + runSend = true; + } + + void printProgress(uint64_t total_time) { + if (send_byte) printf("# Send[%d] %3ld%% %6.2f GB/s (%ld bytes %ld us) Proxy %6.2f GB/s (%d mmts) Kernel %6.2f GB/s (%ld bytes)\n", + id, args.head*100/args.end, (total_time) ? (double)send_byte/total_time/1000.0 : 0, + send_byte, total_time, send_bw_count ? (float)send_bw_cumulative/send_bw_count : 0, send_bw_count, + *sourceCycle ? (double)(*sourceBytes)*sizeof(Pack128)/((double)(*sourceCycle)/VEGA_GPU_RTC_FREQUENCY*1.0E9) : 0, *sourceBytes*sizeof(Pack128)); + } +}; + +class recvChannel : public sendrecvChannel { +private: + struct sockaddr_in netListenAddr; + void* netListenComm; + void* netRecvComm; + int netRecvDev; + char *recvDevBuffer; + char *recvHostBuffer, *d_recvHostBuffer; + void *recvDevHandle; + void *recvHostHandle; + int recvBuffSize; + uint64_t *recvHead, *recvTail, *recvErrorCount, *sinkCycle, *sinkBytes; + struct timeval recv_tvs; + uint64_t recv_sizes; + int recv_active_req; + float recv_bw_cumulative; + int recv_bw_count; + uint64_t recv_byte; + bool runRecv; + bool use_gdr_write; + int sliceSteps; + struct ncclProxyArgs args; + + ncclResult_t listen() { + printf("GDR Write %s\n", use_gdr_write ? "enabled" : "disabled"); + + if (use_gdr_write) { + NCCLCHECK(ncclCudaCalloc(&recvDevBuffer, recvBuffSize, 1)); + printf("Allocated recvDevBuffer %p of %d bytes, sliceSteps %d\n", + recvDevBuffer, recvBuffSize, sliceSteps); + } + else { + NCCLCHECK(ncclCudaHostCalloc(&recvHostBuffer, recvBuffSize)); + d_recvHostBuffer = recvHostBuffer; + int status[1] = {-1}; + if (!move_pages(0, 1, (void **)&recvHostBuffer, NULL, status, 0)) + printf("Allocated recvHostBuffer %p of %d bytes on node %d, sliceSteps %d\n", + recvHostBuffer, recvBuffSize, status[0], sliceSteps); + } + + NCCLCHECK(ncclCudaHostCalloc(&recvHead, 1)); + NCCLCHECK(ncclCudaHostCalloc(&recvTail, 1)); + NCCLCHECK(ncclCudaHostCalloc(&recvErrorCount, 1)); + NCCLCHECK(ncclCudaHostCalloc(&sinkCycle, 1)); + NCCLCHECK(ncclCudaHostCalloc(&sinkBytes, 1)); + netRecvDev = 0; + NCCLCHECK(ncclNetListen(netRecvDev, (void *)&netListenAddr, &netListenComm)); + char ip[INET_ADDRSTRLEN]; + uint16_t port; + inet_ntop(AF_INET, &netListenAddr.sin_addr, ip, sizeof(ip)); + port = htons(netListenAddr.sin_port); + printf("Listening on socket %s:%d\n", ip, port); + + NCCLCHECK(ncclNetAccept(netListenComm, &netRecvComm)); + NCCLCHECK(ncclNetCloseListen(netListenComm)); + + if (use_gdr_write) { + NCCLCHECK(ncclNetRegMr(netRecvComm, recvDevBuffer, recvBuffSize, NCCL_PTR_CUDA, &recvDevHandle)); + } else { + NCCLCHECK(ncclNetRegMr(netRecvComm, recvHostBuffer, recvBuffSize, NCCL_PTR_HOST, &recvHostHandle)); + } + + return ncclSuccess; + } + + ncclResult_t teardown() { + NCCLCHECK(ncclCudaHostFree(sinkCycle)); + NCCLCHECK(ncclCudaHostFree(sinkBytes)); + NCCLCHECK(ncclCudaHostFree(recvErrorCount)); + NCCLCHECK(ncclCudaHostFree(recvHead)); + NCCLCHECK(ncclCudaHostFree(recvTail)); + if (use_gdr_write) { + NCCLCHECK(ncclNetDeregMr(netRecvComm, recvDevHandle)); + CUDACHECK(hipFree(recvDevBuffer)); + } else { + NCCLCHECK(ncclNetDeregMr(netRecvComm, recvHostHandle)); + NCCLCHECK(ncclCudaHostFree(recvHostBuffer)); + } + NCCLCHECK(ncclNetCloseRecv(netRecvComm)); + return ncclSuccess; + } + +public: + recvChannel(int id, bool gdr_write) : sendrecvChannel(id), use_gdr_write(gdr_write) { + recvBuffSize = DEFAULT_BUFFSIZE; + sliceSteps = SLICE_STEPS; + listen(); + } + + ~recvChannel() { + teardown(); + } + + bool netProxy() { + char* localBuff = use_gdr_write ? recvDevBuffer : recvHostBuffer; + void* mhandle = use_gdr_write ? recvDevHandle : recvHostHandle; + int stepSize = recvBuffSize / NCCL_STEPS; + if (!runRecv) return runRecv; + if (args.head < args.end) { + if ((args.tail < args.head + NCCL_STEPS) && (args.tail < LOAD(recvTail) + NCCL_STEPS) && (args.tail < args.end)) { + int buffSlot = args.tail%NCCL_STEPS; + int sliceSize = stepSize * args.sliceSteps; + NCCLCHECK(ncclNetIrecv(netRecvComm, localBuff+buffSlot*stepSize, sliceSize, mhandle, args.requests+buffSlot)); + if (args.requests[buffSlot] != NULL) { + if (recv_active_req == 0) { + gettimeofday(&recv_tvs, NULL); + recv_sizes = 0; + } + recv_active_req ++; + args.tail += args.sliceSteps; + args.idle = 0; + } + } + if (args.tail > args.head) { + int buffSlot = args.head%NCCL_STEPS; + int done, size; + NCCLCHECK(ncclNetTest(args.requests[buffSlot], &done, &size)); + if (done) { + recv_active_req --; + recv_sizes += size; + if (recv_active_req == 0) { + struct timeval tv; + gettimeofday(&tv, NULL); + recv_bw_cumulative += (float)recv_sizes/((tv.tv_sec - recv_tvs.tv_sec)*1000*1000 + tv.tv_usec - recv_tvs.tv_usec)/1000.0; + recv_bw_count ++; + } + args.head += args.sliceSteps; + recv_byte += size; + NCCLCHECK(ncclNetFlush(netRecvComm, localBuff+buffSlot*stepSize, size, mhandle)); + STORE(recvHead, args.head); + args.idle = 0; + } + } + } else { + runRecv = false; + } + return runRecv; + } + + void launchKernel(uint64_t end) { + *recvHead = 0; *recvTail = 0; *recvErrorCount = 0; *sinkCycle = 0, *sinkBytes = 0; + recv_sizes = 0; recv_bw_cumulative = 0; recv_bw_count =0; recv_byte = 0; + memset(&args, 0, sizeof(struct ncclProxyArgs)); + args.head = 0; + args.tail = 0; + args.end = end; + args.sliceSteps = SLICE_STEPS; + hipLaunchKernelGGL(DataSinkKernel, dim3(1, 1, 1), dim3(256, 1, 1), 0, stream, + end, (Pack128 *)(use_gdr_write ? recvDevBuffer : d_recvHostBuffer), recvHead, recvTail, recvErrorCount, sinkCycle, sinkBytes); + runRecv = true; + } + + void printProgress(uint64_t total_time) { + if (recv_byte) printf("# Recv[%d] %3ld%% %6.2f GB/s (%ld bytes %ld us) Proxy %6.2f GB/s (%d mmts) Kernel %6.2f GB/s (%ld bytes) Errors %ld\n", + id, args.head*100/args.end, (total_time) ? (double)recv_byte/total_time/1000.0 : 0, + recv_byte, total_time, recv_bw_count ? (float)recv_bw_cumulative/recv_bw_count : 0, recv_bw_count, + *sinkCycle ? (double)(*sinkBytes)*sizeof(Pack128)/((double)(*sinkCycle)/VEGA_GPU_RTC_FREQUENCY*1.0E9) : 0, *sinkBytes*sizeof(Pack128), + *recvErrorCount); + } +}; + +#define FOR_EACH \ + for (std::list::iterator it = sendrecvChans.begin(); it != sendrecvChans.end(); it++) int main(int argc,char* argv[]) { struct ncclComm *comm; - int sliceSteps = SLICE_STEPS; bool isSource; + bool use_gdr_read = false, use_gdr_write = true; + int port; + std::list sendrecvChans; + int nc = 1; + uint64_t iterations = 1; NCCLCHECK(initNet()); int ndev; @@ -310,14 +528,18 @@ int main(int argc,char* argv[]) else printf("Found %d IB devices\n", ndev); - sendBuffSize = recvBuffSize = DEFAULT_BUFFSIZE; - char *gpu = getCmdOption(argv, argv + argc, "-g"); if (gpu) { printf("Select GPU %s\n", gpu); CUDACHECK(hipSetDevice(atol(gpu))); } + char *c = getCmdOption(argv, argv + argc, "-c"); + if (c) { + printf("Number of channels %s\n", c); + nc = atol(c); + } + char *iters = getCmdOption(argv, argv + argc, "-i"); if (iters) { iterations = atol(iters); @@ -352,243 +574,55 @@ int main(int argc,char* argv[]) if (cmdOptionExists(argv, argv + argc, "-d")) { char *ip = getCmdOption(argv, argv + argc, "-d"); - if (ip) - inet_pton(AF_INET, ip, &netConnectAddr.sin_addr); - char *port = getCmdOption(argv, argv + argc, "-p"); - if (port) - netConnectAddr.sin_port = htons(atoi(port)); - else - netConnectAddr.sin_port = htons(23456); - - netConnectAddr.sin_family = AF_INET; - printf("Connecting to %s:%d\n", ip, ntohs(netConnectAddr.sin_port)); - - printf("GDR Read %s\n", use_gdr_read ? "enabled" : "disabled"); - - if (use_gdr_read) { - NCCLCHECK(ncclCudaCalloc(&sendDevBuffer, sendBuffSize, 1)); - printf("Allocated sendDevBuffer %p of %d bytes, sliceSteps %d\n", - sendDevBuffer, sendBuffSize, sliceSteps); - } - else { -#ifdef USE_MEMALIGN - int page_size = getpagesize(); - posix_memalign((void **)&sendHostBuffer, page_size, sendBuffSize); - CUDACHECK(hipHostRegister(sendHostBuffer, sendBuffSize, hipHostRegisterMapped)); - CUDACHECK(hipHostGetDevicePointer((void **)&d_sendHostBuffer, sendHostBuffer, 0)); -#else - NCCLCHECK(ncclCudaHostCalloc(&sendHostBuffer, sendBuffSize)); - d_sendHostBuffer = sendHostBuffer; -#endif - int status[1] = {-1}; - if (!move_pages(0, 1, (void **)&sendHostBuffer, NULL, status, 0)) - printf("Allocated sendHostBuffer %p of %d bytes on node %d, sliceSteps %d\n", - sendHostBuffer, sendBuffSize, status[0], sliceSteps); - } - - NCCLCHECK(ncclCudaHostCalloc(&sendHead, 1)); - NCCLCHECK(ncclCudaHostCalloc(&sendTail, 1)); - NCCLCHECK(ncclCudaHostCalloc(&sourceCycle, 1)); - NCCLCHECK(ncclCudaHostCalloc(&sourceBytes, 1)); - netSendDev = 0; - NCCLCHECK(ncclNetConnect(netSendDev, &netConnectAddr, &netSendComm)); - - if (use_gdr_read) { - NCCLCHECK(ncclNetRegMr(netSendComm, sendDevBuffer, sendBuffSize, NCCL_PTR_CUDA, &sendDevHandle)); - } else { - NCCLCHECK(ncclNetRegMr(netSendComm, sendHostBuffer, sendBuffSize, NCCL_PTR_HOST, &sendHostHandle)); - } - - hipLaunchKernelGGL(DataSourceKernel, dim3(1, 1, 1), dim3(256, 1, 1), 0, 0, - NCCL_STEPS, (Pack128 *)(use_gdr_read ? sendDevBuffer : d_sendHostBuffer), sendHead, sendTail, sourceCycle, sourceBytes); - - runSend = true; + char *p = getCmdOption(argv, argv + argc, "-p"); + uint16_t port = 23456; + if (p) port = atol(p); + for (int i = 0; i < nc; i++) sendrecvChans.push_back(new sendChannel(i, use_gdr_read, ip, port++)); isSource = true; } else { - printf("GDR Write %s\n", use_gdr_write ? "enabled" : "disabled"); - - if (use_gdr_write) { - NCCLCHECK(ncclCudaCalloc(&recvDevBuffer, recvBuffSize, 1)); - printf("Allocated recvDevBuffer %p of %d bytes, sliceSteps %d\n", - recvDevBuffer, recvBuffSize, sliceSteps); - } - else { -#ifdef USE_MEMALIGN - int page_size = getpagesize(); - posix_memalign((void **)&recvHostBuffer, page_size, recvBuffSize); - CUDACHECK(hipHostRegister(recvHostBuffer, recvBuffSize, hipHostRegisterMapped)); - CUDACHECK(hipHostGetDevicePointer((void **)&d_recvHostBuffer, recvHostBuffer, 0)); -#else - NCCLCHECK(ncclCudaHostCalloc(&recvHostBuffer, recvBuffSize)); - d_recvHostBuffer = recvHostBuffer; -#endif - int status[1] = {-1}; - if (!move_pages(0, 1, (void **)&recvHostBuffer, NULL, status, 0)) - printf("Allocated recvHostBuffer %p of %d bytes on node %d, sliceSteps %d\n", - recvHostBuffer, recvBuffSize, status[0], sliceSteps); - } - - NCCLCHECK(ncclCudaHostCalloc(&recvHead, 1)); - NCCLCHECK(ncclCudaHostCalloc(&recvTail, 1)); - NCCLCHECK(ncclCudaHostCalloc(&recvErrorCount, 1)); - NCCLCHECK(ncclCudaHostCalloc(&sinkCycle, 1)); - NCCLCHECK(ncclCudaHostCalloc(&sinkBytes, 1)); - netRecvDev = 0; - NCCLCHECK(ncclNetListen(netRecvDev, (void *)&netListenAddr, &netListenComm)); - char ip[INET_ADDRSTRLEN]; - uint16_t port; - inet_ntop(AF_INET, &netListenAddr.sin_addr, ip, sizeof(ip)); - port = htons(netListenAddr.sin_port); - printf("Listening on socket %s:%d\n", ip, port); - - NCCLCHECK(ncclNetAccept(netListenComm, &netRecvComm)); - NCCLCHECK(ncclNetCloseListen(netListenComm)); - - if (use_gdr_write) { - NCCLCHECK(ncclNetRegMr(netRecvComm, recvDevBuffer, recvBuffSize, NCCL_PTR_CUDA, &recvDevHandle)); - } else { - NCCLCHECK(ncclNetRegMr(netRecvComm, recvHostBuffer, recvBuffSize, NCCL_PTR_HOST, &recvHostHandle)); - } - - hipLaunchKernelGGL(DataSinkKernel, dim3(1, 1, 1), dim3(256, 1, 1), 0, 0, - NCCL_STEPS, (Pack128 *)(use_gdr_write ? recvDevBuffer : d_recvHostBuffer), recvHead, recvTail, recvErrorCount, sinkCycle, sinkBytes); - - runRecv = true; + for (int i = 0; i < nc; i++) sendrecvChans.push_back(new recvChannel(i, use_gdr_write)); isSource = false; } - struct ncclProxyArgs sendArgs, recvArgs; - - memset(&sendArgs, 0, sizeof(struct ncclProxyArgs)); - sendArgs.head = 0; - sendArgs.tail = 0; - sendArgs.end = NCCL_STEPS; - sendArgs.sliceSteps = sliceSteps; - - memset(&recvArgs, 0, sizeof(struct ncclProxyArgs)); - recvArgs.head = 0; - recvArgs.tail = 0; - recvArgs.end = NCCL_STEPS; - recvArgs.sliceSteps = sliceSteps; - printf("Running warm up..."); + FOR_EACH (*it)->launchKernel(NCCL_STEPS); + + bool run; do { - if (runRecv) - NCCLCHECK(netRecvProxy(&recvArgs)); - if (runSend) - NCCLCHECK(netSendProxy(&sendArgs)); - } while (runSend || runRecv); + run = false; + FOR_EACH run |= (*it)->netProxy(); + } while (run); CUDACHECK(hipDeviceSynchronize()); printf("completed\n"); // reset all counters after warm up cycle - if (isSource) { - *sendHead = 0; *sendTail = 0; *sourceCycle = 0; *sourceBytes = 0; - send_sizes = 0; send_bw_cumulative = 0; send_bw_count =0; send_byte = 0; - hipLaunchKernelGGL(DataSourceKernel, dim3(1, 1, 1), dim3(256, 1, 1), 0, 0, - NCCL_STEPS*iterations*DEFAULT_CYCLES, (Pack128 *)(use_gdr_read ? sendDevBuffer : d_sendHostBuffer), sendHead, sendTail, sourceCycle, sourceBytes); - runSend = true; - } else { - *recvHead = 0; *recvTail = 0; *recvErrorCount = 0; *sinkCycle = 0, *sinkBytes = 0; - recv_sizes = 0; recv_bw_cumulative = 0; recv_bw_count =0; recv_byte = 0; - hipLaunchKernelGGL(DataSinkKernel, dim3(1, 1, 1), dim3(256, 1, 1), 0, 0, - NCCL_STEPS*iterations*DEFAULT_CYCLES, (Pack128 *)(use_gdr_write ? recvDevBuffer : d_recvHostBuffer), recvHead, recvTail, recvErrorCount, sinkCycle, sinkBytes); - runRecv = true; - } + FOR_EACH (*it)->launchKernel(NCCL_STEPS*iterations*DEFAULT_CYCLES); struct timeval tv_start, tv_end, tv_prev; gettimeofday(&tv_start, NULL); gettimeofday(&tv_prev, NULL); - memset(&sendArgs, 0, sizeof(struct ncclProxyArgs)); - sendArgs.head = 0; - sendArgs.tail = 0; - sendArgs.end = NCCL_STEPS*iterations*DEFAULT_CYCLES; - sendArgs.sliceSteps = sliceSteps; - - memset(&recvArgs, 0, sizeof(struct ncclProxyArgs)); - recvArgs.head = 0; - recvArgs.tail = 0; - recvArgs.end = NCCL_STEPS*iterations*DEFAULT_CYCLES; - recvArgs.sliceSteps = sliceSteps; - do { - if (runRecv) - NCCLCHECK(netRecvProxy(&recvArgs)); - if (runSend) - NCCLCHECK(netSendProxy(&sendArgs)); + run = false; + FOR_EACH run |= (*it)->netProxy(); gettimeofday(&tv_end, NULL); uint64_t timelap = ((uint64_t)(tv_end.tv_sec - tv_prev.tv_sec)*1000*1000 + tv_end.tv_usec - tv_prev.tv_usec); if (timelap > 100000UL) { uint64_t total_time = ((uint64_t)(tv_end.tv_sec - tv_start.tv_sec)*1000*1000 + tv_end.tv_usec - tv_start.tv_usec); - if (send_byte) printf("# Send %3ld%% %6.2f GB/s (%ld bytes %ld us) Proxy %6.2f GB/s (%d mmts) Kernel %6.2f GB/s (%ld bytes)\r", - sendArgs.head*100/(NCCL_STEPS*iterations*DEFAULT_CYCLES), (total_time) ? (double)send_byte/total_time/1000.0 : 0, - send_byte, total_time, send_bw_count ? (float)send_bw_cumulative/send_bw_count : 0, send_bw_count, - *sourceCycle ? (double)(*sourceBytes)*sizeof(Pack128)/((double)(*sourceCycle)/VEGA_GPU_RTC_FREQUENCY*1.0E9) : 0, *sourceBytes*sizeof(Pack128)); - if (recv_byte) printf("# Recv %3ld%% %6.2f GB/s (%ld bytes %ld us) Proxy %6.2f GB/s (%d mmts) Kernel %6.2f GB/s (%ld bytes) Errors %ld\r", - recvArgs.head*100/(NCCL_STEPS*iterations*DEFAULT_CYCLES), (total_time) ? (double)recv_byte/total_time/1000.0 : 0, - recv_byte, total_time, recv_bw_count ? (float)recv_bw_cumulative/recv_bw_count : 0, recv_bw_count, - *sinkCycle ? (double)(*sinkBytes)*sizeof(Pack128)/((double)(*sinkCycle)/VEGA_GPU_RTC_FREQUENCY*1.0E9) : 0, *sinkBytes*sizeof(Pack128), - *recvErrorCount); + FOR_EACH (*it)->printProgress(total_time); gettimeofday(&tv_prev, NULL); } - } while (runSend || runRecv); + } while (run); CUDACHECK(hipDeviceSynchronize()); gettimeofday(&tv_end, NULL); uint64_t total_time = ((uint64_t)(tv_end.tv_sec - tv_start.tv_sec)*1000*1000 + tv_end.tv_usec - tv_start.tv_usec); - if (send_byte) printf("# Send %3ld%% %6.2f GB/s (%ld bytes %ld us) Proxy %6.2f GB/s (%d mmts) Kernel %6.2f GB/s (%ld bytes)\n", - sendArgs.head*100/(NCCL_STEPS*iterations*DEFAULT_CYCLES), (total_time) ? (double)send_byte/total_time/1000.0 : 0, - send_byte, total_time, send_bw_count ? (float)send_bw_cumulative/send_bw_count : 0, send_bw_count, - *sourceCycle ? (double)(*sourceBytes)*sizeof(Pack128)/((double)(*sourceCycle)/VEGA_GPU_RTC_FREQUENCY*1.0E9) : 0, *sourceBytes*sizeof(Pack128)); - if (recv_byte) printf("# Recv %3ld%% %6.2f GB/s (%ld bytes %ld us) Proxy %6.2f GB/s (%d mmts) Kernel %6.2f GB/s (%ld bytes) Errors %ld\n", - recvArgs.head*100/(NCCL_STEPS*iterations*DEFAULT_CYCLES), (total_time) ? (double)recv_byte/total_time/1000.0 : 0, - recv_byte, total_time, recv_bw_count ? (float)recv_bw_cumulative/recv_bw_count : 0, recv_bw_count, - *sinkCycle ? (double)(*sinkBytes)*sizeof(Pack128)/((double)(*sinkCycle)/VEGA_GPU_RTC_FREQUENCY*1.0E9) : 0, *sinkBytes*sizeof(Pack128), - *recvErrorCount); - - if (isSource) { - NCCLCHECK(ncclCudaHostFree(sourceCycle)); - NCCLCHECK(ncclCudaHostFree(sourceBytes)); - NCCLCHECK(ncclCudaHostFree(sendHead)); - NCCLCHECK(ncclCudaHostFree(sendTail)); - if (use_gdr_read) { - NCCLCHECK(ncclNetDeregMr(netSendComm, sendDevHandle)); - CUDACHECK(hipFree(sendDevBuffer)); - } else { - NCCLCHECK(ncclNetDeregMr(netSendComm, sendHostHandle)); -#ifdef USE_MEMALIGN - CUDACHECK(hipHostUnregister(sendHostBuffer)); - free(sendHostBuffer); -#else - NCCLCHECK(ncclCudaHostFree(sendHostBuffer)); -#endif - } - NCCLCHECK(ncclNetCloseSend(netSendComm)); - } else { - NCCLCHECK(ncclCudaHostFree(sinkCycle)); - NCCLCHECK(ncclCudaHostFree(sinkBytes)); - NCCLCHECK(ncclCudaHostFree(recvErrorCount)); - NCCLCHECK(ncclCudaHostFree(recvHead)); - NCCLCHECK(ncclCudaHostFree(recvTail)); - if (use_gdr_write) { - NCCLCHECK(ncclNetDeregMr(netRecvComm, recvDevHandle)); - CUDACHECK(hipFree(recvDevBuffer)); - } else { - NCCLCHECK(ncclNetDeregMr(netRecvComm, recvHostHandle)); -#ifdef USE_MEMALIGN - CUDACHECK(hipHostUnregister(recvHostBuffer)); - free(recvHostBuffer); -#else - NCCLCHECK(ncclCudaHostFree(recvHostBuffer)); -#endif - } - NCCLCHECK(ncclNetCloseRecv(netRecvComm)); - } + FOR_EACH (*it)->printProgress(total_time); + FOR_EACH delete *it; return 0; }