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
|
# frozen_string_literal: true
# This is an example and simplified scheduler for test purposes.
# It is not efficient for a large number of file descriptors as it uses IO.select().
# Production Fiber schedulers should use epoll/kqueue/etc.
require 'fiber'
require 'socket'
begin
require 'io/nonblock'
rescue LoadError
# Ignore.
end
class Scheduler
def initialize
@readable = {}
@writable = {}
@waiting = {}
@lock = Mutex.new
@locking = 0
@ready = []
end
attr :readable
attr :writable
attr :waiting
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
@urgent = IO.pipe
while @readable.any? or @writable.any? or @waiting.any? or @locking.positive?
# Can only handle file descriptors up to 1024...
readable, writable = IO.select(@readable.keys + [@urgent.first], @writable.keys, [], next_timeout)
# puts "readable: #{readable}" if readable&.any?
# puts "writable: #{writable}" if writable&.any?
readable&.each do |io|
if fiber = @readable.delete(io)
fiber.resume
elsif io == @urgent.first
@urgent.first.read_nonblock(1024)
end
end
writable&.each do |io|
if fiber = @writable.delete(io)
fiber.resume
end
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
if @ready.any?
ready = nil
@lock.synchronize do
ready, @ready = @ready, []
end
ready.each do |fiber|
fiber.resume
end
end
end
ensure
@urgent.each(&:close)
@urgent = nil
end
def current_time
Process.clock_gettime(Process::CLOCK_MONOTONIC)
end
def io_wait(io, events, duration)
unless (events & IO::READABLE).zero?
@readable[io] = Fiber.current
end
unless (events & IO::WRITABLE).zero?
@writable[io] = Fiber.current
end
Fiber.yield
return true
end
def kernel_sleep(duration = nil)
# p [__method__, duration]
if duration
@waiting[Fiber.current] = current_time + duration
end
Fiber.yield
return true
end
def block(blocker)
# p [__method__, blocker]
@locking += 1
Fiber.yield
ensure
@locking -= 1
end
def unblock(blocker, fiber)
# p [__method__, blocker, fiber]
@lock.synchronize do
@ready << fiber
end
if io = @urgent&.last
io.write_nonblock('.')
end
end
def fiber(&block)
fiber = Fiber.new(blocking: false, &block)
fiber.resume
return fiber
end
end
|
