ext/kgio/write.c

   1 /* we do not modify RSTRING pointers here */
   2 #include "kgio.h"
   3 #include "my_fileno.h"
   4 #include "nonblock.h"
   5 static VALUE sym_wait_writable;
   6
   7 struct wr_args {
   8         VALUE io;
   9         VALUE buf;
  10         const char *ptr;
  11         long len;
  12         int fd;
  13         int flags;
  14 };
  15
  16 static void prepare_write(struct wr_args *a, VALUE io, VALUE str)
  17 {
  18         a->buf = (TYPE(str) == T_STRING) ? str : rb_obj_as_string(str);
  19         a->ptr = RSTRING_PTR(a->buf);
  20         a->len = RSTRING_LEN(a->buf);
  21         a->io = io;
  22         a->fd = my_fileno(io);
  23 }
  24
  25 static int write_check(struct wr_args *a, long n, const char *msg, int io_wait)
  26 {
  27         if (a->len == n) {
  28 done:
  29                 a->buf = Qnil;
  30         } else if (n < 0) {
  31                 if (errno == EINTR) {
  32                         a->fd = my_fileno(a->io);
  33                         return -1;
  34                 }
  35                 if (errno == EAGAIN) {
  36                         long written = RSTRING_LEN(a->buf) - a->len;
  37
  38                         if (io_wait) {
  39                                 (void)kgio_call_wait_writable(a->io);
  40
  41                                 /* buf may be modified in other thread/fiber */
  42                                 a->len = RSTRING_LEN(a->buf) - written;
  43                                 if (a->len <= 0)
  44                                         goto done;
  45                                 a->ptr = RSTRING_PTR(a->buf) + written;
  46                                 return -1;
  47                         } else if (written > 0) {
  48                                 a->buf = MY_STR_SUBSEQ(a->buf, written, a->len);
  49                         } else {
  50                                 a->buf = sym_wait_writable;
  51                         }
  52                         return 0;
  53                 }
  54                 kgio_wr_sys_fail(msg);
  55         } else {
  56                 assert(n >= 0 && n < a->len && "write/send syscall broken?");
  57                 a->ptr += n;
  58                 a->len -= n;
  59                 return -1;
  60         }
  61         return 0;
  62 }
  63
  64 static VALUE my_write(VALUE io, VALUE str, int io_wait)
  65 {
  66         struct wr_args a;
  67         long n;
  68
  69         prepare_write(&a, io, str);
  70         set_nonblocking(a.fd);
  71 retry:
  72         n = (long)write(a.fd, a.ptr, a.len);
  73         if (write_check(&a, n, "write", io_wait) != 0)
  74                 goto retry;
  75         return a.buf;
  76 }
  77
  78 /*
  79  * call-seq:
  80  *
  81  *      io.kgio_write(str)      -> nil
  82  *
  83  * Returns nil when the write completes.
  84  *
  85  * This may block and call any method defined to +kgio_wait_writable+
  86  * for the class.
  87  */
  88 static VALUE kgio_write(VALUE io, VALUE str)
  89 {
  90         return my_write(io, str, 1);
  91 }
  92
  93 /*
  94  * call-seq:
  95  *
  96  *      io.kgio_trywrite(str)   -> nil, String or :wait_writable
  97  *
  98  * Returns nil if the write was completed in full.
  99  *
 100  * Returns a String containing the unwritten portion if EAGAIN
 101  * was encountered, but some portion was successfully written.
 102  *
 103  * Returns :wait_writable if EAGAIN is encountered and nothing
 104  * was written.
 105  */
 106 static VALUE kgio_trywrite(VALUE io, VALUE str)
 107 {
 108         return my_write(io, str, 0);
 109 }
 110
 111 #ifdef USE_MSG_DONTWAIT
 112 /*
 113  * This method behaves like Kgio::PipeMethods#kgio_write, except
 114  * it will use send(2) with the MSG_DONTWAIT flag on sockets to
 115  * avoid unnecessary calls to fcntl(2).
 116  */
 117 static VALUE my_send(VALUE io, VALUE str, int io_wait)
 118 {
 119         struct wr_args a;
 120         long n;
 121
 122         prepare_write(&a, io, str);
 123 retry:
 124         n = (long)send(a.fd, a.ptr, a.len, MSG_DONTWAIT);
 125         if (write_check(&a, n, "send", io_wait) != 0)
 126                 goto retry;
 127         return a.buf;
 128 }
 129
 130 /*
 131  * This method may be optimized on some systems (e.g. GNU/Linux) to use
 132  * MSG_DONTWAIT to avoid explicitly setting the O_NONBLOCK flag via fcntl.
 133  * Otherwise this is the same as Kgio::PipeMethods#kgio_write
 134  */
 135 static VALUE kgio_send(VALUE io, VALUE str)
 136 {
 137         return my_send(io, str, 1);
 138 }
 139
 140 /*
 141  * This method may be optimized on some systems (e.g. GNU/Linux) to use
 142  * MSG_DONTWAIT to avoid explicitly setting the O_NONBLOCK flag via fcntl.
 143  * Otherwise this is the same as Kgio::PipeMethods#kgio_trywrite
 144  */
 145 static VALUE kgio_trysend(VALUE io, VALUE str)
 146 {
 147         return my_send(io, str, 0);
 148 }
 149 #else /* ! USE_MSG_DONTWAIT */
 150 #  define kgio_send kgio_write
 151 #  define kgio_trysend kgio_trywrite
 152 #endif /* ! USE_MSG_DONTWAIT */
 153
 154 #if defined(KGIO_WITHOUT_GVL)
 155 #  include "blocking_io_region.h"
 156 #ifdef MSG_DONTWAIT /* Linux only */
 157 #  define MY_MSG_DONTWAIT (MSG_DONTWAIT)
 158 #else
 159 #  define MY_MSG_DONTWAIT (0)
 160 #endif
 161
 162 static VALUE nogvl_send(void *ptr)
 163 {
 164         struct wr_args *a = ptr;
 165
 166         return (VALUE)send(a->fd, a->ptr, a->len, a->flags);
 167 }
 168 /*
 169  * call-seq:
 170  *
 171  *      io.kgio_syssend(str, flags) -> nil, String or :wait_writable
 172  *
 173  * Returns nil if the write was completed in full.
 174  *
 175  * Returns a String containing the unwritten portion if EAGAIN
 176  * was encountered, but some portion was successfully written.
 177  *
 178  * Returns :wait_writable if EAGAIN is encountered and nothing
 179  * was written.
 180  *
 181  * This method is only available on Ruby 1.9.3 or later.
 182  */
 183 static VALUE kgio_syssend(VALUE io, VALUE str, VALUE flags)
 184 {
 185         struct wr_args a;
 186         long n;
 187
 188         a.flags = NUM2INT(flags);
 189         prepare_write(&a, io, str);
 190         if (a.flags & MY_MSG_DONTWAIT) {
 191                 do {
 192                         n = (long)send(a.fd, a.ptr, a.len, a.flags);
 193                 } while (write_check(&a, n, "send", 0) != 0);
 194         } else {
 195                 do {
 196                         n = (long)rb_thread_io_blocking_region(
 197                                                 nogvl_send, &a, a.fd);
 198                 } while (write_check(&a, n, "send", 0) != 0);
 199         }
 200         return a.buf;
 201 }
 202 #endif /* HAVE_RB_THREAD_IO_BLOCKING_REGION */
 203
 204 /*
 205  * call-seq:
 206  *
 207  *      Kgio.trywrite(io, str)    -> nil, String or :wait_writable
 208  *
 209  * Returns nil if the write was completed in full.
 210  *
 211  * Returns a String containing the unwritten portion if EAGAIN
 212  * was encountered, but some portion was successfully written.
 213  *
 214  * Returns :wait_writable if EAGAIN is encountered and nothing
 215  * was written.
 216  *
 217  * Maybe used in place of PipeMethods#kgio_trywrite for non-Kgio objects
 218  */
 219 static VALUE s_trywrite(VALUE mod, VALUE io, VALUE str)
 220 {
 221         return my_write(io, str, 0);
 222 }
 223
 224 void init_kgio_write(void)
 225 {
 226         VALUE mPipeMethods, mSocketMethods;
 227         VALUE mKgio = rb_define_module("Kgio");
 228
 229         sym_wait_writable = ID2SYM(rb_intern("wait_writable"));
 230
 231         rb_define_singleton_method(mKgio, "trywrite", s_trywrite, 2);
 232
 233         /*
 234          * Document-module: Kgio::PipeMethods
 235          *
 236          * This module may be used used to create classes that respond to
 237          * various Kgio methods for reading and writing.  This is included
 238          * in Kgio::Pipe by default.
 239          */
 240         mPipeMethods = rb_define_module_under(mKgio, "PipeMethods");
 241         rb_define_method(mPipeMethods, "kgio_write", kgio_write, 1);
 242         rb_define_method(mPipeMethods, "kgio_trywrite", kgio_trywrite, 1);
 243
 244         /*
 245          * Document-module: Kgio::SocketMethods
 246          *
 247          * This method behaves like Kgio::PipeMethods, but contains
 248          * optimizations for sockets on certain operating systems
 249          * (e.g. GNU/Linux).
 250          */
 251         mSocketMethods = rb_define_module_under(mKgio, "SocketMethods");
 252         rb_define_method(mSocketMethods, "kgio_write", kgio_send, 1);
 253         rb_define_method(mSocketMethods, "kgio_trywrite", kgio_trysend, 1);
 254
 255 #if defined(KGIO_WITHOUT_GVL)
 256         rb_define_method(mSocketMethods, "kgio_syssend", kgio_syssend, 2);
 257 #endif
 258 }