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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
|
# frozen_string_literal: true
# Released under the MIT License.
# Copyright, 2011-2017, by Tony Arcieri.
# Copyright, 2012, by Logan Bowers.
# Copyright, 2013, by Sadayuki Furuhashi.
# Copyright, 2013, by Stephen von Takach.
# Copyright, 2013, by Tim Carey-Smith.
# Copyright, 2013, by Ravil Bayramgalin.
# Copyright, 2014, by Sergey Avseyev.
# Copyright, 2014, by John Thornton.
# Copyright, 2015, by Vladimir Kochnev.
# Copyright, 2015, by Upekshe Jayasekera.
# Copyright, 2019-2020, by Gregory Longtin.
# Copyright, 2020-2021, by Joao Fernandes.
# Copyright, 2023, by Samuel Williams.
require "set"
module NIO
# Selectors monitor IO objects for events of interest
class Selector
# Return supported backends as symbols
#
# See `#backend` method definition for all possible backends
def self.backends
[:ruby]
end
# Create a new NIO::Selector
def initialize(backend = :ruby)
raise ArgumentError, "unsupported backend: #{backend}" unless [:ruby, nil].include?(backend)
@selectables = {}
@lock = Mutex.new
# Other threads can wake up a selector
@wakeup, @waker = IO.pipe
@closed = false
end
# Return a symbol representing the backend I/O multiplexing mechanism used.
# Supported backends are:
# * :ruby - pure Ruby (i.e IO.select)
# * :java - Java NIO on JRuby
# * :epoll - libev w\ Linux epoll
# * :poll - libev w\ POSIX poll
# * :kqueue - libev w\ BSD kqueue
# * :select - libev w\ SysV select
# * :port - libev w\ I/O completion ports
# * :linuxaio - libev w\ Linux AIO io_submit (experimental)
# * :io_uring - libev w\ Linux io_uring (experimental)
# * :unknown - libev w\ unknown backend
def backend
:ruby
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)
unless defined?(::OpenSSL) && io.is_a?(::OpenSSL::SSL::SSLSocket)
io = IO.try_convert(io)
end
@lock.synchronize do
raise IOError, "selector is closed" if closed?
monitor = @selectables[io]
raise ArgumentError, "already registered as #{monitor.interests.inspect}" if monitor
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.try_convert(io)
monitor.close(false) if monitor && !monitor.closed?
monitor
end
end
# Is the given IO object registered with the selector?
def registered?(io)
@lock.synchronize { @selectables.key? io }
end
# Select which monitors are ready
def select(timeout = nil)
selected_monitors = Set.new
@lock.synchronize do
readers = [@wakeup]
writers = []
@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
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)
else
monitor = @selectables[io]
monitor.readiness = :r
selected_monitors << monitor
end
end
ready_writers.each do |io|
monitor = @selectables[io]
monitor.readiness = monitor.readiness == :r ? :rw : :w
selected_monitors << monitor
end
end
if block_given?
selected_monitors.each { |m| yield m }
selected_monitors.size
else
selected_monitors.to_a
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
begin
@waker.write_nonblock "\0"
rescue IO::WaitWritable
# This indicates the wakeup pipe is full, which means the other thread
# has already received many wakeup calls, but not processed them yet.
# The other thread will completely drain this pipe when it wakes up,
# so it's ok to ignore this exception if it occurs: we know the other
# thread has already been signaled to wake up
end
nil
end
# Close this selector and free its resources
def close
@lock.synchronize do
return if @closed
begin
@wakeup.close
rescue IOError
end
begin
@waker.close
rescue IOError
end
@closed = true
end
end
# Is this selector closed?
def closed?
@closed
end
def empty?
@selectables.empty?
end
end
end
|
