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