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
127
128
129
130
131
|
module NIO
# Selectors monitor IO objects for events of interest
class Selector
# Create a new NIO::Selector
def initialize
@selectables = {}
@lock = Mutex.new
# Other threads can wake up a selector
@wakeup, @waker = IO.pipe
@closed = false
end
# Register interest in an IO object with the selector for the given types
# of events. Valid event types for interest are:
# * :r - is the IO readable?
# * :w - is the IO writeable?
# * :rw - is the IO either readable or writeable?
def register(io, interest)
@lock.synchronize do
if closed?
raise IOError, "selector is closed"
end
if monitor = @selectables[io]
raise ArgumentError, "this IO is already registered with the selector as #{monitor.interests.inspect}"
end
monitor = Monitor.new(io, interest, self)
@selectables[monitor.io] = monitor
monitor
end
end
# Deregister the given IO object from the selector
def deregister(io)
@lock.synchronize do
monitor = @selectables.delete io
monitor.close(false) if monitor and not monitor.closed?
monitor
end
end
# Is the given IO object registered with the selector?
def registered?(io)
@lock.synchronize { @selectables.has_key? io }
end
# Select which monitors are ready
def select(timeout = nil)
@lock.synchronize do
readers, writers = [@wakeup], []
@selectables.each do |io, monitor|
readers << io if monitor.interests == :r || monitor.interests == :rw
writers << io if monitor.interests == :w || monitor.interests == :rw
monitor.readiness = nil
end
ready_readers, ready_writers = Kernel.select readers, writers, [], timeout
return unless ready_readers # timeout or wakeup
selected_monitors = Set.new
ready_readers.each do |io|
if io == @wakeup
# Clear all wakeup signals we've received by reading them
# Wakeups should have level triggered behavior
@wakeup.read(@wakeup.stat.size)
return
else
monitor = @selectables[io]
monitor.readiness = :r
selected_monitors << monitor
end
end
ready_writers.each do |io|
monitor = @selectables[io]
monitor.readiness = case monitor.readiness
when :r
:rw
else
:w
end
selected_monitors << monitor
end
if block_given?
selected_monitors.each do |m|
yield m
end
selected_monitors.size
else
selected_monitors
end
end
end
# Wake up a thread that's in the middle of selecting on this selector, if
# any such thread exists.
#
# Invoking this method more than once between two successive select calls
# has the same effect as invoking it just once. In other words, it provides
# level-triggered behavior.
def wakeup
# Send the selector a signal in the form of writing data to a pipe
@waker.write "\0"
nil
end
# Close this selector and free its resources
def close
@lock.synchronize do
return if @closed
@wakeup.close rescue nil
@waker.close rescue nil
@closed = true
end
end
# Is this selector closed?
def closed?; @closed end
def empty?
@selectables.empty?
end
end
end
|
