Update `rsock_connect` to take `VALUE io` argument. (#11847)

ext/socket/init.c

@@ -578,19 +578,19 @@ socks_connect_blocking(void *data)
 #endif
 
 int
-rsock_connect(int fd, const struct sockaddr *sockaddr, int len, int socks, struct timeval *timeout)
+rsock_connect(VALUE self, const struct sockaddr *sockaddr, int len, int socks, struct timeval *timeout)
 {
-    int status;
+    int descriptor = rb_io_descriptor(self);
     rb_blocking_function_t *func = connect_blocking;
-    struct connect_arg arg;
+    struct connect_arg arg = {.fd = descriptor, .sockaddr = sockaddr, .len = len};
+
+    rb_io_t *fptr;
+    RB_IO_POINTER(self, fptr);
 
-    arg.fd = fd;
-    arg.sockaddr = sockaddr;
-    arg.len = len;
 #if defined(SOCKS) && !defined(SOCKS5)
     if (socks) func = socks_connect_blocking;
 #endif
-    status = (int)BLOCKING_REGION_FD(func, &arg);
+    int status = (int)rb_io_blocking_region(fptr, func, &arg);
 
     if (status < 0) {
         switch (errno) {
@@ -602,7 +602,7 @@ rsock_connect(int fd, const struct sockaddr *sockaddr, int len, int socks, struc
 #ifdef EINPROGRESS
           case EINPROGRESS:
 #endif
-            return wait_connectable(fd, timeout);
+            return wait_connectable(descriptor, timeout);
         }
     }
     return status;

ext/socket/ipsocket.c

@@ -9,16 +9,18 @@
 ************************************************/
 
 #include "rubysocket.h"
+#include <stdio.h>
 
 struct inetsock_arg
 {
-    VALUE sock;
+    VALUE self;
+    VALUE io;
+
     struct {
         VALUE host, serv;
         struct rb_addrinfo *res;
     } remote, local;
     int type;
-    int fd;
     VALUE resolv_timeout;
     VALUE connect_timeout;
 };
@@ -35,8 +37,9 @@ inetsock_cleanup(VALUE v)
         rb_freeaddrinfo(arg->local.res);
         arg->local.res = 0;
     }
-    if (arg->fd >= 0) {
+    if (arg->io != Qnil) {
-        close(arg->fd);
+        rb_io_close(arg->io);
+        arg->io = Qnil;
     }
     return Qnil;
 }
@@ -48,7 +51,7 @@ init_inetsock_internal(VALUE v)
     int error = 0;
     int type = arg->type;
     struct addrinfo *res, *lres;
-    int fd, status = 0, local = 0;
+    int status = 0, local = 0;
     int family = AF_UNSPEC;
     const char *syscall = 0;
     VALUE connect_timeout = arg->connect_timeout;
@@ -74,7 +77,8 @@ init_inetsock_internal(VALUE v)
                                         family, SOCK_STREAM, 0);
     }
 
-    arg->fd = fd = -1;
+    VALUE io = Qnil;
+
     for (res = arg->remote.res->ai; res; res = res->ai_next) {
 #if !defined(INET6) && defined(AF_INET6)
         if (res->ai_family == AF_INET6)
@@ -96,12 +100,14 @@ init_inetsock_internal(VALUE v)
         }
         status = rsock_socket(res->ai_family,res->ai_socktype,res->ai_protocol);
         syscall = "socket(2)";
-        fd = status;
+        if (status < 0) {
-        if (fd < 0) {
             error = errno;
             continue;
         }
-        arg->fd = fd;
+
+        int fd = status;
+        io = arg->io = rsock_init_sock(arg->self, fd);
+
         if (type == INET_SERVER) {
 #if !defined(_WIN32) && !defined(__CYGWIN__)
             status = 1;
@@ -124,20 +130,22 @@ init_inetsock_internal(VALUE v)
             }
 
             if (status >= 0) {
-                status = rsock_connect(fd, res->ai_addr, res->ai_addrlen,
+                status = rsock_connect(io, res->ai_addr, res->ai_addrlen, (type == INET_SOCKS), tv);
-                                       (type == INET_SOCKS), tv);
                 syscall = "connect(2)";
             }
         }
 
         if (status < 0) {
             error = errno;
-            close(fd);
+            arg->io = Qnil;
-            arg->fd = fd = -1;
+            rb_io_close(io);
+            io = Qnil;
             continue;
-        } else
+        } else {
             break;
+        }
     }
+
     if (status < 0) {
         VALUE host, port;
 
@@ -152,28 +160,28 @@ init_inetsock_internal(VALUE v)
         rsock_syserr_fail_host_port(error, syscall, host, port);
     }
 
-    arg->fd = -1;
+    // Don't close the socket in `inetsock_cleanup` if we are returning it:
+    arg->io = Qnil;
 
-    if (type == INET_SERVER) {
+    if (type == INET_SERVER && io != Qnil) {
-        status = listen(fd, SOMAXCONN);
+        status = listen(rb_io_descriptor(io), SOMAXCONN);
         if (status < 0) {
             error = errno;
-            close(fd);
+            rb_io_close(io);
             rb_syserr_fail(error, "listen(2)");
         }
     }
 
     /* create new instance */
-    return rsock_init_sock(arg->sock, fd);
+    return io;
 }
 
 VALUE
-rsock_init_inetsock(VALUE sock, VALUE remote_host, VALUE remote_serv,
+rsock_init_inetsock(VALUE self, VALUE remote_host, VALUE remote_serv, VALUE local_host, VALUE local_serv, int type, VALUE resolv_timeout, VALUE connect_timeout)
-                    VALUE local_host, VALUE local_serv, int type,
-                    VALUE resolv_timeout, VALUE connect_timeout)
 {
     struct inetsock_arg arg;
-    arg.sock = sock;
+    arg.self = self;
+    arg.io = Qnil;
     arg.remote.host = remote_host;
     arg.remote.serv = remote_serv;
     arg.remote.res = 0;
@@ -181,7 +189,6 @@ rsock_init_inetsock(VALUE sock, VALUE remote_host, VALUE remote_serv,
     arg.local.serv = local_serv;
     arg.local.res = 0;
     arg.type = type;
-    arg.fd = -1;
     arg.resolv_timeout = resolv_timeout;
     arg.connect_timeout = connect_timeout;
     return rb_ensure(init_inetsock_internal, (VALUE)&arg,

ext/socket/rubysocket.h

@@ -298,8 +298,6 @@ int Rconnect();
 
 #include "constdefs.h"
 
-#define BLOCKING_REGION_FD(func, arg) (long)rb_thread_io_blocking_region((func), (arg), (arg)->fd)
-
 #define SockAddrStringValue(v) rsock_sockaddr_string_value(&(v))
 #define SockAddrStringValuePtr(v) rsock_sockaddr_string_value_ptr(&(v))
 #define SockAddrStringValueWithAddrinfo(v, rai_ret) rsock_sockaddr_string_value_with_addrinfo(&(v), &(rai_ret))
@@ -381,7 +379,7 @@ VALUE rsock_s_recvfrom_nonblock(VALUE sock, VALUE len, VALUE flg, VALUE str,
                                 VALUE ex, enum sock_recv_type from);
 VALUE rsock_s_recvfrom(VALUE sock, int argc, VALUE *argv, enum sock_recv_type from);
 
-int rsock_connect(int fd, const struct sockaddr *sockaddr, int len, int socks, struct timeval *timeout);
+int rsock_connect(VALUE self, const struct sockaddr *sockaddr, int len, int socks, struct timeval *timeout);
 
 VALUE rsock_s_accept(VALUE klass, VALUE io, struct sockaddr *sockaddr, socklen_t *len);
 VALUE rsock_s_accept_nonblock(VALUE klass, VALUE ex, rb_io_t *fptr,

ext/socket/socket.c

@@ -391,18 +391,17 @@ sock_connect(VALUE sock, VALUE addr)
 {
     VALUE rai;
     rb_io_t *fptr;
-    int fd, n;
 
     SockAddrStringValueWithAddrinfo(addr, rai);
     addr = rb_str_new4(addr);
     GetOpenFile(sock, fptr);
-    fd = fptr->fd;
+
-    n = rsock_connect(fd, (struct sockaddr*)RSTRING_PTR(addr), RSTRING_SOCKLEN(addr), 0, NULL);
+    int result = rsock_connect(sock, (struct sockaddr*)RSTRING_PTR(addr), RSTRING_SOCKLEN(addr), 0, NULL);
-    if (n < 0) {
+    if (result < 0) {
         rsock_sys_fail_raddrinfo_or_sockaddr("connect(2)", addr, rai);
     }
 
-    return INT2FIX(n);
+    return INT2FIX(result);
 }
 
 /* :nodoc: */

ext/socket/udpsocket.c

@@ -45,22 +45,18 @@ udp_init(int argc, VALUE *argv, VALUE sock)
 
 struct udp_arg
 {
+    VALUE io;
     struct rb_addrinfo *res;
-    rb_io_t *fptr;
 };
 
 static VALUE
 udp_connect_internal(VALUE v)
 {
     struct udp_arg *arg = (void *)v;
-    rb_io_t *fptr;
-    int fd;
     struct addrinfo *res;
 
-    rb_io_check_closed(fptr = arg->fptr);
-    fd = fptr->fd;
     for (res = arg->res->ai; res; res = res->ai_next) {
-        if (rsock_connect(fd, res->ai_addr, res->ai_addrlen, 0, NULL) >= 0) {
+        if (rsock_connect(arg->io, res->ai_addr, res->ai_addrlen, 0, NULL) >= 0) {
             return Qtrue;
         }
     }
@@ -84,16 +80,17 @@ udp_connect_internal(VALUE v)
  *
  */
 static VALUE
-udp_connect(VALUE sock, VALUE host, VALUE port)
+udp_connect(VALUE self, VALUE host, VALUE port)
 {
-    struct udp_arg arg;
+    struct udp_arg arg = {.io = self};
-    VALUE ret;
+
+    arg.res = rsock_addrinfo(host, port, rsock_fd_family(rb_io_descriptor(self)), SOCK_DGRAM, 0);
+
+    int result = (int)rb_ensure(udp_connect_internal, (VALUE)&arg, rsock_freeaddrinfo, (VALUE)arg.res);
+    if (!result) {
+        rsock_sys_fail_host_port("connect(2)", host, port);
+    }
 
-    GetOpenFile(sock, arg.fptr);
-    arg.res = rsock_addrinfo(host, port, rsock_fd_family(arg.fptr->fd), SOCK_DGRAM, 0);
-    ret = rb_ensure(udp_connect_internal, (VALUE)&arg,
-                    rsock_freeaddrinfo, (VALUE)arg.res);
-    if (!ret) rsock_sys_fail_host_port("connect(2)", host, port);
     return INT2FIX(0);
 }
 
@@ -101,14 +98,13 @@ static VALUE
 udp_bind_internal(VALUE v)
 {
     struct udp_arg *arg = (void *)v;
-    rb_io_t *fptr;
-    int fd;
     struct addrinfo *res;
 
-    rb_io_check_closed(fptr = arg->fptr);
+    rb_io_t *fptr;
-    fd = fptr->fd;
+    RB_IO_POINTER(arg->io, fptr);
+
     for (res = arg->res->ai; res; res = res->ai_next) {
-        if (bind(fd, res->ai_addr, res->ai_addrlen) < 0) {
+        if (bind(fptr->fd, res->ai_addr, res->ai_addrlen) < 0) {
             continue;
         }
         return Qtrue;
@@ -129,22 +125,23 @@ udp_bind_internal(VALUE v)
  *
  */
 static VALUE
-udp_bind(VALUE sock, VALUE host, VALUE port)
+udp_bind(VALUE self, VALUE host, VALUE port)
 {
-    struct udp_arg arg;
+    struct udp_arg arg = {.io = self};
-    VALUE ret;
+
+    arg.res = rsock_addrinfo(host, port, rsock_fd_family(rb_io_descriptor(self)), SOCK_DGRAM, 0);
+
+    int result = rb_ensure(udp_bind_internal, (VALUE)&arg, rsock_freeaddrinfo, (VALUE)arg.res);
+    if (!result) {
+        rsock_sys_fail_host_port("bind(2)", host, port);
+    }
 
-    GetOpenFile(sock, arg.fptr);
-    arg.res = rsock_addrinfo(host, port, rsock_fd_family(arg.fptr->fd), SOCK_DGRAM, 0);
-    ret = rb_ensure(udp_bind_internal, (VALUE)&arg,
-                    rsock_freeaddrinfo, (VALUE)arg.res);
-    if (!ret) rsock_sys_fail_host_port("bind(2)", host, port);
     return INT2FIX(0);
 }
 
 struct udp_send_arg {
-    struct rb_addrinfo *res;
     rb_io_t *fptr;
+    struct rb_addrinfo *res;
     struct rsock_send_arg sarg;
 };
 
@@ -156,7 +153,6 @@ udp_send_internal(VALUE v)
     struct addrinfo *res;
 
     rb_io_check_closed(fptr = arg->fptr);
-
     for (res = arg->res->ai; res; res = res->ai_next) {
       retry:
         arg->sarg.fd = fptr->fd;

ext/socket/unixsocket.c

@@ -14,15 +14,14 @@
 struct unixsock_arg {
     struct sockaddr_un *sockaddr;
     socklen_t sockaddrlen;
-    int fd;
+    VALUE io;
 };
 
 static VALUE
 unixsock_connect_internal(VALUE a)
 {
     struct unixsock_arg *arg = (struct unixsock_arg *)a;
-    return (VALUE)rsock_connect(arg->fd, (struct sockaddr*)arg->sockaddr,
+    return (VALUE)rsock_connect(arg->io, (struct sockaddr*)arg->sockaddr, arg->sockaddrlen, 0, NULL);
-                                arg->sockaddrlen, 0, NULL);
 }
 
 static VALUE
@@ -51,7 +50,7 @@ unixsock_path_value(VALUE path)
 }
 
 VALUE
-rsock_init_unixsock(VALUE sock, VALUE path, int server)
+rsock_init_unixsock(VALUE self, VALUE path, int server)
 {
     struct sockaddr_un sockaddr;
     socklen_t sockaddrlen;
@@ -73,43 +72,46 @@ rsock_init_unixsock(VALUE sock, VALUE path, int server)
         rsock_sys_fail_path("socket(2)", path);
     }
 
+    VALUE io = rsock_init_sock(self, fd);
+    RB_IO_POINTER(io, fptr);
+
     if (server) {
         status = bind(fd, (struct sockaddr*)&sockaddr, sockaddrlen);
     }
     else {
-        int prot;
+        int error_tag;
         struct unixsock_arg arg;
         arg.sockaddr = &sockaddr;
         arg.sockaddrlen = sockaddrlen;
-        arg.fd = fd;
+        arg.io = io;
-        status = (int)rb_protect(unixsock_connect_internal, (VALUE)&arg, &prot);
+
-        if (prot) {
+        status = (int)rb_protect(unixsock_connect_internal, (VALUE)&arg, &error_tag);
-            close(fd);
+
-            rb_jump_tag(prot);
+        if (error_tag) {
+            rb_io_close(io);
+            rb_jump_tag(error_tag);
         }
     }
 
     if (status < 0) {
         int e = errno;
-        close(fd);
+        rb_io_close(io);
         rsock_syserr_fail_path(e, "connect(2)", path);
     }
 
     if (server) {
         if (listen(fd, SOMAXCONN) < 0) {
             int e = errno;
-            close(fd);
+            rb_io_close(io);
             rsock_syserr_fail_path(e, "listen(2)", path);
         }
     }
 
-    rsock_init_sock(sock, fd);
     if (server) {
-        GetOpenFile(sock, fptr);
         fptr->pathv = rb_str_new_frozen(path);
     }
 
-    return sock;
+    return io;
 }
 
 /*
@@ -125,9 +127,9 @@ rsock_init_unixsock(VALUE sock, VALUE path, int server)
  *
  */
 static VALUE
-unix_init(VALUE sock, VALUE path)
+unix_init(VALUE self, VALUE path)
 {
-    return rsock_init_unixsock(sock, path, 0);
+    return rsock_init_unixsock(self, path, 0);
 }
 
 /*