better handle cross-thread close under Ruby 1.9.3

We don't want to operate on improper FDs in multithreaded
environment.   This affects MRI despite the GVL since file
descriptors are usually allocated without the GVL held
(open()/accept()).

7 files changed, 31 insertions, 15 deletions

diff --git a/ext/sleepy_penguin/epoll.c b/ext/sleepy_penguin/epoll.c
index 3944d2f..225e77e 100644
--- a/ext/sleepy_penguin/epoll.c
+++ b/ext/sleepy_penguin/epoll.c
@@ -118,12 +118,18 @@ static void my_epoll_create(struct rb_epoll *ep)
         ep->flag_cache = rb_ary_new();
 }
 
+static int ep_fd_check(struct rb_epoll *ep)
+{
+        if (ep->fd == -1)
+                rb_raise(rb_eIOError, "closed epoll descriptor");
+        return 1;
+}
+
 static void ep_check(struct rb_epoll *ep)
 {
         if (ep->fd == EP_RECREATE)
                 my_epoll_create(ep);
-        if (ep->fd == -1)
-                rb_raise(rb_eIOError, "closed epoll descriptor");
+        ep_fd_check(ep);
         assert(TYPE(ep->marks) == T_ARRAY && "marks not initialized");
         assert(TYPE(ep->flag_cache) == T_ARRAY && "flag_cache not initialized");
 }
@@ -317,6 +323,7 @@ static int epoll_expired_p(uint64_t expire_at, struct rb_epoll *ep)
 {
         uint64_t now;
 
+        ep_fd_check(ep);
         if (ep->timeout < 0)
                 return 0;
         if (ep->timeout == 0)
@@ -344,7 +351,7 @@ static VALUE real_epwait(struct rb_epoll *ep)
         uint64_t expire_at = ep->timeout > 0 ? now_ms() + ep->timeout : 0;
 
         do {
-                n = (int)rb_sp_io_region(nogvl_wait, ep);
+                n = (int)rb_sp_fd_region(nogvl_wait, ep, ep->fd);
         } while (n == -1 && errno == EINTR && ! epoll_expired_p(expire_at, ep));
 
         return epwait_result(ep, n);
@@ -378,7 +385,7 @@ static int safe_epoll_wait(struct rb_epoll *ep)
                 TRAP_BEG;
                 n = epoll_wait(ep->fd, ep->events, ep->maxevents, 0);
                 TRAP_END;
-        } while (n == -1 && errno == EINTR);
+        } while (n == -1 && errno == EINTR && ep_fd_check(ep));
 
         return n;
 }
@@ -548,10 +555,10 @@ static VALUE epclose(VALUE self)
         }
 
         if (NIL_P(ep->io)) {
+                ep_fd_check(ep);
+
                 if (ep->fd == EP_RECREATE) {
-                        ep->fd = -1;
-                } else if (ep->fd == -1) {
-                        rb_raise(rb_eIOError, "closed epoll descriptor");
+                        ep->fd = -1; /* success */
                 } else {
                         int err;
                         int fd = ep->fd;
diff --git a/ext/sleepy_penguin/eventfd.c b/ext/sleepy_penguin/eventfd.c
index 4da5e45..9d75c91 100644
--- a/ext/sleepy_penguin/eventfd.c
+++ b/ext/sleepy_penguin/eventfd.c
@@ -87,7 +87,7 @@ static VALUE incr(int argc, VALUE *argv, VALUE self)
         RTEST(nonblock) ? rb_sp_set_nonblock(x.fd) : blocking_io_prepare(x.fd);
         x.val = (uint64_t)NUM2ULL(value);
 retry:
-        w = (ssize_t)rb_sp_io_region(efd_write, &x);
+        w = (ssize_t)rb_sp_fd_region(efd_write, &x, x.fd);
         if (w == -1) {
                 if (errno == EAGAIN && RTEST(nonblock))
                         return Qfalse;
@@ -123,11 +123,11 @@ static VALUE getvalue(int argc, VALUE *argv, VALUE self)
         x.fd = rb_sp_fileno(self);
         RTEST(nonblock) ? rb_sp_set_nonblock(x.fd) : blocking_io_prepare(x.fd);
 retry:
-        w = (ssize_t)rb_sp_io_region(efd_read, &x);
+        w = (ssize_t)rb_sp_fd_region(efd_read, &x, x.fd);
         if (w == -1) {
                 if (errno == EAGAIN && RTEST(nonblock))
                         return Qnil;
-                if (rb_io_wait_readable(x.fd))
+                if (rb_io_wait_readable(x.fd = rb_sp_fileno(self)))
                         goto retry;
                 rb_sys_fail("read(eventfd)");
         }
diff --git a/ext/sleepy_penguin/extconf.rb b/ext/sleepy_penguin/extconf.rb
index e089562..ba1db75 100644
--- a/ext/sleepy_penguin/extconf.rb
+++ b/ext/sleepy_penguin/extconf.rb
@@ -9,6 +9,7 @@ have_header('sys/signalfd.h')
 have_header('ruby/io.h') and have_struct_member('rb_io_t', 'fd', 'ruby/io.h')
 have_func('epoll_create1', %w(sys/epoll.h))
 have_func('rb_thread_blocking_region')
+have_func('rb_thread_io_blocking_region')
 have_func('rb_thread_fd_close')
 have_library('pthread')
 create_makefile('sleepy_penguin_ext')
diff --git a/ext/sleepy_penguin/inotify.c b/ext/sleepy_penguin/inotify.c
index 406d6a3..738c02f 100644
--- a/ext/sleepy_penguin/inotify.c
+++ b/ext/sleepy_penguin/inotify.c
@@ -181,7 +181,7 @@ static VALUE take(int argc, VALUE *argv, VALUE self)
         else
                 blocking_io_prepare(args.fd);
         do {
-                r = rb_sp_io_region(inread, &args);
+                r = rb_sp_fd_region(inread, &args, args.fd);
                 if (r == 0 /* Linux < 2.6.21 */
                     ||
                     (r < 0 && errno == EINVAL) /* Linux >= 2.6.21 */
@@ -199,6 +199,7 @@ static VALUE take(int argc, VALUE *argv, VALUE self)
                         if (errno == EAGAIN && RTEST(nonblock)) {
                                 return Qnil;
                         } else {
+                                args.fd = rb_sp_fileno(self);
                                 if (!rb_io_wait_readable(args.fd))
                                         rb_sys_fail("read(inotify)");
                         }
diff --git a/ext/sleepy_penguin/signalfd.c b/ext/sleepy_penguin/signalfd.c
index cdfcd3f..81bb635 100644
--- a/ext/sleepy_penguin/signalfd.c
+++ b/ext/sleepy_penguin/signalfd.c
@@ -195,11 +195,11 @@ static VALUE sfd_take(int argc, VALUE *argv, VALUE self)
                 blocking_io_prepare(fd);
 retry:
         ssi->ssi_fd = fd;
-        r = (ssize_t)rb_sp_io_region(sfd_read, ssi);
+        r = (ssize_t)rb_sp_fd_region(sfd_read, ssi, fd);
         if (r == -1) {
                 if (errno == EAGAIN && RTEST(nonblock))
                         return Qnil;
-                if (rb_io_wait_readable(fd))
+                if (rb_io_wait_readable(fd = rb_sp_fileno(self)))
                         goto retry;
                 rb_sys_fail("read(signalfd)");
         }
diff --git a/ext/sleepy_penguin/sleepy_penguin.h b/ext/sleepy_penguin/sleepy_penguin.h
index 577b7c0..c8adcef 100644
--- a/ext/sleepy_penguin/sleepy_penguin.h
+++ b/ext/sleepy_penguin/sleepy_penguin.h
@@ -30,6 +30,13 @@ VALUE rb_sp_io_region(rb_blocking_function_t *func, void *data);
 #  define blocking_io_prepare(fd) rb_sp_set_nonblock((fd))
 #endif
 
+#ifdef HAVE_RB_THREAD_IO_BLOCKING_REGION
+#  define rb_sp_fd_region(fn,data,fd) \
+          rb_thread_io_blocking_region((fn),(data),(fd))
+#else
+#  define rb_sp_fd_region(fn,data,fd) rb_sp_io_region(fn,data)
+#endif
+
 #define NODOC_CONST(klass,name,value) \
   rb_define_const((klass),(name),(value))
 #endif /* SLEEPY_PENGUIN_H */
diff --git a/ext/sleepy_penguin/timerfd.c b/ext/sleepy_penguin/timerfd.c
index c7378c3..07ac4c2 100644
--- a/ext/sleepy_penguin/timerfd.c
+++ b/ext/sleepy_penguin/timerfd.c
@@ -125,11 +125,11 @@ static VALUE expirations(int argc, VALUE *argv, VALUE self)
         else
                 blocking_io_prepare(fd);
 retry:
-        r = (ssize_t)rb_sp_io_region(tfd_read, &buf);
+        r = (ssize_t)rb_sp_fd_region(tfd_read, &buf, fd);
         if (r == -1) {
                 if (errno == EAGAIN && RTEST(nonblock))
                         return Qnil;
-                if (rb_io_wait_readable(fd))
+                if (rb_io_wait_readable(fd = rb_sp_fileno(self)))
                         goto retry;
                 rb_sys_fail("read(timerfd)");
         }