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