ext/kgio/read.c

   1 /* ref: rubinius b2811f260de16d1e972462e27852470364608de5 */
   2 #define RSTRING_MODIFIED 1
   3 #include "kgio.h"
   4 #include "my_fileno.h"
   5 #include "nonblock.h"
   6 static VALUE sym_wait_readable;
   7
   8 #ifdef USE_MSG_DONTWAIT
   9 static const int peek_flags = MSG_DONTWAIT|MSG_PEEK;
  10 #else
  11 static const int peek_flags = MSG_PEEK;
  12 #endif
  13
  14 struct rd_args {
  15         VALUE io;
  16         VALUE buf;
  17         char *ptr;
  18         long len;
  19         int fd;
  20 };
  21
  22 NORETURN(static void my_eof_error(void));
  23
  24 static void my_eof_error(void)
  25 {
  26         kgio_raise_empty_bt(rb_eEOFError, "end of file reached");
  27 }
  28
  29 static void prepare_read(struct rd_args *a, int argc, VALUE *argv, VALUE io)
  30 {
  31         VALUE length;
  32
  33         a->io = io;
  34         a->fd = my_fileno(io);
  35         rb_scan_args(argc, argv, "11", &length, &a->buf);
  36         a->len = NUM2LONG(length);
  37         if (NIL_P(a->buf)) {
  38                 a->buf = rb_str_new(NULL, a->len);
  39         } else {
  40                 StringValue(a->buf);
  41                 rb_str_modify(a->buf);
  42                 rb_str_resize(a->buf, a->len);
  43         }
  44         a->ptr = RSTRING_PTR(a->buf);
  45 }
  46
  47 static int read_check(struct rd_args *a, long n, const char *msg, int io_wait)
  48 {
  49         if (n < 0) {
  50                 if (errno == EINTR) {
  51                         a->fd = my_fileno(a->io);
  52                         return -1;
  53                 }
  54                 rb_str_set_len(a->buf, 0);
  55                 if (errno == EAGAIN) {
  56                         if (io_wait) {
  57                                 (void)kgio_call_wait_readable(a->io);
  58
  59                                 /* buf may be modified in other thread/fiber */
  60                                 rb_str_modify(a->buf);
  61                                 rb_str_resize(a->buf, a->len);
  62                                 a->ptr = RSTRING_PTR(a->buf);
  63                                 return -1;
  64                         } else {
  65                                 a->buf = sym_wait_readable;
  66                                 return 0;
  67                         }
  68                 }
  69                 kgio_rd_sys_fail(msg);
  70         }
  71         rb_str_set_len(a->buf, n);
  72         if (n == 0)
  73                 a->buf = Qnil;
  74         return 0;
  75 }
  76
  77 static VALUE my_read(int io_wait, int argc, VALUE *argv, VALUE io)
  78 {
  79         struct rd_args a;
  80         long n;
  81
  82         prepare_read(&a, argc, argv, io);
  83
  84         if (a.len > 0) {
  85                 set_nonblocking(a.fd);
  86 retry:
  87                 n = (long)read(a.fd, a.ptr, a.len);
  88                 if (read_check(&a, n, "read", io_wait) != 0)
  89                         goto retry;
  90         }
  91         return a.buf;
  92 }
  93
  94 /*
  95  * call-seq:
  96  *
  97  *      io.kgio_read(maxlen)           ->  buffer
  98  *      io.kgio_read(maxlen, buffer)   ->  buffer
  99  *
 100  * Reads at most maxlen bytes from the stream socket.  Returns with a
 101  * newly allocated buffer, or may reuse an existing buffer if supplied.
 102  *
 103  * This may block and call any method defined to +kgio_wait_readable+
 104  * for the class.
 105  *
 106  * Returns nil on EOF.
 107  *
 108  * This behaves like read(2) and IO#readpartial, NOT fread(3) or
 109  * IO#read which possess read-in-full behavior.
 110  */
 111 static VALUE kgio_read(int argc, VALUE *argv, VALUE io)
 112 {
 113         return my_read(1, argc, argv, io);
 114 }
 115
 116 /*
 117  * Same as Kgio::PipeMethods#kgio_read, except EOFError is raised
 118  * on EOF without a backtrace.  This method is intended as a
 119  * drop-in replacement for places where IO#readpartial is used, and
 120  * may be aliased as such.
 121  */
 122 static VALUE kgio_read_bang(int argc, VALUE *argv, VALUE io)
 123 {
 124         VALUE rv = my_read(1, argc, argv, io);
 125
 126         if (NIL_P(rv)) my_eof_error();
 127         return rv;
 128 }
 129
 130 /*
 131  * call-seq:
 132  *
 133  *      io.kgio_tryread(maxlen)           ->  buffer
 134  *      io.kgio_tryread(maxlen, buffer)   ->  buffer
 135  *
 136  * Reads at most maxlen bytes from the stream socket.  Returns with a
 137  * newly allocated buffer, or may reuse an existing buffer if supplied.
 138  *
 139  * Returns nil on EOF.
 140  *
 141  * Returns :wait_readable if EAGAIN is encountered.
 142  */
 143 static VALUE kgio_tryread(int argc, VALUE *argv, VALUE io)
 144 {
 145         return my_read(0, argc, argv, io);
 146 }
 147
 148 #ifdef USE_MSG_DONTWAIT
 149 static VALUE my_recv(int io_wait, int argc, VALUE *argv, VALUE io)
 150 {
 151         struct rd_args a;
 152         long n;
 153
 154         prepare_read(&a, argc, argv, io);
 155
 156         if (a.len > 0) {
 157 retry:
 158                 n = (long)recv(a.fd, a.ptr, a.len, MSG_DONTWAIT);
 159                 if (read_check(&a, n, "recv", io_wait) != 0)
 160                         goto retry;
 161         }
 162         return a.buf;
 163 }
 164
 165 /*
 166  * This method may be optimized on some systems (e.g. GNU/Linux) to use
 167  * MSG_DONTWAIT to avoid explicitly setting the O_NONBLOCK flag via fcntl.
 168  * Otherwise this is the same as Kgio::PipeMethods#kgio_read
 169  */
 170 static VALUE kgio_recv(int argc, VALUE *argv, VALUE io)
 171 {
 172         return my_recv(1, argc, argv, io);
 173 }
 174
 175 /*
 176  * Same as Kgio::SocketMethods#kgio_read, except EOFError is raised
 177  * on EOF without a backtrace
 178  */
 179 static VALUE kgio_recv_bang(int argc, VALUE *argv, VALUE io)
 180 {
 181         VALUE rv = my_recv(1, argc, argv, io);
 182
 183         if (NIL_P(rv)) my_eof_error();
 184         return rv;
 185 }
 186
 187 /*
 188  * This method may be optimized on some systems (e.g. GNU/Linux) to use
 189  * MSG_DONTWAIT to avoid explicitly setting the O_NONBLOCK flag via fcntl.
 190  * Otherwise this is the same as Kgio::PipeMethods#kgio_tryread
 191  */
 192 static VALUE kgio_tryrecv(int argc, VALUE *argv, VALUE io)
 193 {
 194         return my_recv(0, argc, argv, io);
 195 }
 196 #else /* ! USE_MSG_DONTWAIT */
 197 #  define kgio_recv kgio_read
 198 #  define kgio_recv_bang kgio_read_bang
 199 #  define kgio_tryrecv kgio_tryread
 200 #endif /* USE_MSG_DONTWAIT */
 201
 202 static VALUE my_peek(int io_wait, int argc, VALUE *argv, VALUE io)
 203 {
 204         struct rd_args a;
 205         long n;
 206
 207         prepare_read(&a, argc, argv, io);
 208
 209         if (a.len > 0) {
 210                 if (peek_flags == MSG_PEEK)
 211                         set_nonblocking(a.fd);
 212 retry:
 213                 n = (long)recv(a.fd, a.ptr, a.len, peek_flags);
 214                 if (read_check(&a, n, "recv(MSG_PEEK)", io_wait) != 0)
 215                         goto retry;
 216         }
 217         return a.buf;
 218 }
 219
 220 /*
 221  * call-seq:
 222  *
 223  *      socket.kgio_trypeek(maxlen)           ->  buffer
 224  *      socket.kgio_trypeek(maxlen, buffer)   ->  buffer
 225  *
 226  * Like kgio_tryread, except it uses MSG_PEEK so it does not drain the
 227  * socket buffer.  A subsequent read of any type (including another peek)
 228  * will return the same data.
 229  */
 230 static VALUE kgio_trypeek(int argc, VALUE *argv, VALUE io)
 231 {
 232         return my_peek(0, argc, argv, io);
 233 }
 234
 235 /*
 236  * call-seq:
 237  *
 238  *      socket.kgio_peek(maxlen)           ->  buffer
 239  *      socket.kgio_peek(maxlen, buffer)   ->  buffer
 240  *
 241  * Like kgio_read, except it uses MSG_PEEK so it does not drain the
 242  * socket buffer.  A subsequent read of any type (including another peek)
 243  * will return the same data.
 244  */
 245 static VALUE kgio_peek(int argc, VALUE *argv, VALUE io)
 246 {
 247         return my_peek(1, argc, argv, io);
 248 }
 249
 250 /*
 251  * call-seq:
 252  *
 253  *      Kgio.trypeek(socket, maxlen)           ->  buffer
 254  *      Kgio.trypeek(socket, maxlen, buffer)   ->  buffer
 255  *
 256  * Like Kgio.tryread, except it uses MSG_PEEK so it does not drain the
 257  * socket buffer.  This can only be used on sockets and not pipe objects.
 258  * Maybe used in place of SocketMethods#kgio_trypeek for non-Kgio objects
 259  */
 260 static VALUE s_trypeek(int argc, VALUE *argv, VALUE mod)
 261 {
 262         if (argc <= 1)
 263                 rb_raise(rb_eArgError, "wrong number of arguments");
 264         return my_peek(0, argc - 1, &argv[1], argv[0]);
 265 }
 266
 267 /*
 268  * call-seq:
 269  *
 270  *      Kgio.tryread(io, maxlen)           ->  buffer
 271  *      Kgio.tryread(io, maxlen, buffer)   ->  buffer
 272  *
 273  * Returns nil on EOF.
 274  * Returns :wait_readable if EAGAIN is encountered.
 275  *
 276  * Maybe used in place of PipeMethods#kgio_tryread for non-Kgio objects
 277  */
 278 static VALUE s_tryread(int argc, VALUE *argv, VALUE mod)
 279 {
 280         if (argc <= 1)
 281                 rb_raise(rb_eArgError, "wrong number of arguments");
 282         return my_read(0, argc - 1, &argv[1], argv[0]);
 283 }
 284
 285 void init_kgio_read(void)
 286 {
 287         VALUE mPipeMethods, mSocketMethods;
 288         VALUE mKgio = rb_define_module("Kgio");
 289
 290         sym_wait_readable = ID2SYM(rb_intern("wait_readable"));
 291
 292         rb_define_singleton_method(mKgio, "tryread", s_tryread, -1);
 293         rb_define_singleton_method(mKgio, "trypeek", s_trypeek, -1);
 294
 295         /*
 296          * Document-module: Kgio::PipeMethods
 297          *
 298          * This module may be used used to create classes that respond to
 299          * various Kgio methods for reading and writing.  This is included
 300          * in Kgio::Pipe by default.
 301          */
 302         mPipeMethods = rb_define_module_under(mKgio, "PipeMethods");
 303         rb_define_method(mPipeMethods, "kgio_read", kgio_read, -1);
 304         rb_define_method(mPipeMethods, "kgio_read!", kgio_read_bang, -1);
 305         rb_define_method(mPipeMethods, "kgio_tryread", kgio_tryread, -1);
 306
 307         /*
 308          * Document-module: Kgio::SocketMethods
 309          *
 310          * This method behaves like Kgio::PipeMethods, but contains
 311          * optimizations for sockets on certain operating systems
 312          * (e.g. GNU/Linux).
 313          */
 314         mSocketMethods = rb_define_module_under(mKgio, "SocketMethods");
 315         rb_define_method(mSocketMethods, "kgio_read", kgio_recv, -1);
 316         rb_define_method(mSocketMethods, "kgio_read!", kgio_recv_bang, -1);
 317         rb_define_method(mSocketMethods, "kgio_tryread", kgio_tryrecv, -1);
 318         rb_define_method(mSocketMethods, "kgio_trypeek", kgio_trypeek, -1);
 319         rb_define_method(mSocketMethods, "kgio_peek", kgio_peek, -1);
 320 }