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
|
# frozen_string_literal: true
require 'fiber'
begin
require 'io/nonblock'
rescue LoadError
# Ignore.
end
class Scheduler
def initialize
@readable = {}
@writable = {}
@waiting = {}
@blocking = []
@ios = ObjectSpace::WeakMap.new
end
attr :readable
attr :writable
attr :waiting
attr :blocking
def next_timeout
_fiber, timeout = @waiting.min_by{|key, value| value}
if timeout
offset = timeout - current_time
if offset < 0
return 0
else
return offset
end
end
end
def run
while @readable.any? or @writable.any? or @waiting.any?
# Can only handle file descriptors up to 1024...
readable, writable = IO.select(@readable.keys, @writable.keys, [], next_timeout)
# puts "readable: #{readable}" if readable&.any?
# puts "writable: #{writable}" if writable&.any?
readable&.each do |io|
@readable[io]&.resume
end
writable&.each do |io|
@writable[io]&.resume
end
if @waiting.any?
time = current_time
waiting = @waiting
@waiting = {}
waiting.each do |fiber, timeout|
if timeout <= time
fiber.resume
else
@waiting[fiber] = timeout
end
end
end
end
end
def for_fd(fd)
@ios[fd] ||= ::IO.for_fd(fd, autoclose: false)
end
def wait_readable(io)
@readable[io] = Fiber.current
Fiber.yield
@readable.delete(io)
return true
end
def wait_readable_fd(fd)
wait_readable(
for_fd(fd)
)
end
def wait_writable(io)
@writable[io] = Fiber.current
Fiber.yield
@writable.delete(io)
return true
end
def wait_writable_fd(fd)
wait_writable(
for_fd(fd)
)
end
def current_time
Process.clock_gettime(Process::CLOCK_MONOTONIC)
end
def wait_sleep(duration = nil)
@waiting[Fiber.current] = current_time + duration
Fiber.yield
return true
end
def wait_any(io, events, duration)
unless (events & IO::WAIT_READABLE).zero?
@readable[io] = Fiber.current
end
unless (events & IO::WAIT_WRITABLE).zero?
@writable[io] = Fiber.current
end
Fiber.yield
@readable.delete(io)
@writable.delete(io)
return true
end
def wait_for_single_fd(fd, events, duration)
wait_any(
for_fd(fd),
events,
duration
)
end
def enter_blocking_region
# puts "Enter blocking region: #{caller.first}"
end
def exit_blocking_region
# puts "Exit blocking region: #{caller.first}"
@blocking << caller.first
end
def fiber(&block)
fiber = Fiber.new(blocking: false, &block)
fiber.resume
return fiber
end
end
|
