Revert re-usage of connect and listen ports

[ROCm/rccl commit: 12ac20ade5]

projects/rccl/src/include/socket.h

@@ -77,7 +77,7 @@ ncclResult_t ncclSocketInit(struct ncclSocket* sock, union ncclSocketAddress* ad
 ncclResult_t ncclSocketListen(struct ncclSocket* sock);
 ncclResult_t ncclSocketGetAddr(struct ncclSocket* sock, union ncclSocketAddress* addr);
 // Connect to sock->addr. sock->fd is set after a successful call.
-ncclResult_t ncclSocketConnect(struct ncclSocket* sock, int portReuse = 0);
+ncclResult_t ncclSocketConnect(struct ncclSocket* sock);
 // Return socket connection state.
 ncclResult_t ncclSocketReady(struct ncclSocket* sock, int *running);
 // Accept an incoming connection from listenSock->fd and keep the file descriptor in sock->fd, with the remote side IP/port in sock->addr.

projects/rccl/src/misc/socket.cc

@@ -11,15 +11,9 @@
 #include <unistd.h>
 #include <ifaddrs.h>
 #include <net/if.h>
-#include "param.h"
-
-#include <vector>
-#include <utility>
-#include <unordered_set>
 #include <unistd.h>
 #include <sys/syscall.h>
-
-static std::vector<std::pair<int, std::unordered_set<std::string>>> clientPortPool;
+#include "param.h"
 
 static ncclResult_t socketProgressOpt(int op, struct ncclSocket* sock, void* ptr, int size, int* offset, int block, int* closed) {
   int bytes = 0;
@@ -598,7 +592,7 @@ ncclResult_t ncclSocketReady(struct ncclSocket* sock, int *running) {
   return ncclSuccess;
 }
 
-ncclResult_t ncclSocketConnect(struct ncclSocket* sock, int portReuse) {
+ncclResult_t ncclSocketConnect(struct ncclSocket* sock) {
   char line[SOCKET_NAME_MAXLEN+1];
   const int one = 1;
 
@@ -620,42 +614,6 @@ ncclResult_t ncclSocketConnect(struct ncclSocket* sock, int portReuse) {
 
   SYSCHECK(setsockopt(sock->fd, IPPROTO_TCP, TCP_NODELAY, (char*)&one, sizeof(int)), "setsockopt");
 
-  if (portReuse) {
-    int family = sock->addr.sa.sa_family;
-    if (family != AF_INET && family != AF_INET6) {
-      WARN("Net : connecting to address %s with family %d is neither AF_INET(%d) nor AF_INET6(%d)",
-           ncclSocketToString(&sock->addr, line), family, AF_INET, AF_INET6);
-      return ncclInternalError;
-    }
-    int salen = (family == AF_INET) ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6);    // pre-define ports according to tid, to avoid extra lock for race condition
-
-    if (clientPortPool.size() == 0) {
-      for (int tid = syscall(SYS_gettid), i = 1; i < 5; i++) {
-        clientPortPool.push_back(std::make_pair(60000 + i * 1000 + tid % 1000, std::unordered_set<std::string>()));
-      }
-    }
-    // find a port without conflict (different remote peer) in best effort
-    int reused_port = -1;
-    std::string remote_peer(ncclSocketToString(&sock->addr, line));
-    for (auto& port : clientPortPool) {
-      if (port.second.find(remote_peer) == port.second.end()) {
-        reused_port = port.first;
-        port.second.insert(remote_peer);
-        break;
-      }
-    }
-    // bind the port in fd for connect system call
-    if (reused_port != -1) {
-      int opt = 1;
-      SYSCHECK(setsockopt(sock->fd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt)), "setsockopt");
-      struct sockaddr_in sin;
-      sin.sin_family = family;
-      sin.sin_addr.s_addr = htonl(INADDR_ANY);
-      sin.sin_port = htons(reused_port);
-      SYSCHECK(bind(sock->fd, (struct sockaddr *)&sin, salen), "bind_client_port");
-    }
-  }
-
   sock->state = ncclSocketStateConnecting;
   do {
     NCCLCHECK(socketProgressState(sock));

projects/rccl/src/transport/net_ib.cc

@@ -85,11 +85,6 @@ NCCL_PARAM(IbArThreshold, "IB_AR_THRESHOLD", 8192);
 NCCL_PARAM(IbPciRelaxedOrdering, "IB_PCI_RELAXED_ORDERING", 2);
 NCCL_PARAM(IbAdaptiveRouting, "IB_ADAPTIVE_ROUTING", -2);
 
-NCCL_PARAM(IbSockClientPortReuse, "IB_SOCK_CLIENT_PORT_REUSE", 0);
-NCCL_PARAM(IbSockServerPortReuse, "IB_SOCK_SERVER_PORT_REUSE", 0);
-static thread_local union ncclSocketAddress reusedAddr;
-static thread_local int reusedSockfd = -1;
-
 pthread_t ncclIbAsyncThread;
 static void* ncclIbAsyncThreadMain(void* args) {
   struct ibv_context* context = (struct ibv_context*)args;
@@ -631,19 +626,7 @@ ncclResult_t ncclIbListen(int dev, void* opaqueHandle, void** listenComm) {
   comm->dev = dev;
   handle->magic = NCCL_SOCKET_MAGIC;
   NCCLCHECK(ncclSocketInit(&comm->sock, &ncclIbIfAddr, handle->magic, ncclSocketTypeNetIb, NULL, 1));
-  if (ncclParamIbSockServerPortReuse()) {
-    // reuse the socket address and fd for listen system call
-    if (reusedSockfd == -1) {
-      NCCLCHECK(ncclSocketListen(&comm->sock));
-      memcpy(&reusedAddr, &comm->sock.addr, sizeof(union ncclSocketAddress));
-      reusedSockfd = comm->sock.fd;
-    } else {
-      memcpy(&comm->sock.addr, &reusedAddr, sizeof(union ncclSocketAddress));
-      comm->sock.fd = reusedSockfd;
-    }
-  } else {
-    NCCLCHECK(ncclSocketListen(&comm->sock));
-  }
+  NCCLCHECK(ncclSocketListen(&comm->sock));
   NCCLCHECK(ncclSocketGetAddr(&comm->sock, &handle->connectAddr));
   *listenComm = comm;
   return ncclSuccess;
@@ -669,7 +652,7 @@ ncclResult_t ncclIbConnect(int dev, void* opaqueHandle, void** sendComm, ncclNet
   NCCLCHECK(ncclSocketInit(&comm->sock, &handle->connectAddr, handle->magic, ncclSocketTypeNetIb, NULL, 1));
   stage->comm = comm;
   stage->state = ncclIbCommStateConnect;
-  NCCLCHECK(ncclSocketConnect(&comm->sock, ncclParamIbSockClientPortReuse()));
+  NCCLCHECK(ncclSocketConnect(&comm->sock));
 
 ib_connect_check:
   /* since ncclSocketConnect is async, we must check if connection is complete */
@@ -1413,7 +1396,7 @@ ncclResult_t ncclIbCloseSend(void* sendComm) {
 ncclResult_t ncclIbCloseRecv(void* recvComm) {
   struct ncclIbRecvComm* comm = (struct ncclIbRecvComm*)recvComm;
   if (comm) {
-    if (!ncclParamIbSockServerPortReuse() || reusedSockfd != comm->sock.fd) NCCLCHECK(ncclSocketClose(&comm->sock));
+    NCCLCHECK(ncclSocketClose(&comm->sock));
     for (int q=0; q<comm->nqps; q++)
       if (comm->qps[q] != NULL) NCCLCHECK(wrap_ibv_destroy_qp(comm->qps[q]));
     if (comm->gpuFlush.enabled) {