ext/kgio/read_write.c

   1 /* ref: rubinius b2811f260de16d1e972462e27852470364608de5 */
   2 #define RSTRING_MODIFIED 1
   3 #include "kgio.h"
   4 #include "my_fileno.h"
   5 #include "nonblock.h"
   6 #ifdef HAVE_WRITEV
   7 #  include <sys/uio.h>
   8 #  define USE_WRITEV 1
   9 #else
  10 #  define USE_WRITEV 0
  11 static ssize_t assert_writev(int fd, void* iov, int len)
  12 {
  13         assert(0 && "you should not try to call writev");
  14         return -1;
  15 }
  16 #  define writev assert_writev
  17 #endif
  18 static VALUE sym_wait_readable, sym_wait_writable;
  19 static VALUE eErrno_EPIPE, eErrno_ECONNRESET;
  20 static ID id_set_backtrace;
  21 #ifndef HAVE_RB_STR_SUBSEQ
  22 #define rb_str_subseq rb_str_substr
  23 #endif
  24
  25 #ifndef HAVE_RB_ARY_SUBSEQ
  26 static inline VALUE my_ary_subseq(VALUE ary, long idx, long len)
  27 {
  28         VALUE args[2] = {LONG2FIX(idx), LONG2FIX(len)};
  29         return rb_ary_aref(2, args, ary);
  30 }
  31 #define rb_ary_subseq my_ary_subseq
  32 #endif
  33
  34 /*
  35  * we know MSG_DONTWAIT works properly on all stream sockets under Linux
  36  * we can define this macro for other platforms as people care and
  37  * notice.
  38  */
  39 #if defined(__linux__) && ! defined(USE_MSG_DONTWAIT)
  40 #  define USE_MSG_DONTWAIT
  41 static const int peek_flags = MSG_DONTWAIT|MSG_PEEK;
  42
  43 /* we don't need these variants, we call kgio_autopush_send/recv directly */
  44 static inline void kgio_autopush_read(VALUE io) { }
  45 static inline void kgio_autopush_write(VALUE io) { }
  46
  47 #else
  48 static const int peek_flags = MSG_PEEK;
  49 static inline void kgio_autopush_read(VALUE io) { kgio_autopush_recv(io); }
  50 static inline void kgio_autopush_write(VALUE io) { kgio_autopush_send(io); }
  51 #endif
  52
  53 NORETURN(static void raise_empty_bt(VALUE, const char *));
  54 NORETURN(static void my_eof_error(void));
  55 NORETURN(static void wr_sys_fail(const char *));
  56 NORETURN(static void rd_sys_fail(const char *));
  57
  58 static void raise_empty_bt(VALUE err, const char *msg)
  59 {
  60         VALUE exc = rb_exc_new2(err, msg);
  61         VALUE bt = rb_ary_new();
  62
  63         rb_funcall(exc, id_set_backtrace, 1, bt);
  64         rb_exc_raise(exc);
  65 }
  66
  67 static void my_eof_error(void)
  68 {
  69         raise_empty_bt(rb_eEOFError, "end of file reached");
  70 }
  71
  72 static void wr_sys_fail(const char *msg)
  73 {
  74         switch (errno) {
  75         case EPIPE:
  76                 errno = 0;
  77                 raise_empty_bt(eErrno_EPIPE, msg);
  78         case ECONNRESET:
  79                 errno = 0;
  80                 raise_empty_bt(eErrno_ECONNRESET, msg);
  81         }
  82         rb_sys_fail(msg);
  83 }
  84
  85 static void rd_sys_fail(const char *msg)
  86 {
  87         if (errno == ECONNRESET) {
  88                 errno = 0;
  89                 raise_empty_bt(eErrno_ECONNRESET, msg);
  90         }
  91         rb_sys_fail(msg);
  92 }
  93
  94 static void prepare_read(struct io_args *a, int argc, VALUE *argv, VALUE io)
  95 {
  96         VALUE length;
  97
  98         a->io = io;
  99         a->fd = my_fileno(io);
 100         rb_scan_args(argc, argv, "11", &length, &a->buf);
 101         a->len = NUM2LONG(length);
 102         if (NIL_P(a->buf)) {
 103                 a->buf = rb_str_new(NULL, a->len);
 104         } else {
 105                 StringValue(a->buf);
 106                 rb_str_modify(a->buf);
 107                 rb_str_resize(a->buf, a->len);
 108         }
 109         a->ptr = RSTRING_PTR(a->buf);
 110 }
 111
 112 static int read_check(struct io_args *a, long n, const char *msg, int io_wait)
 113 {
 114         if (n < 0) {
 115                 if (errno == EINTR) {
 116                         a->fd = my_fileno(a->io);
 117                         return -1;
 118                 }
 119                 rb_str_set_len(a->buf, 0);
 120                 if (errno == EAGAIN) {
 121                         if (io_wait) {
 122                                 (void)kgio_call_wait_readable(a->io);
 123
 124                                 /* buf may be modified in other thread/fiber */
 125                                 rb_str_modify(a->buf);
 126                                 rb_str_resize(a->buf, a->len);
 127                                 a->ptr = RSTRING_PTR(a->buf);
 128                                 return -1;
 129                         } else {
 130                                 a->buf = sym_wait_readable;
 131                                 return 0;
 132                         }
 133                 }
 134                 rd_sys_fail(msg);
 135         }
 136         rb_str_set_len(a->buf, n);
 137         if (n == 0)
 138                 a->buf = Qnil;
 139         return 0;
 140 }
 141
 142 static VALUE my_read(int io_wait, int argc, VALUE *argv, VALUE io)
 143 {
 144         struct io_args a;
 145         long n;
 146
 147         prepare_read(&a, argc, argv, io);
 148         kgio_autopush_read(io);
 149
 150         if (a.len > 0) {
 151                 set_nonblocking(a.fd);
 152 retry:
 153                 n = (long)read(a.fd, a.ptr, a.len);
 154                 if (read_check(&a, n, "read", io_wait) != 0)
 155                         goto retry;
 156         }
 157         return a.buf;
 158 }
 159
 160 /*
 161  * call-seq:
 162  *
 163  *      io.kgio_read(maxlen)           ->  buffer
 164  *      io.kgio_read(maxlen, buffer)   ->  buffer
 165  *
 166  * Reads at most maxlen bytes from the stream socket.  Returns with a
 167  * newly allocated buffer, or may reuse an existing buffer if supplied.
 168  *
 169  * This may block and call any method defined to +kgio_wait_readable+
 170  * for the class.
 171  *
 172  * Returns nil on EOF.
 173  *
 174  * This behaves like read(2) and IO#readpartial, NOT fread(3) or
 175  * IO#read which possess read-in-full behavior.
 176  */
 177 static VALUE kgio_read(int argc, VALUE *argv, VALUE io)
 178 {
 179         return my_read(1, argc, argv, io);
 180 }
 181
 182 /*
 183  * Same as Kgio::PipeMethods#kgio_read, except EOFError is raised
 184  * on EOF without a backtrace.  This method is intended as a
 185  * drop-in replacement for places where IO#readpartial is used, and
 186  * may be aliased as such.
 187  */
 188 static VALUE kgio_read_bang(int argc, VALUE *argv, VALUE io)
 189 {
 190         VALUE rv = my_read(1, argc, argv, io);
 191
 192         if (NIL_P(rv)) my_eof_error();
 193         return rv;
 194 }
 195
 196 /*
 197  * call-seq:
 198  *
 199  *      io.kgio_tryread(maxlen)           ->  buffer
 200  *      io.kgio_tryread(maxlen, buffer)   ->  buffer
 201  *
 202  * Reads at most maxlen bytes from the stream socket.  Returns with a
 203  * newly allocated buffer, or may reuse an existing buffer if supplied.
 204  *
 205  * Returns nil on EOF.
 206  *
 207  * Returns :wait_readable if EAGAIN is encountered.
 208  */
 209 static VALUE kgio_tryread(int argc, VALUE *argv, VALUE io)
 210 {
 211         return my_read(0, argc, argv, io);
 212 }
 213
 214 #ifdef USE_MSG_DONTWAIT
 215 static VALUE my_recv(int io_wait, int argc, VALUE *argv, VALUE io)
 216 {
 217         struct io_args a;
 218         long n;
 219
 220         prepare_read(&a, argc, argv, io);
 221         kgio_autopush_recv(io);
 222
 223         if (a.len > 0) {
 224 retry:
 225                 n = (long)recv(a.fd, a.ptr, a.len, MSG_DONTWAIT);
 226                 if (read_check(&a, n, "recv", io_wait) != 0)
 227                         goto retry;
 228         }
 229         return a.buf;
 230 }
 231
 232 /*
 233  * This method may be optimized on some systems (e.g. GNU/Linux) to use
 234  * MSG_DONTWAIT to avoid explicitly setting the O_NONBLOCK flag via fcntl.
 235  * Otherwise this is the same as Kgio::PipeMethods#kgio_read
 236  */
 237 static VALUE kgio_recv(int argc, VALUE *argv, VALUE io)
 238 {
 239         return my_recv(1, argc, argv, io);
 240 }
 241
 242 /*
 243  * Same as Kgio::SocketMethods#kgio_read, except EOFError is raised
 244  * on EOF without a backtrace
 245  */
 246 static VALUE kgio_recv_bang(int argc, VALUE *argv, VALUE io)
 247 {
 248         VALUE rv = my_recv(1, argc, argv, io);
 249
 250         if (NIL_P(rv)) my_eof_error();
 251         return rv;
 252 }
 253
 254 /*
 255  * This method may be optimized on some systems (e.g. GNU/Linux) to use
 256  * MSG_DONTWAIT to avoid explicitly setting the O_NONBLOCK flag via fcntl.
 257  * Otherwise this is the same as Kgio::PipeMethods#kgio_tryread
 258  */
 259 static VALUE kgio_tryrecv(int argc, VALUE *argv, VALUE io)
 260 {
 261         return my_recv(0, argc, argv, io);
 262 }
 263 #else /* ! USE_MSG_DONTWAIT */
 264 #  define kgio_recv kgio_read
 265 #  define kgio_recv_bang kgio_read_bang
 266 #  define kgio_tryrecv kgio_tryread
 267 #endif /* USE_MSG_DONTWAIT */
 268
 269 static VALUE my_peek(int io_wait, int argc, VALUE *argv, VALUE io)
 270 {
 271         struct io_args a;
 272         long n;
 273
 274         prepare_read(&a, argc, argv, io);
 275         kgio_autopush_recv(io);
 276
 277         if (a.len > 0) {
 278                 if (peek_flags == MSG_PEEK)
 279                         set_nonblocking(a.fd);
 280 retry:
 281                 n = (long)recv(a.fd, a.ptr, a.len, peek_flags);
 282                 if (read_check(&a, n, "recv(MSG_PEEK)", io_wait) != 0)
 283                         goto retry;
 284         }
 285         return a.buf;
 286 }
 287
 288 /*
 289  * call-seq:
 290  *
 291  *      socket.kgio_trypeek(maxlen)           ->  buffer
 292  *      socket.kgio_trypeek(maxlen, buffer)   ->  buffer
 293  *
 294  * Like kgio_tryread, except it uses MSG_PEEK so it does not drain the
 295  * socket buffer.  A subsequent read of any type (including another peek)
 296  * will return the same data.
 297  */
 298 static VALUE kgio_trypeek(int argc, VALUE *argv, VALUE io)
 299 {
 300         return my_peek(0, argc, argv, io);
 301 }
 302
 303 /*
 304  * call-seq:
 305  *
 306  *      socket.kgio_peek(maxlen)           ->  buffer
 307  *      socket.kgio_peek(maxlen, buffer)   ->  buffer
 308  *
 309  * Like kgio_read, except it uses MSG_PEEK so it does not drain the
 310  * socket buffer.  A subsequent read of any type (including another peek)
 311  * will return the same data.
 312  */
 313 static VALUE kgio_peek(int argc, VALUE *argv, VALUE io)
 314 {
 315         return my_peek(1, argc, argv, io);
 316 }
 317
 318 /*
 319  * call-seq:
 320  *
 321  *      Kgio.trypeek(socket, maxlen)           ->  buffer
 322  *      Kgio.trypeek(socket, maxlen, buffer)   ->  buffer
 323  *
 324  * Like Kgio.tryread, except it uses MSG_PEEK so it does not drain the
 325  * socket buffer.  This can only be used on sockets and not pipe objects.
 326  * Maybe used in place of SocketMethods#kgio_trypeek for non-Kgio objects
 327  */
 328 static VALUE s_trypeek(int argc, VALUE *argv, VALUE mod)
 329 {
 330         if (argc <= 1)
 331                 rb_raise(rb_eArgError, "wrong number of arguments");
 332         return my_peek(0, argc - 1, &argv[1], argv[0]);
 333 }
 334
 335 static void prepare_write(struct io_args *a, VALUE io, VALUE str)
 336 {
 337         a->buf = (TYPE(str) == T_STRING) ? str : rb_obj_as_string(str);
 338         a->ptr = RSTRING_PTR(a->buf);
 339         a->len = RSTRING_LEN(a->buf);
 340         a->io = io;
 341         a->fd = my_fileno(io);
 342 }
 343
 344 static int write_check(struct io_args *a, long n, const char *msg, int io_wait)
 345 {
 346         if (a->len == n) {
 347 done:
 348                 a->buf = Qnil;
 349         } else if (n < 0) {
 350                 if (errno == EINTR) {
 351                         a->fd = my_fileno(a->io);
 352                         return -1;
 353                 }
 354                 if (errno == EAGAIN) {
 355                         long written = RSTRING_LEN(a->buf) - a->len;
 356
 357                         if (io_wait) {
 358                                 (void)kgio_call_wait_writable(a->io);
 359
 360                                 /* buf may be modified in other thread/fiber */
 361                                 a->len = RSTRING_LEN(a->buf) - written;
 362                                 if (a->len <= 0)
 363                                         goto done;
 364                                 a->ptr = RSTRING_PTR(a->buf) + written;
 365                                 return -1;
 366                         } else if (written > 0) {
 367                                 a->buf = rb_str_subseq(a->buf, written, a->len);
 368                         } else {
 369                                 a->buf = sym_wait_writable;
 370                         }
 371                         return 0;
 372                 }
 373                 wr_sys_fail(msg);
 374         } else {
 375                 assert(n >= 0 && n < a->len && "write/send syscall broken?");
 376                 a->ptr += n;
 377                 a->len -= n;
 378                 return -1;
 379         }
 380         return 0;
 381 }
 382
 383 static VALUE my_write(VALUE io, VALUE str, int io_wait)
 384 {
 385         struct io_args a;
 386         long n;
 387
 388         prepare_write(&a, io, str);
 389         set_nonblocking(a.fd);
 390 retry:
 391         n = (long)write(a.fd, a.ptr, a.len);
 392         if (write_check(&a, n, "write", io_wait) != 0)
 393                 goto retry;
 394         if (TYPE(a.buf) != T_SYMBOL)
 395                 kgio_autopush_write(io);
 396         return a.buf;
 397 }
 398
 399 /*
 400  * call-seq:
 401  *
 402  *      io.kgio_write(str)      -> nil
 403  *
 404  * Returns nil when the write completes.
 405  *
 406  * This may block and call any method defined to +kgio_wait_writable+
 407  * for the class.
 408  */
 409 static VALUE kgio_write(VALUE io, VALUE str)
 410 {
 411         return my_write(io, str, 1);
 412 }
 413
 414 /*
 415  * call-seq:
 416  *
 417  *      io.kgio_trywrite(str)   -> nil, String or :wait_writable
 418  *
 419  * Returns nil if the write was completed in full.
 420  *
 421  * Returns a String containing the unwritten portion if EAGAIN
 422  * was encountered, but some portion was successfully written.
 423  *
 424  * Returns :wait_writable if EAGAIN is encountered and nothing
 425  * was written.
 426  */
 427 static VALUE kgio_trywrite(VALUE io, VALUE str)
 428 {
 429         return my_write(io, str, 0);
 430 }
 431
 432 #ifndef HAVE_WRITEV
 433 #define iovec my_iovec
 434 struct my_iovec {
 435         void  *iov_base;
 436         size_t iov_len;
 437 };
 438 #endif
 439
 440 /* tests for choosing following constants were done on Linux 3.0 x86_64
 441  * (Ubuntu 12.04) Core i3 i3-2330M slowed to 1600MHz
 442  * testing script https://gist.github.com/2850641
 443  * fill free to make more thorough testing and choose better value
 444  */
 445
 446 /* test shows that its meaningless to set WRITEV_MEMLIMIT more that 1M
 447  * even when tcp_wmem set to relatively high value (2M) (in fact, it becomes
 448  * even slower). 512K performs a bit better in average case. */
 449 #define WRITEV_MEMLIMIT (512*1024)
 450 /* same test shows that custom_writev is faster than glibc writev when
 451  * average string is smaller than ~500 bytes and slower when average strings
 452  * is greater then ~600 bytes. 512 bytes were choosen cause current compilers
 453  * turns x/512 into x>>9 */
 454 #define WRITEV_IMPL_THRESHOLD 512
 455
 456 static unsigned int iov_max = 1024; /* this could be overriden in init */
 457
 458 struct io_args_v {
 459         VALUE io;
 460         VALUE buf;
 461         VALUE vec_buf;
 462         struct iovec *vec;
 463         unsigned long iov_cnt;
 464         size_t batch_len;
 465         int something_written;
 466         int fd;
 467 };
 468
 469 static ssize_t custom_writev(int fd, const struct iovec *vec, unsigned int iov_cnt, size_t total_len)
 470 {
 471         unsigned int i;
 472         ssize_t result;
 473         char *buf, *curbuf;
 474         const struct iovec *curvec = vec;
 475
 476         /* we do not want to use ruby's xmalloc because
 477          * it can fire GC, and we'll free buffer shortly anyway */
 478         curbuf = buf = malloc(total_len);
 479         if (buf == NULL) return -1;
 480
 481         for (i = 0; i < iov_cnt; i++, curvec++) {
 482                 memcpy(curbuf, curvec->iov_base, curvec->iov_len);
 483                 curbuf += curvec->iov_len;
 484         }
 485
 486         result = write(fd, buf, total_len);
 487
 488         /* well, it seems that `free` could not change errno
 489          * but lets save it anyway */
 490         i = errno;
 491         free(buf);
 492         errno = i;
 493
 494         return result;
 495 }
 496
 497 static void prepare_writev(struct io_args_v *a, VALUE io, VALUE ary)
 498 {
 499         a->io = io;
 500         a->fd = my_fileno(io);
 501         a->something_written = 0;
 502
 503         if (TYPE(ary) == T_ARRAY)
 504                 /* rb_ary_subseq will not copy array unless it modified */
 505                 a->buf = rb_ary_subseq(ary, 0, RARRAY_LEN(ary));
 506         else
 507                 a->buf = rb_Array(ary);
 508
 509         a->vec_buf = rb_str_new(0, 0);
 510         a->vec = NULL;
 511 }
 512
 513 static void fill_iovec(struct io_args_v *a)
 514 {
 515         unsigned long i;
 516         struct iovec *curvec;
 517
 518         a->iov_cnt = RARRAY_LEN(a->buf);
 519         a->batch_len = 0;
 520         if (a->iov_cnt == 0) return;
 521         if (a->iov_cnt > iov_max) a->iov_cnt = iov_max;
 522         rb_str_resize(a->vec_buf, sizeof(struct iovec) * a->iov_cnt);
 523         curvec = a->vec = (struct iovec*)RSTRING_PTR(a->vec_buf);
 524
 525         for (i=0; i < a->iov_cnt; i++, curvec++) {
 526                 VALUE str = rb_ary_entry(a->buf, i);
 527                 long str_len, next_len;
 528
 529                 if (TYPE(str) != T_STRING) {
 530                         str = rb_obj_as_string(str);
 531                         rb_ary_store(a->buf, i, str);
 532                 }
 533
 534                 str_len = RSTRING_LEN(str);
 535
 536                 /* lets limit total memory to write,
 537                  * but always take first string */
 538                 next_len = a->batch_len + str_len;
 539                 if (i && next_len > WRITEV_MEMLIMIT) {
 540                         a->iov_cnt = i;
 541                         break;
 542                 }
 543                 a->batch_len = next_len;
 544
 545                 curvec->iov_base = RSTRING_PTR(str);
 546                 curvec->iov_len = str_len;
 547         }
 548 }
 549
 550 static long trim_writev_buffer(struct io_args_v *a, long n)
 551 {
 552         long i;
 553         long ary_len = RARRAY_LEN(a->buf);
 554
 555         if (n == (long)a->batch_len) {
 556                 i = a->iov_cnt;
 557                 n = 0;
 558         } else {
 559                 for (i = 0; n && i < ary_len; i++) {
 560                         VALUE entry = rb_ary_entry(a->buf, i);
 561                         n -= RSTRING_LEN(entry);
 562                         if (n < 0) break;
 563                 }
 564         }
 565
 566         /* all done */
 567         if (i == ary_len) {
 568                 assert(n == 0 && "writev system call is broken");
 569                 a->buf = Qnil;
 570                 return 0;
 571         }
 572
 573         /* partially done, remove fully-written buffers */
 574         if (i > 0)
 575                 a->buf = rb_ary_subseq(a->buf, i, ary_len - i);
 576
 577         /* setup+replace partially written buffer */
 578         if (n < 0) {
 579                 VALUE str = rb_ary_entry(a->buf, 0);
 580                 long str_len = RSTRING_LEN(str);
 581                 str = rb_str_subseq(str, str_len + n, -n);
 582                 rb_ary_store(a->buf, 0, str);
 583         }
 584         return RARRAY_LEN(a->buf);
 585 }
 586
 587 static int writev_check(struct io_args_v *a, long n, const char *msg, int io_wait)
 588 {
 589         if (n >= 0) {
 590                 if (n > 0) a->something_written = 1;
 591                 return trim_writev_buffer(a, n);
 592         } else if (n < 0) {
 593                 if (errno == EINTR) {
 594                         a->fd = my_fileno(a->io);
 595                         return -1;
 596                 }
 597                 if (errno == EAGAIN) {
 598                         if (io_wait) {
 599                                 (void)kgio_call_wait_writable(a->io);
 600                                 return -1;
 601                         } else if (!a->something_written) {
 602                                 a->buf = sym_wait_writable;
 603                         }
 604                         return 0;
 605                 }
 606                 wr_sys_fail(msg);
 607         }
 608         return 0;
 609 }
 610
 611 static VALUE my_writev(VALUE io, VALUE ary, int io_wait)
 612 {
 613         struct io_args_v a;
 614         long n;
 615
 616         prepare_writev(&a, io, ary);
 617         set_nonblocking(a.fd);
 618
 619         do {
 620                 fill_iovec(&a);
 621                 if (a.iov_cnt == 0)
 622                         n = 0;
 623                 else if (a.iov_cnt == 1)
 624                         n = (long)write(a.fd, a.vec[0].iov_base,
 625                                         a.vec[0].iov_len);
 626                 /* for big strings use library function */
 627                 else if (USE_WRITEV &&
 628                         ((a.batch_len / WRITEV_IMPL_THRESHOLD) > a.iov_cnt))
 629                         n = (long)writev(a.fd, a.vec, a.iov_cnt);
 630                 else
 631                         n = (long)custom_writev(a.fd, a.vec, a.iov_cnt,
 632                                                 a.batch_len);
 633         } while (writev_check(&a, n, "writev", io_wait) != 0);
 634         rb_str_resize(a.vec_buf, 0);
 635
 636         if (TYPE(a.buf) != T_SYMBOL)
 637                 kgio_autopush_write(io);
 638         return a.buf;
 639 }
 640
 641 /*
 642  * call-seq:
 643  *
 644  *      io.kgio_writev(array)   -> nil
 645  *
 646  * Returns nil when the write completes.
 647  *
 648  * This may block and call any method defined to +kgio_wait_writable+
 649  * for the class.
 650  *
 651  * Note: it uses +Array()+ semantic for converting argument, so that
 652  * it will succeed if you pass something else.
 653  */
 654 static VALUE kgio_writev(VALUE io, VALUE ary)
 655 {
 656         return my_writev(io, ary, 1);
 657 }
 658
 659 /*
 660  * call-seq:
 661  *
 662  *      io.kgio_trywritev(array)        -> nil, Array or :wait_writable
 663  *
 664  * Returns nil if the write was completed in full.
 665  *
 666  * Returns an Array of strings containing the unwritten portion
 667  * if EAGAIN was encountered, but some portion was successfully written.
 668  *
 669  * Returns :wait_writable if EAGAIN is encountered and nothing
 670  * was written.
 671  *
 672  * Note: it uses +Array()+ semantic for converting argument, so that
 673  * it will succeed if you pass something else.
 674  */
 675 static VALUE kgio_trywritev(VALUE io, VALUE ary)
 676 {
 677         return my_writev(io, ary, 0);
 678 }
 679
 680 #ifdef USE_MSG_DONTWAIT
 681 /*
 682  * This method behaves like Kgio::PipeMethods#kgio_write, except
 683  * it will use send(2) with the MSG_DONTWAIT flag on sockets to
 684  * avoid unnecessary calls to fcntl(2).
 685  */
 686 static VALUE my_send(VALUE io, VALUE str, int io_wait)
 687 {
 688         struct io_args a;
 689         long n;
 690
 691         prepare_write(&a, io, str);
 692 retry:
 693         n = (long)send(a.fd, a.ptr, a.len, MSG_DONTWAIT);
 694         if (write_check(&a, n, "send", io_wait) != 0)
 695                 goto retry;
 696         if (TYPE(a.buf) != T_SYMBOL)
 697                 kgio_autopush_send(io);
 698         return a.buf;
 699 }
 700
 701 /*
 702  * This method may be optimized on some systems (e.g. GNU/Linux) to use
 703  * MSG_DONTWAIT to avoid explicitly setting the O_NONBLOCK flag via fcntl.
 704  * Otherwise this is the same as Kgio::PipeMethods#kgio_write
 705  */
 706 static VALUE kgio_send(VALUE io, VALUE str)
 707 {
 708         return my_send(io, str, 1);
 709 }
 710
 711 /*
 712  * This method may be optimized on some systems (e.g. GNU/Linux) to use
 713  * MSG_DONTWAIT to avoid explicitly setting the O_NONBLOCK flag via fcntl.
 714  * Otherwise this is the same as Kgio::PipeMethods#kgio_trywrite
 715  */
 716 static VALUE kgio_trysend(VALUE io, VALUE str)
 717 {
 718         return my_send(io, str, 0);
 719 }
 720 #else /* ! USE_MSG_DONTWAIT */
 721 #  define kgio_send kgio_write
 722 #  define kgio_trysend kgio_trywrite
 723 #endif /* ! USE_MSG_DONTWAIT */
 724
 725 /*
 726  * call-seq:
 727  *
 728  *      Kgio.tryread(io, maxlen)           ->  buffer
 729  *      Kgio.tryread(io, maxlen, buffer)   ->  buffer
 730  *
 731  * Returns nil on EOF.
 732  * Returns :wait_readable if EAGAIN is encountered.
 733  *
 734  * Maybe used in place of PipeMethods#kgio_tryread for non-Kgio objects
 735  */
 736 static VALUE s_tryread(int argc, VALUE *argv, VALUE mod)
 737 {
 738         if (argc <= 1)
 739                 rb_raise(rb_eArgError, "wrong number of arguments");
 740         return my_read(0, argc - 1, &argv[1], argv[0]);
 741 }
 742
 743 /*
 744  * call-seq:
 745  *
 746  *      Kgio.trywrite(io, str)    -> nil, String or :wait_writable
 747  *
 748  * Returns nil if the write was completed in full.
 749  *
 750  * Returns a String containing the unwritten portion if EAGAIN
 751  * was encountered, but some portion was successfully written.
 752  *
 753  * Returns :wait_writable if EAGAIN is encountered and nothing
 754  * was written.
 755  *
 756  * Maybe used in place of PipeMethods#kgio_trywrite for non-Kgio objects
 757  */
 758 static VALUE s_trywrite(VALUE mod, VALUE io, VALUE str)
 759 {
 760         return my_write(io, str, 0);
 761 }
 762
 763 /*
 764  * call-seq:
 765  *
 766  *      Kgio.trywritev(io, array)    -> nil, Array or :wait_writable
 767  *
 768  * Returns nil if the write was completed in full.
 769  *
 770  * Returns a Array of strings containing the unwritten portion if EAGAIN
 771  * was encountered, but some portion was successfully written.
 772  *
 773  * Returns :wait_writable if EAGAIN is encountered and nothing
 774  * was written.
 775  *
 776  * Maybe used in place of PipeMethods#kgio_trywritev for non-Kgio objects
 777  */
 778 static VALUE s_trywritev(VALUE mod, VALUE io, VALUE ary)
 779 {
 780         return kgio_trywritev(io, ary);
 781 }
 782
 783 void init_kgio_read_write(void)
 784 {
 785         VALUE mPipeMethods, mSocketMethods;
 786         VALUE mKgio = rb_define_module("Kgio");
 787         VALUE mWaiters = rb_const_get(mKgio, rb_intern("DefaultWaiters"));
 788
 789         sym_wait_readable = ID2SYM(rb_intern("wait_readable"));
 790         sym_wait_writable = ID2SYM(rb_intern("wait_writable"));
 791
 792         rb_define_singleton_method(mKgio, "tryread", s_tryread, -1);
 793         rb_define_singleton_method(mKgio, "trywrite", s_trywrite, 2);
 794         rb_define_singleton_method(mKgio, "trywritev", s_trywritev, 2);
 795         rb_define_singleton_method(mKgio, "trypeek", s_trypeek, -1);
 796
 797         /*
 798          * Document-module: Kgio::PipeMethods
 799          *
 800          * This module may be used used to create classes that respond to
 801          * various Kgio methods for reading and writing.  This is included
 802          * in Kgio::Pipe by default.
 803          */
 804         mPipeMethods = rb_define_module_under(mKgio, "PipeMethods");
 805         rb_define_method(mPipeMethods, "kgio_read", kgio_read, -1);
 806         rb_define_method(mPipeMethods, "kgio_read!", kgio_read_bang, -1);
 807         rb_define_method(mPipeMethods, "kgio_write", kgio_write, 1);
 808         rb_define_method(mPipeMethods, "kgio_writev", kgio_writev, 1);
 809         rb_define_method(mPipeMethods, "kgio_tryread", kgio_tryread, -1);
 810         rb_define_method(mPipeMethods, "kgio_trywrite", kgio_trywrite, 1);
 811         rb_define_method(mPipeMethods, "kgio_trywritev", kgio_trywritev, 1);
 812
 813         /*
 814          * Document-module: Kgio::SocketMethods
 815          *
 816          * This method behaves like Kgio::PipeMethods, but contains
 817          * optimizations for sockets on certain operating systems
 818          * (e.g. GNU/Linux).
 819          */
 820         mSocketMethods = rb_define_module_under(mKgio, "SocketMethods");
 821         rb_define_method(mSocketMethods, "kgio_read", kgio_recv, -1);
 822         rb_define_method(mSocketMethods, "kgio_read!", kgio_recv_bang, -1);
 823         rb_define_method(mSocketMethods, "kgio_write", kgio_send, 1);
 824         rb_define_method(mSocketMethods, "kgio_writev", kgio_writev, 1);
 825         rb_define_method(mSocketMethods, "kgio_tryread", kgio_tryrecv, -1);
 826         rb_define_method(mSocketMethods, "kgio_trywrite", kgio_trysend, 1);
 827         rb_define_method(mSocketMethods, "kgio_trywritev", kgio_trywritev, 1);
 828         rb_define_method(mSocketMethods, "kgio_trypeek", kgio_trypeek, -1);
 829         rb_define_method(mSocketMethods, "kgio_peek", kgio_peek, -1);
 830
 831         /*
 832          * Returns the client IP address of the socket as a string
 833          * (e.g. "127.0.0.1" or "::1").
 834          * This is always the value of the Kgio::LOCALHOST constant
 835          * for UNIX domain sockets.
 836          */
 837         rb_define_attr(mSocketMethods, "kgio_addr", 1, 1);
 838         id_set_backtrace = rb_intern("set_backtrace");
 839         eErrno_EPIPE = rb_const_get(rb_mErrno, rb_intern("EPIPE"));
 840         eErrno_ECONNRESET = rb_const_get(rb_mErrno, rb_intern("ECONNRESET"));
 841         rb_include_module(mPipeMethods, mWaiters);
 842         rb_include_module(mSocketMethods, mWaiters);
 843
 844 #ifdef HAVE_WRITEV
 845         {
 846 #  ifdef IOV_MAX
 847                 unsigned int sys_iov_max = IOV_MAX;
 848 #  else
 849                 unsigned int sys_iov_max = sysconf(_SC_IOV_MAX);
 850 #  endif
 851                 if (sys_iov_max < iov_max)
 852                         iov_max = sys_iov_max;
 853         }
 854 #endif
 855 }