#include "socket.h" #include #include #include #include #include #include #include #include #include using namespace sccl; static std::vector>> clientPortPool; static scclResult_t socketProgressOpt(int op, struct scclSocket* sock, void* ptr, int size, int* offset, int block, int* closed) { int bytes = 0; *closed = 0; char* data = (char*)ptr; char line[SOCKET_NAME_MAXLEN + 1]; do { if(op == SCCL_SOCKET_RECV) bytes = recv(sock->fd, data + (*offset), size - (*offset), block ? 0 : MSG_DONTWAIT); if(op == SCCL_SOCKET_SEND) bytes = send(sock->fd, data + (*offset), size - (*offset), block ? MSG_NOSIGNAL : MSG_DONTWAIT | MSG_NOSIGNAL); if(op == SCCL_SOCKET_RECV && bytes == 0) { *closed = 1; return scclSuccess; } if(bytes == -1) { if(errno != EINTR && errno != EWOULDBLOCK && errno != EAGAIN) { WARN("socketProgressOpt: Call to recv from %s failed : %s", scclSocketToString(&sock->addr, line), strerror(errno)); return scclRemoteError; } else { bytes = 0; } } (*offset) += bytes; if(sock->abortFlag && *sock->abortFlag != 0) { INFO(SCCL_LOG_CODEALL, "socketProgressOpt: abort called"); return scclInternalError; } } while(bytes > 0 && (*offset) < size); return scclSuccess; } static scclResult_t socketProgress(int op, struct scclSocket* sock, void* ptr, int size, int* offset) { int closed; SCCLCHECK(socketProgressOpt(op, sock, ptr, size, offset, 0 /*block*/, &closed)); if(closed) { char line[SOCKET_NAME_MAXLEN + 1]; WARN("socketProgress: Connection closed by remote peer %s", scclSocketToString(&sock->addr, line, 0)); return scclRemoteError; } return scclSuccess; } static scclResult_t socketWait(int op, struct scclSocket* sock, void* ptr, int size, int* offset) { while(*offset < size) SCCLCHECK(socketProgress(op, sock, ptr, size, offset)); return scclSuccess; } /* Format a string representation of a (union scclSocketAddress *) socket address using getnameinfo() * * Output: "IPv4/IPv6 address" */ const char* scclSocketToString(union scclSocketAddress* addr, char* buf, const int numericHostForm /*= 1*/) { if(buf == NULL || addr == NULL) return NULL; struct sockaddr* saddr = &addr->sa; if(saddr->sa_family != AF_INET && saddr->sa_family != AF_INET6) { buf[0] = '\0'; return buf; } char host[NI_MAXHOST], service[NI_MAXSERV]; /* NI_NUMERICHOST: If set, then the numeric form of the hostname is returned. * (When not set, this will still happen in case the node's name cannot be determined.) */ int flag = NI_NUMERICSERV | (numericHostForm ? NI_NUMERICHOST : 0); (void)getnameinfo(saddr, sizeof(union scclSocketAddress), host, NI_MAXHOST, service, NI_MAXSERV, flag); sprintf(buf, "%s<%s>", host, service); return buf; } static uint16_t socketToPort(union scclSocketAddress* addr) { struct sockaddr* saddr = &addr->sa; return ntohs(saddr->sa_family == AF_INET ? addr->sin.sin_port : addr->sin6.sin6_port); } /* Allow the user to force the IPv4/IPv6 interface selection */ static int envSocketFamily(void) { int family = -1; // Family selection is not forced, will use first one found char* env = getenv("SCCL_SOCKET_FAMILY"); if(env == NULL) return family; INFO(SCCL_LOG_CODEALL, "SCCL_SOCKET_FAMILY set by environment to %s", env); if(strcmp(env, "AF_INET") == 0) family = AF_INET; // IPv4 else if(strcmp(env, "AF_INET6") == 0) family = AF_INET6; // IPv6 return family; } static int findInterfaces(const char* prefixList, char* names, union scclSocketAddress* addrs, int sock_family, int maxIfNameSize, int maxIfs) { struct netIf userIfs[MAX_IFS]; bool searchNot = prefixList && prefixList[0] == '^'; if(searchNot) prefixList++; bool searchExact = prefixList && prefixList[0] == '='; if(searchExact) prefixList++; int nUserIfs = parseStringList(prefixList, userIfs, MAX_IFS); int found = 0; struct ifaddrs *interfaces, *interface; getifaddrs(&interfaces); for(interface = interfaces; interface && found < maxIfs; interface = interface->ifa_next) { if(interface->ifa_addr == NULL) continue; /* We only support IPv4 & IPv6 */ int family = interface->ifa_addr->sa_family; if(family != AF_INET && family != AF_INET6) continue; /* Allow the caller to force the socket family type */ if(sock_family != -1 && family != sock_family) continue; /* We also need to skip IPv6 loopback interfaces */ if(family == AF_INET6) { struct sockaddr_in6* sa = (struct sockaddr_in6*)(interface->ifa_addr); if(IN6_IS_ADDR_LOOPBACK(&sa->sin6_addr)) continue; } // check against user specified interfaces if(!(matchIfList(interface->ifa_name, -1, userIfs, nUserIfs, searchExact) ^ searchNot)) { continue; } // Check that this interface has not already been saved // getifaddrs() normal order appears to be; IPv4, IPv6 Global, IPv6 Link bool duplicate = false; for(int i = 0; i < found; i++) { if(strcmp(interface->ifa_name, names + i * maxIfNameSize) == 0) { duplicate = true; break; } } if(!duplicate) { // Store the interface name strncpy(names + found * maxIfNameSize, interface->ifa_name, maxIfNameSize); // Store the IP address int salen = (family == AF_INET) ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6); memcpy(addrs + found, interface->ifa_addr, salen); found++; } } freeifaddrs(interfaces); return found; } static bool matchSubnet(struct ifaddrs local_if, union scclSocketAddress* remote) { /* Check family first */ int family = local_if.ifa_addr->sa_family; if(family != remote->sa.sa_family) { return false; } if(family == AF_INET) { struct sockaddr_in* local_addr = (struct sockaddr_in*)(local_if.ifa_addr); struct sockaddr_in* mask = (struct sockaddr_in*)(local_if.ifa_netmask); struct sockaddr_in& remote_addr = remote->sin; struct in_addr local_subnet, remote_subnet; local_subnet.s_addr = local_addr->sin_addr.s_addr & mask->sin_addr.s_addr; remote_subnet.s_addr = remote_addr.sin_addr.s_addr & mask->sin_addr.s_addr; return (local_subnet.s_addr ^ remote_subnet.s_addr) ? false : true; } else if(family == AF_INET6) { struct sockaddr_in6* local_addr = (struct sockaddr_in6*)(local_if.ifa_addr); struct sockaddr_in6* mask = (struct sockaddr_in6*)(local_if.ifa_netmask); struct sockaddr_in6& remote_addr = remote->sin6; struct in6_addr& local_in6 = local_addr->sin6_addr; struct in6_addr& mask_in6 = mask->sin6_addr; struct in6_addr& remote_in6 = remote_addr.sin6_addr; bool same = true; int len = 16; // IPv6 address is 16 unsigned char for(int c = 0; c < len; c++) { // Network byte order is big-endian char c1 = local_in6.s6_addr[c] & mask_in6.s6_addr[c]; char c2 = remote_in6.s6_addr[c] & mask_in6.s6_addr[c]; if(c1 ^ c2) { same = false; break; } } // At last, we need to compare scope id // Two Link-type addresses can have the same subnet address even though they are not in the same scope // For Global type, this field is 0, so a comparison wouldn't matter same &= (local_addr->sin6_scope_id == remote_addr.sin6_scope_id); return same; } else { WARN("Net : Unsupported address family type"); return false; } } int scclFindInterfaceMatchSubnet(char* ifNames, union scclSocketAddress* localAddrs, union scclSocketAddress* remoteAddr, int ifNameMaxSize, int maxIfs) { char line_a[SOCKET_NAME_MAXLEN + 1]; int found = 0; struct ifaddrs *interfaces, *interface; getifaddrs(&interfaces); for(interface = interfaces; interface && !found; interface = interface->ifa_next) { if(interface->ifa_addr == NULL) continue; /* We only support IPv4 & IPv6 */ int family = interface->ifa_addr->sa_family; if(family != AF_INET && family != AF_INET6) continue; // check against user specified interfaces if(!matchSubnet(*interface, remoteAddr)) { continue; } // Store the local IP address int salen = (family == AF_INET) ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6); memcpy(localAddrs + found, interface->ifa_addr, salen); // Store the interface name strncpy(ifNames + found * ifNameMaxSize, interface->ifa_name, ifNameMaxSize); found++; if(found == maxIfs) break; } if(found == 0) { WARN("Net : No interface found in the same subnet as remote address %s", scclSocketToString(remoteAddr, line_a)); } freeifaddrs(interfaces); return found; } scclResult_t scclSocketGetAddrFromString(union scclSocketAddress* ua, const char* ip_port_pair) { if(!(ip_port_pair && strlen(ip_port_pair) > 1)) { WARN("Net : string is null"); return scclInvalidArgument; } bool ipv6 = ip_port_pair[0] == '['; /* Construct the sockaddress structure */ if(!ipv6) { struct netIf ni; // parse : string, expect one pair if(parseStringList(ip_port_pair, &ni, 1) != 1) { WARN("Net : No valid : pair found"); return scclInvalidArgument; } struct addrinfo hints, *p; int rv; memset(&hints, 0, sizeof(hints)); hints.ai_family = AF_UNSPEC; hints.ai_socktype = SOCK_STREAM; if((rv = getaddrinfo(ni.prefix, NULL, &hints, &p)) != 0) { WARN("Net : error encountered when getting address info : %s", gai_strerror(rv)); return scclInvalidArgument; } // use the first if(p->ai_family == AF_INET) { struct sockaddr_in& sin = ua->sin; memcpy(&sin, p->ai_addr, sizeof(struct sockaddr_in)); sin.sin_family = AF_INET; // IPv4 // inet_pton(AF_INET, ni.prefix, &(sin.sin_addr)); // IP address sin.sin_port = htons(ni.port); // port } else if(p->ai_family == AF_INET6) { struct sockaddr_in6& sin6 = ua->sin6; memcpy(&sin6, p->ai_addr, sizeof(struct sockaddr_in6)); sin6.sin6_family = AF_INET6; // IPv6 sin6.sin6_port = htons(ni.port); // port sin6.sin6_flowinfo = 0; // needed by IPv6, but possibly obsolete sin6.sin6_scope_id = 0; // should be global scope, set to 0 } else { WARN("Net : unsupported IP family"); return scclInvalidArgument; } freeaddrinfo(p); // all done with this structure } else { int i, j = -1, len = strlen(ip_port_pair); for(i = 1; i < len; i++) { if(ip_port_pair[i] == '%') j = i; if(ip_port_pair[i] == ']') break; } if(i == len) { WARN("Net : No valid [IPv6]:port pair found"); return scclInvalidArgument; } bool global_scope = (j == -1 ? true : false); // If no % found, global scope; otherwise, link scope char ip_str[NI_MAXHOST], port_str[NI_MAXSERV], if_name[IFNAMSIZ]; memset(ip_str, '\0', sizeof(ip_str)); memset(port_str, '\0', sizeof(port_str)); memset(if_name, '\0', sizeof(if_name)); strncpy(ip_str, ip_port_pair + 1, global_scope ? i - 1 : j - 1); strncpy(port_str, ip_port_pair + i + 2, len - i - 1); int port = atoi(port_str); if(!global_scope) strncpy(if_name, ip_port_pair + j + 1, i - j - 1); // If not global scope, we need the intf name struct sockaddr_in6& sin6 = ua->sin6; sin6.sin6_family = AF_INET6; // IPv6 inet_pton(AF_INET6, ip_str, &(sin6.sin6_addr)); // IP address sin6.sin6_port = htons(port); // port sin6.sin6_flowinfo = 0; // needed by IPv6, but possibly obsolete sin6.sin6_scope_id = global_scope ? 0 : if_nametoindex(if_name); // 0 if global scope; intf index if link scope } return scclSuccess; } int scclFindInterfaces(char* ifNames, union scclSocketAddress* ifAddrs, int ifNameMaxSize, int maxIfs) { static int shownIfName = 0; int nIfs = 0; // Allow user to force the INET socket family selection int sock_family = envSocketFamily(); // User specified interface char* env = getenv("SCCL_SOCKET_IFNAME"); if(env && strlen(env) > 1) { INFO(SCCL_LOG_CODEALL, "SCCL_SOCKET_IFNAME set by environment to %s", env); // Specified by user : find or fail if(shownIfName++ == 0) INFO(SCCL_LOG_CODEALL, "SCCL_SOCKET_IFNAME set to %s", env); nIfs = findInterfaces(env, ifNames, ifAddrs, sock_family, ifNameMaxSize, maxIfs); } else { // Try to automatically pick the right one // Start with IB nIfs = findInterfaces("ib", ifNames, ifAddrs, sock_family, ifNameMaxSize, maxIfs); // else see if we can get some hint from COMM ID if(nIfs == 0) { char* commId = getenv("SCCL_COMM_ID"); if(commId && strlen(commId) > 1) { INFO(SCCL_LOG_CODEALL, "SCCL_COMM_ID set by environment to %s", commId); // Try to find interface that is in the same subnet as the IP in comm id union scclSocketAddress idAddr; scclSocketGetAddrFromString(&idAddr, commId); nIfs = scclFindInterfaceMatchSubnet(ifNames, ifAddrs, &idAddr, ifNameMaxSize, maxIfs); } } // Then look for anything else (but not docker or lo) if(nIfs == 0) nIfs = findInterfaces("^docker,lo", ifNames, ifAddrs, sock_family, ifNameMaxSize, maxIfs); // Finally look for docker, then lo. if(nIfs == 0) nIfs = findInterfaces("docker", ifNames, ifAddrs, sock_family, ifNameMaxSize, maxIfs); if(nIfs == 0) nIfs = findInterfaces("lo", ifNames, ifAddrs, sock_family, ifNameMaxSize, maxIfs); } return nIfs; } scclResult_t scclSocketListen(struct scclSocket* sock) { if(sock == NULL) { WARN("scclSocketListen: pass NULL socket"); return scclInvalidArgument; } if(sock->fd == -1) { WARN("scclSocketListen: file descriptor is -1"); return scclInvalidArgument; } if(socketToPort(&sock->addr)) { // Port is forced by env. Make sure we get the port. int opt = 1; #if defined(SO_REUSEPORT) SYSCHECK(setsockopt(sock->fd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt)), "setsockopt"); #else SYSCHECK(setsockopt(sock->fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)), "setsockopt"); #endif } // addr port should be 0 (Any port) SYSCHECK(bind(sock->fd, &sock->addr.sa, sock->salen), "bind"); /* Get the assigned Port */ socklen_t size = sock->salen; SYSCHECK(getsockname(sock->fd, &sock->addr.sa, &size), "getsockname"); /* Put the socket in listen mode * NB: The backlog will be silently truncated to the value in /proc/sys/net/core/somaxconn */ SYSCHECK(listen(sock->fd, 16384), "listen"); sock->state = scclSocketStateReady; return scclSuccess; } scclResult_t scclSocketGetAddr(struct scclSocket* sock, union scclSocketAddress* addr) { if(sock == NULL) { WARN("scclSocketGetAddr: pass NULL socket"); return scclInvalidArgument; } if(sock->state != scclSocketStateReady) return scclInternalError; memcpy(addr, &sock->addr, sizeof(union scclSocketAddress)); return scclSuccess; } static scclResult_t socketTryAccept(struct scclSocket* sock) { socklen_t socklen = sizeof(union scclSocketAddress); sock->fd = accept(sock->acceptFd, &sock->addr.sa, &socklen); if(sock->fd != -1) { sock->state = scclSocketStateAccepted; } else if(errno != EAGAIN && errno != EWOULDBLOCK) { WARN("socketTryAccept: Accept failed: %s", strerror(errno)); return scclSystemError; } return scclSuccess; } static scclResult_t socketFinalizeAccept(struct scclSocket* sock) { uint64_t magic; enum scclSocketType type; int received = 0; const int one = 1; SYSCHECK(setsockopt(sock->fd, IPPROTO_TCP, TCP_NODELAY, (char*)&one, sizeof(int)), "setsockopt"); SCCLCHECK(scclSocketProgress(SCCL_SOCKET_RECV, sock, &magic, sizeof(magic), &received)); if(received == 0) return scclSuccess; SCCLCHECK(socketWait(SCCL_SOCKET_RECV, sock, &magic, sizeof(magic), &received)); if(magic != sock->magic) { WARN("socketFinalizeAccept: wrong magic %lx != %lx", magic, sock->magic); close(sock->fd); sock->fd = -1; // Ignore spurious connection and accept again sock->state = scclSocketStateAccepting; return scclSuccess; } else { received = 0; SCCLCHECK(socketWait(SCCL_SOCKET_RECV, sock, &type, sizeof(type), &received)); if(type != sock->type) { WARN("socketFinalizeAccept: wrong type %d != %d", type, sock->type); sock->state = scclSocketStateError; close(sock->fd); sock->fd = -1; return scclInternalError; } else { sock->state = scclSocketStateReady; } } return scclSuccess; } static scclResult_t socketStartConnect(struct scclSocket* sock) { /* blocking/non-blocking connect() is determined by asyncFlag. */ int ret = connect(sock->fd, &sock->addr.sa, sock->salen); if(ret == 0) { sock->state = scclSocketStateConnected; return scclSuccess; } else if(errno == EINPROGRESS) { sock->state = scclSocketStateConnectPolling; return scclSuccess; } else if(errno == ECONNREFUSED) { if(++sock->refusedRetries == RETRY_REFUSED_TIMES) { sock->state = scclSocketStateError; WARN("socketStartConnect: exceeded retries (%d)", sock->refusedRetries); return scclRemoteError; } usleep(SLEEP_INT); if(sock->refusedRetries % 1000 == 0) INFO(SCCL_LOG_CODEALL, "Call to connect returned %s, retrying", strerror(errno)); return scclSuccess; } else if(errno == ETIMEDOUT) { if(++sock->timedOutRetries == RETRY_TIMEDOUT_TIMES) { sock->state = scclSocketStateError; WARN("socketStartConnect: exceeded timeouts (%d)", sock->timedOutRetries); return scclRemoteError; } usleep(SLEEP_INT); return scclSuccess; } else { char line[SOCKET_NAME_MAXLEN + 1]; sock->state = scclSocketStateError; WARN("socketStartConnect: Connect to %s failed : %s", scclSocketToString(&sock->addr, line), strerror(errno)); return scclSystemError; } } static scclResult_t socketPollConnect(struct scclSocket* sock) { struct pollfd pfd; int timeout = 1, ret; socklen_t rlen = sizeof(int); memset(&pfd, 0, sizeof(struct pollfd)); pfd.fd = sock->fd; pfd.events = POLLOUT; ret = poll(&pfd, 1, timeout); if(ret == 0 || (ret < 0 && errno == EINTR)) { return scclSuccess; } else if(ret < 0) { WARN("socketPollConnect poll() failed with error %s", strerror(errno)); return scclRemoteError; } else { EQCHECK(ret == 1 && (pfd.revents & POLLOUT), 0); } /* check socket status */ SYSCHECK(getsockopt(sock->fd, SOL_SOCKET, SO_ERROR, (void*)&ret, &rlen), "getsockopt"); if(ret == 0) { sock->state = scclSocketStateConnected; } else if(ret == ECONNREFUSED) { if(++sock->refusedRetries == RETRY_REFUSED_TIMES) { sock->state = scclSocketStateError; WARN("socketPollConnect: exceeded retries (%d)", sock->refusedRetries); return scclRemoteError; } if(sock->refusedRetries % 1000 == 0) INFO(SCCL_LOG_CODEALL, "Call to connect returned %s, retrying", strerror(errno)); usleep(SLEEP_INT); sock->state = scclSocketStateConnecting; } else if(ret == ETIMEDOUT) { if(++sock->timedOutRetries == RETRY_TIMEDOUT_TIMES) { sock->state = scclSocketStateError; WARN("socketPollConnect: exceeded timeouts (%d)", sock->timedOutRetries); return scclRemoteError; } usleep(SLEEP_INT); sock->state = scclSocketStateConnecting; } else if(ret != EINPROGRESS) { sock->state = scclSocketStateError; return scclSystemError; } return scclSuccess; } scclResult_t scclSocketPollConnect(struct scclSocket* sock) { if(sock == NULL) { WARN("scclSocketPollConnect: pass NULL socket"); return scclInvalidArgument; } SCCLCHECK(socketPollConnect(sock)); return scclSuccess; } static scclResult_t socketFinalizeConnect(struct scclSocket* sock) { int sent = 0; SCCLCHECK(socketProgress(SCCL_SOCKET_SEND, sock, &sock->magic, sizeof(sock->magic), &sent)); if(sent == 0) return scclSuccess; SCCLCHECK(socketWait(SCCL_SOCKET_SEND, sock, &sock->magic, sizeof(sock->magic), &sent)); sent = 0; SCCLCHECK(socketWait(SCCL_SOCKET_SEND, sock, &sock->type, sizeof(sock->type), &sent)); sock->state = scclSocketStateReady; return scclSuccess; } static scclResult_t socketProgressState(struct scclSocket* sock) { if(sock->state == scclSocketStateAccepting) { SCCLCHECK(socketTryAccept(sock)); } if(sock->state == scclSocketStateAccepted) { SCCLCHECK(socketFinalizeAccept(sock)); } if(sock->state == scclSocketStateConnecting) { SCCLCHECK(socketStartConnect(sock)); } if(sock->state == scclSocketStateConnectPolling) { SCCLCHECK(socketPollConnect(sock)); } if(sock->state == scclSocketStateConnected) { SCCLCHECK(socketFinalizeConnect(sock)); } return scclSuccess; } scclResult_t scclSocketReady(struct scclSocket* sock, int* running) { if(sock == NULL) { *running = 0; return scclSuccess; } if(sock->state == scclSocketStateError || sock->state == scclSocketStateClosed) { WARN("scclSocketReady: unexpected socket state %d", sock->state); return scclRemoteError; } *running = (sock->state == scclSocketStateReady) ? 1 : 0; if(*running == 0) { SCCLCHECK(socketProgressState(sock)); *running = (sock->state == scclSocketStateReady) ? 1 : 0; } return scclSuccess; } scclResult_t scclSocketConnect(struct scclSocket* sock, int portReuse) { char line[SOCKET_NAME_MAXLEN + 1]; const int one = 1; if(sock == NULL) { WARN("scclSocketConnect: pass NULL socket"); return scclInvalidArgument; } if(sock->fd == -1) { WARN("scclSocketConnect: file descriptor is -1"); return scclInvalidArgument; } if(sock->state != scclSocketStateInitialized) { WARN("scclSocketConnect: wrong socket state %d", sock->state); if(sock->state == scclSocketStateError) return scclRemoteError; return scclInternalError; } 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)", scclSocketToString(&sock->addr, line), family, AF_INET, AF_INET6); return scclInternalError; } 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(scclSocketToString(&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 = scclSocketStateConnecting; do { SCCLCHECK(socketProgressState(sock)); } while(sock->asyncFlag == 0 && (sock->abortFlag == NULL || *sock->abortFlag == 0) && (sock->state == scclSocketStateConnecting || sock->state == scclSocketStateConnectPolling || sock->state == scclSocketStateConnected)); if(sock->abortFlag && *sock->abortFlag != 0) return scclInternalError; switch(sock->state) { case scclSocketStateConnecting: case scclSocketStateConnectPolling: case scclSocketStateConnected: case scclSocketStateReady: return scclSuccess; case scclSocketStateError: return scclSystemError; default: WARN("scclSocketConnect: wrong socket state %d", sock->state); return scclInternalError; } } scclResult_t scclSocketAccept(struct scclSocket* sock, struct scclSocket* listenSock) { scclResult_t ret = scclSuccess; if(listenSock == NULL || sock == NULL) { WARN("scclSocketAccept: pass NULL socket"); ret = scclInvalidArgument; goto exit; } if(listenSock->state != scclSocketStateReady) { WARN("scclSocketAccept: wrong socket state %d", listenSock->state); if(listenSock->state == scclSocketStateError) ret = scclSystemError; else ret = scclInternalError; goto exit; } if(sock->acceptFd == -1) { memcpy(sock, listenSock, sizeof(struct scclSocket)); sock->acceptFd = listenSock->fd; sock->state = scclSocketStateAccepting; } do { SCCLCHECKGOTO(socketProgressState(sock), ret, exit); } while(sock->asyncFlag == 0 && (sock->abortFlag == NULL || *sock->abortFlag == 0) && (sock->state == scclSocketStateAccepting || sock->state == scclSocketStateAccepted)); if(sock->abortFlag && *sock->abortFlag != 0) return scclInternalError; switch(sock->state) { case scclSocketStateAccepting: case scclSocketStateAccepted: case scclSocketStateReady: ret = scclSuccess; break; case scclSocketStateError: ret = scclSystemError; break; default: WARN("scclSocketAccept: wrong socket state %d", sock->state); ret = scclInternalError; break; } exit: return ret; } scclResult_t scclSocketInit(struct scclSocket* sock, union scclSocketAddress* addr, uint64_t magic, enum scclSocketType type, volatile uint32_t* abortFlag, int asyncFlag) { scclResult_t ret = scclSuccess; if(sock == NULL) goto exit; sock->timedOutRetries = 0; sock->refusedRetries = 0; sock->abortFlag = abortFlag; sock->asyncFlag = asyncFlag; sock->state = scclSocketStateInitialized; sock->magic = magic; sock->type = type; sock->fd = -1; sock->acceptFd = -1; if(addr) { /* IPv4/IPv6 support */ int family; memcpy(&sock->addr, addr, sizeof(union scclSocketAddress)); family = sock->addr.sa.sa_family; if(family != AF_INET && family != AF_INET6) { char line[SOCKET_NAME_MAXLEN + 1]; WARN("scclSocketInit: connecting to address %s with family %d is neither AF_INET(%d) nor AF_INET6(%d)", scclSocketToString(&sock->addr, line), family, AF_INET, AF_INET6); ret = scclInternalError; goto fail; } sock->salen = (family == AF_INET) ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6); /* Connect to a hostname / port */ sock->fd = socket(family, SOCK_STREAM, 0); if(sock->fd == -1) { WARN("scclSocketInit: Socket creation failed : %s", strerror(errno)); ret = scclSystemError; goto fail; } } else { memset(&sock->addr, 0, sizeof(union scclSocketAddress)); } /* 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); } exit: return ret; fail: goto exit; } scclResult_t scclSocketProgress(int op, struct scclSocket* sock, void* ptr, int size, int* offset) { if(sock == NULL) { WARN("scclSocketProgress: pass NULL socket"); return scclInvalidArgument; } SCCLCHECK(socketProgress(op, sock, ptr, size, offset)); return scclSuccess; } scclResult_t scclSocketWait(int op, struct scclSocket* sock, void* ptr, int size, int* offset) { if(sock == NULL) { WARN("scclSocketWait: pass NULL socket"); return scclInvalidArgument; } SCCLCHECK(socketWait(op, sock, ptr, size, offset)); return scclSuccess; } scclResult_t scclSocketSend(struct scclSocket* sock, void* ptr, int size) { int offset = 0; if(sock == NULL) { WARN("scclSocketSend: pass NULL socket"); return scclInvalidArgument; } if(sock->state != scclSocketStateReady) { WARN("scclSocketSend: socket state (%d) is not ready", sock->state); return scclInternalError; } SCCLCHECK(socketWait(SCCL_SOCKET_SEND, sock, ptr, size, &offset)); return scclSuccess; } scclResult_t scclSocketRecv(struct scclSocket* sock, void* ptr, int size) { int offset = 0; if(sock == NULL) { WARN("scclSocketRecv: pass NULL socket"); return scclInvalidArgument; } if(sock->state != scclSocketStateReady) { WARN("scclSocketRecv: socket state (%d) is not ready", sock->state); return scclInternalError; } SCCLCHECK(socketWait(SCCL_SOCKET_RECV, sock, ptr, size, &offset)); return scclSuccess; } // Receive or detect connection closed scclResult_t scclSocketTryRecv(struct scclSocket* sock, void* ptr, int size, int* closed, bool blocking) { int offset = 0; if(sock == NULL) { WARN("scclSocketTryRecv: pass NULL socket"); return scclInvalidArgument; } *closed = 0; // Block until connection closes or nbytes received if(blocking) { while(offset < size) { SCCLCHECK(socketProgressOpt(SCCL_SOCKET_RECV, sock, ptr, size, &offset, 0, closed)); if(*closed) return scclSuccess; } } else { SCCLCHECK(socketProgressOpt(SCCL_SOCKET_RECV, sock, ptr, size, &offset, 0, closed)); if(*closed) return scclSuccess; // If any bytes were received, block waiting for the rest if(offset > 0) { while(offset < size) { SCCLCHECK(socketProgressOpt(SCCL_SOCKET_RECV, sock, ptr, size, &offset, 0, closed)); if(*closed) return scclSuccess; } // No bytes were received, return scclInProgress } else { return scclInProgress; } } return scclSuccess; } scclResult_t scclSocketClose(struct scclSocket* sock) { if(sock != NULL) { if(sock->fd >= 0) { /* shutdown() is needed to send FIN packet to proxy thread; shutdown() is not affected * by refcount of fd, but close() is. close() won't close a fd and send FIN packet if * the fd is duplicated (e.g. fork()). So shutdown() guarantees the correct and graceful * connection close here. */ shutdown(sock->fd, SHUT_RDWR); close(sock->fd); } sock->state = scclSocketStateClosed; sock->fd = -1; } return scclSuccess; } scclResult_t scclSocketGetFd(struct scclSocket* sock, int* fd) { if(sock == NULL) { WARN("scclSocketGetFd: pass NULL socket"); return scclInvalidArgument; } if(fd) *fd = sock->fd; return scclSuccess; } scclResult_t scclSocketSetFd(int fd, struct scclSocket* sock) { if(sock == NULL) { WARN("scclSocketGetFd: pass NULL socket"); return scclInvalidArgument; } sock->fd = fd; return scclSuccess; }