diff --git a/lib/timeout.rb b/lib/timeout.rb index 2aad1d7..2da28f0 100644 --- a/lib/timeout.rb +++ b/lib/timeout.rb @@ -50,87 +50,135 @@ def exception(*) end # :stopdoc: - CONDVAR = ConditionVariable.new - QUEUE = Queue.new - QUEUE_MUTEX = Mutex.new - TIMEOUT_THREAD_MUTEX = Mutex.new - @timeout_thread = nil - private_constant :CONDVAR, :QUEUE, :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 - end - end + def alive? + !@queue.closed? && @timeout_thread.alive? end - def finished - @mutex.synchronize do - @done = true + 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 - end - end - 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 - requests << req unless req.done? + + def done? + @mutex.synchronize do + @done end - closest_deadline = requests.min_by(&:deadline).deadline + end - 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) + 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 + private_constant :Request + + 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? + + 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) + 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 + Kernel.at_exit do + executor&.shutdown + end + + @@executor = executor end end end @@ -177,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