2 static VALUE mKgio_WaitReadable
, mKgio_WaitWritable
;
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 *));
18 static void raise_empty_bt(VALUE err
, const char *msg
)
20 VALUE exc
= rb_exc_new2(err
, msg
);
21 VALUE bt
= rb_ary_new();
23 rb_funcall(exc
, rb_intern("set_backtrace"), 1, bt
);
27 static void my_eof_error(void)
29 raise_empty_bt(rb_eEOFError
, "");
32 static void wr_sys_fail(const char *msg
)
37 raise_empty_bt(eErrno_EPIPE
, msg
);
40 raise_empty_bt(eErrno_ECONNRESET
, msg
);
45 static void prepare_read(struct io_args
*a
, int argc
, VALUE
*argv
, VALUE io
)
50 a
->fd
= my_fileno(io
);
51 rb_scan_args(argc
, argv
, "11", &length
, &a
->buf
);
52 a
->len
= NUM2LONG(length
);
54 a
->buf
= rb_str_new(NULL
, a
->len
);
57 rb_str_resize(a
->buf
, a
->len
);
59 a
->ptr
= RSTRING_PTR(a
->buf
);
62 static int read_check(struct io_args
*a
, long n
, const char *msg
, int io_wait
)
67 rb_str_set_len(a
->buf
, 0);
68 if (errno
== EAGAIN
) {
70 kgio_wait_readable(a
->io
, a
->fd
);
72 /* buf may be modified in other thread/fiber */
73 rb_str_resize(a
->buf
, a
->len
);
74 a
->ptr
= RSTRING_PTR(a
->buf
);
77 a
->buf
= mKgio_WaitReadable
;
83 rb_str_set_len(a
->buf
, n
);
89 static VALUE
my_read(int io_wait
, int argc
, VALUE
*argv
, VALUE io
)
94 prepare_read(&a
, argc
, argv
, io
);
95 set_nonblocking(a
.fd
);
97 n
= (long)read(a
.fd
, a
.ptr
, a
.len
);
98 if (read_check(&a
, n
, "read", io_wait
) != 0)
106 * io.kgio_read(maxlen) -> buffer
107 * io.kgio_read(maxlen, buffer) -> buffer
109 * Reads at most maxlen bytes from the stream socket. Returns with a
110 * newly allocated buffer, or may reuse an existing buffer if supplied.
112 * Calls the method assigned to Kgio.wait_readable, or blocks in a
113 * thread-safe manner for writability.
115 * Returns nil on EOF.
117 * This behaves like read(2) and IO#readpartial, NOT fread(3) or
118 * IO#read which possess read-in-full behavior.
120 static VALUE
kgio_read(int argc
, VALUE
*argv
, VALUE io
)
122 return my_read(1, argc
, argv
, io
);
126 * Same as Kgio::PipeMethods#kgio_read, except EOFError is raised
127 * on EOF without a backtrace
129 static VALUE
kgio_read_bang(int argc
, VALUE
*argv
, VALUE io
)
131 VALUE rv
= my_read(1, argc
, argv
, io
);
133 if (NIL_P(rv
)) my_eof_error();
140 * io.kgio_tryread(maxlen) -> buffer
141 * io.kgio_tryread(maxlen, buffer) -> buffer
143 * Reads at most maxlen bytes from the stream socket. Returns with a
144 * newly allocated buffer, or may reuse an existing buffer if supplied.
146 * Returns nil on EOF.
148 * Returns Kgio::WaitReadable if EAGAIN is encountered.
150 static VALUE
kgio_tryread(int argc
, VALUE
*argv
, VALUE io
)
152 return my_read(0, argc
, argv
, io
);
155 #ifdef USE_MSG_DONTWAIT
156 static VALUE
my_recv(int io_wait
, int argc
, VALUE
*argv
, VALUE io
)
161 prepare_read(&a
, argc
, argv
, io
);
163 n
= (long)recv(a
.fd
, a
.ptr
, a
.len
, MSG_DONTWAIT
);
164 if (read_check(&a
, n
, "recv", io_wait
) != 0)
170 * This method may be optimized on some systems (e.g. GNU/Linux) to use
171 * MSG_DONTWAIT to avoid explicitly setting the O_NONBLOCK flag via fcntl.
172 * Otherwise this is the same as Kgio::PipeMethods#kgio_read
174 static VALUE
kgio_recv(int argc
, VALUE
*argv
, VALUE io
)
176 return my_recv(1, argc
, argv
, io
);
180 * Same as Kgio::SocketMethods#kgio_read, except EOFError is raised
181 * on EOF without a backtrace
183 static VALUE
kgio_recv_bang(int argc
, VALUE
*argv
, VALUE io
)
185 VALUE rv
= my_recv(1, argc
, argv
, io
);
187 if (NIL_P(rv
)) my_eof_error();
192 * This method may be optimized on some systems (e.g. GNU/Linux) to use
193 * MSG_DONTWAIT to avoid explicitly setting the O_NONBLOCK flag via fcntl.
194 * Otherwise this is the same as Kgio::PipeMethods#kgio_tryread
196 static VALUE
kgio_tryrecv(int argc
, VALUE
*argv
, VALUE io
)
198 return my_recv(0, argc
, argv
, io
);
200 #else /* ! USE_MSG_DONTWAIT */
201 # define kgio_recv kgio_read
202 # define kgio_recv_bang kgio_read_bang
203 # define kgio_tryrecv kgio_tryread
204 #endif /* USE_MSG_DONTWAIT */
206 static void prepare_write(struct io_args
*a
, VALUE io
, VALUE str
)
208 a
->buf
= (TYPE(str
) == T_STRING
) ? str
: rb_obj_as_string(str
);
209 a
->ptr
= RSTRING_PTR(a
->buf
);
210 a
->len
= RSTRING_LEN(a
->buf
);
212 a
->fd
= my_fileno(io
);
215 static int write_check(struct io_args
*a
, long n
, const char *msg
, int io_wait
)
220 } else if (n
== -1) {
223 if (errno
== EAGAIN
) {
224 long written
= RSTRING_LEN(a
->buf
) - a
->len
;
227 kgio_wait_writable(a
->io
, a
->fd
);
229 /* buf may be modified in other thread/fiber */
230 a
->len
= RSTRING_LEN(a
->buf
) - written
;
233 a
->ptr
= RSTRING_PTR(a
->buf
) + written
;
235 } else if (written
> 0) {
236 a
->buf
= rb_str_new(a
->ptr
+ n
, a
->len
- n
);
238 a
->buf
= mKgio_WaitWritable
;
244 assert(n
>= 0 && n
< a
->len
&& "write/send syscall broken?");
252 static VALUE
my_write(VALUE io
, VALUE str
, int io_wait
)
257 prepare_write(&a
, io
, str
);
258 set_nonblocking(a
.fd
);
260 n
= (long)write(a
.fd
, a
.ptr
, a
.len
);
261 if (write_check(&a
, n
, "write", io_wait
) != 0)
269 * io.kgio_write(str) -> nil
271 * Returns nil when the write completes.
273 * Calls the method Kgio.wait_writable if it is set. Otherwise this
274 * blocks in a thread-safe manner until all data is written or a
275 * fatal error occurs.
277 static VALUE
kgio_write(VALUE io
, VALUE str
)
279 return my_write(io
, str
, 1);
285 * io.kgio_trywrite(str) -> nil or Kgio::WaitWritable
287 * Returns nil if the write was completed in full.
289 * Returns a String containing the unwritten portion if EAGAIN
290 * was encountered, but some portion was successfully written.
292 * Returns Kgio::WaitWritable if EAGAIN is encountered and nothing
295 static VALUE
kgio_trywrite(VALUE io
, VALUE str
)
297 return my_write(io
, str
, 0);
300 #ifdef USE_MSG_DONTWAIT
302 * This method behaves like Kgio::PipeMethods#kgio_write, except
303 * it will use send(2) with the MSG_DONTWAIT flag on sockets to
304 * avoid unnecessary calls to fcntl(2).
306 static VALUE
my_send(VALUE io
, VALUE str
, int io_wait
)
311 prepare_write(&a
, io
, str
);
313 n
= (long)send(a
.fd
, a
.ptr
, a
.len
, MSG_DONTWAIT
);
314 if (write_check(&a
, n
, "send", io_wait
) != 0)
320 * This method may be optimized on some systems (e.g. GNU/Linux) to use
321 * MSG_DONTWAIT to avoid explicitly setting the O_NONBLOCK flag via fcntl.
322 * Otherwise this is the same as Kgio::PipeMethods#kgio_write
324 static VALUE
kgio_send(VALUE io
, VALUE str
)
326 return my_send(io
, str
, 1);
330 * This method may be optimized on some systems (e.g. GNU/Linux) to use
331 * MSG_DONTWAIT to avoid explicitly setting the O_NONBLOCK flag via fcntl.
332 * Otherwise this is the same as Kgio::PipeMethods#kgio_trywrite
334 static VALUE
kgio_trysend(VALUE io
, VALUE str
)
336 return my_send(io
, str
, 0);
338 #else /* ! USE_MSG_DONTWAIT */
339 # define kgio_send kgio_write
340 # define kgio_trysend kgio_trywrite
341 #endif /* ! USE_MSG_DONTWAIT */
343 void init_kgio_read_write(VALUE mKgio
)
345 VALUE mPipeMethods
, mSocketMethods
;
347 mKgio_WaitReadable
= rb_const_get(mKgio
, rb_intern("WaitReadable"));
348 mKgio_WaitWritable
= rb_const_get(mKgio
, rb_intern("WaitWritable"));
351 * Document-module: Kgio::PipeMethods
353 * This module may be used used to create classes that respond to
354 * various Kgio methods for reading and writing. This is included
355 * in Kgio::Pipe by default.
357 mPipeMethods
= rb_define_module_under(mKgio
, "PipeMethods");
358 rb_define_method(mPipeMethods
, "kgio_read", kgio_read
, -1);
359 rb_define_method(mPipeMethods
, "kgio_read!", kgio_read_bang
, -1);
360 rb_define_method(mPipeMethods
, "kgio_write", kgio_write
, 1);
361 rb_define_method(mPipeMethods
, "kgio_tryread", kgio_tryread
, -1);
362 rb_define_method(mPipeMethods
, "kgio_trywrite", kgio_trywrite
, 1);
365 * Document-module: Kgio::SocketMethods
367 * This method behaves like Kgio::PipeMethods, but contains
368 * optimizations for sockets on certain operating systems
371 mSocketMethods
= rb_define_module_under(mKgio
, "SocketMethods");
372 rb_define_method(mSocketMethods
, "kgio_read", kgio_recv
, -1);
373 rb_define_method(mSocketMethods
, "kgio_read!", kgio_recv_bang
, -1);
374 rb_define_method(mSocketMethods
, "kgio_write", kgio_send
, 1);
375 rb_define_method(mSocketMethods
, "kgio_tryread", kgio_tryrecv
, -1);
376 rb_define_method(mSocketMethods
, "kgio_trywrite", kgio_trysend
, 1);
379 * Returns the client IPv4 address of the socket in dotted quad
380 * form as a string. This is always the value of the
381 * Kgio::LOCALHOST constant for UNIX domain sockets.
383 rb_define_attr(mSocketMethods
, "kgio_addr", 1, 1);
385 eErrno_EPIPE
= rb_const_get(rb_mErrno
, rb_intern("EPIPE"));
386 eErrno_ECONNRESET
= rb_const_get(rb_mErrno
, rb_intern("ECONNRESET"));