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