From ab6746c23d73de4027e2bef14cd0c9e7f8a951f2 Mon Sep 17 00:00:00 2001 From: Charles Oliver Nutter Date: Wed, 31 Aug 2022 13:27:24 -0500 Subject: [PATCH 1/3] Terminate the timeout thread at exit In some runtimes, the process may not go away but the Ruby runtime should be shut down. This thread does not clean itself up, which leads to such runtimes leaning on a best-effort kill+join of the dangling threads late in the shutdown process. As a result these threads may linger slightly after shutdown and fail expectations that they will have completed and finalized. This patch adds an `at_exit` for the timeout library to terminate its own thread, which would happen earlier in the shutdown process (`at_exit` execution is considered part of the "user" shutdown phase in JRuby). This appears to fix issues running JRuby's embedding API tests that check for timeout threads to have been cleanly terminated and finalized. --- lib/timeout.rb | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/lib/timeout.rb b/lib/timeout.rb index 2aad1d7..f525846 100644 --- a/lib/timeout.rb +++ b/lib/timeout.rb @@ -103,6 +103,9 @@ def self.create_timeout_thread while true until QUEUE.empty? and !requests.empty? # wait to have at least one request req = QUEUE.pop + + Thread.current.kill if req.nil? + requests << req unless req.done? end closest_deadline = requests.min_by(&:deadline).deadline @@ -131,6 +134,12 @@ def self.ensure_timeout_thread_created TIMEOUT_THREAD_MUTEX.synchronize do unless @timeout_thread and @timeout_thread.alive? @timeout_thread = create_timeout_thread + + # shut down timeout queue and wait for thread termination at exit + Kernel.at_exit do + QUEUE.close + @timeout_thread.join + end end end end From af4d8c51232c694ef90963d1628e95392ee1791f Mon Sep 17 00:00:00 2001 From: Charles Oliver Nutter Date: Wed, 7 Sep 2022 14:29:00 -0500 Subject: [PATCH 2/3] Allow restarting timeout thread and queue If the shutdown at_exit runs before other at_exit or termination logic that still depends on the timeout thread, those subsequent uses of timeout may fail. This additional patch allows the timeout thread and queue to be restarted after the at_exit has fired, with the side effect that it will not try to shutdown a second time. This addresses review comments from #21. --- lib/timeout.rb | 22 +++++++++++++--------- 1 file changed, 13 insertions(+), 9 deletions(-) diff --git a/lib/timeout.rb b/lib/timeout.rb index f525846..2a0a3e7 100644 --- a/lib/timeout.rb +++ b/lib/timeout.rb @@ -51,11 +51,11 @@ def exception(*) # :stopdoc: CONDVAR = ConditionVariable.new - QUEUE = Queue.new QUEUE_MUTEX = Mutex.new TIMEOUT_THREAD_MUTEX = Mutex.new + @@queue = Queue.new @timeout_thread = nil - private_constant :CONDVAR, :QUEUE, :QUEUE_MUTEX, :TIMEOUT_THREAD_MUTEX + private_constant :CONDVAR, :QUEUE_MUTEX, :TIMEOUT_THREAD_MUTEX class Request attr_reader :deadline @@ -101,8 +101,8 @@ def self.create_timeout_thread watcher = Thread.new do requests = [] while true - until QUEUE.empty? and !requests.empty? # wait to have at least one request - req = QUEUE.pop + until @@queue.empty? and !requests.empty? # wait to have at least one request + req = @@queue.pop Thread.current.kill if req.nil? @@ -112,7 +112,7 @@ def self.create_timeout_thread now = 0.0 QUEUE_MUTEX.synchronize do - while (now = GET_TIME.call(Process::CLOCK_MONOTONIC)) < closest_deadline and QUEUE.empty? + while (now = GET_TIME.call(Process::CLOCK_MONOTONIC)) < closest_deadline and @@queue.empty? CONDVAR.wait(QUEUE_MUTEX, closest_deadline - now) end end @@ -136,9 +136,13 @@ def self.ensure_timeout_thread_created @timeout_thread = create_timeout_thread # shut down timeout queue and wait for thread termination at exit - Kernel.at_exit do - QUEUE.close - @timeout_thread.join + @shutdown_proc ||= Kernel.at_exit do + QUEUE_MUTEX.synchronize do + @@queue.close + @timeout_thread&.join + @@queue = Queue.new + @timeout_thread = nil + end end end end @@ -190,7 +194,7 @@ def timeout(sec, klass = nil, message = nil, &block) #:yield: +sec+ perform = Proc.new do |exc| request = Request.new(Thread.current, sec, exc, message) QUEUE_MUTEX.synchronize do - QUEUE << request + @@queue << request CONDVAR.signal end begin From b0150d098edef5a6e2459bfc71befd5a5d9f6196 Mon Sep 17 00:00:00 2001 From: Charles Oliver Nutter Date: Wed, 7 Sep 2022 15:21:40 -0500 Subject: [PATCH 3/3] Move executor logic into a new hidden class This allows more atomic shutdown; either the queue close or the thread terminating indicates the executor has been shut down, and a new one can be created and put into action without interfering with the shutdown of a previous one. --- lib/timeout.rb | 193 +++++++++++++++++++++++++++---------------------- 1 file changed, 105 insertions(+), 88 deletions(-) diff --git a/lib/timeout.rb b/lib/timeout.rb index 2a0a3e7..2da28f0 100644 --- a/lib/timeout.rb +++ b/lib/timeout.rb @@ -50,100 +50,135 @@ def exception(*) end # :stopdoc: - CONDVAR = ConditionVariable.new - QUEUE_MUTEX = Mutex.new - TIMEOUT_THREAD_MUTEX = Mutex.new - @@queue = Queue.new - @timeout_thread = nil - private_constant :CONDVAR, :QUEUE_MUTEX, :TIMEOUT_THREAD_MUTEX - - class Request - attr_reader :deadline - - def initialize(thread, timeout, exception_class, message) - @thread = thread - @deadline = GET_TIME.call(Process::CLOCK_MONOTONIC) + timeout - @exception_class = exception_class - @message = message - - @mutex = Mutex.new - @done = false # protected by @mutex + class Executor + CONDVAR = ConditionVariable.new + QUEUE_MUTEX = Mutex.new + private_constant :CONDVAR, :QUEUE_MUTEX + + def initialize + @queue = Queue.new + @timeout_thread = create_timeout_thread end - def done? - @mutex.synchronize do - @done + def submit(sec, klass, message) + perform = Proc.new do |exc| + request = Request.new(Thread.current, sec, exc, message) + QUEUE_MUTEX.synchronize do + @queue << request + CONDVAR.signal + end + begin + return yield(sec) + ensure + request.finished + end + end + + if klass + perform.call(klass) + else + backtrace = Error.catch(&perform) + raise Error, message, backtrace end end - def expired?(now) - now >= @deadline + def shutdown + @queue.close + @timeout_thread&.join end - def interrupt - @mutex.synchronize do - unless @done - @thread.raise @exception_class, @message - @done = true + def alive? + !@queue.closed? && @timeout_thread.alive? + end + + class Request + attr_reader :deadline + + def initialize(thread, timeout, exception_class, message) + @thread = thread + @deadline = GET_TIME.call(Process::CLOCK_MONOTONIC) + timeout + @exception_class = exception_class + @message = message + + @mutex = Mutex.new + @done = false # protected by @mutex + end + + def done? + @mutex.synchronize do + @done end end - end - def finished - @mutex.synchronize do - @done = true + def expired?(now) + now >= @deadline + end + + def interrupt + @mutex.synchronize do + unless @done + @thread.raise @exception_class, @message + @done = true + end + end + end + + def finished + @mutex.synchronize do + @done = true + end end end - end - private_constant :Request + private_constant :Request - def self.create_timeout_thread - watcher = Thread.new do - requests = [] - while true - until @@queue.empty? and !requests.empty? # wait to have at least one request - req = @@queue.pop + def create_timeout_thread + watcher = Thread.new do + requests = [] + while true + until @queue.empty? and !requests.empty? # wait to have at least one request + req = @queue.pop - Thread.current.kill if req.nil? + Thread.current.kill if req.nil? - requests << req unless req.done? - end - closest_deadline = requests.min_by(&:deadline).deadline + requests << req unless req.done? + end + closest_deadline = requests.min_by(&:deadline).deadline - now = 0.0 - QUEUE_MUTEX.synchronize do - while (now = GET_TIME.call(Process::CLOCK_MONOTONIC)) < closest_deadline and @@queue.empty? - CONDVAR.wait(QUEUE_MUTEX, closest_deadline - now) + now = 0.0 + QUEUE_MUTEX.synchronize do + while (now = GET_TIME.call(Process::CLOCK_MONOTONIC)) < closest_deadline and @queue.empty? + CONDVAR.wait(QUEUE_MUTEX, closest_deadline - now) + end end - end - requests.each do |req| - req.interrupt if req.expired?(now) + requests.each do |req| + req.interrupt if req.expired?(now) + end + requests.reject!(&:done?) end - requests.reject!(&:done?) end + watcher.name = "Timeout stdlib thread" + watcher.thread_variable_set(:"\0__detached_thread__", true) + watcher end - watcher.name = "Timeout stdlib thread" - watcher.thread_variable_set(:"\0__detached_thread__", true) - watcher end - private_class_method :create_timeout_thread - def self.ensure_timeout_thread_created - unless @timeout_thread and @timeout_thread.alive? - TIMEOUT_THREAD_MUTEX.synchronize do - unless @timeout_thread and @timeout_thread.alive? - @timeout_thread = create_timeout_thread + EXECUTOR_MUTEX = Mutex.new + @@executor = nil + private_constant :EXECUTOR_MUTEX + + def self.ensure_executor + unless @@executor && @@executor.alive? + EXECUTOR_MUTEX.synchronize do + unless @@executor && @@executor.alive? + executor = Executor.new # shut down timeout queue and wait for thread termination at exit - @shutdown_proc ||= Kernel.at_exit do - QUEUE_MUTEX.synchronize do - @@queue.close - @timeout_thread&.join - @@queue = Queue.new - @timeout_thread = nil - end + Kernel.at_exit do + executor&.shutdown end + + @@executor = executor end end end @@ -190,26 +225,8 @@ def timeout(sec, klass = nil, message = nil, &block) #:yield: +sec+ return scheduler.timeout_after(sec, klass || Error, message, &block) end - Timeout.ensure_timeout_thread_created - perform = Proc.new do |exc| - request = Request.new(Thread.current, sec, exc, message) - QUEUE_MUTEX.synchronize do - @@queue << request - CONDVAR.signal - end - begin - return yield(sec) - ensure - request.finished - end - end - - if klass - perform.call(klass) - else - backtrace = Error.catch(&perform) - raise Error, message, backtrace - end + Timeout.ensure_executor + @@executor.submit(sec, klass, message, &block) end module_function :timeout end