AI/rocm-systems
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'nccl/master' into develop

Browse Source 
[ROCm/rccl commit: 0b2062c560]
This commit is contained in:
BertanDogancay
2025-03-27 12:51:55 -05:00
parent f0c853438c 60240fec77
commit 8ed27fde74
92 changed files with 7322 additions and 2168 deletions
Download Patch File Download Diff File Expand all files Collapse all files
projects/rccl/src/misc/socket.cc
+13 -7
View File
@@ -289,6 +289,7 @@ ncclResult_t ncclSocketGetAddrFromString(union ncclSocketAddress* ua, const char
sin6.sin6_scope_id = 0; // should be global scope, set to 0
} else {
WARN("Net : unsupported IP family");
freeaddrinfo(p);
return ncclInvalidArgument;
}
@@ -413,7 +414,7 @@ ncclResult_t ncclSocketGetAddr(struct ncclSocket* sock, union ncclSocketAddress*
static ncclResult_t socketTryAccept(struct ncclSocket* sock) {
socklen_t socklen = sizeof(union ncclSocketAddress);
sock->fd = accept(sock->acceptFd, &sock->addr.sa, &socklen);
sock->fd = accept(sock->acceptFd, (struct sockaddr*)&sock->addr, &socklen);
if (sock->fd != -1) {
sock->state = ncclSocketStateAccepted;
} else if (errno != EAGAIN && errno != EWOULDBLOCK) {
@@ -506,8 +507,9 @@ static ncclResult_t socketPollConnect(struct ncclSocket* sock) {
} else if (ret < 0) {
WARN("socketPollConnect poll() failed with error %s", strerror(errno));
return ncclRemoteError;
} else {
EQCHECK(ret == 1 && (pfd.revents & POLLOUT), 0);
} else if (ret != 1 || (pfd.revents & POLLOUT) == 0) {
WARN("socketPollConnect poll() returned %d%s", ret, (pfd.revents & POLLOUT) ? "" : ", no POLLOUT events");
return ncclSystemError;
}
/* check socket status */
@@ -734,12 +736,12 @@ ncclResult_t ncclSocketInit(struct ncclSocket* sock, union ncclSocketAddress* ad
// [RCCL] Runtime socket options
if (rcclParamSocketReuseAddr()) {
int opt = 1;
SYSCHECKGOTO(setsockopt(sock->fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)), ret, fail);
SYSCHECKGOTO(setsockopt(sock->fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)), "setsockopt", ret, fail);
}
int lingerParam = (int)rcclParamSocketLinger();
if (lingerParam > -1) {
linger linger_opt = { 1, lingerParam };
SYSCHECKGOTO(setsockopt(sock->fd, SOL_SOCKET, SO_LINGER, &linger_opt, sizeof(linger_opt)), ret, fail);
SYSCHECKGOTO(setsockopt(sock->fd, SOL_SOCKET, SO_LINGER, &linger_opt, sizeof(linger_opt)), "setsockopt", ret, fail);
}
} else {
memset(&sock->addr, 0, sizeof(union ncclSocketAddress));
@@ -748,13 +750,17 @@ ncclResult_t ncclSocketInit(struct ncclSocket* sock, union ncclSocketAddress* ad
/* Set socket as non-blocking if async or if we need to be able to abort */
if ((sock->asyncFlag || sock->abortFlag) && sock->fd >= 0) {
int flags;
EQCHECKGOTO(flags = fcntl(sock->fd, F_GETFL), -1, ret, fail);
SYSCHECKGOTO(fcntl(sock->fd, F_SETFL, flags | O_NONBLOCK), ret, fail);
SYSCHECKGOTO(flags = fcntl(sock->fd, F_GETFL), "fcntl", ret, fail);
SYSCHECKGOTO(fcntl(sock->fd, F_SETFL, flags | O_NONBLOCK), "fcntl", ret, fail);
}
exit:
return ret;
fail:
if (sock->fd != -1) {
close(sock->fd);
sock->fd = -1;
}
goto exit;
}