ext/kgio/read_write.c

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