ext/io_splice/io_splice_ext.c

   1 #include "ruby.h"
   2 #ifdef HAVE_RUBY_IO_H
   3 #  include "ruby/io.h"
   4 #else
   5 #  include "rubyio.h"
   6 #endif
   7 #include <errno.h>
   8 #include <fcntl.h>
   9 #include <assert.h>
  10 #include <sys/uio.h>
  11 #include <limits.h>
  12 #include <alloca.h>
  13
  14 #if ! HAVE_RB_IO_T
  15 #  define rb_io_t OpenFile
  16 #endif
  17
  18 #ifdef GetReadFile
  19 #  define FPTR_TO_FD(fptr) (fileno(GetReadFile(fptr)))
  20 #else
  21 #  if !HAVE_RB_IO_T || (RUBY_VERSION_MAJOR == 1 && RUBY_VERSION_MINOR == 8)
  22 #    define FPTR_TO_FD(fptr) fileno(fptr->f)
  23 #  else
  24 #    define FPTR_TO_FD(fptr) fptr->fd
  25 #  endif
  26 #endif
  27
  28 static int my_fileno(VALUE io)
  29 {
  30         rb_io_t *fptr;
  31
  32         for (;;) {
  33                 switch (TYPE(io)) {
  34                 case T_FIXNUM: return NUM2INT(io);
  35                 case T_FILE: {
  36                         GetOpenFile(io, fptr);
  37                         return FPTR_TO_FD(fptr);
  38                 }
  39                 default:
  40                         io = rb_convert_type(io, T_FILE, "IO", "to_io");
  41                         /* retry */
  42                 }
  43         }
  44 }
  45 #ifndef HAVE_RB_THREAD_BLOCKING_REGION
  46 /* partial emulation of the 1.9 rb_thread_blocking_region under 1.8 */
  47 #  include <rubysig.h>
  48 #  define RUBY_UBF_IO ((rb_unblock_function_t *)-1)
  49 typedef void rb_unblock_function_t(void *);
  50 typedef VALUE rb_blocking_function_t(void *);
  51 static VALUE
  52 rb_thread_blocking_region(
  53         rb_blocking_function_t *fn, void *data1,
  54         rb_unblock_function_t *ubf, void *data2)
  55 {
  56         VALUE rv;
  57
  58         assert(RUBY_UBF_IO == ubf && "RUBY_UBF_IO required for emulation");
  59
  60         TRAP_BEG;
  61         rv = fn(data1);
  62         TRAP_END;
  63
  64         return rv;
  65 }
  66 #endif /* ! HAVE_RB_THREAD_BLOCKING_REGION */
  67
  68 #ifndef RSTRING_PTR
  69 #  define RSTRING_PTR(s) (RSTRING(s)->ptr)
  70 #endif
  71 #ifndef RSTRING_LEN
  72 #  define RSTRING_LEN(s) (RSTRING(s)->len)
  73 #endif
  74 #ifndef RARRAY_PTR
  75 #  define RARRAY_PTR(s) (RARRAY(s)->ptr)
  76 #endif
  77 #ifndef RARRAY_LEN
  78 #  define RARRAY_LEN(s) (RARRAY(s)->len)
  79 #endif
  80
  81 /*
  82  * Releases GVL only iff blocking I/O is used.
  83  * We'll trust programmers who use non-blocking I/O explicitly to
  84  * want the fastest possible performance without resorting to threads,
  85  * so releasing and them immediately reacquiring the GVL would be
  86  * a waste of time.
  87  */
  88 static VALUE nb_io_run(rb_blocking_function_t *fn, void *data, unsigned flags)
  89 {
  90         if (flags & SPLICE_F_NONBLOCK)
  91                 return fn(data);
  92         return rb_thread_blocking_region(fn, data, RUBY_UBF_IO, 0);
  93 }
  94
  95 struct splice_args {
  96         int fd_in;
  97         off_t *off_in;
  98         int fd_out;
  99         off_t *off_out;
 100         size_t len;
 101         unsigned flags;
 102 };
 103
 104 static VALUE nogvl_splice(void *ptr)
 105 {
 106         struct splice_args *a = ptr;
 107
 108         return (VALUE)splice(a->fd_in, a->off_in, a->fd_out, a->off_out,
 109                              a->len, a->flags);
 110 }
 111
 112 /*
 113  * call-seq:
 114  *    IO.splice(fd_in, off_in, fd_out, off_out, len, flags) => integer
 115  *
 116  * Splice +len+ bytes from/to a pipe.  Either +fd_in+ or +fd_out+
 117  * MUST be a pipe.  +fd_in+ and +fd_out+ may BOTH be pipes as of
 118  * Linux 2.6.31 or later.
 119  *
 120  * +off_in+ and +off_out+ if non-nil may be used to
 121  * specify an offset for the non-pipe file descriptor.
 122  *
 123  * +flags+ may be a bitmask of the following flags:
 124  *
 125  * IO::Splice::F_MOVE, IO::Splice::F_NONBLOCK, IO::Splice::F_MORE
 126  *
 127  * Returns the number of bytes spliced.
 128  * Raises EOFError when +fd_in+ has reached end of file.
 129  * Raises Errno::EAGAIN if the IO::Splice::F_NONBLOCK flag is set
 130  * and the pipe has no data to read from or space to write to.  May
 131  * also raise Errno::EAGAIN if the non-pipe descriptor has no data
 132  * to read from or space to write to.
 133  *
 134  *     rd, wr = (pipe = IO.pipe).map { |io| io.fileno }
 135  *     src_io, dst_io = File.open("/path/to/src"), File.open("/path/to/dst")
 136  *     src, dst = src_io.fileno, dst_io.fileno
 137  *
 138  *     nr = IO.splice(src, nil, wr, nil, IO::Splice::PIPE_CAPA, 0)
 139  *     IO.splice(rd, nil, dst, nil, nr, 0)
 140  *
 141  * As splice never exposes buffers to userspace, it will not take
 142  * into account userspace buffering done by Ruby or stdio.  It is
 143  * also not subject to encoding/decoding filters under Ruby 1.9.
 144  *
 145  * See manpage for full documentation:
 146  * http://kernel.org/doc/man-pages/online/pages/man2/splice.2.html
 147  */
 148 static VALUE my_splice(VALUE self,
 149         VALUE fd_in, VALUE off_in,
 150         VALUE fd_out, VALUE off_out,
 151         VALUE len, VALUE flags)
 152 {
 153         off_t i, o;
 154         long n;
 155         struct splice_args a = {
 156                 .off_in = NIL_P(off_in) ? NULL : (i = NUM2OFFT(off_in), &i),
 157                 .off_out = NIL_P(off_out) ? NULL : (o = NUM2OFFT(off_out), &o),
 158                 .fd_in = my_fileno(fd_in),
 159                 .fd_out = my_fileno(fd_out),
 160                 .len = (size_t)NUM2ULONG(len),
 161                 .flags = NUM2UINT(flags),
 162         };
 163
 164         n = (long)rb_thread_blocking_region(nogvl_splice, &a, RUBY_UBF_IO, 0);
 165         if (n == 0)
 166                 rb_eof_error();
 167         if (n < 0)
 168                 rb_sys_fail("splice");
 169         return LONG2NUM(n);
 170 }
 171
 172 struct tee_args {
 173         int fd_in;
 174         int fd_out;
 175         size_t len;
 176         unsigned flags;
 177 };
 178
 179 /* runs without GVL */
 180 static VALUE nogvl_tee(void *ptr)
 181 {
 182         struct tee_args *a = ptr;
 183
 184         return (VALUE)tee(a->fd_in, a->fd_out, a->len, a->flags);
 185 }
 186
 187 /*
 188  * call-seq:
 189  *   IO.tee(fd_in, fd_out, len, flags) => integer
 190  *
 191  * Copies up to +len+ bytes of data from +fd_in+ to +fd_out+.  +fd_in+
 192  * and +fd_out+ must both refer to pipe descriptors.  +fd_in+ and +fd_out+
 193  * may not be endpoints of the same pipe.
 194  *
 195  * +flags+ may be zero or IO::Splice::F_NONBLOCK
 196  * Other IO::Splice flags are currently unimplemented or have no effect.
 197  *
 198  * Returns the number of bytes duplicated if successful.
 199  * Raises EOFError when +fd_in+ is closed and emptied.
 200  * Raises Errno::EAGAIN when +fd_in+ is empty and/or +fd_out+ is full
 201  * and +flags+ contains IO::Splice::F_NONBLOCK
 202  *
 203  * See manpage for full documentation:
 204  * http://kernel.org/doc/man-pages/online/pages/man2/tee.2.html
 205  */
 206 static VALUE my_tee(VALUE self,
 207         VALUE fd_in, VALUE fd_out,
 208         VALUE len, VALUE flags)
 209 {
 210         long n;
 211         struct tee_args a = {
 212                 .fd_in = my_fileno(fd_in),
 213                 .fd_out = my_fileno(fd_out),
 214                 .len = (size_t)NUM2ULONG(len),
 215                 .flags = NUM2UINT(flags),
 216         };
 217
 218         n = (long)nb_io_run(nogvl_tee, &a, a.flags);
 219         if (n == 0)
 220                 rb_eof_error();
 221         if (n < 0)
 222                 rb_sys_fail("tee");
 223
 224         return LONG2NUM(n);
 225 }
 226
 227 struct vmsplice_args {
 228         int fd;
 229         struct iovec *iov;
 230         unsigned long nr_segs;
 231         unsigned flags;
 232 };
 233
 234 static VALUE nogvl_vmsplice(void *ptr)
 235 {
 236         struct vmsplice_args *a = ptr;
 237
 238         return (VALUE)vmsplice(a->fd, a->iov, a->nr_segs, a->flags);
 239 }
 240
 241 /* this can't be a function since we use alloca() */
 242 #define ARY2IOVEC(iov,iovcnt,expect,ary) \
 243 do { \
 244         VALUE *cur; \
 245         struct iovec *tmp; \
 246         long n; \
 247         cur = RARRAY_PTR(ary); \
 248         n = RARRAY_LEN(ary); \
 249         if (n > IOV_MAX) \
 250                 rb_raise(rb_eArgError, "array is larger than IOV_MAX"); \
 251         iov = tmp = alloca(sizeof(struct iovec) * n); \
 252         expect = 0; \
 253         iovcnt = n; \
 254         for (; --n >= 0; tmp++, cur++) { \
 255                 Check_Type(*cur, T_STRING); \
 256                 tmp->iov_base = RSTRING_PTR(*cur); \
 257                 tmp->iov_len = RSTRING_LEN(*cur); \
 258                 expect += tmp->iov_len; \
 259         } \
 260 } while (0)
 261
 262 static void advance_vmsplice_args(struct vmsplice_args *a, long n)
 263 {
 264         struct iovec *new_iov = a->iov;
 265         int i;
 266
 267         /* skip over iovecs we've already written completely */
 268         for (i = 0; i < a->nr_segs; i++, new_iov++) {
 269                 if (n == 0)
 270                         break;
 271                 /*
 272                  * partially written iov,
 273                  * modify and retry with current iovec in
 274                  * front
 275                  */
 276                 if (new_iov->iov_len > (size_t)n) {
 277                         VALUE base = (VALUE)new_iov->iov_base;
 278
 279                         new_iov->iov_len -= n;
 280                         new_iov->iov_base = (void *)(base + n);
 281                         break;
 282                 }
 283
 284                 n -= new_iov->iov_len;
 285         }
 286
 287         /* setup to retry without the already-written iovecs */
 288         a->nr_segs -= i;
 289         a->iov = new_iov;
 290 }
 291
 292 /*
 293  * call-seq:
 294  *   IO.vmsplice(fd, string_array, flags) => integer
 295  *   IO.vmsplice(fd, string, flags) => integer
 296  *
 297  * Transfers an array of strings into the pipe descriptor given by fd.
 298  * +fd+ must be the writable end of a pipe.
 299  *
 300  * This may allow the kernel to avoid data copies in some cases.
 301  * but is (probably) of limited usefulness in Ruby.
 302  *
 303  * See manpage for full documentation:
 304  * http://kernel.org/doc/man-pages/online/pages/man2/vmsplice.2.html
 305  */
 306 static VALUE my_vmsplice(VALUE self, VALUE fd, VALUE data, VALUE flags)
 307 {
 308         long rv = 0;
 309         ssize_t left;
 310         struct vmsplice_args a;
 311
 312         switch (TYPE(data)) {
 313         case T_STRING: {
 314                 struct iovec iov;
 315
 316                 iov.iov_base = RSTRING_PTR(data);
 317                 iov.iov_len = (size_t)(left = (ssize_t)RSTRING_LEN(data));
 318                 a.iov = &iov;
 319                 a.nr_segs = 1;
 320                 }
 321                 break;
 322         case T_ARRAY:
 323                 ARY2IOVEC(a.iov, a.nr_segs, left, data);
 324                 break;
 325         default:
 326                 rb_raise(rb_eTypeError, "wrong argument type %s "
 327                          "(expected a String or Array of strings)",
 328                          rb_obj_classname(data));
 329         }
 330         a.fd = my_fileno(fd);
 331         a.flags = NUM2UINT(flags);
 332
 333         for (;;) {
 334                 long n = (long)nb_io_run(nogvl_vmsplice, &a, a.flags);
 335
 336                 if (n < 0) {
 337                         if (errno == EAGAIN) {
 338                                 if (a.flags & SPLICE_F_NONBLOCK)
 339                                         rb_sys_fail("vmsplice");
 340                                 else if (rb_io_wait_writable(a.fd))
 341                                         continue;
 342                                 /* fall through on error */
 343                         }
 344                         /*
 345                          * unlikely to hit this case, return the
 346                          * already written bytes, we'll let the next
 347                          * write (or close) fail instead
 348                          */
 349                         if (rv > 0)
 350                                 break;
 351                         rb_sys_fail("vmsplice");
 352                 }
 353
 354                 rv += n;
 355                 left -= n;
 356                 if (left == 0)
 357                         break;
 358                 advance_vmsplice_args(&a, n);
 359         }
 360
 361         return LONG2NUM(rv);
 362 }
 363
 364 void Init_io_splice_ext(void)
 365 {
 366         VALUE mSplice = rb_define_module_under(rb_cIO, "Splice");
 367
 368         rb_define_singleton_method(rb_cIO, "splice", my_splice, 6);
 369         rb_define_singleton_method(rb_cIO, "tee", my_tee, 4);
 370         rb_define_singleton_method(rb_cIO, "vmsplice", my_vmsplice, 3);
 371
 372         /*
 373          * Attempt to move pages instead of copying.  This is only a hint
 374          * and support for it was removed in Linux 2.6.21.  It will be
 375          * re-added for FUSE devices only in Linux 2.6.35.
 376          */
 377         rb_define_const(mSplice, "F_MOVE", UINT2NUM(SPLICE_F_MOVE));
 378
 379         /*
 380          * Do not block on pipe I/O.  This flag only affects the pipe(s)
 381          * being spliced from/to and has no effect on the non-pipe
 382          * descriptor (which requires non-blocking operation to be set
 383          * explicitly).
 384          *
 385          * The non-blocking flag (O_NONBLOCK) on the pipe descriptors
 386          * themselves are ignored by this family of functions, and
 387          * using this flag is the only way to get non-blocking operation
 388          * out of them.
 389          */
 390         rb_define_const(mSplice, "F_NONBLOCK", UINT2NUM(SPLICE_F_NONBLOCK));
 391
 392         /*
 393          * Indicate that there may be more data coming into the outbound
 394          * descriptor.  This can allow the kernel to avoid sending partial
 395          * frames from sockets.  Currently only used with splice.
 396          */
 397         rb_define_const(mSplice, "F_MORE", UINT2NUM(SPLICE_F_MORE));
 398
 399         /*
 400          * Only usable by vmsplice.  This flag probably not useful in the
 401          * context of Ruby applications which cannot control alignment.
 402          */
 403         rb_define_const(mSplice, "F_GIFT", UINT2NUM(SPLICE_F_GIFT));
 404
 405 #ifdef F_GETPIPE_SZ
 406         /* :nodoc: */
 407         rb_define_const(mSplice, "F_GETPIPE_SZ", UINT2NUM(F_GETPIPE_SZ));
 408 #endif
 409 #ifdef F_SETPIPE_SZ
 410         /* :nodoc: */
 411         rb_define_const(mSplice, "F_SETPIPE_SZ", UINT2NUM(F_SETPIPE_SZ));
 412 #endif
 413
 414         /*
 415          * The maximum size of an atomic write to a pipe
 416          * POSIX requires this to be at least 512 bytes.
 417          * Under Linux, this is 4096 bytes.
 418          */
 419         rb_define_const(mSplice, "PIPE_BUF", UINT2NUM(PIPE_BUF));
 420 }