ext/sleepy_penguin/epoll.c

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