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 #include <sys/utsname.h>
  14
  15 static VALUE sym_EAGAIN;
  16
  17 #ifndef F_LINUX_SPECIFIC_BASE
  18 #  define F_LINUX_SPECIFIC_BASE 1024
  19 #endif
  20
  21 #ifndef F_GETPIPE_SZ
  22 #  define F_SETPIPE_SZ    (F_LINUX_SPECIFIC_BASE + 7)
  23 #  define F_GETPIPE_SZ    (F_LINUX_SPECIFIC_BASE + 8)
  24 #endif
  25
  26 #if ! HAVE_RB_IO_T
  27 #  define rb_io_t OpenFile
  28 #endif
  29
  30 #ifdef GetReadFile
  31 #  define FPTR_TO_FD(fptr) (fileno(GetReadFile(fptr)))
  32 #else
  33 #  if !HAVE_RB_IO_T || (RUBY_VERSION_MAJOR == 1 && RUBY_VERSION_MINOR == 8)
  34 #    define FPTR_TO_FD(fptr) fileno(fptr->f)
  35 #  else
  36 #    define FPTR_TO_FD(fptr) fptr->fd
  37 #  endif
  38 #endif
  39
  40 static int my_fileno(VALUE io)
  41 {
  42         rb_io_t *fptr;
  43
  44         for (;;) {
  45                 switch (TYPE(io)) {
  46                 case T_FIXNUM: return NUM2INT(io);
  47                 case T_FILE: {
  48                         GetOpenFile(io, fptr);
  49                         return FPTR_TO_FD(fptr);
  50                 }
  51                 default:
  52                         io = rb_convert_type(io, T_FILE, "IO", "to_io");
  53                         /* retry */
  54                 }
  55         }
  56 }
  57 #ifndef HAVE_RB_THREAD_BLOCKING_REGION
  58 /* partial emulation of the 1.9 rb_thread_blocking_region under 1.8 */
  59 #  include <rubysig.h>
  60 #  define RUBY_UBF_IO ((rb_unblock_function_t *)-1)
  61 typedef void rb_unblock_function_t(void *);
  62 typedef VALUE rb_blocking_function_t(void *);
  63 static VALUE
  64 rb_thread_blocking_region(
  65         rb_blocking_function_t *fn, void *data1,
  66         rb_unblock_function_t *ubf, void *data2)
  67 {
  68         VALUE rv;
  69
  70         assert(RUBY_UBF_IO == ubf && "RUBY_UBF_IO required for emulation");
  71
  72         TRAP_BEG;
  73         rv = fn(data1);
  74         TRAP_END;
  75
  76         return rv;
  77 }
  78 #endif /* ! HAVE_RB_THREAD_BLOCKING_REGION */
  79
  80 #ifndef RSTRING_PTR
  81 #  define RSTRING_PTR(s) (RSTRING(s)->ptr)
  82 #endif
  83 #ifndef RSTRING_LEN
  84 #  define RSTRING_LEN(s) (RSTRING(s)->len)
  85 #endif
  86 #ifndef RARRAY_PTR
  87 #  define RARRAY_PTR(s) (RARRAY(s)->ptr)
  88 #endif
  89 #ifndef RARRAY_LEN
  90 #  define RARRAY_LEN(s) (RARRAY(s)->len)
  91 #endif
  92
  93 static VALUE io_run(rb_blocking_function_t *fn, void *data)
  94 {
  95         return rb_thread_blocking_region(fn, data, RUBY_UBF_IO, 0);
  96 }
  97
  98 /*
  99  * Releases GVL only iff blocking I/O is used.
 100  * Only use this if all file descriptors in data are pipes.
 101  * We'll trust programmers who use non-blocking I/O explicitly to
 102  * want the fastest possible performance without resorting to threads,
 103  * so releasing and them immediately reacquiring the GVL would be
 104  * a waste of time.
 105  */
 106 static VALUE nb_io_run(rb_blocking_function_t *fn, void *data, unsigned flags)
 107 {
 108         if (flags & SPLICE_F_NONBLOCK)
 109                 return fn(data);
 110         return io_run(fn, data);
 111 }
 112
 113 struct splice_args {
 114         int fd_in;
 115         off_t *off_in;
 116         int fd_out;
 117         off_t *off_out;
 118         size_t len;
 119         unsigned flags;
 120 };
 121
 122 static VALUE nogvl_splice(void *ptr)
 123 {
 124         struct splice_args *a = ptr;
 125
 126         return (VALUE)splice(a->fd_in, a->off_in, a->fd_out, a->off_out,
 127                              a->len, a->flags);
 128 }
 129
 130 static long do_splice(int argc, VALUE *argv, unsigned dflags)
 131 {
 132         off_t i, o;
 133         VALUE fd_in, off_in, fd_out, off_out, len, flags;
 134         struct splice_args a;
 135
 136         rb_scan_args(argc, argv, "51",
 137                      &fd_in, &off_in, &fd_out, &off_out, &len, &flags);
 138
 139         a.off_in = NIL_P(off_in) ? NULL : (i = NUM2OFFT(off_in), &i);
 140         a.off_out = NIL_P(off_out) ? NULL : (o = NUM2OFFT(off_out), &o);
 141         a.fd_in = my_fileno(fd_in);
 142         a.fd_out = my_fileno(fd_out);
 143         a.len = (size_t)NUM2ULONG(len);
 144         a.flags = NIL_P(flags) ? dflags : NUM2UINT(flags) | dflags;
 145
 146         return (long)io_run(nogvl_splice, &a);
 147 }
 148
 149 /*
 150  * call-seq:
 151  *    IO.splice(fd_in, off_in, fd_out, off_out, len, flags) => integer
 152  *
 153  * Splice +len+ bytes from/to a pipe.  Either +fd_in+ or +fd_out+
 154  * MUST be a pipe.  +fd_in+ and +fd_out+ may BOTH be pipes as of
 155  * Linux 2.6.31 or later.
 156  *
 157  * +off_in+ and +off_out+ if non-nil may be used to
 158  * specify an offset for the non-pipe file descriptor.
 159  *
 160  * +flags+ may be a bitmask of the following flags:
 161  *
 162  * IO::Splice::F_MOVE, IO::Splice::F_NONBLOCK, IO::Splice::F_MORE
 163  *
 164  * Returns the number of bytes spliced.
 165  * Raises EOFError when +fd_in+ has reached end of file.
 166  * Raises Errno::EAGAIN if the IO::Splice::F_NONBLOCK flag is set
 167  * and the pipe has no data to read from or space to write to.  May
 168  * also raise Errno::EAGAIN if the non-pipe descriptor has no data
 169  * to read from or space to write to.
 170  *
 171  *     rd, wr = (pipe = IO.pipe).map { |io| io.fileno }
 172  *     src_io, dst_io = File.open("/path/to/src"), File.open("/path/to/dst")
 173  *     src, dst = src_io.fileno, dst_io.fileno
 174  *
 175  *     nr = IO.splice(src, nil, wr, nil, IO::Splice::PIPE_CAPA, 0)
 176  *     IO.splice(rd, nil, dst, nil, nr, 0)
 177  *
 178  * As splice never exposes buffers to userspace, it will not take
 179  * into account userspace buffering done by Ruby or stdio.  It is
 180  * also not subject to encoding/decoding filters under Ruby 1.9.
 181  *
 182  * See manpage for full documentation:
 183  * http://kernel.org/doc/man-pages/online/pages/man2/splice.2.html
 184  */
 185 static VALUE my_splice(int argc, VALUE *argv, VALUE self)
 186 {
 187         long n = do_splice(argc, argv, 0);
 188
 189         if (n == 0)
 190                 rb_eof_error();
 191         if (n < 0)
 192                 rb_sys_fail("splice");
 193         return LONG2NUM(n);
 194 }
 195
 196 static VALUE trysplice(int argc, VALUE *argv, VALUE self)
 197 {
 198         long n = do_splice(argc, argv, SPLICE_F_NONBLOCK);
 199
 200         if (n == 0)
 201                 return Qnil;
 202         if (n < 0) {
 203                 if (errno == EAGAIN)
 204                         return sym_EAGAIN;
 205                 rb_sys_fail("splice");
 206         }
 207         return LONG2NUM(n);
 208 }
 209
 210 struct tee_args {
 211         int fd_in;
 212         int fd_out;
 213         size_t len;
 214         unsigned flags;
 215 };
 216
 217 /* runs without GVL */
 218 static VALUE nogvl_tee(void *ptr)
 219 {
 220         struct tee_args *a = ptr;
 221
 222         return (VALUE)tee(a->fd_in, a->fd_out, a->len, a->flags);
 223 }
 224
 225 /*
 226  * call-seq:
 227  *   IO.tee(fd_in, fd_out, len, flags) => integer
 228  *
 229  * Copies up to +len+ bytes of data from +fd_in+ to +fd_out+.  +fd_in+
 230  * and +fd_out+ must both refer to pipe descriptors.  +fd_in+ and +fd_out+
 231  * may not be endpoints of the same pipe.
 232  *
 233  * +flags+ may be zero or IO::Splice::F_NONBLOCK
 234  * Other IO::Splice flags are currently unimplemented or have no effect.
 235  *
 236  * Returns the number of bytes duplicated if successful.
 237  * Raises EOFError when +fd_in+ is closed and emptied.
 238  * Raises Errno::EAGAIN when +fd_in+ is empty and/or +fd_out+ is full
 239  * and +flags+ contains IO::Splice::F_NONBLOCK
 240  *
 241  * See manpage for full documentation:
 242  * http://kernel.org/doc/man-pages/online/pages/man2/tee.2.html
 243  */
 244 static VALUE my_tee(VALUE self,
 245         VALUE fd_in, VALUE fd_out,
 246         VALUE len, VALUE flags)
 247 {
 248         long n;
 249         struct tee_args a = {
 250                 .fd_in = my_fileno(fd_in),
 251                 .fd_out = my_fileno(fd_out),
 252                 .len = (size_t)NUM2ULONG(len),
 253                 .flags = NUM2UINT(flags),
 254         };
 255
 256         n = (long)nb_io_run(nogvl_tee, &a, a.flags);
 257         if (n == 0)
 258                 rb_eof_error();
 259         if (n < 0)
 260                 rb_sys_fail("tee");
 261
 262         return LONG2NUM(n);
 263 }
 264
 265 struct vmsplice_args {
 266         int fd;
 267         struct iovec *iov;
 268         unsigned long nr_segs;
 269         unsigned flags;
 270 };
 271
 272 static VALUE nogvl_vmsplice(void *ptr)
 273 {
 274         struct vmsplice_args *a = ptr;
 275
 276         return (VALUE)vmsplice(a->fd, a->iov, a->nr_segs, a->flags);
 277 }
 278
 279 /* this can't be a function since we use alloca() */
 280 #define ARY2IOVEC(iov,iovcnt,expect,ary) \
 281 do { \
 282         VALUE *cur; \
 283         struct iovec *tmp; \
 284         long n; \
 285         cur = RARRAY_PTR(ary); \
 286         n = RARRAY_LEN(ary); \
 287         if (n > IOV_MAX) \
 288                 rb_raise(rb_eArgError, "array is larger than IOV_MAX"); \
 289         iov = tmp = alloca(sizeof(struct iovec) * n); \
 290         expect = 0; \
 291         iovcnt = n; \
 292         for (; --n >= 0; tmp++, cur++) { \
 293                 Check_Type(*cur, T_STRING); \
 294                 tmp->iov_base = RSTRING_PTR(*cur); \
 295                 tmp->iov_len = RSTRING_LEN(*cur); \
 296                 expect += tmp->iov_len; \
 297         } \
 298 } while (0)
 299
 300 static void advance_vmsplice_args(struct vmsplice_args *a, long n)
 301 {
 302         struct iovec *new_iov = a->iov;
 303         unsigned long i;
 304
 305         /* skip over iovecs we've already written completely */
 306         for (i = 0; i < a->nr_segs; i++, new_iov++) {
 307                 if (n == 0)
 308                         break;
 309                 /*
 310                  * partially written iov,
 311                  * modify and retry with current iovec in
 312                  * front
 313                  */
 314                 if (new_iov->iov_len > (size_t)n) {
 315                         VALUE base = (VALUE)new_iov->iov_base;
 316
 317                         new_iov->iov_len -= n;
 318                         new_iov->iov_base = (void *)(base + n);
 319                         break;
 320                 }
 321
 322                 n -= new_iov->iov_len;
 323         }
 324
 325         /* setup to retry without the already-written iovecs */
 326         a->nr_segs -= i;
 327         a->iov = new_iov;
 328 }
 329
 330 /*
 331  * call-seq:
 332  *   IO.vmsplice(fd, string_array, flags) => integer
 333  *   IO.vmsplice(fd, string, flags) => integer
 334  *
 335  * Transfers an array of strings into the pipe descriptor given by fd.
 336  * +fd+ must be the writable end of a pipe.
 337  *
 338  * This may allow the kernel to avoid data copies in some cases.
 339  * but is (probably) of limited usefulness in Ruby.
 340  *
 341  * See manpage for full documentation:
 342  * http://kernel.org/doc/man-pages/online/pages/man2/vmsplice.2.html
 343  */
 344 static VALUE my_vmsplice(VALUE self, VALUE fd, VALUE data, VALUE flags)
 345 {
 346         long rv = 0;
 347         ssize_t left;
 348         struct vmsplice_args a;
 349
 350         switch (TYPE(data)) {
 351         case T_STRING: {
 352                 struct iovec iov;
 353
 354                 iov.iov_base = RSTRING_PTR(data);
 355                 iov.iov_len = (size_t)(left = (ssize_t)RSTRING_LEN(data));
 356                 a.iov = &iov;
 357                 a.nr_segs = 1;
 358                 }
 359                 break;
 360         case T_ARRAY:
 361                 ARY2IOVEC(a.iov, a.nr_segs, left, data);
 362                 break;
 363         default:
 364                 rb_raise(rb_eTypeError, "wrong argument type %s "
 365                          "(expected a String or Array of strings)",
 366                          rb_obj_classname(data));
 367         }
 368         a.fd = my_fileno(fd);
 369         a.flags = NUM2UINT(flags);
 370
 371         for (;;) {
 372                 long n = (long)nb_io_run(nogvl_vmsplice, &a, a.flags);
 373
 374                 if (n < 0) {
 375                         if (errno == EAGAIN) {
 376                                 if (a.flags & SPLICE_F_NONBLOCK)
 377                                         rb_sys_fail("vmsplice");
 378                                 else if (rb_io_wait_writable(a.fd))
 379                                         continue;
 380                                 /* fall through on error */
 381                         }
 382                         /*
 383                          * unlikely to hit this case, return the
 384                          * already written bytes, we'll let the next
 385                          * write (or close) fail instead
 386                          */
 387                         if (rv > 0)
 388                                 break;
 389                         rb_sys_fail("vmsplice");
 390                 }
 391
 392                 rv += n;
 393                 left -= n;
 394                 if (left == 0)
 395                         break;
 396                 advance_vmsplice_args(&a, n);
 397         }
 398
 399         return LONG2NUM(rv);
 400 }
 401
 402 /*
 403  * call-seq:
 404  *   reader, writer = IO.pipe
 405  *   reader.pipe_size => integer
 406  *
 407  * Returns the pipe capacity of the underlying pipe in bytes.  The
 408  * default capacity is 65536 bytes since Linux 2.6.11, and 4096 bytes
 409  * in previous kernels.
 410  *
 411  * Since the pipe is a circular buffer in the same kernel, the size
 412  * of the reader is exactly the same as the size of the writer.
 413  *
 414  * This method is only exposed on Linux 2.6.35 or later.
 415  */
 416 static VALUE pipe_size(VALUE self)
 417 {
 418         int size = fcntl(my_fileno(self),  F_GETPIPE_SZ);
 419
 420         if (size < 0)
 421                 rb_sys_fail("fcntl(F_GETPIPE_SZ)");
 422
 423         return INT2NUM(size);
 424 }
 425
 426 /*
 427  * call-seq:
 428  *   reader, writer = IO.pipe
 429  *   reader.pipe_size = integer
 430  *
 431  * Sets and returns the pipe capacity of the underlying pipe in bytes.
 432  *
 433  * This MUST be a power-of-two, or Errno::EINVAL will be raised.
 434  * Linux will silently increase this to be equal to the page size
 435  * (4096 bytes on most architectures) if the specified value is
 436  * less than the size of a page.
 437  *
 438  * For users without CAP_SYS_RESOURCE, this raises Errno::EPERM when
 439  * attempting to specify a value greater than the value in
 440  * /proc/sys/fs/pipe-max-size.
 441  *
 442  * Since the pipe is a circular buffer in the same kernel, the size
 443  * of the reader is exactly the same as the size of the writer.
 444  *
 445  * Raises Errno::EBUSY if the assigned value is less than
 446  * the currently filled portion of the pipe.
 447  *
 448  * This method is only exposed on Linux 2.6.35 or later.
 449  */
 450 static VALUE set_pipe_size(VALUE self, VALUE size)
 451 {
 452         int fd = my_fileno(self);
 453         int bytes = NUM2INT(size);
 454         int rv = fcntl(fd, F_SETPIPE_SZ, bytes);
 455
 456         if (rv < 0) {
 457                 if (errno == ENOMEM) {
 458                         rb_gc();
 459                         rv = fcntl(fd, F_SETPIPE_SZ, bytes);
 460                 }
 461                 if (rv < 0)
 462                         rb_sys_fail("fcntl(F_SETPIPE_SZ)");
 463         }
 464
 465         return size;
 466 }
 467
 468 void Init_io_splice_ext(void)
 469 {
 470         VALUE mSplice = rb_define_module_under(rb_cIO, "Splice");
 471         struct utsname utsname;
 472
 473         rb_define_singleton_method(rb_cIO, "splice", my_splice, -1);
 474         rb_define_singleton_method(rb_cIO, "trysplice", trysplice, -1);
 475         rb_define_singleton_method(rb_cIO, "tee", my_tee, 4);
 476         rb_define_singleton_method(rb_cIO, "vmsplice", my_vmsplice, 3);
 477
 478         /*
 479          * Attempt to move pages instead of copying.  This is only a hint
 480          * and support for it was removed in Linux 2.6.21.  It will be
 481          * re-added for FUSE filesystems only in Linux 2.6.35.
 482          */
 483         rb_define_const(mSplice, "F_MOVE", UINT2NUM(SPLICE_F_MOVE));
 484
 485         /*
 486          * Do not block on pipe I/O.  This flag only affects the pipe(s)
 487          * being spliced from/to and has no effect on the non-pipe
 488          * descriptor (which requires non-blocking operation to be set
 489          * explicitly).
 490          *
 491          * The non-blocking flag (O_NONBLOCK) on the pipe descriptors
 492          * themselves are ignored by this family of functions, and
 493          * using this flag is the only way to get non-blocking operation
 494          * out of them.
 495          */
 496         rb_define_const(mSplice, "F_NONBLOCK", UINT2NUM(SPLICE_F_NONBLOCK));
 497
 498         /*
 499          * Indicate that there may be more data coming into the outbound
 500          * descriptor.  This can allow the kernel to avoid sending partial
 501          * frames from sockets.  Currently only used with splice.
 502          */
 503         rb_define_const(mSplice, "F_MORE", UINT2NUM(SPLICE_F_MORE));
 504
 505         /*
 506          * Only usable by vmsplice.  This flag probably not useful in the
 507          * context of Ruby applications which cannot control alignment.
 508          */
 509         rb_define_const(mSplice, "F_GIFT", UINT2NUM(SPLICE_F_GIFT));
 510
 511         /*
 512          * The maximum size of an atomic write to a pipe
 513          * POSIX requires this to be at least 512 bytes.
 514          * Under Linux, this is 4096 bytes.
 515          */
 516         rb_define_const(mSplice, "PIPE_BUF", UINT2NUM(PIPE_BUF));
 517
 518         if (uname(&utsname) == -1)
 519                 rb_sys_fail("uname");
 520
 521         /* includes 2.6.35-rc[1-6] */
 522         if (strcmp(utsname.release, "2.6.35") >= 0) {
 523                 rb_define_method(rb_cIO, "pipe_size", pipe_size, 0);
 524                 rb_define_method(rb_cIO, "pipe_size=", set_pipe_size, 1);
 525
 526                 /*
 527                  * fcntl() command constant used to return the size of a pipe.
 528                  * This constant is only defined when running Linux 2.6.35
 529                  * or later.  For convenience, use IO#pipe_size instead.
 530                  */
 531                 rb_define_const(mSplice, "F_GETPIPE_SZ",
 532                                 UINT2NUM(F_GETPIPE_SZ));
 533
 534                 /*
 535                  * fcntl() command constant used to set the size of a pipe.
 536                  * This constant is only defined when running Linux 2.6.35
 537                  * or later.  For convenience, use IO#pipe_size= instead.
 538                  */
 539                 rb_define_const(mSplice, "F_SETPIPE_SZ",
 540                                 UINT2NUM(F_SETPIPE_SZ));
 541         }
 542
 543         sym_EAGAIN = ID2SYM(rb_intern("EAGAIN"));
 544 }