Skip to content

Commit 3222824

Browse files
committed
support Ractor
1. Introduce State to store all status. 2. Store State instance to the Ractor local storage if possible 3. Make `GET_TIME` (Method object) shareable if possible 3 is supporeted Ruby 4.0 and later, so the Rator support is works only on Ruby 4.0 and later.
1 parent cd51eac commit 3222824

2 files changed

Lines changed: 125 additions & 59 deletions

File tree

lib/timeout.rb

Lines changed: 105 additions & 59 deletions
Original file line numberDiff line numberDiff line change
@@ -44,12 +44,105 @@ def self.handle_timeout(message) # :nodoc:
4444
end
4545

4646
# :stopdoc:
47-
CONDVAR = ConditionVariable.new
48-
QUEUE = Queue.new
49-
QUEUE_MUTEX = Mutex.new
50-
TIMEOUT_THREAD_MUTEX = Mutex.new
51-
@timeout_thread = nil
52-
private_constant :CONDVAR, :QUEUE, :QUEUE_MUTEX, :TIMEOUT_THREAD_MUTEX
47+
48+
# We keep a private reference so that time mocking libraries won't break
49+
# Timeout.
50+
GET_TIME =
51+
if defined?(Ractor.make_shareable)
52+
begin
53+
Ractor.make_shareable(Process.method(:clock_gettime))
54+
rescue # failed on Ruby 3.4
55+
Process.method(:clock_gettime)
56+
end
57+
else
58+
Process.method(:clock_gettime)
59+
end
60+
61+
private_constant :GET_TIME
62+
63+
class State
64+
attr_reader :condvar, :queue, :queue_mutex # shared with Timeout.timeout()
65+
66+
def initialize
67+
@condvar = ConditionVariable.new
68+
@queue = Queue.new
69+
@queue_mutex = Mutex.new
70+
71+
@timeout_thread = nil
72+
@timeout_thread_mutex = Mutex.new
73+
end
74+
75+
if defined?(Ractor.store_if_absent) &&
76+
defined?(Ractor.shareble?) && Ractor.shareable?(GET_TIME)
77+
78+
# Ractor support if
79+
# 1. Ractor.store_if_absent is available
80+
# 2. Method object can be shareable (4.0~)
81+
82+
Ractor.store_if_absent :timeout_gem_state do
83+
State.new
84+
end
85+
86+
def self.instance
87+
Ractor[:timeout_gem_state]
88+
end
89+
90+
::Timeout::RACTOR_SUPPORT = true # for test
91+
else
92+
def self.instance
93+
@GLOBAL_STATE ||= State.new
94+
end
95+
end
96+
97+
def create_timeout_thread
98+
watcher = Thread.new do
99+
requests = []
100+
while true
101+
until @queue.empty? and !requests.empty? # wait to have at least one request
102+
req = @queue.pop
103+
requests << req unless req.done?
104+
end
105+
closest_deadline = requests.min_by(&:deadline).deadline
106+
107+
now = 0.0
108+
@queue_mutex.synchronize do
109+
while (now = GET_TIME.call(Process::CLOCK_MONOTONIC)) < closest_deadline and @queue.empty?
110+
@condvar.wait(@queue_mutex, closest_deadline - now)
111+
end
112+
end
113+
114+
requests.each do |req|
115+
req.interrupt if req.expired?(now)
116+
end
117+
requests.reject!(&:done?)
118+
end
119+
end
120+
121+
if !watcher.group.enclosed? && (!defined?(Ractor.main?) || Ractor.main?)
122+
ThreadGroup::Default.add(watcher)
123+
end
124+
125+
watcher.name = "Timeout stdlib thread"
126+
watcher.thread_variable_set(:"\0__detached_thread__", true)
127+
watcher
128+
end
129+
130+
def ensure_timeout_thread_created
131+
unless @timeout_thread&.alive?
132+
# If the Mutex is already owned we are in a signal handler.
133+
# In that case, just return and let the main thread create the Timeout thread.
134+
return if @timeout_thread_mutex.owned?
135+
136+
@timeout_thread_mutex.synchronize do
137+
unless @timeout_thread&.alive?
138+
@timeout_thread = create_timeout_thread
139+
end
140+
end
141+
end
142+
end
143+
end
144+
145+
private_constant :State
53146

54147
class Request
55148
attr_reader :deadline
@@ -91,55 +184,6 @@ def finished
91184
end
92185
private_constant :Request
93186

94-
def self.create_timeout_thread
95-
watcher = Thread.new do
96-
requests = []
97-
while true
98-
until QUEUE.empty? and !requests.empty? # wait to have at least one request
99-
req = QUEUE.pop
100-
requests << req unless req.done?
101-
end
102-
closest_deadline = requests.min_by(&:deadline).deadline
103-
104-
now = 0.0
105-
QUEUE_MUTEX.synchronize do
106-
while (now = GET_TIME.call(Process::CLOCK_MONOTONIC)) < closest_deadline and QUEUE.empty?
107-
CONDVAR.wait(QUEUE_MUTEX, closest_deadline - now)
108-
end
109-
end
110-
111-
requests.each do |req|
112-
req.interrupt if req.expired?(now)
113-
end
114-
requests.reject!(&:done?)
115-
end
116-
end
117-
ThreadGroup::Default.add(watcher) unless watcher.group.enclosed?
118-
watcher.name = "Timeout stdlib thread"
119-
watcher.thread_variable_set(:"\0__detached_thread__", true)
120-
watcher
121-
end
122-
private_class_method :create_timeout_thread
123-
124-
def self.ensure_timeout_thread_created
125-
unless @timeout_thread and @timeout_thread.alive?
126-
# If the Mutex is already owned we are in a signal handler.
127-
# In that case, just return and let the main thread create the @timeout_thread.
128-
return if TIMEOUT_THREAD_MUTEX.owned?
129-
TIMEOUT_THREAD_MUTEX.synchronize do
130-
unless @timeout_thread and @timeout_thread.alive?
131-
@timeout_thread = create_timeout_thread
132-
end
133-
end
134-
end
135-
end
136-
private_class_method :ensure_timeout_thread_created
137-
138-
# We keep a private reference so that time mocking libraries won't break
139-
# Timeout.
140-
GET_TIME = Process.method(:clock_gettime)
141-
private_constant :GET_TIME
142-
143187
# :startdoc:
144188

145189
# Perform an operation in a block, raising an error if it takes longer than
@@ -178,12 +222,14 @@ def self.timeout(sec, klass = nil, message = nil, &block) #:yield: +sec+
178222
return scheduler.timeout_after(sec, klass || Error, message, &block)
179223
end
180224

181-
ensure_timeout_thread_created
225+
state = State.instance
226+
state.ensure_timeout_thread_created
227+
182228
perform = Proc.new do |exc|
183229
request = Request.new(Thread.current, sec, exc, message)
184-
QUEUE_MUTEX.synchronize do
185-
QUEUE << request
186-
CONDVAR.signal
230+
state.queue_mutex.synchronize do
231+
state.queue << request
232+
state.condvar.signal
187233
end
188234
begin
189235
return yield(sec)

test/test_timeout.rb

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -280,4 +280,24 @@ def test_handling_enclosed_threadgroup
280280
}.join
281281
end;
282282
end
283+
284+
def test_ractor
285+
assert_separately(%w[-rtimeout -W0], <<-'end;')
286+
r = Ractor.new do
287+
Timeout.timeout(1) { 42 }
288+
end.value
289+
290+
assert_equal 42, r
291+
292+
r = Ractor.new do
293+
begin
294+
Timeout.timeout(0.1) { sleep }
295+
rescue Timeout::Error
296+
:ok
297+
end
298+
end.value
299+
300+
assert_equal :ok, r
301+
end;
302+
end if Timeout.const_defined?(:RACTOR_SUPPORT)
283303
end

0 commit comments

Comments
 (0)