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
|
# frozen_string_literal: true
require 'fiber'
require 'socket'
begin
require 'io/nonblock'
rescue LoadError
# Ignore.
end
class Scheduler
def initialize
@readable = {}
@writable = {}
@waiting = {}
@urgent = nil
@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|
@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
if @ready.any?
# Clear out the urgent notification pipe.
@urgent.first.read_nonblock(1024)
ready = nil
@lock.synchronize do
ready, @ready = @ready, Array.new
end
ready.each do |fiber|
fiber.resume
end
end
end
ensure
@urgent.each(&:close)
end
def current_time
Process.clock_gettime(Process::CLOCK_MONOTONIC)
end
def kernel_sleep(duration = nil)
@waiting[Fiber.current] = current_time + duration
Fiber.yield
return true
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
@readable.delete(io)
@writable.delete(io)
return true
end
def mutex_lock(mutex)
@locking += 1
Fiber.yield
ensure
@locking -= 1
end
def mutex_unlock(mutex, fiber)
@lock.synchronize do
@ready << fiber
if @urgent
@urgent.last.write('.')
end
end
end
def fiber(&block)
fiber = Fiber.new(blocking: false, &block)
fiber.resume
return fiber
end
end
|