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         if (TYPE(a.buf) != T_SYMBOL)
  76                 kgio_autopush_write(io);
  77         return a.buf;
  78 }
  79
  80 /*
  81  * call-seq:
  82  *
  83  *      io.kgio_write(str)      -> nil
  84  *
  85  * Returns nil when the write completes.
  86  *
  87  * This may block and call any method defined to +kgio_wait_writable+
  88  * for the class.
  89  */
  90 static VALUE kgio_write(VALUE io, VALUE str)
  91 {
  92         return my_write(io, str, 1);
  93 }
  94
  95 /*
  96  * call-seq:
  97  *
  98  *      io.kgio_trywrite(str)   -> nil, String or :wait_writable
  99  *
 100  * Returns nil if the write was completed in full.
 101  *
 102  * Returns a String containing the unwritten portion if EAGAIN
 103  * was encountered, but some portion was successfully written.
 104  *
 105  * Returns :wait_writable if EAGAIN is encountered and nothing
 106  * was written.
 107  */
 108 static VALUE kgio_trywrite(VALUE io, VALUE str)
 109 {
 110         return my_write(io, str, 0);
 111 }
 112
 113 #ifdef USE_MSG_DONTWAIT
 114 /*
 115  * This method behaves like Kgio::PipeMethods#kgio_write, except
 116  * it will use send(2) with the MSG_DONTWAIT flag on sockets to
 117  * avoid unnecessary calls to fcntl(2).
 118  */
 119 static VALUE my_send(VALUE io, VALUE str, int io_wait)
 120 {
 121         struct wr_args a;
 122         long n;
 123
 124         prepare_write(&a, io, str);
 125 retry:
 126         n = (long)send(a.fd, a.ptr, a.len, MSG_DONTWAIT);
 127         if (write_check(&a, n, "send", io_wait) != 0)
 128                 goto retry;
 129         if (TYPE(a.buf) != T_SYMBOL)
 130                 kgio_autopush_send(io);
 131         return a.buf;
 132 }
 133
 134 /*
 135  * This method may be optimized on some systems (e.g. GNU/Linux) to use
 136  * MSG_DONTWAIT to avoid explicitly setting the O_NONBLOCK flag via fcntl.
 137  * Otherwise this is the same as Kgio::PipeMethods#kgio_write
 138  */
 139 static VALUE kgio_send(VALUE io, VALUE str)
 140 {
 141         return my_send(io, str, 1);
 142 }
 143
 144 /*
 145  * This method may be optimized on some systems (e.g. GNU/Linux) to use
 146  * MSG_DONTWAIT to avoid explicitly setting the O_NONBLOCK flag via fcntl.
 147  * Otherwise this is the same as Kgio::PipeMethods#kgio_trywrite
 148  */
 149 static VALUE kgio_trysend(VALUE io, VALUE str)
 150 {
 151         return my_send(io, str, 0);
 152 }
 153 #else /* ! USE_MSG_DONTWAIT */
 154 #  define kgio_send kgio_write
 155 #  define kgio_trysend kgio_trywrite
 156 #endif /* ! USE_MSG_DONTWAIT */
 157
 158 #if defined(KGIO_WITHOUT_GVL)
 159 #  include "blocking_io_region.h"
 160 #ifdef MSG_DONTWAIT /* Linux only */
 161 #  define MY_MSG_DONTWAIT (MSG_DONTWAIT)
 162 #else
 163 #  define MY_MSG_DONTWAIT (0)
 164 #endif
 165
 166 static VALUE nogvl_send(void *ptr)
 167 {
 168         struct wr_args *a = ptr;
 169
 170         return (VALUE)send(a->fd, a->ptr, a->len, a->flags);
 171 }
 172 /*
 173  * call-seq:
 174  *
 175  *      io.kgio_syssend(str, flags) -> nil, String or :wait_writable
 176  *
 177  * Returns nil if the write was completed in full.
 178  *
 179  * Returns a String containing the unwritten portion if EAGAIN
 180  * was encountered, but some portion was successfully written.
 181  *
 182  * Returns :wait_writable if EAGAIN is encountered and nothing
 183  * was written.
 184  *
 185  * This method is only available on Ruby 1.9.3 or later.
 186  */
 187 static VALUE kgio_syssend(VALUE io, VALUE str, VALUE flags)
 188 {
 189         struct wr_args a;
 190         long n;
 191
 192         a.flags = NUM2INT(flags);
 193         prepare_write(&a, io, str);
 194         if (a.flags & MY_MSG_DONTWAIT) {
 195                 do {
 196                         n = (long)send(a.fd, a.ptr, a.len, a.flags);
 197                 } while (write_check(&a, n, "send", 0) != 0);
 198         } else {
 199                 do {
 200                         n = (long)rb_thread_io_blocking_region(
 201                                                 nogvl_send, &a, a.fd);
 202                 } while (write_check(&a, n, "send", 0) != 0);
 203         }
 204         return a.buf;
 205 }
 206 #endif /* HAVE_RB_THREAD_IO_BLOCKING_REGION */
 207
 208 /*
 209  * call-seq:
 210  *
 211  *      Kgio.trywrite(io, str)    -> nil, String or :wait_writable
 212  *
 213  * Returns nil if the write was completed in full.
 214  *
 215  * Returns a String containing the unwritten portion if EAGAIN
 216  * was encountered, but some portion was successfully written.
 217  *
 218  * Returns :wait_writable if EAGAIN is encountered and nothing
 219  * was written.
 220  *
 221  * Maybe used in place of PipeMethods#kgio_trywrite for non-Kgio objects
 222  */
 223 static VALUE s_trywrite(VALUE mod, VALUE io, VALUE str)
 224 {
 225         return my_write(io, str, 0);
 226 }
 227
 228 void init_kgio_write(void)
 229 {
 230         VALUE mPipeMethods, mSocketMethods;
 231         VALUE mKgio = rb_define_module("Kgio");
 232
 233         sym_wait_writable = ID2SYM(rb_intern("wait_writable"));
 234
 235         rb_define_singleton_method(mKgio, "trywrite", s_trywrite, 2);
 236
 237         /*
 238          * Document-module: Kgio::PipeMethods
 239          *
 240          * This module may be used used to create classes that respond to
 241          * various Kgio methods for reading and writing.  This is included
 242          * in Kgio::Pipe by default.
 243          */
 244         mPipeMethods = rb_define_module_under(mKgio, "PipeMethods");
 245         rb_define_method(mPipeMethods, "kgio_write", kgio_write, 1);
 246         rb_define_method(mPipeMethods, "kgio_trywrite", kgio_trywrite, 1);
 247
 248         /*
 249          * Document-module: Kgio::SocketMethods
 250          *
 251          * This method behaves like Kgio::PipeMethods, but contains
 252          * optimizations for sockets on certain operating systems
 253          * (e.g. GNU/Linux).
 254          */
 255         mSocketMethods = rb_define_module_under(mKgio, "SocketMethods");
 256         rb_define_method(mSocketMethods, "kgio_write", kgio_send, 1);
 257         rb_define_method(mSocketMethods, "kgio_trywrite", kgio_trysend, 1);
 258
 259 #if defined(KGIO_WITHOUT_GVL)
 260         rb_define_method(mSocketMethods, "kgio_syssend", kgio_syssend, 2);
 261 #endif
 262 }