2 static VALUE sym_wait_readable
, sym_wait_writable
;
3 static VALUE eErrno_EPIPE
, eErrno_ECONNRESET
;
6 * we know MSG_DONTWAIT works properly on all stream sockets under Linux
7 * we can define this macro for other platforms as people care and
10 #if defined(__linux__) && ! defined(USE_MSG_DONTWAIT)
11 # define USE_MSG_DONTWAIT
14 NORETURN(static void raise_empty_bt(VALUE
, const char *));
15 NORETURN(static void my_eof_error(void));
16 NORETURN(static void wr_sys_fail(const char *));
17 NORETURN(static void rd_sys_fail(const char *));
19 static void raise_empty_bt(VALUE err
, const char *msg
)
21 VALUE exc
= rb_exc_new2(err
, msg
);
22 VALUE bt
= rb_ary_new();
24 rb_funcall(exc
, rb_intern("set_backtrace"), 1, bt
);
28 static void my_eof_error(void)
30 raise_empty_bt(rb_eEOFError
, "end of file reached");
33 static void wr_sys_fail(const char *msg
)
38 raise_empty_bt(eErrno_EPIPE
, msg
);
41 raise_empty_bt(eErrno_ECONNRESET
, msg
);
46 static void rd_sys_fail(const char *msg
)
48 if (errno
== ECONNRESET
) {
50 raise_empty_bt(eErrno_ECONNRESET
, msg
);
55 static void prepare_read(struct io_args
*a
, int argc
, VALUE
*argv
, VALUE io
)
60 a
->fd
= my_fileno(io
);
61 rb_scan_args(argc
, argv
, "11", &length
, &a
->buf
);
62 a
->len
= NUM2LONG(length
);
64 a
->buf
= rb_str_new(NULL
, a
->len
);
67 rb_str_resize(a
->buf
, a
->len
);
69 a
->ptr
= RSTRING_PTR(a
->buf
);
72 static int read_check(struct io_args
*a
, long n
, const char *msg
, int io_wait
)
77 rb_str_set_len(a
->buf
, 0);
78 if (errno
== EAGAIN
) {
80 (void)kgio_call_wait_readable(a
->io
);
82 /* buf may be modified in other thread/fiber */
83 rb_str_resize(a
->buf
, a
->len
);
84 a
->ptr
= RSTRING_PTR(a
->buf
);
87 a
->buf
= sym_wait_readable
;
93 rb_str_set_len(a
->buf
, n
);
99 static VALUE
my_read(int io_wait
, int argc
, VALUE
*argv
, VALUE io
)
104 prepare_read(&a
, argc
, argv
, io
);
107 set_nonblocking(a
.fd
);
109 n
= (long)read(a
.fd
, a
.ptr
, a
.len
);
110 if (read_check(&a
, n
, "read", io_wait
) != 0)
119 * io.kgio_read(maxlen) -> buffer
120 * io.kgio_read(maxlen, buffer) -> buffer
122 * Reads at most maxlen bytes from the stream socket. Returns with a
123 * newly allocated buffer, or may reuse an existing buffer if supplied.
125 * Calls whatever is is defined to be the kgio_wait_readable method
128 * Returns nil on EOF.
130 * This behaves like read(2) and IO#readpartial, NOT fread(3) or
131 * IO#read which possess read-in-full behavior.
133 static VALUE
kgio_read(int argc
, VALUE
*argv
, VALUE io
)
135 return my_read(1, argc
, argv
, io
);
139 * Same as Kgio::PipeMethods#kgio_read, except EOFError is raised
140 * on EOF without a backtrace. This method is intended as a
141 * drop-in replacement for places where IO#readpartial is used, and
142 * may be aliased as such.
144 static VALUE
kgio_read_bang(int argc
, VALUE
*argv
, VALUE io
)
146 VALUE rv
= my_read(1, argc
, argv
, io
);
148 if (NIL_P(rv
)) my_eof_error();
155 * io.kgio_tryread(maxlen) -> buffer
156 * io.kgio_tryread(maxlen, buffer) -> buffer
158 * Reads at most maxlen bytes from the stream socket. Returns with a
159 * newly allocated buffer, or may reuse an existing buffer if supplied.
161 * Returns nil on EOF.
163 * Returns :wait_readable if EAGAIN is encountered.
165 static VALUE
kgio_tryread(int argc
, VALUE
*argv
, VALUE io
)
167 return my_read(0, argc
, argv
, io
);
170 #ifdef USE_MSG_DONTWAIT
171 static VALUE
my_recv(int io_wait
, int argc
, VALUE
*argv
, VALUE io
)
176 prepare_read(&a
, argc
, argv
, io
);
177 kgio_autopush_recv(io
);
181 n
= (long)recv(a
.fd
, a
.ptr
, a
.len
, MSG_DONTWAIT
);
182 if (read_check(&a
, n
, "recv", io_wait
) != 0)
189 * This method may be optimized on some systems (e.g. GNU/Linux) to use
190 * MSG_DONTWAIT to avoid explicitly setting the O_NONBLOCK flag via fcntl.
191 * Otherwise this is the same as Kgio::PipeMethods#kgio_read
193 static VALUE
kgio_recv(int argc
, VALUE
*argv
, VALUE io
)
195 return my_recv(1, argc
, argv
, io
);
199 * Same as Kgio::SocketMethods#kgio_read, except EOFError is raised
200 * on EOF without a backtrace
202 static VALUE
kgio_recv_bang(int argc
, VALUE
*argv
, VALUE io
)
204 VALUE rv
= my_recv(1, argc
, argv
, io
);
206 if (NIL_P(rv
)) my_eof_error();
211 * This method may be optimized on some systems (e.g. GNU/Linux) to use
212 * MSG_DONTWAIT to avoid explicitly setting the O_NONBLOCK flag via fcntl.
213 * Otherwise this is the same as Kgio::PipeMethods#kgio_tryread
215 static VALUE
kgio_tryrecv(int argc
, VALUE
*argv
, VALUE io
)
217 return my_recv(0, argc
, argv
, io
);
219 #else /* ! USE_MSG_DONTWAIT */
220 # define kgio_recv kgio_read
221 # define kgio_recv_bang kgio_read_bang
222 # define kgio_tryrecv kgio_tryread
223 #endif /* USE_MSG_DONTWAIT */
225 static void prepare_write(struct io_args
*a
, VALUE io
, VALUE str
)
227 a
->buf
= (TYPE(str
) == T_STRING
) ? str
: rb_obj_as_string(str
);
228 a
->ptr
= RSTRING_PTR(a
->buf
);
229 a
->len
= RSTRING_LEN(a
->buf
);
231 a
->fd
= my_fileno(io
);
234 static int write_check(struct io_args
*a
, long n
, const char *msg
, int io_wait
)
239 } else if (n
== -1) {
242 if (errno
== EAGAIN
) {
243 long written
= RSTRING_LEN(a
->buf
) - a
->len
;
246 (void)kgio_call_wait_writable(a
->io
);
248 /* buf may be modified in other thread/fiber */
249 a
->len
= RSTRING_LEN(a
->buf
) - written
;
252 a
->ptr
= RSTRING_PTR(a
->buf
) + written
;
254 } else if (written
> 0) {
255 a
->buf
= rb_str_new(a
->ptr
, a
->len
);
257 a
->buf
= sym_wait_writable
;
263 assert(n
>= 0 && n
< a
->len
&& "write/send syscall broken?");
271 static VALUE
my_write(VALUE io
, VALUE str
, int io_wait
)
276 prepare_write(&a
, io
, str
);
277 set_nonblocking(a
.fd
);
279 n
= (long)write(a
.fd
, a
.ptr
, a
.len
);
280 if (write_check(&a
, n
, "write", io_wait
) != 0)
288 * io.kgio_write(str) -> nil
290 * Returns nil when the write completes.
292 * Calls whatever is is defined to be the kgio_wait_writable method
295 static VALUE
kgio_write(VALUE io
, VALUE str
)
297 return my_write(io
, str
, 1);
303 * io.kgio_trywrite(str) -> nil or :wait_writable
305 * Returns nil if the write was completed in full.
307 * Returns a String containing the unwritten portion if EAGAIN
308 * was encountered, but some portion was successfully written.
310 * Returns :wait_writable if EAGAIN is encountered and nothing
313 static VALUE
kgio_trywrite(VALUE io
, VALUE str
)
315 return my_write(io
, str
, 0);
318 #ifdef USE_MSG_DONTWAIT
320 * This method behaves like Kgio::PipeMethods#kgio_write, except
321 * it will use send(2) with the MSG_DONTWAIT flag on sockets to
322 * avoid unnecessary calls to fcntl(2).
324 static VALUE
my_send(VALUE io
, VALUE str
, int io_wait
)
329 prepare_write(&a
, io
, str
);
331 n
= (long)send(a
.fd
, a
.ptr
, a
.len
, MSG_DONTWAIT
);
332 if (write_check(&a
, n
, "send", io_wait
) != 0)
334 if (TYPE(a
.buf
) != T_SYMBOL
)
335 kgio_autopush_send(io
);
340 * This method may be optimized on some systems (e.g. GNU/Linux) to use
341 * MSG_DONTWAIT to avoid explicitly setting the O_NONBLOCK flag via fcntl.
342 * Otherwise this is the same as Kgio::PipeMethods#kgio_write
344 static VALUE
kgio_send(VALUE io
, VALUE str
)
346 return my_send(io
, str
, 1);
350 * This method may be optimized on some systems (e.g. GNU/Linux) to use
351 * MSG_DONTWAIT to avoid explicitly setting the O_NONBLOCK flag via fcntl.
352 * Otherwise this is the same as Kgio::PipeMethods#kgio_trywrite
354 static VALUE
kgio_trysend(VALUE io
, VALUE str
)
356 return my_send(io
, str
, 0);
358 #else /* ! USE_MSG_DONTWAIT */
359 # define kgio_send kgio_write
360 # define kgio_trysend kgio_trywrite
361 #endif /* ! USE_MSG_DONTWAIT */
366 * Kgio.tryread(io, maxlen) -> buffer
367 * Kgio.tryread(io, maxlen, buffer) -> buffer
369 * Returns nil on EOF.
370 * Returns :wait_readable if EAGAIN is encountered.
372 * Maybe used in place of PipeMethods#kgio_tryread for non-Kgio objects
374 static VALUE
s_tryread(int argc
, VALUE
*argv
, VALUE mod
)
377 rb_raise(rb_eArgError
, "wrong number of arguments");
378 return my_read(0, argc
- 1, &argv
[1], argv
[0]);
384 * Kgio.trywrite(io, str) -> nil or :wait_writable
386 * Returns nil if the write was completed in full.
388 * Returns a String containing the unwritten portion if EAGAIN
389 * was encountered, but some portion was successfully written.
391 * Returns :wait_writable if EAGAIN is encountered and nothing
394 * Maybe used in place of PipeMethods#kgio_trywrite for non-Kgio objects
396 static VALUE
s_trywrite(VALUE mod
, VALUE io
, VALUE str
)
398 return my_write(io
, str
, 0);
401 void init_kgio_read_write(void)
403 VALUE mPipeMethods
, mSocketMethods
;
404 VALUE mKgio
= rb_define_module("Kgio");
405 VALUE mWaiters
= rb_const_get(mKgio
, rb_intern("DefaultWaiters"));
407 sym_wait_readable
= ID2SYM(rb_intern("wait_readable"));
408 sym_wait_writable
= ID2SYM(rb_intern("wait_writable"));
410 rb_define_singleton_method(mKgio
, "tryread", s_tryread
, -1);
411 rb_define_singleton_method(mKgio
, "trywrite", s_trywrite
, 2);
414 * Document-module: Kgio::PipeMethods
416 * This module may be used used to create classes that respond to
417 * various Kgio methods for reading and writing. This is included
418 * in Kgio::Pipe by default.
420 mPipeMethods
= rb_define_module_under(mKgio
, "PipeMethods");
421 rb_define_method(mPipeMethods
, "kgio_read", kgio_read
, -1);
422 rb_define_method(mPipeMethods
, "kgio_read!", kgio_read_bang
, -1);
423 rb_define_method(mPipeMethods
, "kgio_write", kgio_write
, 1);
424 rb_define_method(mPipeMethods
, "kgio_tryread", kgio_tryread
, -1);
425 rb_define_method(mPipeMethods
, "kgio_trywrite", kgio_trywrite
, 1);
428 * Document-module: Kgio::SocketMethods
430 * This method behaves like Kgio::PipeMethods, but contains
431 * optimizations for sockets on certain operating systems
434 mSocketMethods
= rb_define_module_under(mKgio
, "SocketMethods");
435 rb_define_method(mSocketMethods
, "kgio_read", kgio_recv
, -1);
436 rb_define_method(mSocketMethods
, "kgio_read!", kgio_recv_bang
, -1);
437 rb_define_method(mSocketMethods
, "kgio_write", kgio_send
, 1);
438 rb_define_method(mSocketMethods
, "kgio_tryread", kgio_tryrecv
, -1);
439 rb_define_method(mSocketMethods
, "kgio_trywrite", kgio_trysend
, 1);
442 * Returns the client IPv4 address of the socket in dotted quad
443 * form as a string. This is always the value of the
444 * Kgio::LOCALHOST constant for UNIX domain sockets.
446 rb_define_attr(mSocketMethods
, "kgio_addr", 1, 1);
448 eErrno_EPIPE
= rb_const_get(rb_mErrno
, rb_intern("EPIPE"));
449 eErrno_ECONNRESET
= rb_const_get(rb_mErrno
, rb_intern("ECONNRESET"));
450 rb_include_module(mPipeMethods
, mWaiters
);
451 rb_include_module(mSocketMethods
, mWaiters
);