* [sleepy.penguin] [PATCH 0/6] epoll wrapper cleanups
@ 2013-04-11 4:17 Eric Wong
2013-04-11 4:17 ` [sleepy.penguin] [PATCH 1/6] test_epoll: fix timing error in test Eric Wong
` (5 more replies)
0 siblings, 6 replies; 9+ messages in thread
From: Eric Wong @ 2013-04-11 4:17 UTC (permalink / raw)
To: sleepy.penguin
I've decided to provide lower-level access to epoll users without the
extra overhead of automatic GC marking and Epoll#set management.
The normal Epoll class is now implemented in Ruby, while the low-level
Epoll::IO class is a lightweight C wrapper. This should make code
easier to maintain, and folks that really want good concurrency can
use Epoll::IO directly (without the GC-safety normal Epoll provides).
I'm also deprecating the Epoll#set and Epoll#delete APIs, since
they're expensive to maintain. They will be removed in
sleepy_penguin 4.0 (probably 12 months or so).
I've also made Epoll thread-safe for Rubinius, as Rubinius C extensions
do not (or will not) hold any global lock, so the @marks array must be
protected. Epoll::IO does not need additional locking to be
thread-safe, it uses thread-local storage and the kernel eventpoll
struct is already protected by its internal locking.
Most of these changes have been my experiences in using epoll in recent
years (mostly outside of Ruby) and also close inspection (and some
hacking :) of the epoll implementation in Linux.
I'll document these patches more, later, and push.
ext/sleepy_penguin/epoll.c | 633 ++++++---------------------------------
ext/sleepy_penguin/epoll_green.h | 8 +-
ext/sleepy_penguin/inotify.c | 11 +-
lib/sleepy_penguin.rb | 1 +
lib/sleepy_penguin/epoll.rb | 266 ++++++++++++++++
test/test_epoll.rb | 31 +-
test/test_epoll_io.rb | 24 ++
test/test_epoll_optimizations.rb | 2 +-
8 files changed, 418 insertions(+), 558 deletions(-)
^ permalink raw reply [flat|nested] 9+ messages in thread* [sleepy.penguin] [PATCH 1/6] test_epoll: fix timing error in test 2013-04-11 4:17 [sleepy.penguin] [PATCH 0/6] epoll wrapper cleanups Eric Wong @ 2013-04-11 4:17 ` Eric Wong 2013-04-11 4:17 ` [sleepy.penguin] [PATCH 2/6] test_epoll: synchronize writes to the pipe array Eric Wong ` (4 subsequent siblings) 5 siblings, 0 replies; 9+ messages in thread From: Eric Wong @ 2013-04-11 4:17 UTC (permalink / raw) To: sleepy.penguin We need to record the time before the thread is spawned. --- test/test_epoll.rb | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/test/test_epoll.rb b/test/test_epoll.rb index fd22654..1e9a068 100644 --- a/test/test_epoll.rb +++ b/test/test_epoll.rb @@ -26,12 +26,12 @@ def test_constants def test_cross_thread tmp = [] - Thread.new { sleep 0.100; @ep.add(@wr, Epoll::OUT) } t0 = Time.now + Thread.new { sleep 0.100; @ep.add(@wr, Epoll::OUT) } @ep.wait { |flags,obj| tmp << [ flags, obj ] } elapsed = Time.now - t0 assert elapsed >= 0.100 - assert_equal [[Epoll::OUT, @wr]], tmp + assert_equal [[Epoll::OUT, @wr]], tmp, tmp.inspect end def test_fork_safe -- 1.8.2.279.g631bc94 ^ permalink raw reply related [flat|nested] 9+ messages in thread
* [sleepy.penguin] [PATCH 2/6] test_epoll: synchronize writes to the pipe array 2013-04-11 4:17 [sleepy.penguin] [PATCH 0/6] epoll wrapper cleanups Eric Wong 2013-04-11 4:17 ` [sleepy.penguin] [PATCH 1/6] test_epoll: fix timing error in test Eric Wong @ 2013-04-11 4:17 ` Eric Wong 2013-04-11 4:17 ` [sleepy.penguin] [PATCH 3/6] split Epoll and Epoll::IO, rewrite Epoll in Ruby Eric Wong ` (3 subsequent siblings) 5 siblings, 0 replies; 9+ messages in thread From: Eric Wong @ 2013-04-11 4:17 UTC (permalink / raw) To: sleepy.penguin Concurrent modification of the array is not thread-safe. --- test/test_epoll.rb | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/test/test_epoll.rb b/test/test_epoll.rb index 1e9a068..cd50cff 100644 --- a/test/test_epoll.rb +++ b/test/test_epoll.rb @@ -473,22 +473,26 @@ def test_wait_one_event_per_thread nr = 10 nr.times do r, w = IO.pipe - pipes[r] = w + lock.synchronize { pipes[r] = w } @ep.add(r, Epoll::IN | Epoll::ET | Epoll::ONESHOT) t = Thread.new do sleep 2 events = 0 @ep.wait(maxevents) do |_,obj| - assert pipes.include?(obj), "#{obj.inspect} is unknown" - lock.synchronize { ok << obj } + lock.synchronize do + assert pipes.include?(obj), "#{obj.inspect} is unknown" + ok << obj + end events += 1 end events end thr << t end - pipes.each_value { |w| w.syswrite '.' } + lock.synchronize do + pipes.each_value { |w| w.syswrite '.' } + end thr.each do |t| begin t.run -- 1.8.2.279.g631bc94 ^ permalink raw reply related [flat|nested] 9+ messages in thread
* [sleepy.penguin] [PATCH 3/6] split Epoll and Epoll::IO, rewrite Epoll in Ruby 2013-04-11 4:17 [sleepy.penguin] [PATCH 0/6] epoll wrapper cleanups Eric Wong 2013-04-11 4:17 ` [sleepy.penguin] [PATCH 1/6] test_epoll: fix timing error in test Eric Wong 2013-04-11 4:17 ` [sleepy.penguin] [PATCH 2/6] test_epoll: synchronize writes to the pipe array Eric Wong @ 2013-04-11 4:17 ` Eric Wong 2013-04-12 20:38 ` Eric Wong 2013-04-11 4:17 ` [sleepy.penguin] [PATCH 4/6] epoll: implement thread-safety for mark/flag arrays Eric Wong ` (2 subsequent siblings) 5 siblings, 1 reply; 9+ messages in thread From: Eric Wong @ 2013-04-11 4:17 UTC (permalink / raw) To: sleepy.penguin Epoll::IO is a dangerous, low-level class which is intended for users aware of the GC and fork behavior of epoll in the Linux kernel. Rewriting the higher-level Epoll in Ruby makes it easier to maintain, especially since Rubinius has no GVL while running C extensions. --- ext/sleepy_penguin/epoll.c | 601 ++++++--------------------------------- ext/sleepy_penguin/epoll_green.h | 8 +- lib/sleepy_penguin.rb | 1 + lib/sleepy_penguin/epoll.rb | 228 +++++++++++++++ test/test_epoll.rb | 15 +- test/test_epoll_io.rb | 24 ++ test/test_epoll_optimizations.rb | 2 +- 7 files changed, 356 insertions(+), 523 deletions(-) create mode 100644 lib/sleepy_penguin/epoll.rb create mode 100644 test/test_epoll_io.rb diff --git a/ext/sleepy_penguin/epoll.c b/ext/sleepy_penguin/epoll.c index 3dcd357..64df698 100644 --- a/ext/sleepy_penguin/epoll.c +++ b/ext/sleepy_penguin/epoll.c @@ -3,20 +3,12 @@ #include <pthread.h> #include <time.h> #include "missing_epoll.h" -#ifdef HAVE_RUBY_ST_H -# include <ruby/st.h> -#else -# include <st.h> -#endif #include "missing_rb_thread_fd_close.h" #include "missing_rb_update_max_fd.h" -#define EP_RECREATE (-2) static pthread_key_t epoll_key; -static st_table *active; -static const int step = 64; /* unlikely to grow unless you're huge */ -static VALUE cEpoll_IO; static ID id_for_fd; +static VALUE cEpoll; static uint64_t now_ms(void) { @@ -47,27 +39,26 @@ static VALUE unpack_event_data(struct epoll_event *event) # endif #endif -struct rb_epoll { - int fd; - VALUE io; - VALUE marks; - VALUE flag_cache; - int flags; -}; - struct ep_per_thread { - struct rb_epoll *ep; + VALUE io; + int fd; int timeout; int maxevents; int capa; struct epoll_event events[FLEX_ARRAY]; }; -static struct ep_per_thread *ept_get(int maxevents) +/* this will raise if the IO is closed */ +static void ep_fd_check(struct ep_per_thread *ept) +{ + ept->fd = rb_sp_fileno(ept->io); +} + +static struct ep_per_thread *ept_get(VALUE self, int maxevents) { struct ep_per_thread *ept = pthread_getspecific(epoll_key); - int err; size_t size; + int err; if (ept && ept->capa >= maxevents) goto out; @@ -87,253 +78,72 @@ static struct ep_per_thread *ept_get(int maxevents) ept->capa = maxevents; out: ept->maxevents = maxevents; + ept->io = self; + ep_fd_check(ept); return ept; } -static struct rb_epoll *ep_get(VALUE self) -{ - struct rb_epoll *ep; - - Data_Get_Struct(self, struct rb_epoll, ep); - - return ep; -} - -static void gcmark(void *ptr) -{ - struct rb_epoll *ep = ptr; - - rb_gc_mark(ep->io); - rb_gc_mark(ep->marks); - rb_gc_mark(ep->flag_cache); -} - -static void gcfree(void *ptr) -{ - struct rb_epoll *ep = ptr; - - if (ep->fd >= 0) { - st_data_t key = ep->fd; - st_delete(active, &key, NULL); - } - if (NIL_P(ep->io) && ep->fd >= 0) { - /* can't raise during GC, and close() never fails in Linux */ - (void)close(ep->fd); - errno = 0; - } - /* let GC take care of the underlying IO object if there is one */ - - xfree(ep); -} - -static VALUE alloc(VALUE klass) -{ - struct rb_epoll *ep; - VALUE self; - - self = Data_Make_Struct(klass, struct rb_epoll, gcmark, gcfree, ep); - ep->fd = -1; - ep->io = Qnil; - ep->marks = Qnil; - ep->flag_cache = Qnil; - ep->flags = 0; - - return self; -} - -static void my_epoll_create(struct rb_epoll *ep) -{ - ep->fd = epoll_create1(ep->flags); - - if (ep->fd == -1) { - if (errno == EMFILE || errno == ENFILE || errno == ENOMEM) { - rb_gc(); - ep->fd = epoll_create1(ep->flags); - } - if (ep->fd == -1) - rb_sys_fail("epoll_create1"); - } - rb_update_max_fd(ep->fd); - st_insert(active, (st_data_t)ep->fd, (st_data_t)ep); - ep->marks = rb_ary_new(); - 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); - ep_fd_check(ep); - assert(TYPE(ep->marks) == T_ARRAY && "marks not initialized"); - assert(TYPE(ep->flag_cache) == T_ARRAY && "flag_cache not initialized"); -} - /* * call-seq: - * SleepyPenguin::Epoll.new([flags]) -> Epoll object + * SleepyPenguin::Epoll::IO.new(flags) -> Epoll::IO object * - * Creates a new Epoll object with an optional +flags+ argument. - * +flags+ may currently be +:CLOEXEC+ or +0+ (or +nil+). + * Creates a new Epoll::IO object with the given +flags+ argument. + * +flags+ may currently be +CLOEXEC+ or +0+. */ -static VALUE init(int argc, VALUE *argv, VALUE self) -{ - struct rb_epoll *ep = ep_get(self); - VALUE fl; - - rb_scan_args(argc, argv, "01", &fl); - ep->flags = rb_sp_get_flags(self, fl); - my_epoll_create(ep); - - return self; -} - -static VALUE ctl(VALUE self, VALUE io, VALUE flags, int op) +static VALUE s_new(VALUE klass, VALUE _flags) { - struct epoll_event event; - struct rb_epoll *ep = ep_get(self); - int fd = rb_sp_fileno(io); - int rv; + int flags = rb_sp_get_flags(klass, _flags); + int fd = epoll_create1(flags); + VALUE rv; - ep_check(ep); - event.events = rb_sp_get_uflags(self, flags); - pack_event_data(&event, io); - - rv = epoll_ctl(ep->fd, op, fd, &event); - if (rv == -1) { - if (errno == ENOMEM) { + if (fd < 0) { + if (errno == EMFILE || errno == ENFILE || errno == ENOMEM) { rb_gc(); - rv = epoll_ctl(ep->fd, op, fd, &event); + fd = epoll_create1(flags); } - if (rv == -1) - rb_sys_fail("epoll_ctl"); - } - switch (op) { - case EPOLL_CTL_ADD: - rb_ary_store(ep->marks, fd, io); - /* fall-through */ - case EPOLL_CTL_MOD: - flags = UINT2NUM(event.events); - rb_ary_store(ep->flag_cache, fd, flags); - break; - case EPOLL_CTL_DEL: - rb_ary_store(ep->marks, fd, Qnil); - rb_ary_store(ep->flag_cache, fd, Qnil); + if (fd == -1) + rb_sys_fail("epoll_create1"); } - return INT2NUM(rv); + rv = INT2FIX(fd); + return rb_call_super(1, &rv); } /* * call-seq: - * ep.set(io, flags) -> 0 + * epoll_io.epoll_ctl(op, io, events) -> nil * - * Used to avoid exceptions when your app is too lazy to check - * what state a descriptor is in, this sets the epoll descriptor - * to watch an +io+ with the given +flags+ + * Register, modify, or register a watch for a given +io+ for events. * - * +flags+ may be an array of symbols or an unsigned Integer bit mask: + * +op+ may be one of +EPOLL_CTL_ADD+, +EPOLL_CTL_MOD+, or +EPOLL_CTL_DEL+ + * +io+ is an IO object or one which proxies via the +to_io+ method. + * +events+ is an integer mask of events to watch for. * - * - flags = [ :IN, :ET ] - * - flags = SleepyPenguin::Epoll::IN | SleepyPenguin::Epoll::ET - * - * See constants in Epoll for more information. + * Returns nil on success. */ -static VALUE set(VALUE self, VALUE io, VALUE flags) +static VALUE epctl(VALUE self, VALUE _op, VALUE io, VALUE events) { struct epoll_event event; - struct rb_epoll *ep = ep_get(self); + int epfd = rb_sp_fileno(self); int fd = rb_sp_fileno(io); + int op = NUM2INT(_op); int rv; - VALUE cur_io = rb_ary_entry(ep->marks, fd); - ep_check(ep); - event.events = rb_sp_get_uflags(self, flags); + event.events = NUM2UINT(events); pack_event_data(&event, io); - 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; - -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; - } - } 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); - } - flags = UINT2NUM(event.events); - rb_ary_store(ep->flag_cache, fd, flags); - - return INT2NUM(rv); -} - -/* - * call-seq: - * epoll.delete(io) -> io or nil - * - * Stops an +io+ object from being monitored. This is like Epoll#del - * but returns +nil+ on ENOENT instead of raising an error. This is - * useful for apps that do not care to track the status of an - * epoll object itself. - */ -static VALUE delete(VALUE self, VALUE io) -{ - struct rb_epoll *ep = ep_get(self); - int fd = rb_sp_fileno(io); - int rv; - VALUE cur_io; - - ep_check(ep); - if (rb_sp_io_closed(io)) - goto out; - - cur_io = rb_ary_entry(ep->marks, fd); - if (NIL_P(cur_io) || rb_sp_io_closed(cur_io)) - return Qnil; - - rv = epoll_ctl(ep->fd, EPOLL_CTL_DEL, fd, NULL); - if (rv == -1) { - /* beware of IO.for_fd-created descriptors */ - if (errno == ENOENT || errno == EBADF) { - errno = 0; - io = Qnil; - } else { - rb_sys_fail("epoll_ctl - del"); + rv = epoll_ctl(epfd, op, fd, &event); + if (rv < 0) { + if (errno == ENOMEM) { + rb_gc(); + rv = epoll_ctl(epfd, op, fd, &event); } + if (rv < 0) + rb_sys_fail("epoll_ctl"); } -out: - rb_ary_store(ep->marks, fd, Qnil); - rb_ary_store(ep->flag_cache, fd, Qnil); - return io; + return Qnil; } static VALUE epwait_result(struct ep_per_thread *ept, int n) @@ -358,10 +168,11 @@ static int epoll_resume_p(uint64_t expire_at, struct ep_per_thread *ept) { uint64_t now; - ep_fd_check(ept->ep); - if (errno != EINTR) return 0; + + ep_fd_check(ept); + if (ept->timeout < 0) return 1; now = now_ms(); @@ -373,8 +184,7 @@ static int epoll_resume_p(uint64_t expire_at, struct ep_per_thread *ept) static VALUE nogvl_wait(void *args) { struct ep_per_thread *ept = args; - int fd = ept->ep->fd; - int n = epoll_wait(fd, ept->events, ept->maxevents, ept->timeout); + int n = epoll_wait(ept->fd, ept->events, ept->maxevents, ept->timeout); return (VALUE)n; } @@ -385,7 +195,7 @@ static VALUE real_epwait(struct ep_per_thread *ept) uint64_t expire_at = ept->timeout > 0 ? now_ms() + ept->timeout : 0; do { - n = (int)rb_sp_fd_region(nogvl_wait, ept, ept->ep->fd); + n = (int)rb_sp_fd_region(nogvl_wait, ept, ept->fd); } while (n == -1 && epoll_resume_p(expire_at, ept)); return epwait_result(ept, n); @@ -396,11 +206,11 @@ static VALUE real_epwait(struct ep_per_thread *ept) /* * call-seq: - * epoll.wait([maxevents[, timeout]]) { |flags, io| ... } + * ep_io.epoll_wait([maxevents[, timeout]]) { |events, io| ... } * - * Calls epoll_wait(2) and yields Integer +flags+ and IO objects watched + * Calls epoll_wait(2) and yields Integer +events+ and IO objects watched * for. +maxevents+ is the maximum number of events to process at once, - * lower numbers may prevent starvation when used by Epoll#wait in multiple + * lower numbers may prevent starvation when used by epoll_wait in multiple * threads. Larger +maxevents+ reduces syscall overhead for * single-threaded applications. +maxevents+ defaults to 64 events. * +timeout+ is specified in milliseconds, +nil+ @@ -409,259 +219,17 @@ static VALUE real_epwait(struct ep_per_thread *ept) static VALUE epwait(int argc, VALUE *argv, VALUE self) { VALUE timeout, maxevents; - struct rb_epoll *ep = ep_get(self); struct ep_per_thread *ept; - ep_check(ep); rb_need_block(); rb_scan_args(argc, argv, "02", &maxevents, &timeout); - ept = ept_get(NIL_P(maxevents) ? 64 : NUM2INT(maxevents)); + + ept = ept_get(self, NIL_P(maxevents) ? 64 : NUM2INT(maxevents)); ept->timeout = NIL_P(timeout) ? -1 : NUM2INT(timeout); - ept->ep = ep; return real_epwait(ept); } -/* - * call-seq: - * epoll.add(io, flags) -> 0 - * - * Starts watching a given +io+ object with +flags+ which may be an Integer - * bitmask or Array representing arrays to watch for. Consider Epoll#set - * instead as it is easier to use. - */ -static VALUE add(VALUE self, VALUE io, VALUE flags) -{ - return ctl(self, io, flags, EPOLL_CTL_ADD); -} - -/* - * call-seq: - * epoll.del(io) -> 0 - * - * Disables an IO object from being watched. Consider Epoll#delete as - * it is easier to use. - */ -static VALUE del(VALUE self, VALUE io) -{ - return ctl(self, io, INT2FIX(0), EPOLL_CTL_DEL); -} - -/* - * call-seq: - * epoll.mod(io, flags) -> 0 - * - * Changes the watch for an existing IO object based on +flags+. - * Consider Epoll#set instead as it is easier to use. - */ -static VALUE mod(VALUE self, VALUE io, VALUE flags) -{ - return ctl(self, io, flags, EPOLL_CTL_MOD); -} - -/* - * call-seq: - * epoll.to_io -> Epoll::IO object - * - * Used to expose the given Epoll object as an Epoll::IO object for IO.select - * or IO#stat. This is unlikely to be useful directly, but is used internally - * by IO.select. - */ -static VALUE to_io(VALUE self) -{ - struct rb_epoll *ep = ep_get(self); - - ep_check(ep); - - if (NIL_P(ep->io)) - ep->io = rb_funcall(cEpoll_IO, id_for_fd, 1, INT2NUM(ep->fd)); - - return ep->io; -} - -/* - * call-seq: - * epoll.close -> nil - * - * Closes an existing Epoll object and returns memory back to the kernel. - * Raises IOError if object is already closed. - */ -static VALUE epclose(VALUE self) -{ - struct rb_epoll *ep = ep_get(self); - - if (ep->fd >= 0) { - st_data_t key = ep->fd; - st_delete(active, &key, NULL); - } - - if (NIL_P(ep->io)) { - ep_fd_check(ep); - - if (ep->fd == EP_RECREATE) { - ep->fd = -1; /* success */ - } else { - int err; - int fd = ep->fd; - - ep->fd = -1; - rb_thread_fd_close(fd); - err = close(fd); - if (err == -1) - rb_sys_fail("close"); - } - } else { - ep->fd = -1; - rb_io_close(ep->io); - } - - return Qnil; -} - -/* - * call-seq: - * epoll.closed? -> true or false - * - * Returns whether or not an Epoll object is closed. - */ -static VALUE epclosed(VALUE self) -{ - struct rb_epoll *ep = ep_get(self); - - return ep->fd == -1 ? Qtrue : Qfalse; -} - -static int cloexec_dup(struct rb_epoll *ep) -{ -#ifdef F_DUPFD_CLOEXEC - int flags = ep->flags & EPOLL_CLOEXEC ? F_DUPFD_CLOEXEC : F_DUPFD; - int fd = fcntl(ep->fd, flags, 0); -#else /* potentially racy on GVL-free systems: */ - int fd = dup(ep->fd); - if (fd >= 0) - (void)fcntl(fd, F_SETFD, FD_CLOEXEC); -#endif - return fd; -} - -/* - * call-seq: - * epoll.dup -> another Epoll object - * - * Duplicates an Epoll object and userspace buffers related to this library. - * Since SleepyPenguin 3.1.0, this is no longer needed for multi-threaded - * Epoll#wait. - */ -static VALUE init_copy(VALUE copy, VALUE orig) -{ - struct rb_epoll *a = ep_get(orig); - struct rb_epoll *b = ep_get(copy); - - assert(NIL_P(b->io) && "Ruby broken?"); - - 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) { - if (errno == ENFILE || errno == EMFILE) { - rb_gc(); - b->fd = cloexec_dup(a); - } - if (b->fd == -1) - rb_sys_fail("dup"); - } - st_insert(active, (st_data_t)b->fd, (st_data_t)b); - - return copy; -} - -/* occasionally it's still useful to lookup aliased IO objects - * based on for debugging */ -static int my_fileno(VALUE obj) -{ - if (T_FIXNUM == TYPE(obj)) - return FIX2INT(obj); - return rb_sp_fileno(obj); -} - -/* - * call-seq: - * epoll.io_for(io) -> object - * - * Returns the given IO object currently being watched for. Different - * IO objects may internally refer to the same process file descriptor. - * Mostly used for debugging. - */ -static VALUE io_for(VALUE self, VALUE obj) -{ - struct rb_epoll *ep = ep_get(self); - - return rb_ary_entry(ep->marks, my_fileno(obj)); -} - -/* - * call-seq: - * epoll.flags_for(io) -> Integer - * - * Returns the flags currently watched for in current Epoll object. - * Mostly used for debugging. - */ -static VALUE flags_for(VALUE self, VALUE obj) -{ - struct rb_epoll *ep = ep_get(self); - - return rb_ary_entry(ep->flag_cache, my_fileno(obj)); -} - -/* - * call-seq: - * epoll.include?(io) => true or false - * - * Returns whether or not a given IO is watched and prevented from being - * garbage-collected by the current Epoll object. This may include - * closed IO objects. - */ -static VALUE include_p(VALUE self, VALUE obj) -{ - struct rb_epoll *ep = ep_get(self); - - return NIL_P(rb_ary_entry(ep->marks, my_fileno(obj))) ? Qfalse : Qtrue; -} - -/* - * we close (or lose to GC) epoll descriptors at fork to avoid leakage - * and invalid objects being referenced later in the child - */ -static int ep_atfork(st_data_t key, st_data_t value, void *ignored) -{ - struct rb_epoll *ep = (struct rb_epoll *)value; - - if (NIL_P(ep->io)) { - if (ep->fd >= 0) - (void)close(ep->fd); - } else { - ep->io = Qnil; /* must let GC take care of it later :< */ - } - ep->fd = EP_RECREATE; - - return ST_CONTINUE; -} - -static void atfork_child(void) -{ - st_table *old = active; - - active = st_init_numtable(); - st_foreach(old, ep_atfork, (st_data_t)NULL); - st_free_table(old); -} - static void epoll_once(void) { int err = pthread_key_create(&epoll_key, free); @@ -670,19 +238,17 @@ static void epoll_once(void) errno = err; rb_sys_fail("pthread_key_create"); } +} - active = st_init_numtable(); - - if (pthread_atfork(NULL, NULL, atfork_child) != 0) { - rb_gc(); - if (pthread_atfork(NULL, NULL, atfork_child) != 0) - rb_memerror(); - } +/* :nodoc: */ +static VALUE event_flags(VALUE self, VALUE flags) +{ + return UINT2NUM(rb_sp_get_uflags(self, flags)); } void sleepy_penguin_init_epoll(void) { - VALUE mSleepyPenguin, cEpoll; + VALUE mSleepyPenguin, cEpoll_IO; static pthread_once_t once = PTHREAD_ONCE_INIT; int err = pthread_once(&once, epoll_once); @@ -708,6 +274,7 @@ void sleepy_penguin_init_epoll(void) * And then access classes via: * * - SP::Epoll + * - SP::Epoll::IO * - SP::EventFD * - SP::Inotify * - SP::TimerFD @@ -717,36 +284,36 @@ void sleepy_penguin_init_epoll(void) /* * Document-class: SleepyPenguin::Epoll * - * The Epoll class provides access to epoll(7) functionality in the - * Linux 2.6 kernel. It provides fork and GC-safety for Ruby - * objects stored within the IO object and may be passed as an - * argument to IO.select. + * The Epoll class provides high-level access to epoll(7) + * functionality in the Linux 2.6 and later kernels. It provides + * fork and GC-safety for Ruby objects stored within the IO object + * and may be passed as an argument to IO.select. */ cEpoll = rb_define_class_under(mSleepyPenguin, "Epoll", rb_cObject); /* * Document-class: SleepyPenguin::Epoll::IO * - * Epoll::IO is an internal class. Its only purpose is to be - * compatible with IO.select and related methods and should - * never be used directly, use Epoll instead. + * Epoll::IO is a low-level class. It does not provide fork nor + * GC-safety, so Ruby IO objects added via epoll_ctl must be retained + * by the application until IO#close is called. */ cEpoll_IO = rb_define_class_under(cEpoll, "IO", rb_cIO); - rb_define_method(cEpoll, "initialize", init, -1); - rb_define_method(cEpoll, "initialize_copy", init_copy, 1); - rb_define_alloc_func(cEpoll, alloc); - rb_define_method(cEpoll, "to_io", to_io, 0); - rb_define_method(cEpoll, "close", epclose, 0); - rb_define_method(cEpoll, "closed?", epclosed, 0); - rb_define_method(cEpoll, "add", add, 2); - rb_define_method(cEpoll, "mod", mod, 2); - rb_define_method(cEpoll, "del", del, 1); - rb_define_method(cEpoll, "delete", delete, 1); - rb_define_method(cEpoll, "io_for", io_for, 1); - rb_define_method(cEpoll, "flags_for", flags_for, 1); - rb_define_method(cEpoll, "include?", include_p, 1); - rb_define_method(cEpoll, "set", set, 2); - rb_define_method(cEpoll, "wait", epwait, -1); + rb_define_singleton_method(cEpoll_IO, "new", s_new, 1); + + rb_define_method(cEpoll_IO, "epoll_ctl", epctl, 3); + rb_define_method(cEpoll_IO, "epoll_wait", epwait, -1); + + rb_define_method(cEpoll, "__event_flags", event_flags, 1); + + /* registers an IO object via epoll_ctl */ + rb_define_const(cEpoll, "CTL_ADD", INT2NUM(EPOLL_CTL_ADD)); + + /* unregisters an IO object via epoll_ctl */ + rb_define_const(cEpoll, "CTL_DEL", INT2NUM(EPOLL_CTL_DEL)); + + /* modifies the registration of an IO object via epoll_ctl */ + rb_define_const(cEpoll, "CTL_MOD", INT2NUM(EPOLL_CTL_MOD)); /* specifies whether close-on-exec flag is set for Epoll.new */ rb_define_const(cEpoll, "CLOEXEC", INT2NUM(EPOLL_CLOEXEC)); diff --git a/ext/sleepy_penguin/epoll_green.h b/ext/sleepy_penguin/epoll_green.h index 276a545..4331227 100644 --- a/ext/sleepy_penguin/epoll_green.h +++ b/ext/sleepy_penguin/epoll_green.h @@ -26,9 +26,9 @@ static int safe_epoll_wait(struct ep_per_thread *ept) do { TRAP_BEG; - n = epoll_wait(ept->ep->fd, ept->events, ept->maxevents, 0); + n = epoll_wait(ept->fd, ept->events, ept->maxevents, 0); TRAP_END; - } while (n == -1 && errno == EINTR && ep_fd_check(ept->ep)); + } while (n == -1 && errno == EINTR && (ep_fd_check(ept), 1)); return n; } @@ -38,7 +38,7 @@ static int epwait_forever(struct ep_per_thread *ept) int n; do { - (void)rb_io_wait_readable(ept->ep->fd); + (void)rb_io_wait_readable(ept->fd); n = safe_epoll_wait(ept); } while (n == 0); @@ -55,7 +55,7 @@ static int epwait_timed(struct ep_per_thread *ept) for (;;) { struct timeval t0, now, diff; int n; - int fd = ept->ep->fd; + int fd = ept->fd; fd_set rfds; FD_ZERO(&rfds); diff --git a/lib/sleepy_penguin.rb b/lib/sleepy_penguin.rb index 3a189b1..c13eb0c 100644 --- a/lib/sleepy_penguin.rb +++ b/lib/sleepy_penguin.rb @@ -5,3 +5,4 @@ module SleepyPenguin SLEEPY_PENGUIN_VERSION = '3.1.0' end require 'sleepy_penguin_ext' +require 'sleepy_penguin/epoll' diff --git a/lib/sleepy_penguin/epoll.rb b/lib/sleepy_penguin/epoll.rb new file mode 100644 index 0000000..845dcf0 --- /dev/null +++ b/lib/sleepy_penguin/epoll.rb @@ -0,0 +1,228 @@ +class SleepyPenguin::Epoll + + # Epoll objects may be watched by IO.select and similar methods + attr_reader :to_io + + # call-seq: + # SleepyPenguin::Epoll.new([flags]) -> Epoll object + # + # Creates a new Epoll object with an optional +flags+ argument. + # +flags+ may currently be +:CLOEXEC+ or +0+ (or +nil+). + def initialize(create_flags = nil) + @to_io = SleepyPenguin::Epoll::IO.new(create_flags) + @events = [] + @marks = [] + @pid = $$ + @create_flags = create_flags + @copies = { @to_io => self } + end + + def __ep_reinit # :nodoc: + @events.clear + @marks.clear + @to_io = SleepyPenguin::Epoll::IO.new(@create_flags) + end + + # auto-reinitialize the Epoll object after forking + def __ep_check # :nodoc: + return if @pid == $$ + return if @to_io.closed? + objects = @copies.values + @copies.each_key { |epio| epio.close } + @copies.clear + __ep_reinit + objects.each do |obj| + io_dup = @to_io.dup + @copies[io_dup] = obj + end + @pid = $$ + end + + # Calls epoll_wait(2) and yields Integer +events+ and IO objects watched + # for. +maxevents+ is the maximum number of events to process at once, + # lower numbers may prevent starvation when used by epoll_wait in multiple + # threads. Larger +maxevents+ reduces syscall overhead for + # single-threaded applications. +maxevents+ defaults to 64 events. + # +timeout+ is specified in milliseconds, +nil+ + # (the default) meaning it will block and wait indefinitely. + def wait(maxevents = 64, timeout = nil) + __ep_check + @to_io.epoll_wait(maxevents, timeout) { |events, io| yield(events, io) } + end + + # Starts watching a given +io+ object with +events+ which may be an Integer + # bitmask or Array representing arrays to watch for. + def add(io, events) + __ep_check + fd = io.to_io.fileno + events = __event_flags(events) + @to_io.epoll_ctl(CTL_ADD, io, events) + @marks[fd] = io + @events[fd] = events + 0 + end + + # call-seq: + # ep.del(io) -> 0 + # + # Disables an IO object from being watched. + def del(io) + __ep_check + fd = io.to_io.fileno + rv = @to_io.epoll_ctl(CTL_DEL, io, 0) + @marks[fd] = @events[fd] = nil + rv + end + + # call-seq: + # ep.delete(io) -> io or nil + # + # This method is deprecated and will be removed in sleepy_penguin 4.x + # + # Stops an +io+ object from being monitored. This is like Epoll#del + # but returns +nil+ on ENOENT instead of raising an error. This is + # useful for apps that do not care to track the status of an + # epoll object itself. + # + # This method is deprecated and will be removed in sleepy_penguin 4.x + def delete(io) + __ep_check + fd = io.to_io.fileno + cur_io = @marks[fd] + return if nil == cur_io || cur_io.to_io.closed? + @marks[fd] = @events[fd] = nil + @to_io.epoll_ctl(CTL_DEL, io, 0) + io + rescue Errno::ENOENT, Errno::EBADF + end + + # call-seq: + # epoll.mod(io, flags) -> 0 + # + # Changes the watch for an existing IO object based on +events+. + # Returns zero on success, will raise SystemError on failure. + def mod(io, events) + __ep_check + fd = io.to_io.fileno + events = __event_flags(events) + rv = @to_io.epoll_ctl(CTL_MOD, io, events) + @marks[fd] = io + @events[fd] = events + rv + end + + # call-seq: + # ep.set(io, flags) -> 0 + # + # This method is deprecated and will be removed in sleepy_penguin 4.x + # + # Used to avoid exceptions when your app is too lazy to check + # what state a descriptor is in, this sets the epoll descriptor + # to watch an +io+ with the given +events+ + # + # +events+ may be an array of symbols or an unsigned Integer bit mask: + # + # - events = [ :IN, :ET ] + # - events = SleepyPenguin::Epoll::IN | SleepyPenguin::Epoll::ET + # + # See constants in Epoll for more information. + # + # This method is deprecated and will be removed in sleepy_penguin 4.x + def set(io, events) + __ep_check + fd = io.to_io.fileno + cur_io = @marks[fd] + if cur_io == io + cur_events = @events[fd] + return 0 if (cur_events & ONESHOT) == 0 && cur_events == events + begin + @to_io.epoll_ctl(CTL_MOD, io, events) + rescue Errno::ENOENT + warn "epoll flag cache failed (mod -> add)" + @to_io.epoll_ctl(CTL_ADD, io, events) + @marks[fd] = io + end + else + begin + @to_io.epoll_ctl(CTL_ADD, io, events) + rescue Errno::EEXIST + warn "epoll flag cache failed (add -> mod)" + @to_io.epoll_ctl(CTL_MOD, io, events) + end + @marks[fd] = io + end + @events[fd] = events + 0 + end + + # call-seq: + # ep.close -> nil + # + # Closes an existing Epoll object and returns memory back to the kernel. + # Raises IOError if object is already closed. + def close + __ep_check + @copies.delete(@to_io) + @to_io.close + end + + # call-seq: + # ep.closed? -> true or false + # + # Returns whether or not an Epoll object is closed. + def closed? + __ep_check + @to_io.closed? + end + + # we still support integer FDs for some debug functions + def __fileno(io) # :nodoc: + Integer === io ? io : io.to_io.fileno + end + + # call-seq: + # ep.io_for(io) -> object + # + # Returns the given IO object currently being watched for. Different + # IO objects may internally refer to the same process file descriptor. + # Mostly used for debugging. + def io_for(io) + __ep_check + @marks[__fileno(io)] + end + + # call-seq: + # epoll.events_for(io) -> Integer + # + # Returns the events currently watched for in current Epoll object. + # Mostly used for debugging. + def events_for(io) + __ep_check + @events[__fileno(io)] + end + + # backwards compatibility, to be removed in 4.x + alias flags_for events_for + + # call-seq: + # epoll.include?(io) -> true or false + # + # Returns whether or not a given IO is watched and prevented from being + # garbage-collected by the current Epoll object. This may include + # closed IO objects. + def include?(io) + __ep_check + @marks[__fileno(io)] ? true : nil + end + + def initialize_copy(src) # :nodoc: + __ep_check + rv = super + unless @to_io.closed? + @to_io = @to_io.dup + @copies[@to_io] = self + end + + rv + end +end diff --git a/test/test_epoll.rb b/test/test_epoll.rb index cd50cff..1a99dfd 100644 --- a/test/test_epoll.rb +++ b/test/test_epoll.rb @@ -48,6 +48,19 @@ def test_fork_safe assert_equal [[Epoll::IN, @rd]], tmp end + def test_dup_and_fork + epdup = @ep.dup + @ep.close + assert ! epdup.closed? + pid = fork do + exit(!epdup.closed? && @ep.closed?) + end + _, status = Process.waitpid2(pid) + assert status.success?, status.inspect + ensure + epdup.close + end + def test_after_fork_usability fork { @ep.add(@rd, Epoll::IN); exit!(0) } fork { @ep.set(@rd, Epoll::IN); exit!(0) } @@ -399,7 +412,7 @@ def test_flags_for_sym_ary def test_include? assert ! @ep.include?(@rd) @ep.add @rd, Epoll::IN - assert @ep.include?(@rd) + assert @ep.include?(@rd), @ep.instance_variable_get(:@marks).inspect assert @ep.include?(@rd.fileno) assert ! @ep.include?(@wr) assert ! @ep.include?(@wr.fileno) diff --git a/test/test_epoll_io.rb b/test/test_epoll_io.rb new file mode 100644 index 0000000..8aca155 --- /dev/null +++ b/test/test_epoll_io.rb @@ -0,0 +1,24 @@ +require 'test/unit' +require 'fcntl' +require 'socket' +require 'thread' +$-w = true +Thread.abort_on_exception = true +require 'sleepy_penguin' + +class TestEpollIO < Test::Unit::TestCase + include SleepyPenguin + RBX = defined?(RUBY_ENGINE) && (RUBY_ENGINE == 'rbx') + + def setup + @rd, @wr = IO.pipe + @epio = Epoll::IO.new(nil) + end + + def test_add_wait + @epio.epoll_ctl(Epoll::CTL_ADD, @wr, Epoll::OUT) + ev = [] + @epio.epoll_wait { |events, obj| ev << [ events, obj ] } + assert_equal([[Epoll::OUT, @wr]], ev) + end +end diff --git a/test/test_epoll_optimizations.rb b/test/test_epoll_optimizations.rb index bd77397..f5970fd 100644 --- a/test/test_epoll_optimizations.rb +++ b/test/test_epoll_optimizations.rb @@ -28,7 +28,7 @@ def test_set end assert_nil err lines = io.readlines; io.close - assert_equal 1, lines.grep(/^epoll_ctl/).size + assert_equal 1, lines.grep(/^epoll_ctl/).size, lines.inspect assert_match(/EPOLL_CTL_ADD/, lines.grep(/^epoll_ctl/)[0]) io, err = Strace.me { @ep.set(@wr, Epoll::OUT | Epoll::ONESHOT) } -- 1.8.2.279.g631bc94 ^ permalink raw reply related [flat|nested] 9+ messages in thread
* Re: [sleepy.penguin] [PATCH 3/6] split Epoll and Epoll::IO, rewrite Epoll in Ruby 2013-04-11 4:17 ` [sleepy.penguin] [PATCH 3/6] split Epoll and Epoll::IO, rewrite Epoll in Ruby Eric Wong @ 2013-04-12 20:38 ` Eric Wong 0 siblings, 0 replies; 9+ messages in thread From: Eric Wong @ 2013-04-12 20:38 UTC (permalink / raw) To: sleepy.penguin Eric Wong <normalperson@yhbt.net> wrote: > --- a/ext/sleepy_penguin/epoll.c > +++ b/ext/sleepy_penguin/epoll.c > @@ -358,10 +168,11 @@ static int epoll_resume_p(uint64_t expire_at, struct ep_per_thread *ept) > { > uint64_t now; > > - ep_fd_check(ept->ep); > - > if (errno != EINTR) > return 0; > + > + ep_fd_check(ept); > + > if (ept->timeout < 0) > return 1; > now = now_ms(); The above hunk was buggy and inconsistent with the green-thread-friendly variant. Below is an updated patch: --------------------------------8<------------------------------- Subject: [PATCH] split Epoll and Epoll::IO, rewrite Epoll in Ruby Epoll::IO is a dangerous, low-level class which is intended for users aware of the GC and fork behavior of epoll in the Linux kernel. Rewriting the higher-level Epoll in Ruby makes it easier to maintain, especially since Rubinius has no GVL while running C extensions. --- ext/sleepy_penguin/epoll.c | 603 ++++++--------------------------------- ext/sleepy_penguin/epoll_green.h | 8 +- lib/sleepy_penguin.rb | 1 + lib/sleepy_penguin/epoll.rb | 228 +++++++++++++++ test/test_epoll.rb | 15 +- test/test_epoll_io.rb | 24 ++ test/test_epoll_optimizations.rb | 2 +- 7 files changed, 359 insertions(+), 522 deletions(-) create mode 100644 lib/sleepy_penguin/epoll.rb create mode 100644 test/test_epoll_io.rb diff --git a/ext/sleepy_penguin/epoll.c b/ext/sleepy_penguin/epoll.c index 3dcd357..2ddc71d 100644 --- a/ext/sleepy_penguin/epoll.c +++ b/ext/sleepy_penguin/epoll.c @@ -3,20 +3,12 @@ #include <pthread.h> #include <time.h> #include "missing_epoll.h" -#ifdef HAVE_RUBY_ST_H -# include <ruby/st.h> -#else -# include <st.h> -#endif #include "missing_rb_thread_fd_close.h" #include "missing_rb_update_max_fd.h" -#define EP_RECREATE (-2) static pthread_key_t epoll_key; -static st_table *active; -static const int step = 64; /* unlikely to grow unless you're huge */ -static VALUE cEpoll_IO; static ID id_for_fd; +static VALUE cEpoll; static uint64_t now_ms(void) { @@ -47,27 +39,31 @@ static VALUE unpack_event_data(struct epoll_event *event) # endif #endif -struct rb_epoll { - int fd; - VALUE io; - VALUE marks; - VALUE flag_cache; - int flags; -}; - struct ep_per_thread { - struct rb_epoll *ep; + VALUE io; + int fd; int timeout; int maxevents; int capa; struct epoll_event events[FLEX_ARRAY]; }; -static struct ep_per_thread *ept_get(int maxevents) +/* this will raise if the IO is closed */ +static int ep_fd_check(struct ep_per_thread *ept) +{ + int save_errno = errno; + + ept->fd = rb_sp_fileno(ept->io); + errno = save_errno; + + return 1; +} + +static struct ep_per_thread *ept_get(VALUE self, int maxevents) { struct ep_per_thread *ept = pthread_getspecific(epoll_key); - int err; size_t size; + int err; if (ept && ept->capa >= maxevents) goto out; @@ -87,253 +83,72 @@ static struct ep_per_thread *ept_get(int maxevents) ept->capa = maxevents; out: ept->maxevents = maxevents; + ept->io = self; + ept->fd = rb_sp_fileno(ept->io); return ept; } -static struct rb_epoll *ep_get(VALUE self) -{ - struct rb_epoll *ep; - - Data_Get_Struct(self, struct rb_epoll, ep); - - return ep; -} - -static void gcmark(void *ptr) -{ - struct rb_epoll *ep = ptr; - - rb_gc_mark(ep->io); - rb_gc_mark(ep->marks); - rb_gc_mark(ep->flag_cache); -} - -static void gcfree(void *ptr) -{ - struct rb_epoll *ep = ptr; - - if (ep->fd >= 0) { - st_data_t key = ep->fd; - st_delete(active, &key, NULL); - } - if (NIL_P(ep->io) && ep->fd >= 0) { - /* can't raise during GC, and close() never fails in Linux */ - (void)close(ep->fd); - errno = 0; - } - /* let GC take care of the underlying IO object if there is one */ - - xfree(ep); -} - -static VALUE alloc(VALUE klass) -{ - struct rb_epoll *ep; - VALUE self; - - self = Data_Make_Struct(klass, struct rb_epoll, gcmark, gcfree, ep); - ep->fd = -1; - ep->io = Qnil; - ep->marks = Qnil; - ep->flag_cache = Qnil; - ep->flags = 0; - - return self; -} - -static void my_epoll_create(struct rb_epoll *ep) -{ - ep->fd = epoll_create1(ep->flags); - - if (ep->fd == -1) { - if (errno == EMFILE || errno == ENFILE || errno == ENOMEM) { - rb_gc(); - ep->fd = epoll_create1(ep->flags); - } - if (ep->fd == -1) - rb_sys_fail("epoll_create1"); - } - rb_update_max_fd(ep->fd); - st_insert(active, (st_data_t)ep->fd, (st_data_t)ep); - ep->marks = rb_ary_new(); - 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); - ep_fd_check(ep); - assert(TYPE(ep->marks) == T_ARRAY && "marks not initialized"); - assert(TYPE(ep->flag_cache) == T_ARRAY && "flag_cache not initialized"); -} - /* * call-seq: - * SleepyPenguin::Epoll.new([flags]) -> Epoll object + * SleepyPenguin::Epoll::IO.new(flags) -> Epoll::IO object * - * Creates a new Epoll object with an optional +flags+ argument. - * +flags+ may currently be +:CLOEXEC+ or +0+ (or +nil+). + * Creates a new Epoll::IO object with the given +flags+ argument. + * +flags+ may currently be +CLOEXEC+ or +0+. */ -static VALUE init(int argc, VALUE *argv, VALUE self) -{ - struct rb_epoll *ep = ep_get(self); - VALUE fl; - - rb_scan_args(argc, argv, "01", &fl); - ep->flags = rb_sp_get_flags(self, fl); - my_epoll_create(ep); - - return self; -} - -static VALUE ctl(VALUE self, VALUE io, VALUE flags, int op) +static VALUE s_new(VALUE klass, VALUE _flags) { - struct epoll_event event; - struct rb_epoll *ep = ep_get(self); - int fd = rb_sp_fileno(io); - int rv; + int flags = rb_sp_get_flags(klass, _flags); + int fd = epoll_create1(flags); + VALUE rv; - ep_check(ep); - event.events = rb_sp_get_uflags(self, flags); - pack_event_data(&event, io); - - rv = epoll_ctl(ep->fd, op, fd, &event); - if (rv == -1) { - if (errno == ENOMEM) { + if (fd < 0) { + if (errno == EMFILE || errno == ENFILE || errno == ENOMEM) { rb_gc(); - rv = epoll_ctl(ep->fd, op, fd, &event); + fd = epoll_create1(flags); } - if (rv == -1) - rb_sys_fail("epoll_ctl"); - } - switch (op) { - case EPOLL_CTL_ADD: - rb_ary_store(ep->marks, fd, io); - /* fall-through */ - case EPOLL_CTL_MOD: - flags = UINT2NUM(event.events); - rb_ary_store(ep->flag_cache, fd, flags); - break; - case EPOLL_CTL_DEL: - rb_ary_store(ep->marks, fd, Qnil); - rb_ary_store(ep->flag_cache, fd, Qnil); + if (fd == -1) + rb_sys_fail("epoll_create1"); } - return INT2NUM(rv); + rv = INT2FIX(fd); + return rb_call_super(1, &rv); } /* * call-seq: - * ep.set(io, flags) -> 0 + * epoll_io.epoll_ctl(op, io, events) -> nil * - * Used to avoid exceptions when your app is too lazy to check - * what state a descriptor is in, this sets the epoll descriptor - * to watch an +io+ with the given +flags+ + * Register, modify, or register a watch for a given +io+ for events. * - * +flags+ may be an array of symbols or an unsigned Integer bit mask: + * +op+ may be one of +EPOLL_CTL_ADD+, +EPOLL_CTL_MOD+, or +EPOLL_CTL_DEL+ + * +io+ is an IO object or one which proxies via the +to_io+ method. + * +events+ is an integer mask of events to watch for. * - * - flags = [ :IN, :ET ] - * - flags = SleepyPenguin::Epoll::IN | SleepyPenguin::Epoll::ET - * - * See constants in Epoll for more information. + * Returns nil on success. */ -static VALUE set(VALUE self, VALUE io, VALUE flags) +static VALUE epctl(VALUE self, VALUE _op, VALUE io, VALUE events) { struct epoll_event event; - struct rb_epoll *ep = ep_get(self); + int epfd = rb_sp_fileno(self); int fd = rb_sp_fileno(io); + int op = NUM2INT(_op); int rv; - VALUE cur_io = rb_ary_entry(ep->marks, fd); - ep_check(ep); - event.events = rb_sp_get_uflags(self, flags); + event.events = NUM2UINT(events); pack_event_data(&event, io); - 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; - -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; - } - } 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); - } - flags = UINT2NUM(event.events); - rb_ary_store(ep->flag_cache, fd, flags); - - return INT2NUM(rv); -} - -/* - * call-seq: - * epoll.delete(io) -> io or nil - * - * Stops an +io+ object from being monitored. This is like Epoll#del - * but returns +nil+ on ENOENT instead of raising an error. This is - * useful for apps that do not care to track the status of an - * epoll object itself. - */ -static VALUE delete(VALUE self, VALUE io) -{ - struct rb_epoll *ep = ep_get(self); - int fd = rb_sp_fileno(io); - int rv; - VALUE cur_io; - - ep_check(ep); - if (rb_sp_io_closed(io)) - goto out; - - cur_io = rb_ary_entry(ep->marks, fd); - if (NIL_P(cur_io) || rb_sp_io_closed(cur_io)) - return Qnil; - - rv = epoll_ctl(ep->fd, EPOLL_CTL_DEL, fd, NULL); - if (rv == -1) { - /* beware of IO.for_fd-created descriptors */ - if (errno == ENOENT || errno == EBADF) { - errno = 0; - io = Qnil; - } else { - rb_sys_fail("epoll_ctl - del"); + rv = epoll_ctl(epfd, op, fd, &event); + if (rv < 0) { + if (errno == ENOMEM) { + rb_gc(); + rv = epoll_ctl(epfd, op, fd, &event); } + if (rv < 0) + rb_sys_fail("epoll_ctl"); } -out: - rb_ary_store(ep->marks, fd, Qnil); - rb_ary_store(ep->flag_cache, fd, Qnil); - return io; + return Qnil; } static VALUE epwait_result(struct ep_per_thread *ept, int n) @@ -358,7 +173,7 @@ static int epoll_resume_p(uint64_t expire_at, struct ep_per_thread *ept) { uint64_t now; - ep_fd_check(ept->ep); + ep_fd_check(ept); /* may raise IOError */ if (errno != EINTR) return 0; @@ -373,8 +188,7 @@ static int epoll_resume_p(uint64_t expire_at, struct ep_per_thread *ept) static VALUE nogvl_wait(void *args) { struct ep_per_thread *ept = args; - int fd = ept->ep->fd; - int n = epoll_wait(fd, ept->events, ept->maxevents, ept->timeout); + int n = epoll_wait(ept->fd, ept->events, ept->maxevents, ept->timeout); return (VALUE)n; } @@ -385,7 +199,7 @@ static VALUE real_epwait(struct ep_per_thread *ept) uint64_t expire_at = ept->timeout > 0 ? now_ms() + ept->timeout : 0; do { - n = (int)rb_sp_fd_region(nogvl_wait, ept, ept->ep->fd); + n = (int)rb_sp_fd_region(nogvl_wait, ept, ept->fd); } while (n == -1 && epoll_resume_p(expire_at, ept)); return epwait_result(ept, n); @@ -396,11 +210,11 @@ static VALUE real_epwait(struct ep_per_thread *ept) /* * call-seq: - * epoll.wait([maxevents[, timeout]]) { |flags, io| ... } + * ep_io.epoll_wait([maxevents[, timeout]]) { |events, io| ... } * - * Calls epoll_wait(2) and yields Integer +flags+ and IO objects watched + * Calls epoll_wait(2) and yields Integer +events+ and IO objects watched * for. +maxevents+ is the maximum number of events to process at once, - * lower numbers may prevent starvation when used by Epoll#wait in multiple + * lower numbers may prevent starvation when used by epoll_wait in multiple * threads. Larger +maxevents+ reduces syscall overhead for * single-threaded applications. +maxevents+ defaults to 64 events. * +timeout+ is specified in milliseconds, +nil+ @@ -409,259 +223,17 @@ static VALUE real_epwait(struct ep_per_thread *ept) static VALUE epwait(int argc, VALUE *argv, VALUE self) { VALUE timeout, maxevents; - struct rb_epoll *ep = ep_get(self); struct ep_per_thread *ept; - ep_check(ep); rb_need_block(); rb_scan_args(argc, argv, "02", &maxevents, &timeout); - ept = ept_get(NIL_P(maxevents) ? 64 : NUM2INT(maxevents)); + + ept = ept_get(self, NIL_P(maxevents) ? 64 : NUM2INT(maxevents)); ept->timeout = NIL_P(timeout) ? -1 : NUM2INT(timeout); - ept->ep = ep; return real_epwait(ept); } -/* - * call-seq: - * epoll.add(io, flags) -> 0 - * - * Starts watching a given +io+ object with +flags+ which may be an Integer - * bitmask or Array representing arrays to watch for. Consider Epoll#set - * instead as it is easier to use. - */ -static VALUE add(VALUE self, VALUE io, VALUE flags) -{ - return ctl(self, io, flags, EPOLL_CTL_ADD); -} - -/* - * call-seq: - * epoll.del(io) -> 0 - * - * Disables an IO object from being watched. Consider Epoll#delete as - * it is easier to use. - */ -static VALUE del(VALUE self, VALUE io) -{ - return ctl(self, io, INT2FIX(0), EPOLL_CTL_DEL); -} - -/* - * call-seq: - * epoll.mod(io, flags) -> 0 - * - * Changes the watch for an existing IO object based on +flags+. - * Consider Epoll#set instead as it is easier to use. - */ -static VALUE mod(VALUE self, VALUE io, VALUE flags) -{ - return ctl(self, io, flags, EPOLL_CTL_MOD); -} - -/* - * call-seq: - * epoll.to_io -> Epoll::IO object - * - * Used to expose the given Epoll object as an Epoll::IO object for IO.select - * or IO#stat. This is unlikely to be useful directly, but is used internally - * by IO.select. - */ -static VALUE to_io(VALUE self) -{ - struct rb_epoll *ep = ep_get(self); - - ep_check(ep); - - if (NIL_P(ep->io)) - ep->io = rb_funcall(cEpoll_IO, id_for_fd, 1, INT2NUM(ep->fd)); - - return ep->io; -} - -/* - * call-seq: - * epoll.close -> nil - * - * Closes an existing Epoll object and returns memory back to the kernel. - * Raises IOError if object is already closed. - */ -static VALUE epclose(VALUE self) -{ - struct rb_epoll *ep = ep_get(self); - - if (ep->fd >= 0) { - st_data_t key = ep->fd; - st_delete(active, &key, NULL); - } - - if (NIL_P(ep->io)) { - ep_fd_check(ep); - - if (ep->fd == EP_RECREATE) { - ep->fd = -1; /* success */ - } else { - int err; - int fd = ep->fd; - - ep->fd = -1; - rb_thread_fd_close(fd); - err = close(fd); - if (err == -1) - rb_sys_fail("close"); - } - } else { - ep->fd = -1; - rb_io_close(ep->io); - } - - return Qnil; -} - -/* - * call-seq: - * epoll.closed? -> true or false - * - * Returns whether or not an Epoll object is closed. - */ -static VALUE epclosed(VALUE self) -{ - struct rb_epoll *ep = ep_get(self); - - return ep->fd == -1 ? Qtrue : Qfalse; -} - -static int cloexec_dup(struct rb_epoll *ep) -{ -#ifdef F_DUPFD_CLOEXEC - int flags = ep->flags & EPOLL_CLOEXEC ? F_DUPFD_CLOEXEC : F_DUPFD; - int fd = fcntl(ep->fd, flags, 0); -#else /* potentially racy on GVL-free systems: */ - int fd = dup(ep->fd); - if (fd >= 0) - (void)fcntl(fd, F_SETFD, FD_CLOEXEC); -#endif - return fd; -} - -/* - * call-seq: - * epoll.dup -> another Epoll object - * - * Duplicates an Epoll object and userspace buffers related to this library. - * Since SleepyPenguin 3.1.0, this is no longer needed for multi-threaded - * Epoll#wait. - */ -static VALUE init_copy(VALUE copy, VALUE orig) -{ - struct rb_epoll *a = ep_get(orig); - struct rb_epoll *b = ep_get(copy); - - assert(NIL_P(b->io) && "Ruby broken?"); - - 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) { - if (errno == ENFILE || errno == EMFILE) { - rb_gc(); - b->fd = cloexec_dup(a); - } - if (b->fd == -1) - rb_sys_fail("dup"); - } - st_insert(active, (st_data_t)b->fd, (st_data_t)b); - - return copy; -} - -/* occasionally it's still useful to lookup aliased IO objects - * based on for debugging */ -static int my_fileno(VALUE obj) -{ - if (T_FIXNUM == TYPE(obj)) - return FIX2INT(obj); - return rb_sp_fileno(obj); -} - -/* - * call-seq: - * epoll.io_for(io) -> object - * - * Returns the given IO object currently being watched for. Different - * IO objects may internally refer to the same process file descriptor. - * Mostly used for debugging. - */ -static VALUE io_for(VALUE self, VALUE obj) -{ - struct rb_epoll *ep = ep_get(self); - - return rb_ary_entry(ep->marks, my_fileno(obj)); -} - -/* - * call-seq: - * epoll.flags_for(io) -> Integer - * - * Returns the flags currently watched for in current Epoll object. - * Mostly used for debugging. - */ -static VALUE flags_for(VALUE self, VALUE obj) -{ - struct rb_epoll *ep = ep_get(self); - - return rb_ary_entry(ep->flag_cache, my_fileno(obj)); -} - -/* - * call-seq: - * epoll.include?(io) => true or false - * - * Returns whether or not a given IO is watched and prevented from being - * garbage-collected by the current Epoll object. This may include - * closed IO objects. - */ -static VALUE include_p(VALUE self, VALUE obj) -{ - struct rb_epoll *ep = ep_get(self); - - return NIL_P(rb_ary_entry(ep->marks, my_fileno(obj))) ? Qfalse : Qtrue; -} - -/* - * we close (or lose to GC) epoll descriptors at fork to avoid leakage - * and invalid objects being referenced later in the child - */ -static int ep_atfork(st_data_t key, st_data_t value, void *ignored) -{ - struct rb_epoll *ep = (struct rb_epoll *)value; - - if (NIL_P(ep->io)) { - if (ep->fd >= 0) - (void)close(ep->fd); - } else { - ep->io = Qnil; /* must let GC take care of it later :< */ - } - ep->fd = EP_RECREATE; - - return ST_CONTINUE; -} - -static void atfork_child(void) -{ - st_table *old = active; - - active = st_init_numtable(); - st_foreach(old, ep_atfork, (st_data_t)NULL); - st_free_table(old); -} - static void epoll_once(void) { int err = pthread_key_create(&epoll_key, free); @@ -670,19 +242,17 @@ static void epoll_once(void) errno = err; rb_sys_fail("pthread_key_create"); } +} - active = st_init_numtable(); - - if (pthread_atfork(NULL, NULL, atfork_child) != 0) { - rb_gc(); - if (pthread_atfork(NULL, NULL, atfork_child) != 0) - rb_memerror(); - } +/* :nodoc: */ +static VALUE event_flags(VALUE self, VALUE flags) +{ + return UINT2NUM(rb_sp_get_uflags(self, flags)); } void sleepy_penguin_init_epoll(void) { - VALUE mSleepyPenguin, cEpoll; + VALUE mSleepyPenguin, cEpoll_IO; static pthread_once_t once = PTHREAD_ONCE_INIT; int err = pthread_once(&once, epoll_once); @@ -708,6 +278,7 @@ void sleepy_penguin_init_epoll(void) * And then access classes via: * * - SP::Epoll + * - SP::Epoll::IO * - SP::EventFD * - SP::Inotify * - SP::TimerFD @@ -717,36 +288,36 @@ void sleepy_penguin_init_epoll(void) /* * Document-class: SleepyPenguin::Epoll * - * The Epoll class provides access to epoll(7) functionality in the - * Linux 2.6 kernel. It provides fork and GC-safety for Ruby - * objects stored within the IO object and may be passed as an - * argument to IO.select. + * The Epoll class provides high-level access to epoll(7) + * functionality in the Linux 2.6 and later kernels. It provides + * fork and GC-safety for Ruby objects stored within the IO object + * and may be passed as an argument to IO.select. */ cEpoll = rb_define_class_under(mSleepyPenguin, "Epoll", rb_cObject); /* * Document-class: SleepyPenguin::Epoll::IO * - * Epoll::IO is an internal class. Its only purpose is to be - * compatible with IO.select and related methods and should - * never be used directly, use Epoll instead. + * Epoll::IO is a low-level class. It does not provide fork nor + * GC-safety, so Ruby IO objects added via epoll_ctl must be retained + * by the application until IO#close is called. */ cEpoll_IO = rb_define_class_under(cEpoll, "IO", rb_cIO); - rb_define_method(cEpoll, "initialize", init, -1); - rb_define_method(cEpoll, "initialize_copy", init_copy, 1); - rb_define_alloc_func(cEpoll, alloc); - rb_define_method(cEpoll, "to_io", to_io, 0); - rb_define_method(cEpoll, "close", epclose, 0); - rb_define_method(cEpoll, "closed?", epclosed, 0); - rb_define_method(cEpoll, "add", add, 2); - rb_define_method(cEpoll, "mod", mod, 2); - rb_define_method(cEpoll, "del", del, 1); - rb_define_method(cEpoll, "delete", delete, 1); - rb_define_method(cEpoll, "io_for", io_for, 1); - rb_define_method(cEpoll, "flags_for", flags_for, 1); - rb_define_method(cEpoll, "include?", include_p, 1); - rb_define_method(cEpoll, "set", set, 2); - rb_define_method(cEpoll, "wait", epwait, -1); + rb_define_singleton_method(cEpoll_IO, "new", s_new, 1); + + rb_define_method(cEpoll_IO, "epoll_ctl", epctl, 3); + rb_define_method(cEpoll_IO, "epoll_wait", epwait, -1); + + rb_define_method(cEpoll, "__event_flags", event_flags, 1); + + /* registers an IO object via epoll_ctl */ + rb_define_const(cEpoll, "CTL_ADD", INT2NUM(EPOLL_CTL_ADD)); + + /* unregisters an IO object via epoll_ctl */ + rb_define_const(cEpoll, "CTL_DEL", INT2NUM(EPOLL_CTL_DEL)); + + /* modifies the registration of an IO object via epoll_ctl */ + rb_define_const(cEpoll, "CTL_MOD", INT2NUM(EPOLL_CTL_MOD)); /* specifies whether close-on-exec flag is set for Epoll.new */ rb_define_const(cEpoll, "CLOEXEC", INT2NUM(EPOLL_CLOEXEC)); diff --git a/ext/sleepy_penguin/epoll_green.h b/ext/sleepy_penguin/epoll_green.h index 276a545..e3414eb 100644 --- a/ext/sleepy_penguin/epoll_green.h +++ b/ext/sleepy_penguin/epoll_green.h @@ -26,9 +26,9 @@ static int safe_epoll_wait(struct ep_per_thread *ept) do { TRAP_BEG; - n = epoll_wait(ept->ep->fd, ept->events, ept->maxevents, 0); + n = epoll_wait(ept->fd, ept->events, ept->maxevents, 0); TRAP_END; - } while (n == -1 && errno == EINTR && ep_fd_check(ept->ep)); + } while (n == -1 && ep_fd_check(ept) && errno == EINTR); return n; } @@ -38,7 +38,7 @@ static int epwait_forever(struct ep_per_thread *ept) int n; do { - (void)rb_io_wait_readable(ept->ep->fd); + (void)rb_io_wait_readable(ept->fd); n = safe_epoll_wait(ept); } while (n == 0); @@ -55,7 +55,7 @@ static int epwait_timed(struct ep_per_thread *ept) for (;;) { struct timeval t0, now, diff; int n; - int fd = ept->ep->fd; + int fd = ept->fd; fd_set rfds; FD_ZERO(&rfds); diff --git a/lib/sleepy_penguin.rb b/lib/sleepy_penguin.rb index 3a189b1..c13eb0c 100644 --- a/lib/sleepy_penguin.rb +++ b/lib/sleepy_penguin.rb @@ -5,3 +5,4 @@ module SleepyPenguin SLEEPY_PENGUIN_VERSION = '3.1.0' end require 'sleepy_penguin_ext' +require 'sleepy_penguin/epoll' diff --git a/lib/sleepy_penguin/epoll.rb b/lib/sleepy_penguin/epoll.rb new file mode 100644 index 0000000..845dcf0 --- /dev/null +++ b/lib/sleepy_penguin/epoll.rb @@ -0,0 +1,228 @@ +class SleepyPenguin::Epoll + + # Epoll objects may be watched by IO.select and similar methods + attr_reader :to_io + + # call-seq: + # SleepyPenguin::Epoll.new([flags]) -> Epoll object + # + # Creates a new Epoll object with an optional +flags+ argument. + # +flags+ may currently be +:CLOEXEC+ or +0+ (or +nil+). + def initialize(create_flags = nil) + @to_io = SleepyPenguin::Epoll::IO.new(create_flags) + @events = [] + @marks = [] + @pid = $$ + @create_flags = create_flags + @copies = { @to_io => self } + end + + def __ep_reinit # :nodoc: + @events.clear + @marks.clear + @to_io = SleepyPenguin::Epoll::IO.new(@create_flags) + end + + # auto-reinitialize the Epoll object after forking + def __ep_check # :nodoc: + return if @pid == $$ + return if @to_io.closed? + objects = @copies.values + @copies.each_key { |epio| epio.close } + @copies.clear + __ep_reinit + objects.each do |obj| + io_dup = @to_io.dup + @copies[io_dup] = obj + end + @pid = $$ + end + + # Calls epoll_wait(2) and yields Integer +events+ and IO objects watched + # for. +maxevents+ is the maximum number of events to process at once, + # lower numbers may prevent starvation when used by epoll_wait in multiple + # threads. Larger +maxevents+ reduces syscall overhead for + # single-threaded applications. +maxevents+ defaults to 64 events. + # +timeout+ is specified in milliseconds, +nil+ + # (the default) meaning it will block and wait indefinitely. + def wait(maxevents = 64, timeout = nil) + __ep_check + @to_io.epoll_wait(maxevents, timeout) { |events, io| yield(events, io) } + end + + # Starts watching a given +io+ object with +events+ which may be an Integer + # bitmask or Array representing arrays to watch for. + def add(io, events) + __ep_check + fd = io.to_io.fileno + events = __event_flags(events) + @to_io.epoll_ctl(CTL_ADD, io, events) + @marks[fd] = io + @events[fd] = events + 0 + end + + # call-seq: + # ep.del(io) -> 0 + # + # Disables an IO object from being watched. + def del(io) + __ep_check + fd = io.to_io.fileno + rv = @to_io.epoll_ctl(CTL_DEL, io, 0) + @marks[fd] = @events[fd] = nil + rv + end + + # call-seq: + # ep.delete(io) -> io or nil + # + # This method is deprecated and will be removed in sleepy_penguin 4.x + # + # Stops an +io+ object from being monitored. This is like Epoll#del + # but returns +nil+ on ENOENT instead of raising an error. This is + # useful for apps that do not care to track the status of an + # epoll object itself. + # + # This method is deprecated and will be removed in sleepy_penguin 4.x + def delete(io) + __ep_check + fd = io.to_io.fileno + cur_io = @marks[fd] + return if nil == cur_io || cur_io.to_io.closed? + @marks[fd] = @events[fd] = nil + @to_io.epoll_ctl(CTL_DEL, io, 0) + io + rescue Errno::ENOENT, Errno::EBADF + end + + # call-seq: + # epoll.mod(io, flags) -> 0 + # + # Changes the watch for an existing IO object based on +events+. + # Returns zero on success, will raise SystemError on failure. + def mod(io, events) + __ep_check + fd = io.to_io.fileno + events = __event_flags(events) + rv = @to_io.epoll_ctl(CTL_MOD, io, events) + @marks[fd] = io + @events[fd] = events + rv + end + + # call-seq: + # ep.set(io, flags) -> 0 + # + # This method is deprecated and will be removed in sleepy_penguin 4.x + # + # Used to avoid exceptions when your app is too lazy to check + # what state a descriptor is in, this sets the epoll descriptor + # to watch an +io+ with the given +events+ + # + # +events+ may be an array of symbols or an unsigned Integer bit mask: + # + # - events = [ :IN, :ET ] + # - events = SleepyPenguin::Epoll::IN | SleepyPenguin::Epoll::ET + # + # See constants in Epoll for more information. + # + # This method is deprecated and will be removed in sleepy_penguin 4.x + def set(io, events) + __ep_check + fd = io.to_io.fileno + cur_io = @marks[fd] + if cur_io == io + cur_events = @events[fd] + return 0 if (cur_events & ONESHOT) == 0 && cur_events == events + begin + @to_io.epoll_ctl(CTL_MOD, io, events) + rescue Errno::ENOENT + warn "epoll flag cache failed (mod -> add)" + @to_io.epoll_ctl(CTL_ADD, io, events) + @marks[fd] = io + end + else + begin + @to_io.epoll_ctl(CTL_ADD, io, events) + rescue Errno::EEXIST + warn "epoll flag cache failed (add -> mod)" + @to_io.epoll_ctl(CTL_MOD, io, events) + end + @marks[fd] = io + end + @events[fd] = events + 0 + end + + # call-seq: + # ep.close -> nil + # + # Closes an existing Epoll object and returns memory back to the kernel. + # Raises IOError if object is already closed. + def close + __ep_check + @copies.delete(@to_io) + @to_io.close + end + + # call-seq: + # ep.closed? -> true or false + # + # Returns whether or not an Epoll object is closed. + def closed? + __ep_check + @to_io.closed? + end + + # we still support integer FDs for some debug functions + def __fileno(io) # :nodoc: + Integer === io ? io : io.to_io.fileno + end + + # call-seq: + # ep.io_for(io) -> object + # + # Returns the given IO object currently being watched for. Different + # IO objects may internally refer to the same process file descriptor. + # Mostly used for debugging. + def io_for(io) + __ep_check + @marks[__fileno(io)] + end + + # call-seq: + # epoll.events_for(io) -> Integer + # + # Returns the events currently watched for in current Epoll object. + # Mostly used for debugging. + def events_for(io) + __ep_check + @events[__fileno(io)] + end + + # backwards compatibility, to be removed in 4.x + alias flags_for events_for + + # call-seq: + # epoll.include?(io) -> true or false + # + # Returns whether or not a given IO is watched and prevented from being + # garbage-collected by the current Epoll object. This may include + # closed IO objects. + def include?(io) + __ep_check + @marks[__fileno(io)] ? true : nil + end + + def initialize_copy(src) # :nodoc: + __ep_check + rv = super + unless @to_io.closed? + @to_io = @to_io.dup + @copies[@to_io] = self + end + + rv + end +end diff --git a/test/test_epoll.rb b/test/test_epoll.rb index cd50cff..1a99dfd 100644 --- a/test/test_epoll.rb +++ b/test/test_epoll.rb @@ -48,6 +48,19 @@ def test_fork_safe assert_equal [[Epoll::IN, @rd]], tmp end + def test_dup_and_fork + epdup = @ep.dup + @ep.close + assert ! epdup.closed? + pid = fork do + exit(!epdup.closed? && @ep.closed?) + end + _, status = Process.waitpid2(pid) + assert status.success?, status.inspect + ensure + epdup.close + end + def test_after_fork_usability fork { @ep.add(@rd, Epoll::IN); exit!(0) } fork { @ep.set(@rd, Epoll::IN); exit!(0) } @@ -399,7 +412,7 @@ def test_flags_for_sym_ary def test_include? assert ! @ep.include?(@rd) @ep.add @rd, Epoll::IN - assert @ep.include?(@rd) + assert @ep.include?(@rd), @ep.instance_variable_get(:@marks).inspect assert @ep.include?(@rd.fileno) assert ! @ep.include?(@wr) assert ! @ep.include?(@wr.fileno) diff --git a/test/test_epoll_io.rb b/test/test_epoll_io.rb new file mode 100644 index 0000000..8aca155 --- /dev/null +++ b/test/test_epoll_io.rb @@ -0,0 +1,24 @@ +require 'test/unit' +require 'fcntl' +require 'socket' +require 'thread' +$-w = true +Thread.abort_on_exception = true +require 'sleepy_penguin' + +class TestEpollIO < Test::Unit::TestCase + include SleepyPenguin + RBX = defined?(RUBY_ENGINE) && (RUBY_ENGINE == 'rbx') + + def setup + @rd, @wr = IO.pipe + @epio = Epoll::IO.new(nil) + end + + def test_add_wait + @epio.epoll_ctl(Epoll::CTL_ADD, @wr, Epoll::OUT) + ev = [] + @epio.epoll_wait { |events, obj| ev << [ events, obj ] } + assert_equal([[Epoll::OUT, @wr]], ev) + end +end diff --git a/test/test_epoll_optimizations.rb b/test/test_epoll_optimizations.rb index bd77397..f5970fd 100644 --- a/test/test_epoll_optimizations.rb +++ b/test/test_epoll_optimizations.rb @@ -28,7 +28,7 @@ def test_set end assert_nil err lines = io.readlines; io.close - assert_equal 1, lines.grep(/^epoll_ctl/).size + assert_equal 1, lines.grep(/^epoll_ctl/).size, lines.inspect assert_match(/EPOLL_CTL_ADD/, lines.grep(/^epoll_ctl/)[0]) io, err = Strace.me { @ep.set(@wr, Epoll::OUT | Epoll::ONESHOT) } -- Eric Wong ^ permalink raw reply related [flat|nested] 9+ messages in thread
* [sleepy.penguin] [PATCH 4/6] epoll: implement thread-safety for mark/flag arrays 2013-04-11 4:17 [sleepy.penguin] [PATCH 0/6] epoll wrapper cleanups Eric Wong ` (2 preceding siblings ...) 2013-04-11 4:17 ` [sleepy.penguin] [PATCH 3/6] split Epoll and Epoll::IO, rewrite Epoll in Ruby Eric Wong @ 2013-04-11 4:17 ` Eric Wong 2013-04-12 21:18 ` Eric Wong 2013-04-11 4:17 ` [sleepy.penguin] [PATCH 5/6] epoll: cache alignment for per-thread structure Eric Wong 2013-04-11 4:17 ` [sleepy.penguin] [PATCH 6/6] avoid ENOMEM checking in common code paths Eric Wong 5 siblings, 1 reply; 9+ messages in thread From: Eric Wong @ 2013-04-11 4:17 UTC (permalink / raw) To: sleepy.penguin Concurrent modification of Arrays is thread-unsafe and must be protected by a Mutex. eventpoll objects inside the Linux kernel are similarly protected by a (kernel) mutex, and do not need additional locking. --- lib/sleepy_penguin/epoll.rb | 148 ++++++++++++++++++++++++++++---------------- 1 file changed, 93 insertions(+), 55 deletions(-) diff --git a/lib/sleepy_penguin/epoll.rb b/lib/sleepy_penguin/epoll.rb index 845dcf0..62664d6 100644 --- a/lib/sleepy_penguin/epoll.rb +++ b/lib/sleepy_penguin/epoll.rb @@ -1,3 +1,4 @@ +require 'thread' class SleepyPenguin::Epoll # Epoll objects may be watched by IO.select and similar methods @@ -10,6 +11,7 @@ class SleepyPenguin::Epoll # +flags+ may currently be +:CLOEXEC+ or +0+ (or +nil+). def initialize(create_flags = nil) @to_io = SleepyPenguin::Epoll::IO.new(create_flags) + @mtx = Mutex.new @events = [] @marks = [] @pid = $$ @@ -46,19 +48,34 @@ def __ep_check # :nodoc: # +timeout+ is specified in milliseconds, +nil+ # (the default) meaning it will block and wait indefinitely. def wait(maxevents = 64, timeout = nil) - __ep_check + # snapshot the marks so we do can sit this thread on epoll_wait while other + # threads may call epoll_ctl. People say RCU is a poor man's GC, but our + # (ab)use of GC here is inspired by RCU... + snapshot = @mtx.synchronize do + __ep_check + @marks.dup + end + + # we keep a snapshot of @marks around in case another thread closes + # the IO while it is being transferred to userspace. We release mtx + # so another thread may add events to us while we're sleeping. @to_io.epoll_wait(maxevents, timeout) { |events, io| yield(events, io) } + ensure + # hopefully Ruby does not optimize this array away... + snapshot[0] end # Starts watching a given +io+ object with +events+ which may be an Integer # bitmask or Array representing arrays to watch for. def add(io, events) - __ep_check fd = io.to_io.fileno events = __event_flags(events) + @mtx.synchronize do + __ep_check + @events[fd] = events + @marks[fd] = io + end @to_io.epoll_ctl(CTL_ADD, io, events) - @marks[fd] = io - @events[fd] = events 0 end @@ -67,11 +84,13 @@ def add(io, events) # # Disables an IO object from being watched. def del(io) - __ep_check fd = io.to_io.fileno - rv = @to_io.epoll_ctl(CTL_DEL, io, 0) - @marks[fd] = @events[fd] = nil - rv + @mtx.synchronize do + __ep_check + @events[fd] = @marks[fd] = nil + end + @to_io.epoll_ctl(CTL_DEL, io, 0) + 0 end # call-seq: @@ -86,11 +105,13 @@ def del(io) # # This method is deprecated and will be removed in sleepy_penguin 4.x def delete(io) - __ep_check fd = io.to_io.fileno - cur_io = @marks[fd] - return if nil == cur_io || cur_io.to_io.closed? - @marks[fd] = @events[fd] = nil + @mtx.synchronize do + __ep_check + cur_io = @marks[fd] + return if nil == cur_io || cur_io.to_io.closed? + @events[fd] = @marks[fd] = nil + end @to_io.epoll_ctl(CTL_DEL, io, 0) io rescue Errno::ENOENT, Errno::EBADF @@ -102,13 +123,14 @@ def delete(io) # Changes the watch for an existing IO object based on +events+. # Returns zero on success, will raise SystemError on failure. def mod(io, events) - __ep_check - fd = io.to_io.fileno events = __event_flags(events) - rv = @to_io.epoll_ctl(CTL_MOD, io, events) - @marks[fd] = io - @events[fd] = events - rv + fd = io.to_io.fileno + @mtx.synchronize do + __ep_check + @marks[fd] = io # may be a different object with same fd/file + @events[fd] = events + end + @to_io.epoll_ctl(CTL_MOD, io, events) end # call-seq: @@ -129,29 +151,31 @@ def mod(io, events) # # This method is deprecated and will be removed in sleepy_penguin 4.x def set(io, events) - __ep_check fd = io.to_io.fileno - cur_io = @marks[fd] - if cur_io == io - cur_events = @events[fd] - return 0 if (cur_events & ONESHOT) == 0 && cur_events == events - begin - @to_io.epoll_ctl(CTL_MOD, io, events) - rescue Errno::ENOENT - warn "epoll flag cache failed (mod -> add)" - @to_io.epoll_ctl(CTL_ADD, io, events) + @mtx.synchronize do + __ep_check + cur_io = @marks[fd] + if cur_io == io + cur_events = @events[fd] + return 0 if (cur_events & ONESHOT) == 0 && cur_events == events + begin + @to_io.epoll_ctl(CTL_MOD, io, events) + rescue Errno::ENOENT + warn "epoll flag cache failed (mod -> add)" + @to_io.epoll_ctl(CTL_ADD, io, events) + @marks[fd] = io + end + else + begin + @to_io.epoll_ctl(CTL_ADD, io, events) + rescue Errno::EEXIST + warn "epoll flag cache failed (add -> mod)" + @to_io.epoll_ctl(CTL_MOD, io, events) + end @marks[fd] = io end - else - begin - @to_io.epoll_ctl(CTL_ADD, io, events) - rescue Errno::EEXIST - warn "epoll flag cache failed (add -> mod)" - @to_io.epoll_ctl(CTL_MOD, io, events) - end - @marks[fd] = io + @events[fd] = events end - @events[fd] = events 0 end @@ -161,9 +185,11 @@ def set(io, events) # Closes an existing Epoll object and returns memory back to the kernel. # Raises IOError if object is already closed. def close - __ep_check - @copies.delete(@to_io) - @to_io.close + @mtx.synchronize do + __ep_check + @copies.delete(@to_io) + @to_io.close + end end # call-seq: @@ -171,8 +197,10 @@ def close # # Returns whether or not an Epoll object is closed. def closed? - __ep_check - @to_io.closed? + @mtx.synchronize do + __ep_check + @to_io.closed? + end end # we still support integer FDs for some debug functions @@ -187,8 +215,11 @@ def __fileno(io) # :nodoc: # IO objects may internally refer to the same process file descriptor. # Mostly used for debugging. def io_for(io) - __ep_check - @marks[__fileno(io)] + fd = __fileno(io) + @mtx.synchronize do + __ep_check + @marks[fd] + end end # call-seq: @@ -197,8 +228,11 @@ def io_for(io) # Returns the events currently watched for in current Epoll object. # Mostly used for debugging. def events_for(io) - __ep_check - @events[__fileno(io)] + fd = __fileno(io) + @mtx.synchronize do + __ep_check + @events[fd] + end end # backwards compatibility, to be removed in 4.x @@ -211,18 +245,22 @@ def events_for(io) # garbage-collected by the current Epoll object. This may include # closed IO objects. def include?(io) - __ep_check - @marks[__fileno(io)] ? true : nil + fd = __fileno(io) + @mtx.synchronize do + __ep_check + @marks[fd] ? true : nil + end end def initialize_copy(src) # :nodoc: - __ep_check - rv = super - unless @to_io.closed? - @to_io = @to_io.dup - @copies[@to_io] = self + @mtx.synchronize do + __ep_check + rv = super + unless @to_io.closed? + @to_io = @to_io.dup + @copies[@to_io] = self + end + rv end - - rv end end -- 1.8.2.279.g631bc94 ^ permalink raw reply related [flat|nested] 9+ messages in thread
* Re: [sleepy.penguin] [PATCH 4/6] epoll: implement thread-safety for mark/flag arrays 2013-04-11 4:17 ` [sleepy.penguin] [PATCH 4/6] epoll: implement thread-safety for mark/flag arrays Eric Wong @ 2013-04-12 21:18 ` Eric Wong 0 siblings, 0 replies; 9+ messages in thread From: Eric Wong @ 2013-04-12 21:18 UTC (permalink / raw) To: sleepy.penguin Eric Wong <normalperson@yhbt.net> wrote: > def add(io, events) > - __ep_check > fd = io.to_io.fileno > events = __event_flags(events) > + @mtx.synchronize do > + __ep_check > + @events[fd] = events > + @marks[fd] = io > + end > @to_io.epoll_ctl(CTL_ADD, io, events) > - @marks[fd] = io > - @events[fd] = events > 0 > end The above ordering is incorrect and the marks must be set after the epoll_ctl syscall in the presence of concurrent epoll_wait callers. Since epoll_ctl is uninterruptible, there is no real problem in holding @mtx across epoll_ctl. So the correct ordering should be something like this: @mtx.synchronize do __ep_check @to_io.epoll_ctl(CTL_ADD, io, events) @events[fd] = events @marks[fd] = io end Yes, lock nesting happens with our epoll_ctl call; but even inside the current 3.9 Linux kernel there is internal lock nesting (both epmutex and ep->mtx are held) for EPOLL_CTL_ADD/DEL operations. Ditto for the other epoll_ctl wrapper methods. --------------------------------8<--------------------------------------- Subject: [PATCH] epoll: implement thread-safety for mark/flag arrays Concurrent modification of Arrays is thread-unsafe and must be protected by a Mutex. eventpoll objects inside the Linux kernel are similarly protected by a (kernel) mutex, and do not need additional locking. --- lib/sleepy_penguin/epoll.rb | 150 +++++++++++++++++++++++++++----------------- 1 file changed, 93 insertions(+), 57 deletions(-) diff --git a/lib/sleepy_penguin/epoll.rb b/lib/sleepy_penguin/epoll.rb index 845dcf0..8d78e46 100644 --- a/lib/sleepy_penguin/epoll.rb +++ b/lib/sleepy_penguin/epoll.rb @@ -1,3 +1,4 @@ +require 'thread' class SleepyPenguin::Epoll # Epoll objects may be watched by IO.select and similar methods @@ -10,6 +11,7 @@ class SleepyPenguin::Epoll # +flags+ may currently be +:CLOEXEC+ or +0+ (or +nil+). def initialize(create_flags = nil) @to_io = SleepyPenguin::Epoll::IO.new(create_flags) + @mtx = Mutex.new @events = [] @marks = [] @pid = $$ @@ -46,19 +48,34 @@ def __ep_check # :nodoc: # +timeout+ is specified in milliseconds, +nil+ # (the default) meaning it will block and wait indefinitely. def wait(maxevents = 64, timeout = nil) - __ep_check + # snapshot the marks so we do can sit this thread on epoll_wait while other + # threads may call epoll_ctl. People say RCU is a poor man's GC, but our + # (ab)use of GC here is inspired by RCU... + snapshot = @mtx.synchronize do + __ep_check + @marks.dup + end + + # we keep a snapshot of @marks around in case another thread closes + # the IO while it is being transferred to userspace. We release mtx + # so another thread may add events to us while we're sleeping. @to_io.epoll_wait(maxevents, timeout) { |events, io| yield(events, io) } + ensure + # hopefully Ruby does not optimize this array away... + snapshot[0] end # Starts watching a given +io+ object with +events+ which may be an Integer # bitmask or Array representing arrays to watch for. def add(io, events) - __ep_check fd = io.to_io.fileno events = __event_flags(events) - @to_io.epoll_ctl(CTL_ADD, io, events) - @marks[fd] = io - @events[fd] = events + @mtx.synchronize do + __ep_check + @to_io.epoll_ctl(CTL_ADD, io, events) + @events[fd] = events + @marks[fd] = io + end 0 end @@ -67,11 +84,13 @@ def add(io, events) # # Disables an IO object from being watched. def del(io) - __ep_check fd = io.to_io.fileno - rv = @to_io.epoll_ctl(CTL_DEL, io, 0) - @marks[fd] = @events[fd] = nil - rv + @mtx.synchronize do + __ep_check + @to_io.epoll_ctl(CTL_DEL, io, 0) + @events[fd] = @marks[fd] = nil + end + 0 end # call-seq: @@ -86,12 +105,14 @@ def del(io) # # This method is deprecated and will be removed in sleepy_penguin 4.x def delete(io) - __ep_check fd = io.to_io.fileno - cur_io = @marks[fd] - return if nil == cur_io || cur_io.to_io.closed? - @marks[fd] = @events[fd] = nil - @to_io.epoll_ctl(CTL_DEL, io, 0) + @mtx.synchronize do + __ep_check + cur_io = @marks[fd] + return if nil == cur_io || cur_io.to_io.closed? + @to_io.epoll_ctl(CTL_DEL, io, 0) + @events[fd] = @marks[fd] = nil + end io rescue Errno::ENOENT, Errno::EBADF end @@ -102,13 +123,14 @@ def delete(io) # Changes the watch for an existing IO object based on +events+. # Returns zero on success, will raise SystemError on failure. def mod(io, events) - __ep_check - fd = io.to_io.fileno events = __event_flags(events) - rv = @to_io.epoll_ctl(CTL_MOD, io, events) - @marks[fd] = io - @events[fd] = events - rv + fd = io.to_io.fileno + @mtx.synchronize do + __ep_check + @to_io.epoll_ctl(CTL_MOD, io, events) + @marks[fd] = io # may be a different object with same fd/file + @events[fd] = events + end end # call-seq: @@ -129,29 +151,31 @@ def mod(io, events) # # This method is deprecated and will be removed in sleepy_penguin 4.x def set(io, events) - __ep_check fd = io.to_io.fileno - cur_io = @marks[fd] - if cur_io == io - cur_events = @events[fd] - return 0 if (cur_events & ONESHOT) == 0 && cur_events == events - begin - @to_io.epoll_ctl(CTL_MOD, io, events) - rescue Errno::ENOENT - warn "epoll flag cache failed (mod -> add)" - @to_io.epoll_ctl(CTL_ADD, io, events) + @mtx.synchronize do + __ep_check + cur_io = @marks[fd] + if cur_io == io + cur_events = @events[fd] + return 0 if (cur_events & ONESHOT) == 0 && cur_events == events + begin + @to_io.epoll_ctl(CTL_MOD, io, events) + rescue Errno::ENOENT + warn "epoll event cache failed (mod -> add)" + @to_io.epoll_ctl(CTL_ADD, io, events) + @marks[fd] = io + end + else + begin + @to_io.epoll_ctl(CTL_ADD, io, events) + rescue Errno::EEXIST + warn "epoll event cache failed (add -> mod)" + @to_io.epoll_ctl(CTL_MOD, io, events) + end @marks[fd] = io end - else - begin - @to_io.epoll_ctl(CTL_ADD, io, events) - rescue Errno::EEXIST - warn "epoll flag cache failed (add -> mod)" - @to_io.epoll_ctl(CTL_MOD, io, events) - end - @marks[fd] = io + @events[fd] = events end - @events[fd] = events 0 end @@ -161,9 +185,10 @@ def set(io, events) # Closes an existing Epoll object and returns memory back to the kernel. # Raises IOError if object is already closed. def close - __ep_check - @copies.delete(@to_io) - @to_io.close + @mtx.synchronize do + @copies.delete(@to_io) + @to_io.close + end end # call-seq: @@ -171,8 +196,9 @@ def close # # Returns whether or not an Epoll object is closed. def closed? - __ep_check - @to_io.closed? + @mtx.synchronize do + @to_io.closed? + end end # we still support integer FDs for some debug functions @@ -187,8 +213,11 @@ def __fileno(io) # :nodoc: # IO objects may internally refer to the same process file descriptor. # Mostly used for debugging. def io_for(io) - __ep_check - @marks[__fileno(io)] + fd = __fileno(io) + @mtx.synchronize do + __ep_check + @marks[fd] + end end # call-seq: @@ -197,8 +226,11 @@ def io_for(io) # Returns the events currently watched for in current Epoll object. # Mostly used for debugging. def events_for(io) - __ep_check - @events[__fileno(io)] + fd = __fileno(io) + @mtx.synchronize do + __ep_check + @events[fd] + end end # backwards compatibility, to be removed in 4.x @@ -211,18 +243,22 @@ def events_for(io) # garbage-collected by the current Epoll object. This may include # closed IO objects. def include?(io) - __ep_check - @marks[__fileno(io)] ? true : nil + fd = __fileno(io) + @mtx.synchronize do + __ep_check + @marks[fd] ? true : false + end end def initialize_copy(src) # :nodoc: - __ep_check - rv = super - unless @to_io.closed? - @to_io = @to_io.dup - @copies[@to_io] = self + @mtx.synchronize do + __ep_check + rv = super + unless @to_io.closed? + @to_io = @to_io.dup + @copies[@to_io] = self + end + rv end - - rv end end -- Eric Wong ^ permalink raw reply related [flat|nested] 9+ messages in thread
* [sleepy.penguin] [PATCH 5/6] epoll: cache alignment for per-thread structure 2013-04-11 4:17 [sleepy.penguin] [PATCH 0/6] epoll wrapper cleanups Eric Wong ` (3 preceding siblings ...) 2013-04-11 4:17 ` [sleepy.penguin] [PATCH 4/6] epoll: implement thread-safety for mark/flag arrays Eric Wong @ 2013-04-11 4:17 ` Eric Wong 2013-04-11 4:17 ` [sleepy.penguin] [PATCH 6/6] avoid ENOMEM checking in common code paths Eric Wong 5 siblings, 0 replies; 9+ messages in thread From: Eric Wong @ 2013-04-11 4:17 UTC (permalink / raw) To: sleepy.penguin This probably won't make a huge difference in Ruby, but perhaps one day the unnecessary dirtying of cache lines will affect performance (and we'll be ready when that day comes). While we're at it, remove usage of pthread* functions for thread-local variables. The __thread construct from GCC (and also implemented by clang) is much easier-to-use than the pthread_*specific API. --- ext/sleepy_penguin/epoll.c | 46 +++++++++++++++++++++------------------------- 1 file changed, 21 insertions(+), 25 deletions(-) diff --git a/ext/sleepy_penguin/epoll.c b/ext/sleepy_penguin/epoll.c index 64df698..70a77f6 100644 --- a/ext/sleepy_penguin/epoll.c +++ b/ext/sleepy_penguin/epoll.c @@ -1,14 +1,15 @@ #include "sleepy_penguin.h" #include <sys/epoll.h> -#include <pthread.h> +#include <unistd.h> #include <time.h> #include "missing_epoll.h" #include "missing_rb_thread_fd_close.h" #include "missing_rb_update_max_fd.h" +#define L1_CACHE_LINE_MAX 128 /* largest I've seen (Pentium 4) */ -static pthread_key_t epoll_key; static ID id_for_fd; static VALUE cEpoll; +static size_t l1_cache_line_size; static uint64_t now_ms(void) { @@ -56,9 +57,10 @@ static void ep_fd_check(struct ep_per_thread *ept) static struct ep_per_thread *ept_get(VALUE self, int maxevents) { - struct ep_per_thread *ept = pthread_getspecific(epoll_key); + static __thread struct ep_per_thread *ept; size_t size; int err; + void *ptr; if (ept && ept->capa >= maxevents) goto out; @@ -67,14 +69,12 @@ static struct ep_per_thread *ept_get(VALUE self, int maxevents) sizeof(struct epoll_event) * maxevents; free(ept); /* free(NULL) is POSIX and works on glibc */ - ept = malloc(size); - if (ept == NULL) - rb_memerror(); - err = pthread_setspecific(epoll_key, ept); - if (err != 0) { + err = posix_memalign(&ptr, l1_cache_line_size, size); + if (err) { errno = err; - rb_sys_fail("pthread_setspecific"); + rb_memerror(); } + ept = ptr; ept->capa = maxevents; out: ept->maxevents = maxevents; @@ -230,32 +230,28 @@ static VALUE epwait(int argc, VALUE *argv, VALUE self) return real_epwait(ept); } -static void epoll_once(void) -{ - int err = pthread_key_create(&epoll_key, free); - - if (err) { - errno = err; - rb_sys_fail("pthread_key_create"); - } -} - /* :nodoc: */ static VALUE event_flags(VALUE self, VALUE flags) { return UINT2NUM(rb_sp_get_uflags(self, flags)); } +static size_t l1_cache_line_size_detect(void) +{ +#ifdef _SC_LEVEL1_DCACHE_LINESIZE + long tmp = sysconf(_SC_LEVEL1_DCACHE_LINESIZE); + + if (tmp > 0 && tmp <= L1_CACHE_LINE_MAX) + return (size_t)tmp; +#endif /* _SC_LEVEL1_DCACHE_LINESIZE */ + return L1_CACHE_LINE_MAX; +} + void sleepy_penguin_init_epoll(void) { VALUE mSleepyPenguin, cEpoll_IO; - static pthread_once_t once = PTHREAD_ONCE_INIT; - int err = pthread_once(&once, epoll_once); - if (err) { - errno = err; - rb_sys_fail("pthread_once(.., epoll_once)"); - } + l1_cache_line_size = l1_cache_line_size_detect(); /* * Document-module: SleepyPenguin -- 1.8.2.279.g631bc94 ^ permalink raw reply related [flat|nested] 9+ messages in thread
* [sleepy.penguin] [PATCH 6/6] avoid ENOMEM checking in common code paths 2013-04-11 4:17 [sleepy.penguin] [PATCH 0/6] epoll wrapper cleanups Eric Wong ` (4 preceding siblings ...) 2013-04-11 4:17 ` [sleepy.penguin] [PATCH 5/6] epoll: cache alignment for per-thread structure Eric Wong @ 2013-04-11 4:17 ` Eric Wong 5 siblings, 0 replies; 9+ messages in thread From: Eric Wong @ 2013-04-11 4:17 UTC (permalink / raw) To: sleepy.penguin ENOMEM from syscalls such as inotify_add_watch and epoll_ctl are from the lack of kernel memory, so even a successful rb_gc() is unlikely to be able to reap memory taken from those slab caches. --- ext/sleepy_penguin/epoll.c | 10 ++-------- ext/sleepy_penguin/inotify.c | 11 +++-------- 2 files changed, 5 insertions(+), 16 deletions(-) diff --git a/ext/sleepy_penguin/epoll.c b/ext/sleepy_penguin/epoll.c index 70a77f6..179fe06 100644 --- a/ext/sleepy_penguin/epoll.c +++ b/ext/sleepy_penguin/epoll.c @@ -134,14 +134,8 @@ static VALUE epctl(VALUE self, VALUE _op, VALUE io, VALUE events) pack_event_data(&event, io); rv = epoll_ctl(epfd, op, fd, &event); - if (rv < 0) { - if (errno == ENOMEM) { - rb_gc(); - rv = epoll_ctl(epfd, op, fd, &event); - } - if (rv < 0) - rb_sys_fail("epoll_ctl"); - } + if (rv < 0) + rb_sys_fail("epoll_ctl"); return Qnil; } diff --git a/ext/sleepy_penguin/inotify.c b/ext/sleepy_penguin/inotify.c index f858dc4..4c606a4 100644 --- a/ext/sleepy_penguin/inotify.c +++ b/ext/sleepy_penguin/inotify.c @@ -89,14 +89,9 @@ static VALUE add_watch(VALUE self, VALUE path, VALUE vmask) uint32_t mask = rb_sp_get_uflags(self, vmask); int rc = inotify_add_watch(fd, pathname, mask); - if (rc == -1) { - if (errno == ENOMEM) { - rb_gc(); - rc = inotify_add_watch(fd, pathname, mask); - } - if (rc == -1) - rb_sys_fail("inotify_add_watch"); - } + if (rc == -1) + rb_sys_fail("inotify_add_watch"); + return UINT2NUM((uint32_t)rc); } -- 1.8.2.279.g631bc94 ^ permalink raw reply related [flat|nested] 9+ messages in thread
end of thread, other threads:[~2013-04-12 21:18 UTC | newest] Thread overview: 9+ messages (download: mbox.gz / follow: Atom feed) -- links below jump to the message on this page -- 2013-04-11 4:17 [sleepy.penguin] [PATCH 0/6] epoll wrapper cleanups Eric Wong 2013-04-11 4:17 ` [sleepy.penguin] [PATCH 1/6] test_epoll: fix timing error in test Eric Wong 2013-04-11 4:17 ` [sleepy.penguin] [PATCH 2/6] test_epoll: synchronize writes to the pipe array Eric Wong 2013-04-11 4:17 ` [sleepy.penguin] [PATCH 3/6] split Epoll and Epoll::IO, rewrite Epoll in Ruby Eric Wong 2013-04-12 20:38 ` Eric Wong 2013-04-11 4:17 ` [sleepy.penguin] [PATCH 4/6] epoll: implement thread-safety for mark/flag arrays Eric Wong 2013-04-12 21:18 ` Eric Wong 2013-04-11 4:17 ` [sleepy.penguin] [PATCH 5/6] epoll: cache alignment for per-thread structure Eric Wong 2013-04-11 4:17 ` [sleepy.penguin] [PATCH 6/6] avoid ENOMEM checking in common code paths Eric Wong
Code repositories for project(s) associated with this public inbox https://yhbt.net/sleepy_penguin.git/ This is a public inbox, see mirroring instructions for how to clone and mirror all data and code used for this inbox; as well as URLs for read-only IMAP folder(s) and NNTP newsgroup(s).