diff --git a/projects/rccl/src/include/socket.h b/projects/rccl/src/include/socket.h index 9f271798bc..9e51372892 100644 --- a/projects/rccl/src/include/socket.h +++ b/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. diff --git a/projects/rccl/src/misc/socket.cc b/projects/rccl/src/misc/socket.cc index 6928694c19..bb3353a8e6 100644 --- a/projects/rccl/src/misc/socket.cc +++ b/projects/rccl/src/misc/socket.cc @@ -11,15 +11,9 @@ #include #include #include -#include "param.h" - -#include -#include -#include #include #include - -static std::vector>> 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())); - } - } - // 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)); diff --git a/projects/rccl/src/transport/net_ib.cc b/projects/rccl/src/transport/net_ib.cc index dea2e233db..0a80859651 100644 --- a/projects/rccl/src/transport/net_ib.cc +++ b/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; qnqps; q++) if (comm->qps[q] != NULL) NCCLCHECK(wrap_ibv_destroy_qp(comm->qps[q])); if (comm->gpuFlush.enabled) {