1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
|
#include "kgio.h"
#include "my_fileno.h"
static ID id_wait_rd, id_wait_wr;
#if defined(HAVE_RB_TIME_INTERVAL) && defined(HAVE_RB_WAIT_FOR_SINGLE_FD)
static int kgio_timedwait(VALUE self, VALUE timeout, int write_p)
{
struct timeval tv = rb_time_interval(timeout);
int events = write_p ? RB_WAITFD_OUT : RB_WAITFD_IN;
return rb_wait_for_single_fd(my_fileno(self), events, &tv);
}
#else /* ! (HAVE_RB_TIME_INTERVAL && HAVE_RB_WAIT_FOR_SINGLE_FD) */
static int kgio_timedwait(VALUE self, VALUE timeout, int write_p)
{
VALUE argv[4];
VALUE set = rb_ary_new3(1, self);
argv[0] = write_p ? Qnil : set;
argv[1] = write_p ? set : Qnil;
argv[2] = Qnil;
argv[3] = timeout;
set = rb_funcall2(rb_cIO, rb_intern("select"), 4, argv);
return NIL_P(set) ? 0 : 1;
}
#endif /* ! (HAVE_RB_TIME_INTERVAL && HAVE_RB_WAIT_FOR_SINGLE_FD) */
static int kgio_wait(int argc, VALUE *argv, VALUE self, int write_p)
{
int fd;
VALUE timeout;
if (rb_scan_args(argc, argv, "01", &timeout) == 1 && !NIL_P(timeout))
return kgio_timedwait(self, timeout, write_p);
fd = my_fileno(self);
errno = EAGAIN;
write_p ? rb_io_wait_writable(fd) : rb_io_wait_readable(fd);
return 1;
}
/*
* call-seq:
*
* io.kgio_wait_readable -> IO
* io.kgio_wait_readable(timeout) -> IO or nil
*
* Blocks the running Thread indefinitely until the IO object is readable
* or if +timeout+ expires. If +timeout+ is specified and expires, +nil+
* is returned.
*
* This method is automatically called (without timeout argument) by default
* whenever kgio_read needs to block on input.
*
* Users of alternative threading/fiber libraries are
* encouraged to override this method in their subclasses or modules to
* work with their threading/blocking methods.
*/
static VALUE kgio_wait_readable(int argc, VALUE *argv, VALUE self)
{
int r = kgio_wait(argc, argv, self, 0);
if (r < 0) rb_sys_fail("kgio_wait_readable");
return r == 0 ? Qnil : self;
}
/*
* call-seq:
*
* io.kgio_wait_writable -> IO
* io.kgio_wait_writable(timeout) -> IO or nil
*
* Blocks the running Thread indefinitely until the IO object is writable
* or if +timeout+ expires. If +timeout+ is specified and expires, +nil+
* is returned.
*
* This method is automatically called (without timeout argument) by default
* whenever kgio_write needs to block on output.
*
* Users of alternative threading/fiber libraries are
* encouraged to override this method in their subclasses or modules to
* work with their threading/blocking methods.
*/
static VALUE kgio_wait_writable(int argc, VALUE *argv, VALUE self)
{
int r = kgio_wait(argc, argv, self, 1);
if (r < 0) rb_sys_fail("kgio_wait_writable");
return r == 0 ? Qnil : self;
}
VALUE kgio_call_wait_writable(VALUE io)
{
return rb_funcall(io, id_wait_wr, 0);
}
VALUE kgio_call_wait_readable(VALUE io)
{
return rb_funcall(io, id_wait_rd, 0);
}
void init_kgio_wait(void)
{
VALUE mKgio = rb_define_module("Kgio");
/*
* Document-module: Kgio::DefaultWaiters
*
* This module contains default kgio_wait_readable and
* kgio_wait_writable methods that block indefinitely (in a
* thread-safe manner) until an IO object is read or writable.
* This module is included in the Kgio::PipeMethods and
* Kgio::SocketMethods modules used by all bundled IO-derived
* objects.
*/
VALUE mWaiters = rb_define_module_under(mKgio, "DefaultWaiters");
id_wait_rd = rb_intern("kgio_wait_readable");
id_wait_wr = rb_intern("kgio_wait_writable");
rb_define_method(mWaiters, "kgio_wait_readable",
kgio_wait_readable, -1);
rb_define_method(mWaiters, "kgio_wait_writable",
kgio_wait_writable, -1);
}
|
