ext/sleepy_penguin/epoll.c

   1 #include "sleepy_penguin.h"
   2 #ifdef HAVE_SYS_EPOLL_H
   3 #include <sys/epoll.h>
   4 #include <unistd.h>
   5 #include <time.h>
   6 #include "missing_epoll.h"
   7 #include "missing_rb_thread_fd_close.h"
   8 #include "missing_rb_update_max_fd.h"
   9
  10 static ID id_for_fd;
  11 static VALUE cEpoll;
  12
  13 static uint64_t now_ms(void)
  14 {
  15         struct timespec now;
  16
  17         clock_gettime(CLOCK_MONOTONIC, &now);
  18
  19         return now.tv_sec * 1000 + (now.tv_nsec + 500000) / 1000000;
  20 }
  21
  22 static void pack_event_data(struct epoll_event *event, VALUE obj)
  23 {
  24         event->data.ptr = (void *)obj;
  25 }
  26
  27 static VALUE unpack_event_data(struct epoll_event *event)
  28 {
  29         return (VALUE)event->data.ptr;
  30 }
  31
  32 struct ep_per_thread {
  33         VALUE io;
  34         int fd;
  35         int timeout;
  36         int maxevents;
  37         int capa;
  38         struct epoll_event events[FLEX_ARRAY];
  39 };
  40
  41 /* this will raise if the IO is closed */
  42 static int ep_fd_check(struct ep_per_thread *ept)
  43 {
  44         int save_errno = errno;
  45
  46         ept->fd = rb_sp_fileno(ept->io);
  47         errno = save_errno;
  48
  49         return 1;
  50 }
  51
  52 static struct ep_per_thread *ept_get(VALUE self, int maxevents)
  53 {
  54         static __thread struct ep_per_thread *ept;
  55         size_t size;
  56         int err;
  57         void *ptr;
  58
  59         /* error check here to prevent OOM from posix_memalign */
  60         if (maxevents <= 0) {
  61                 errno = EINVAL;
  62                 rb_sys_fail("epoll_wait maxevents <= 0");
  63         }
  64
  65         if (ept && ept->capa >= maxevents)
  66                 goto out;
  67
  68         size = sizeof(struct ep_per_thread) +
  69                sizeof(struct epoll_event) * maxevents;
  70
  71         free(ept); /* free(NULL) is POSIX and works on glibc */
  72         err = posix_memalign(&ptr, rb_sp_l1_cache_line_size, size);
  73         if (err) {
  74                 errno = err;
  75                 rb_memerror();
  76         }
  77         ept = ptr;
  78         ept->capa = maxevents;
  79 out:
  80         ept->maxevents = maxevents;
  81         ept->io = self;
  82         ept->fd = rb_sp_fileno(ept->io);
  83
  84         return ept;
  85 }
  86
  87 /*
  88  * call-seq:
  89  *      SleepyPenguin::Epoll::IO.new(flags)     -> Epoll::IO object
  90  *
  91  * Creates a new Epoll::IO object with the given +flags+ argument.
  92  * +flags+ may currently be +CLOEXEC+ or +0+.
  93  */
  94 static VALUE s_new(VALUE klass, VALUE _flags)
  95 {
  96         int default_flags = RB_SP_CLOEXEC(EPOLL_CLOEXEC);
  97         int flags = rb_sp_get_flags(klass, _flags, default_flags);
  98         int fd = epoll_create1(flags);
  99         VALUE rv;
 100
 101         if (fd < 0) {
 102                 if (errno == EMFILE || errno == ENFILE || errno == ENOMEM) {
 103                         rb_gc();
 104                         fd = epoll_create1(flags);
 105                 }
 106                 if (fd < 0)
 107                         rb_sys_fail("epoll_create1");
 108         }
 109
 110         rv = INT2FIX(fd);
 111         return rb_call_super(1, &rv);
 112 }
 113
 114 /*
 115  * call-seq:
 116  *      epoll_io.epoll_ctl(op, io, events)      -> nil
 117  *
 118  * Register, modify, or register a watch for a given +io+ for events.
 119  *
 120  * +op+ may be one of +EPOLL_CTL_ADD+, +EPOLL_CTL_MOD+, or +EPOLL_CTL_DEL+
 121  * +io+ is an IO object or one which proxies via the +to_io+ method.
 122  * +events+ is an integer mask of events to watch for.
 123  *
 124  * Returns nil on success.
 125  */
 126 static VALUE epctl(VALUE self, VALUE _op, VALUE io, VALUE events)
 127 {
 128         struct epoll_event event;
 129         int epfd = rb_sp_fileno(self);
 130         int fd = rb_sp_fileno(io);
 131         int op = NUM2INT(_op);
 132         int rv;
 133
 134         event.events = NUM2UINT(events);
 135         pack_event_data(&event, io);
 136
 137         rv = epoll_ctl(epfd, op, fd, &event);
 138         if (rv < 0)
 139                 rb_sys_fail("epoll_ctl");
 140
 141         return Qnil;
 142 }
 143
 144 static VALUE epwait_result(struct ep_per_thread *ept, int n)
 145 {
 146         int i;
 147         struct epoll_event *epoll_event = ept->events;
 148         VALUE obj_events, obj;
 149
 150         if (n < 0)
 151                 rb_sys_fail("epoll_wait");
 152
 153         for (i = n; --i >= 0; epoll_event++) {
 154                 obj_events = UINT2NUM(epoll_event->events);
 155                 obj = unpack_event_data(epoll_event);
 156                 rb_yield_values(2, obj_events, obj);
 157         }
 158
 159         return INT2NUM(n);
 160 }
 161
 162 static int epoll_resume_p(uint64_t expire_at, struct ep_per_thread *ept)
 163 {
 164         uint64_t now;
 165
 166         ep_fd_check(ept); /* may raise IOError */
 167
 168         if (errno != EINTR)
 169                 return 0;
 170         if (ept->timeout < 0)
 171                 return 1;
 172         now = now_ms();
 173         ept->timeout = now > expire_at ? 0 : (int)(expire_at - now);
 174         return 1;
 175 }
 176
 177 static VALUE nogvl_wait(void *args)
 178 {
 179         struct ep_per_thread *ept = args;
 180         int n = epoll_wait(ept->fd, ept->events, ept->maxevents, ept->timeout);
 181
 182         return (VALUE)n;
 183 }
 184
 185 static VALUE real_epwait(struct ep_per_thread *ept)
 186 {
 187         long n;
 188         uint64_t expire_at = ept->timeout > 0 ? now_ms() + ept->timeout : 0;
 189
 190         do {
 191                 n = (long)rb_sp_fd_region(nogvl_wait, ept, ept->fd);
 192         } while (n < 0 && epoll_resume_p(expire_at, ept));
 193
 194         return epwait_result(ept, (int)n);
 195 }
 196
 197 /*
 198  * call-seq:
 199  *      ep_io.epoll_wait([maxevents[, timeout]]) { |events, io| ... }
 200  *
 201  * Calls epoll_wait(2) and yields Integer +events+ and IO objects watched
 202  * for.  +maxevents+ is the maximum number of events to process at once,
 203  * lower numbers may prevent starvation when used by epoll_wait in multiple
 204  * threads.  Larger +maxevents+ reduces syscall overhead for
 205  * single-threaded applications. +maxevents+ defaults to 64 events.
 206  * +timeout+ is specified in milliseconds, +nil+
 207  * (the default) meaning it will block and wait indefinitely.
 208  */
 209 static VALUE epwait(int argc, VALUE *argv, VALUE self)
 210 {
 211         VALUE timeout, maxevents;
 212         struct ep_per_thread *ept;
 213
 214         rb_need_block();
 215         rb_scan_args(argc, argv, "02", &maxevents, &timeout);
 216
 217         ept = ept_get(self, NIL_P(maxevents) ? 64 : NUM2INT(maxevents));
 218         ept->timeout = NIL_P(timeout) ? -1 : NUM2INT(timeout);
 219
 220         return real_epwait(ept);
 221 }
 222
 223 /* :nodoc: */
 224 static VALUE event_flags(VALUE self, VALUE flags)
 225 {
 226         return UINT2NUM(rb_sp_get_uflags(self, flags));
 227 }
 228
 229 void sleepy_penguin_init_epoll(void)
 230 {
 231         VALUE mSleepyPenguin, cEpoll_IO;
 232
 233         /*
 234          * Document-module: SleepyPenguin
 235          *
 236          *      require "sleepy_penguin"
 237          *      include SleepyPenguin
 238          *
 239          * The SleepyPenguin namespace includes the Epoll, Inotify,
 240          * TimerFD, EventFD classes in its top level and no other constants.
 241          *
 242          * If you are uncomfortable including SleepyPenguin, you may also
 243          * use the "SP" alias if it doesn't conflict with existing code:
 244          *
 245          *      require "sleepy_penguin/sp"
 246          *
 247          * And then access classes via:
 248          *
 249          * - SP::Epoll
 250          * - SP::Epoll::IO
 251          * - SP::EventFD
 252          * - SP::Inotify
 253          * - SP::TimerFD
 254          */
 255         mSleepyPenguin = rb_define_module("SleepyPenguin");
 256
 257         /*
 258          * Document-class: SleepyPenguin::Epoll
 259          *
 260          * The Epoll class provides high-level access to epoll(7)
 261          * functionality in the Linux 2.6 and later kernels.  It provides
 262          * fork and GC-safety for Ruby objects stored within the IO object
 263          * and may be passed as an argument to IO.select.
 264          */
 265         cEpoll = rb_define_class_under(mSleepyPenguin, "Epoll", rb_cObject);
 266
 267         /*
 268          * Document-class: SleepyPenguin::Epoll::IO
 269          *
 270          * Epoll::IO is a low-level class.  It does not provide fork nor
 271          * GC-safety, so Ruby IO objects added via epoll_ctl must be retained
 272          * by the application until IO#close is called.
 273          */
 274         cEpoll_IO = rb_define_class_under(cEpoll, "IO", rb_cIO);
 275         rb_define_singleton_method(cEpoll_IO, "new", s_new, 1);
 276
 277         rb_define_method(cEpoll_IO, "epoll_ctl", epctl, 3);
 278         rb_define_method(cEpoll_IO, "epoll_wait", epwait, -1);
 279
 280         rb_define_method(cEpoll, "__event_flags", event_flags, 1);
 281
 282         /* registers an IO object via epoll_ctl */
 283         rb_define_const(cEpoll, "CTL_ADD", INT2NUM(EPOLL_CTL_ADD));
 284
 285         /* unregisters an IO object via epoll_ctl */
 286         rb_define_const(cEpoll, "CTL_DEL", INT2NUM(EPOLL_CTL_DEL));
 287
 288         /* modifies the registration of an IO object via epoll_ctl */
 289         rb_define_const(cEpoll, "CTL_MOD", INT2NUM(EPOLL_CTL_MOD));
 290
 291         /* specifies whether close-on-exec flag is set for Epoll.new */
 292         rb_define_const(cEpoll, "CLOEXEC", INT2NUM(EPOLL_CLOEXEC));
 293
 294         /* watch for read/recv operations */
 295         rb_define_const(cEpoll, "IN", UINT2NUM(EPOLLIN));
 296
 297         /* watch for write/send operations */
 298         rb_define_const(cEpoll, "OUT", UINT2NUM(EPOLLOUT));
 299
 300 #ifdef EPOLLRDHUP
 301         /*
 302          * Watch a specified io for shutdown(SHUT_WR) on the remote-end.
 303          * Available since Linux 2.6.17.
 304          */
 305         rb_define_const(cEpoll, "RDHUP", UINT2NUM(EPOLLRDHUP));
 306 #endif
 307
 308 #ifdef EPOLLWAKEUP
 309         /*
 310          * This prevents system suspend while event is ready.
 311          * This requires the caller to have the CAP_BLOCK_SUSPEND capability
 312          * Available since Linux 3.5
 313          */
 314         rb_define_const(cEpoll, "WAKEUP", UINT2NUM(EPOLLWAKEUP));
 315 #endif
 316
 317         /* watch for urgent read(2) data */
 318         rb_define_const(cEpoll, "PRI", UINT2NUM(EPOLLPRI));
 319
 320         /*
 321          * watch for errors, there is no need to specify this,
 322          * it is always monitored when an IO is watched
 323          */
 324         rb_define_const(cEpoll, "ERR", UINT2NUM(EPOLLERR));
 325
 326         /*
 327          * watch for hangups, there is no need to specify this,
 328          * it is always monitored when an IO is watched
 329          */
 330         rb_define_const(cEpoll, "HUP", UINT2NUM(EPOLLHUP));
 331
 332         /* notifications are only Edge Triggered, see epoll(7) */
 333         rb_define_const(cEpoll, "ET", UINT2NUM((uint32_t)EPOLLET));
 334
 335         /* unwatch the descriptor once any event has fired */
 336         rb_define_const(cEpoll, "ONESHOT", UINT2NUM(EPOLLONESHOT));
 337
 338         id_for_fd = rb_intern("for_fd");
 339
 340         if (RB_SP_GREEN_THREAD)
 341                 rb_require("sleepy_penguin/epoll/io");
 342
 343         /* the high-level interface is implemented in Ruby: */
 344         rb_require("sleepy_penguin/epoll");
 345 }
 346 #endif /* HAVE_SYS_EPOLL_H */