Direct io for accept, send, sendmsg, recvfrom, and related methods.

Merged: https://github.com/ruby/ruby/pull/4477

ext/socket/basicsocket.c

@@ -566,7 +566,7 @@ rsock_bsock_send(int argc, VALUE *argv, VALUE sock)
     arg.flags = NUM2INT(flags);
     while (rsock_maybe_fd_writable(arg.fd),
 	   (n = (ssize_t)BLOCKING_REGION_FD(func, &arg)) < 0) {
-	if (rb_io_wait_writable(arg.fd)) {
+	if (rb_io_maybe_wait_writable(errno, sock, Qnil)) {
 	    continue;
 	}
 	rb_sys_fail(funcname);

ext/socket/init.c

@@ -166,7 +166,7 @@ recvfrom_locktmp(VALUE v)
 }
 
 VALUE
-rsock_s_recvfrom(VALUE sock, int argc, VALUE *argv, enum sock_recv_type from)
+rsock_s_recvfrom(VALUE socket, int argc, VALUE *argv, enum sock_recv_type from)
 {
     rb_io_t *fptr;
     VALUE str;
@@ -177,28 +177,36 @@ rsock_s_recvfrom(VALUE sock, int argc, VALUE *argv, enum sock_recv_type from)
 
     rb_scan_args(argc, argv, "12", &len, &flg, &str);
 
-    if (flg == Qnil) arg.flags = 0;
+    if (flg == Qnil)
-    else             arg.flags = NUM2INT(flg);
+        arg.flags = 0;
+    else
+        arg.flags = NUM2INT(flg);
+
     buflen = NUM2INT(len);
     str = rsock_strbuf(str, buflen);
 
-    GetOpenFile(sock, fptr);
+    RB_IO_POINTER(socket, fptr);
+
     if (rb_io_read_pending(fptr)) {
         rb_raise(rb_eIOError, "recv for buffered IO");
     }
+
     arg.fd = fptr->fd;
     arg.alen = (socklen_t)sizeof(arg.buf);
     arg.str = str;
     arg.length = buflen;
 
-    while (rb_io_check_closed(fptr),
+    while (true) {
-	   rsock_maybe_wait_fd(arg.fd),
+        rb_io_check_closed(fptr);
-	   (slen = (long)rb_str_locktmp_ensure(str, recvfrom_locktmp,
+        rsock_maybe_wait_fd(arg.fd);
-	                                       (VALUE)&arg)) < 0) {
+
-        if (!rb_io_wait_readable(fptr->fd)) {
+        slen = (long)rb_str_locktmp_ensure(str, recvfrom_locktmp, (VALUE)&arg);
+
+        if (slen >= 0) break;
+
+        if (!rb_io_maybe_wait_readable(errno, socket, Qnil))
             rb_sys_fail("recvfrom(2)");
     }
-    }
 
     /* Resize the string to the amount of data received */
     rb_str_set_len(str, slen);
@@ -221,7 +229,7 @@ rsock_s_recvfrom(VALUE sock, int argc, VALUE *argv, enum sock_recv_type from)
         return rb_assoc_new(str, rsock_unixaddr(&arg.buf.un, arg.alen));
 #endif
       case RECV_SOCKET:
-	return rb_assoc_new(str, rsock_io_socket_addrinfo(sock, &arg.buf.addr, arg.alen));
+	return rb_assoc_new(str, rsock_io_socket_addrinfo(socket, &arg.buf.addr, arg.alen));
       default:
 	rb_bug("rsock_s_recvfrom called with bad value");
     }
@@ -682,21 +690,26 @@ accept_blocking(void *data)
 }
 
 VALUE
-rsock_s_accept(VALUE klass, int fd, struct sockaddr *sockaddr, socklen_t *len)
+rsock_s_accept(VALUE klass, VALUE io, struct sockaddr *sockaddr, socklen_t *len)
 {
-    int fd2;
+    rb_io_t *fptr = NULL;
-    int retry = 0;
+    RB_IO_POINTER(io, fptr);
-    struct accept_arg arg;
+
+    struct accept_arg accept_arg = {
+      .fd = fptr->fd,
+      .sockaddr = sockaddr,
+      .len = len
+    };
+
+    int retry = 0;
 
-    arg.fd = fd;
-    arg.sockaddr = sockaddr;
-    arg.len = len;
   retry:
-    rsock_maybe_wait_fd(fd);
+    rsock_maybe_wait_fd(accept_arg.fd);
-    fd2 = (int)BLOCKING_REGION_FD(accept_blocking, &arg);
+    int peer = (int)BLOCKING_REGION_FD(accept_blocking, &accept_arg);
-    if (fd2 < 0) {
+    if (peer < 0) {
-	int e = errno;
+        int error = errno;
-	switch (e) {
+
+        switch (error) {
           case EMFILE:
           case ENFILE:
           case ENOMEM:
@@ -705,15 +718,19 @@ rsock_s_accept(VALUE klass, int fd, struct sockaddr *sockaddr, socklen_t *len)
             retry = 1;
             goto retry;
           default:
-	    if (!rb_io_wait_readable(fd)) break;
+            if (!rb_io_maybe_wait_readable(error, io, Qnil)) break;
             retry = 0;
             goto retry;
         }
-	rb_syserr_fail(e, "accept(2)");
+
+        rb_syserr_fail(error, "accept(2)");
     }
-    rb_update_max_fd(fd2);
+
-    if (!klass) return INT2NUM(fd2);
+    rb_update_max_fd(peer);
-    return rsock_init_sock(rb_obj_alloc(klass), fd2);
+
+    if (!klass) return INT2NUM(peer);
+
+    return rsock_init_sock(rb_obj_alloc(klass), peer);
 }
 
 int

ext/socket/rubysocket.h

@@ -373,7 +373,7 @@ VALUE rsock_s_recvfrom(VALUE sock, int argc, VALUE *argv, enum sock_recv_type fr
 
 int rsock_connect(int fd, const struct sockaddr *sockaddr, int len, int socks, struct timeval *timeout);
 
-VALUE rsock_s_accept(VALUE klass, int fd, struct sockaddr *sockaddr, socklen_t *len);
+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,
 			      struct sockaddr *sockaddr, socklen_t *len);
 VALUE rsock_sock_listen(VALUE sock, VALUE log);

ext/socket/socket.c

@@ -750,17 +750,14 @@ sock_recvfrom_nonblock(VALUE sock, VALUE len, VALUE flg, VALUE str, VALUE ex)
  *
  */
 static VALUE
-sock_accept(VALUE sock)
+sock_accept(VALUE server)
 {
-    rb_io_t *fptr;
+    union_sockaddr buffer;
-    VALUE sock2;
+    socklen_t length = (socklen_t)sizeof(buffer);
-    union_sockaddr buf;
-    socklen_t len = (socklen_t)sizeof buf;
 
-    GetOpenFile(sock, fptr);
+    VALUE peer = rsock_s_accept(rb_cSocket, server, &buffer.addr, &length);
-    sock2 = rsock_s_accept(rb_cSocket,fptr->fd,&buf.addr,&len);
 
-    return rb_assoc_new(sock2, rsock_io_socket_addrinfo(sock2, &buf.addr, len));
+    return rb_assoc_new(peer, rsock_io_socket_addrinfo(peer, &buffer.addr, length));
 }
 
 /* :nodoc: */
@@ -820,17 +817,14 @@ sock_accept_nonblock(VALUE sock, VALUE ex)
  * * Socket#accept
  */
 static VALUE
-sock_sysaccept(VALUE sock)
+sock_sysaccept(VALUE server)
 {
-    rb_io_t *fptr;
+    union_sockaddr buffer;
-    VALUE sock2;
+    socklen_t length = (socklen_t)sizeof(buffer);
-    union_sockaddr buf;
-    socklen_t len = (socklen_t)sizeof buf;
 
-    GetOpenFile(sock, fptr);
+    VALUE peer = rsock_s_accept(0, server, &buffer.addr, &length);
-    sock2 = rsock_s_accept(0,fptr->fd,&buf.addr,&len);
 
-    return rb_assoc_new(sock2, rsock_io_socket_addrinfo(sock2, &buf.addr, len));
+    return rb_assoc_new(peer, rsock_io_socket_addrinfo(peer, &buffer.addr, length));
 }
 
 #ifdef HAVE_GETHOSTNAME

ext/socket/tcpserver.c

@@ -53,15 +53,12 @@ tcp_svr_init(int argc, VALUE *argv, VALUE sock)
  *
  */
 static VALUE
-tcp_accept(VALUE sock)
+tcp_accept(VALUE server)
 {
-    rb_io_t *fptr;
+    union_sockaddr buffer;
-    union_sockaddr from;
+    socklen_t length = sizeof(buffer);
-    socklen_t fromlen;
 
-    GetOpenFile(sock, fptr);
+    return rsock_s_accept(rb_cTCPSocket, server, &buffer.addr, &length);
-    fromlen = (socklen_t)sizeof(from);
-    return rsock_s_accept(rb_cTCPSocket, fptr->fd, &from.addr, &fromlen);
 }
 
 /* :nodoc: */
@@ -91,15 +88,12 @@ tcp_accept_nonblock(VALUE sock, VALUE ex)
  *
  */
 static VALUE
-tcp_sysaccept(VALUE sock)
+tcp_sysaccept(VALUE server)
 {
-    rb_io_t *fptr;
+    union_sockaddr buffer;
-    union_sockaddr from;
+    socklen_t length = sizeof(buffer);
-    socklen_t fromlen;
 
-    GetOpenFile(sock, fptr);
+    return rsock_s_accept(0, server, &buffer.addr, &length);
-    fromlen = (socklen_t)sizeof(from);
-    return rsock_s_accept(0, fptr->fd, &from.addr, &fromlen);
 }
 
 void

ext/socket/unixserver.c

@@ -47,16 +47,12 @@ unix_svr_init(VALUE sock, VALUE path)
  *
  */
 static VALUE
-unix_accept(VALUE sock)
+unix_accept(VALUE server)
 {
-    rb_io_t *fptr;
+    struct sockaddr_un buffer;
-    struct sockaddr_un from;
+    socklen_t length = sizeof(buffer);
-    socklen_t fromlen;
 
-    GetOpenFile(sock, fptr);
+    return rsock_s_accept(rb_cUNIXSocket, server, (struct sockaddr*)&buffer, &length);
-    fromlen = (socklen_t)sizeof(struct sockaddr_un);
-    return rsock_s_accept(rb_cUNIXSocket, fptr->fd,
-		          (struct sockaddr*)&from, &fromlen);
 }
 
 /* :nodoc: */
@@ -92,15 +88,12 @@ unix_accept_nonblock(VALUE sock, VALUE ex)
  *
  */
 static VALUE
-unix_sysaccept(VALUE sock)
+unix_sysaccept(VALUE server)
 {
-    rb_io_t *fptr;
+    struct sockaddr_un buffer;
-    struct sockaddr_un from;
+    socklen_t length = sizeof(buffer);
-    socklen_t fromlen;
 
-    GetOpenFile(sock, fptr);
+    return rsock_s_accept(0, server, (struct sockaddr*)&buffer, &length);
-    fromlen = (socklen_t)sizeof(struct sockaddr_un);
-    return rsock_s_accept(0, fptr->fd, (struct sockaddr*)&from, &fromlen);
 }
 
 #endif

include/ruby/io.h

@@ -159,6 +159,9 @@ int rb_io_wait_writable(int fd);
 int rb_wait_for_single_fd(int fd, int events, struct timeval *tv);
 
 VALUE rb_io_wait(VALUE io, VALUE events, VALUE timeout);
+VALUE rb_io_maybe_wait(int error, VALUE io, VALUE events, VALUE timeout);
+int rb_io_maybe_wait_readable(int error, VALUE io, VALUE timeout);
+int rb_io_maybe_wait_writable(int error, VALUE io, VALUE timeout);
 
 /* compatibility for ruby 1.8 and older */
 #define rb_io_mode_flags(modestr) [<"rb_io_mode_flags() is obsolete; use rb_io_modestr_fmode()">]

io.c

@@ -1392,6 +1392,42 @@ rb_wait_for_single_fd(int fd, int events, struct timeval *timeout)
     return rb_thread_wait_for_single_fd(fd, events, timeout);
 }
 
+VALUE rb_io_maybe_wait(int error, VALUE io, VALUE events, VALUE timeout)
+{
+    switch (error) {
+      case EINTR:
+#if defined(ERESTART)
+      case ERESTART:
+#endif
+        // We might have pending interrupts since the previous syscall was interrupted:
+        rb_thread_check_ints();
+
+        // The operation was interrupted, so retry it immediately:
+        return events;
+
+      case EAGAIN:
+#if defined(EWOULDBLOCK) && EWOULDBLOCK != EAGAIN
+      case EWOULDBLOCK:
+#endif
+        // The operation would block, so wait for the specified events:
+        return rb_io_wait(io, events, timeout);
+
+      default:
+        // Non-specific error, no event is ready:
+        return RB_INT2NUM(0);
+    }
+}
+
+int rb_io_maybe_wait_readable(int error, VALUE io, VALUE timeout)
+{
+    return RB_NUM2INT(rb_io_maybe_wait(error, io, RB_INT2NUM(RUBY_IO_READABLE), timeout));
+}
+
+int rb_io_maybe_wait_writable(int error, VALUE io, VALUE timeout)
+{
+    return RB_NUM2INT(rb_io_maybe_wait(error, io, RB_INT2NUM(RUBY_IO_WRITABLE), timeout));
+}
+
 static void
 make_writeconv(rb_io_t *fptr)
 {