use flag cache and marks array to optimize away syscalls

We can have simplify user code by caching flags and using
the GC mark array to avoid syscalls and unnecessary
errors/exceptions.

6 files changed, 204 insertions, 15 deletions

diff --git a/ext/sleepy_penguin/epoll.c b/ext/sleepy_penguin/epoll.c
index b8cfa3a..2361e18 100644
--- a/ext/sleepy_penguin/epoll.c
+++ b/ext/sleepy_penguin/epoll.c
@@ -51,6 +51,7 @@ struct rb_epoll {
         struct epoll_event *events;
         VALUE io;
         VALUE marks;
+        VALUE flag_cache;
         int flags;
 };
 
@@ -95,6 +96,7 @@ static void gcmark(void *ptr)
 
         rb_gc_mark(ep->io);
         rb_gc_mark(ep->marks);
+        rb_gc_mark(ep->flag_cache);
 }
 
 static void gcfree(void *ptr)
@@ -125,6 +127,7 @@ static VALUE alloc(VALUE klass)
         ep->fd = -1;
         ep->io = Qnil;
         ep->marks = Qnil;
+        ep->flag_cache = Qnil;
         ep->capa = step;
         ep->flags = EPOLL_CLOEXEC;
         ep->events = xmalloc(sizeof(struct epoll_event) * ep->capa);
@@ -146,6 +149,7 @@ static void my_epoll_create(struct rb_epoll *ep)
         }
         st_insert(active, (st_data_t)ep->fd, (st_data_t)ep);
         ep->marks = rb_ary_new();
+        ep->flag_cache = rb_ary_new();
 }
 
 static void ep_check(struct rb_epoll *ep)
@@ -154,6 +158,8 @@ static void ep_check(struct rb_epoll *ep)
                 my_epoll_create(ep);
         if (ep->fd == -1)
                 rb_raise(rb_eIOError, "closed");
+        assert(TYPE(ep->marks) == T_ARRAY && "marks not initialized");
+        assert(TYPE(ep->flag_cache) == T_ARRAY && "flag_cache not initialized");
 }
 
 /*
@@ -205,8 +211,16 @@ static VALUE ctl(VALUE self, VALUE io, VALUE flags, int op)
                 if (rv == -1)
                         rb_sys_fail("epoll_ctl");
         }
-        if (op == EPOLL_CTL_ADD)
+        switch (op) {
+        case EPOLL_CTL_ADD:
                 rb_ary_store(ep->marks, fd, io);
+                /* fall-through */
+        case EPOLL_CTL_MOD:
+                rb_ary_store(ep->flag_cache, fd, flags);
+                break;
+        case EPOLL_CTL_DEL:
+                rb_ary_store(ep->marks, fd, Qnil);
+        }
 
         return INT2NUM(rv);
 }
@@ -221,23 +235,44 @@ static VALUE set(VALUE self, VALUE io, VALUE flags)
         struct rb_epoll *ep = ep_get(self);
         int fd = my_fileno(io);
         int rv;
+        VALUE cur_io = rb_ary_entry(ep->marks, fd);
 
         ep_check(ep);
         event.events = NUM2UINT(flags);
         pack_event_data(&event, io);
 
-        rv = epoll_ctl(ep->fd, EPOLL_CTL_MOD, fd, &event);
-        if (rv == -1) {
-                if (errno == ENOENT) {
-                        rv = epoll_ctl(ep->fd, EPOLL_CTL_ADD, fd, &event);
-                        if (rv == -1)
-                                rb_sys_fail("epoll_ctl - add");
+        if (cur_io == io) {
+                VALUE cur_flags = rb_ary_entry(ep->flag_cache, fd);
+                uint32_t cur_events;
+
+                assert(!NIL_P(cur_flags) && "cur_flags nil but cur_io is not");
+                cur_events = NUM2UINT(cur_flags);
+
+                if (!(cur_events & EPOLLONESHOT) && cur_events == event.events)
+                        return Qnil;
 
-                        rb_ary_store(ep->marks, fd, io);
-                        return INT2NUM(rv);
+fallback_mod:
+                rv = epoll_ctl(ep->fd, EPOLL_CTL_MOD, fd, &event);
+                if (rv == -1) {
+                        if (errno != ENOENT)
+                                rb_sys_fail("epoll_ctl - mod");
+                        errno = 0;
+                        rb_warn("epoll flag_cache failed (mod -> add)");
+                        goto fallback_add;
                 }
-                rb_sys_fail("epoll_ctl - mod");
+        } else {
+fallback_add:
+                rv = epoll_ctl(ep->fd, EPOLL_CTL_ADD, fd, &event);
+                if (rv == -1) {
+                        if (errno != EEXIST)
+                                rb_sys_fail("epoll_ctl - add");
+                        errno = 0;
+                        rb_warn("epoll flag_cache failed (add -> mod)");
+                        goto fallback_mod;
+                }
+                rb_ary_store(ep->marks, fd, io);
         }
+        rb_ary_store(ep->flag_cache, fd, flags);
 
         return INT2NUM(rv);
 }
@@ -253,16 +288,26 @@ static VALUE delete(VALUE self, VALUE io)
         struct rb_epoll *ep = ep_get(self);
         int fd = my_fileno(io);
         int rv;
+        VALUE cur_io;
 
         ep_check(ep);
+        cur_io = rb_ary_entry(ep->marks, fd);
+        if (NIL_P(cur_io) || my_io_closed(cur_io))
+                return Qnil;
+
         rv = epoll_ctl(ep->fd, EPOLL_CTL_DEL, fd, NULL);
         if (rv == -1) {
-                if (errno != ENOENT)
+                /* beware of IO.for_fd-created descriptors */
+                if (errno == ENOENT || errno == EBADF) {
+                        errno = 0;
+                        io = Qnil;
+                } else {
                         rb_sys_fail("epoll_ctl - del");
-                errno = 0;
-                return Qnil;
+                }
         }
-        return INT2NUM(rv);
+        rb_ary_store(ep->marks, fd, Qnil);
+
+        return io;
 }
 
 static VALUE epwait_result(struct rb_epoll *ep, int n)
@@ -523,8 +568,11 @@ static VALUE init_copy(VALUE copy, VALUE orig)
 
         ep_check(a);
         assert(NIL_P(b->marks) && "mark array not nil");
+        assert(NIL_P(b->flag_cache) && "flag_cache not nil");
         b->marks = a->marks;
+        b->flag_cache = a->flag_cache;
         assert(TYPE(b->marks) == T_ARRAY && "mark array not initialized");
+        assert(TYPE(b->flag_cache) == T_ARRAY && "flag_cache not initialized");
         b->flags = a->flags;
         b->fd = cloexec_dup(a);
         if (b->fd == -1) {
diff --git a/ext/sleepy_penguin/extconf.rb b/ext/sleepy_penguin/extconf.rb
index d753498..50eb9a7 100644
--- a/ext/sleepy_penguin/extconf.rb
+++ b/ext/sleepy_penguin/extconf.rb
@@ -4,6 +4,7 @@ have_header("pthread.h") or abort 'pthread.h not found'
 have_header('sys/eventfd.h')
 have_header('sys/signalfd.h')
 have_header('sys/timerfd.h')
+have_header('ruby/io.h') and have_struct_member('rb_io_t', 'fd', 'ruby/io.h')
 have_func('rb_memerror')
 have_func('rb_io_close')
 have_func('epoll_create1', %w(sys/epoll.h))
diff --git a/ext/sleepy_penguin/sleepy_penguin.h b/ext/sleepy_penguin/sleepy_penguin.h
index 6e43113..c4461b9 100644
--- a/ext/sleepy_penguin/sleepy_penguin.h
+++ b/ext/sleepy_penguin/sleepy_penguin.h
@@ -26,6 +26,18 @@
 #  endif
 #endif
 
+#if defined(RFILE) && defined(HAVE_ST_FD)
+static int my_io_closed(VALUE io)
+{
+        return RFILE(io)->fptr->fd < 0;
+}
+#else
+static int my_io_closed(VALUE io)
+{
+        return rb_funcall(io, rb_intern("closed?"), 0) == Qtrue;
+}
+#endif
+
 static int my_fileno(VALUE io)
 {
         rb_io_t *fptr;
diff --git a/sleepy_penguin.gemspec b/sleepy_penguin.gemspec
index 26fcbda..17fecf4 100644
--- a/sleepy_penguin.gemspec
+++ b/sleepy_penguin.gemspec
@@ -21,6 +21,7 @@ Gem::Specification.new do |s|
   s.test_files = Dir['test/test_*.rb']
   s.extensions = %w(ext/sleepy_penguin/extconf.rb)
   s.add_development_dependency('wrongdoc', '~> 1.3')
+  s.add_development_dependency('strace_me', '~> 1.0')
 
   # s.license = %w(LGPL) # disabled for compatibility with older RubyGems
 end
diff --git a/test/test_epoll.rb b/test/test_epoll.rb
index a541a2c..b73cb02 100644
--- a/test/test_epoll.rb
+++ b/test/test_epoll.rb
@@ -320,7 +320,7 @@ class TestEpoll < Test::Unit::TestCase
     assert_nil @ep.delete(@rd)
     assert_nil @ep.delete(@wr)
     assert_nothing_raised { @ep.add @rd, Epoll::IN }
-    assert_equal 0, @ep.delete(@rd)
+    assert_equal @rd, @ep.delete(@rd)
     assert_nil @ep.delete(@rd)
   end
 end
diff --git a/test/test_epoll_optimizations.rb b/test/test_epoll_optimizations.rb
new file mode 100644
index 0000000..c8df6f3
--- /dev/null
+++ b/test/test_epoll_optimizations.rb
@@ -0,0 +1,127 @@
+require 'test/unit'
+require 'strace'
+$-w = true
+
+require 'sleepy_penguin'
+
+class TestEpollOptimizations < Test::Unit::TestCase
+  include SleepyPenguin
+  RBX = defined?(RUBY_ENGINE) && (RUBY_ENGINE == 'rbx')
+  IO_PURGATORY = []
+
+  def setup
+    @rd, @wr = IO.pipe
+    @ep = Epoll.new
+  end
+
+  def teardown
+    [ @ep, @rd, @wr ].each { |io| io.close unless io.closed? }
+  end
+
+  def test_set
+    io, err = Strace.me do
+      @ep.set(@wr, Epoll::OUT)
+      @ep.set(@wr, Epoll::OUT)
+    end
+    assert_nil err
+    lines = io.readlines; io.close
+    assert_equal 1, lines.grep(/^epoll_ctl/).size
+    assert_match /EPOLL_CTL_ADD/, lines.grep(/^epoll_ctl/)[0]
+
+    io, err = Strace.me { @ep.set(@wr, Epoll::OUT | Epoll::ONESHOT) }
+    assert_nil err
+    lines = io.readlines; io.close
+    assert_equal 1, lines.grep(/^epoll_ctl/).size
+    assert_match /EPOLL_CTL_MOD/, lines.grep(/^epoll_ctl/)[0]
+
+    io, err = Strace.me { @ep.set(@wr, Epoll::OUT) }
+    assert_nil err
+    lines = io.readlines; io.close
+    assert_equal 1, lines.grep(/^epoll_ctl/).size
+    assert_match /EPOLL_CTL_MOD/, lines.grep(/^epoll_ctl/)[0]
+    @wr.close
+    @rd.close
+
+    @rd, @wr = IO.pipe
+    io, err = Strace.me { @ep.set(@wr, Epoll::OUT) }
+    assert_nil err
+    lines = io.readlines; io.close
+    assert_equal 1, lines.grep(/^epoll_ctl/).size
+    assert_match /EPOLL_CTL_ADD/, lines.grep(/^epoll_ctl/)[0]
+  end
+
+  def test_delete
+    @ep.set(@wr, Epoll::OUT)
+    rv = true
+    io, err = Strace.me { rv = @ep.delete(@wr) }
+    assert_equal @wr, rv
+    assert_nil err
+    lines = io.readlines; io.close
+    assert_equal 1, lines.grep(/^epoll_ctl/).size
+    assert_match %r{=\s+0$}, lines.grep(/^epoll_ctl/)[0]
+
+    rv = true
+    io, err = Strace.me { rv = @ep.delete(@wr) }
+    assert_nil rv
+    assert_nil err
+    lines = io.readlines; io.close
+    assert_equal 0, lines.grep(/^epoll_ctl/).size
+  end
+
+  def test_delete_closed
+    a = @wr.fileno
+    @ep.set(@wr, Epoll::OUT)
+    @rd.close
+    @wr.close
+    @rd, @wr = IO.pipe
+    assert_equal a, @wr.fileno
+    rv = true
+    io, err = Strace.me { rv = @ep.delete(@wr) }
+    lines = io.readlines; io.close
+    assert_nil err
+    assert_nil rv
+    assert_equal 0, lines.grep(/^epoll_ctl/).size
+  end
+
+  def test_delete_aliased_a
+    tmp = IO.for_fd @wr.fileno
+    IO_PURGATORY << tmp
+    @ep.set(tmp, Epoll::OUT)
+    rv = nil
+    io, err = Strace.me { rv = @ep.delete(@wr) }
+    lines = io.readlines; io.close
+    assert_equal @wr, rv
+    assert_nil err
+    assert_equal 1, lines.grep(/^epoll_ctl/).size
+    assert_match %r{=\s+0$}, lines.grep(/^epoll_ctl/)[0]
+    assert_equal 0, lines.grep(/ENOENT/).size
+  end
+
+  def test_delete_aliased_b
+    tmp = IO.for_fd @wr.fileno
+    IO_PURGATORY << tmp
+    @ep.set(@wr, Epoll::OUT)
+    rv = nil
+    io, err = Strace.me { rv = @ep.delete(tmp) }
+    lines = io.readlines; io.close
+    assert_equal tmp, rv
+    assert_nil err
+    assert_equal 1, lines.grep(/^epoll_ctl/).size
+    assert_match %r{=\s+0$}, lines.grep(/^epoll_ctl/)[0]
+    assert_equal 0, lines.grep(/ENOENT/).size
+  end
+
+  def test_delete_aliased_closed
+    tmp = IO.for_fd @wr.fileno
+    IO_PURGATORY << tmp
+    @ep.set(tmp, Epoll::OUT)
+    @wr.close
+    rv = nil
+    io, err = Strace.me { rv = @ep.delete(tmp) }
+    lines = io.readlines; io.close
+    assert_nil rv
+    assert_nil err
+    assert_equal 1, lines.grep(/^epoll_ctl/).size
+    assert_equal 1, lines.grep(/EBADF/).size
+  end
+end