diff --git a/lib/cli/ui/spinner/spin_group.rb b/lib/cli/ui/spinner/spin_group.rb index ab459921..03f69473 100644 --- a/lib/cli/ui/spinner/spin_group.rb +++ b/lib/cli/ui/spinner/spin_group.rb @@ -143,7 +143,12 @@ def check result = @future.value @success = true @success = false if result == TASK_FAILED - rescue => exc + rescue Interrupt, SystemExit + raise + rescue Exception => exc # rubocop:disable Lint/RescueException + # Any other exception (including ScriptError and friends) is a task + # failure, reported through the normal debrief rather than raised + # out of the middle of SpinGroup#wait's render loop. @exception = exc @success = false end @@ -439,6 +444,11 @@ def wait(to: $stdout) @work_queue.interrupt debrief(to: to) if @interrupt_debrief stopped? ? false : raise + rescue SystemExit + # An internal queue has no owner left to stop its remaining workers. + # A shared queue remains the caller's responsibility. + @work_queue.interrupt if @internal_work_queue + raise end #: (String message) -> void diff --git a/lib/cli/ui/work_queue.rb b/lib/cli/ui/work_queue.rb index 47650e17..27865cb7 100644 --- a/lib/cli/ui/work_queue.rb +++ b/lib/cli/ui/work_queue.rb @@ -4,6 +4,10 @@ module CLI module UI class WorkQueue + # Raised into a future whose worker exited without settling it: Thread#raise + # and Thread#kill can land outside the worker's rescues. + class AbandonedTaskError < StandardError; end + class Future #: -> void def initialize @@ -17,21 +21,27 @@ def initialize #: (untyped result) -> void def complete(result) - @mutex.synchronize do - @completed = true - @result = result - @condition.broadcast + Thread.handle_interrupt(Exception => :never) do + @mutex.synchronize do + return if @completed + + @result = result + @completed = true + @condition.broadcast + end end end #: (Exception error) -> void def fail(error) - @mutex.synchronize do - return if @completed + Thread.handle_interrupt(Exception => :never) do + @mutex.synchronize do + return if @completed - @completed = true - @error = error - @condition.broadcast + @error = error + @completed = true + @condition.broadcast + end end end @@ -66,76 +76,132 @@ def start #: (Integer max_concurrent) -> void def initialize(max_concurrent) - @max_concurrent = max_concurrent + @max_concurrent = [max_concurrent, 1].max #: Integer @queue = Queue.new #: Queue @mutex = Mutex.new #: Mutex - @condition = ConditionVariable.new #: ConditionVariable @workers = [] #: Array[Thread] + @active_workers = 0 #: Integer end #: { -> untyped } -> Future def enqueue(&block) future = Future.new @mutex.synchronize do - start_worker if @workers.size < @max_concurrent + @queue.push([future, block]) + start_worker if @active_workers < @max_concurrent end - @queue.push([future, block]) future end #: -> void def close - @queue.close + @mutex.synchronize { @queue.close } end #: -> void def wait - @queue.close - @workers.each(&:join) + close + + joined = 0 + loop do + workers = @mutex.synchronize { @workers.drop(joined) } + break if workers.empty? + + workers.each(&:join) + joined += workers.size + end end #: -> void def interrupt - @mutex.synchronize do + workers = @mutex.synchronize do @queue.close # Fail any remaining tasks in the queue until @queue.empty? future, _block = @queue.pop(true) future&.fail(Interrupt.new) end - # Interrupt all worker threads - @workers.each { |worker| worker.raise(Interrupt) if worker.alive? } - @workers.each(&:join) + @workers.dup + end + + current_worker = workers.include?(Thread.current) + other_workers = workers.reject { |worker| worker == Thread.current } + + # Interrupt worker threads without holding @mutex: workers retire under + # that mutex, so joining them while holding it would deadlock. + other_workers.each do |worker| + worker.raise(Interrupt) if worker.alive? + rescue ThreadError + # The worker exited between alive? and raise. + end + other_workers.each do |worker| + worker.join + rescue Interrupt + # The Interrupt raised above can land after a worker has left its + # rescues (it was already terminating); the worker then dies with + # it and join re-raises it here. That must not replace whatever + # this thread is propagating (e.g. the SystemExit that triggered + # this interrupt). + end + + @mutex.synchronize do @workers.clear end + + raise Interrupt if current_worker end private #: -> void def start_worker - @workers << Thread.new do - loop do - work = @queue.pop - break if work.nil? - - future, block = work - - begin - future.start - result = block.call - future.complete(result) - rescue Interrupt => e - future.fail(e) - raise # Always re-raise interrupts to terminate the worker - rescue StandardError => e - future.fail(e) - # Don't re-raise standard errors - allow worker to continue + worker = Thread.new do + Thread.handle_interrupt(Exception => :never) do + loop do + future = nil #: Future? + + begin + work = nil #: untyped + Thread.handle_interrupt(Exception => :on_blocking) do + work = @queue.pop + end + + break if work.nil? + + future, block = work + future.start + result = Thread.handle_interrupt(Exception => :immediate) { block.call } + future.complete(result) + rescue Interrupt => e + future&.fail(e) + raise + rescue StandardError => e + future&.fail(e) + # Ordinary task failures do not poison the worker. + rescue Exception => e # rubocop:disable Lint/RescueException + future&.fail(e) + raise + ensure + # A worker must never abandon a future: an unsettled future + # blocks Future#value forever. No-op once it is settled. + future&.fail(AbandonedTaskError.new('worker exited before completing this task')) + end end end rescue Interrupt # Clean exit on interrupt + rescue Exception # rubocop:disable Lint/RescueException + # The future carries the exception to its caller. Fatal exceptions + # terminate this worker; the ensure below replaces it if needed. + ensure + @mutex.synchronize do + @active_workers -= 1 + start_worker if @active_workers < @max_concurrent && !@queue.empty? + end end + + @workers << worker + @active_workers += 1 end end end diff --git a/test/cli/ui/spinner/spin_group_test.rb b/test/cli/ui/spinner/spin_group_test.rb index 4f262568..8f7c2a4a 100644 --- a/test/cli/ui/spinner/spin_group_test.rb +++ b/test/cli/ui/spinner/spin_group_test.rb @@ -1,6 +1,7 @@ # frozen_string_literal: true require 'test_helper' +require 'timeout' module CLI module UI @@ -36,6 +37,111 @@ def test_spin_group_auto_debrief_false assert_equal('', err) end + def test_spin_group_non_standard_error_is_reported_as_a_task_failure + capture_io do + CLI::UI::StdoutRouter.ensure_activated + + failures = [] + sibling_ran = false + + sg = SpinGroup.new + sg.failure_debrief { |title, exception, _out, _err| failures << [title, exception] } + sg.add('raises') { raise NotImplementedError, 'not done yet' } + sg.add('sibling') { sibling_ran = true } + + # Before WorkQueue settled futures for non-StandardError exceptions + # this spun forever; the timeout keeps a regression from wedging the + # suite instead of failing it. + refute(Timeout.timeout(10) { sg.wait }) + + assert(sibling_ran, 'sibling task should still run to completion') + assert_equal(1, failures.size) + title, error = failures.first + assert_equal('raises', title) + assert_instance_of(NotImplementedError, error) + assert_equal('not done yet', error.message) + end + end + + def test_spin_group_system_exit_propagates_and_interrupts_remaining_work + capture_io do + CLI::UI::StdoutRouter.ensure_activated + + slow_started = Queue.new + interrupted = Queue.new + sg = SpinGroup.new(auto_debrief: false) + sg.add('exits') do + slow_started.pop + exit(1) + end + sg.add('slow') do + slow_started << true + sleep(5) + rescue Interrupt + interrupted << true + raise + end + + error = Timeout.timeout(10) { assert_raises(SystemExit) { sg.wait } } + + assert_equal(1, error.status) + assert(Timeout.timeout(1) { interrupted.pop }, 'remaining workers should observe Interrupt') + workers = sg.instance_variable_get(:@work_queue).instance_variable_get(:@workers) + refute(workers.any?(&:alive?), 'remaining workers should be terminated before wait raises') + end + end + + def test_spin_group_zero_max_concurrent_uses_the_default + capture_io do + CLI::UI::StdoutRouter.ensure_activated + + sg = SpinGroup.new(max_concurrent: 0, auto_debrief: false) + sg.add('s') { true } + + assert(Timeout.timeout(10) { sg.wait }) + end + end + + def test_spin_group_negative_max_concurrent_runs_serially + capture_io do + CLI::UI::StdoutRouter.ensure_activated + + mutex = Mutex.new + running = 0 + max_running = 0 + sg = SpinGroup.new(max_concurrent: -1, auto_debrief: false) + 3.times do |i| + sg.add("task #{i}") do + mutex.synchronize do + running += 1 + max_running = [max_running, running].max + end + sleep(0.01) + mutex.synchronize { running -= 1 } + end + end + + assert(Timeout.timeout(10) { sg.wait }) + assert_equal(1, max_running) + end + end + + def test_spin_group_system_exit_does_not_interrupt_a_shared_work_queue + capture_io do + CLI::UI::StdoutRouter.ensure_activated + + work_queue = WorkQueue.new(1) + sg = SpinGroup.new(auto_debrief: false, work_queue: work_queue) + sg.add('exits') { exit(1) } + + assert_raises(SystemExit) { sg.wait } + + followup = work_queue.enqueue { :ran } + work_queue.wait + assert_equal(:ran, followup.value) + end + end + def test_spin_group_success_debrief capture_io do CLI::UI::StdoutRouter.ensure_activated diff --git a/test/cli/ui/work_queue_test.rb b/test/cli/ui/work_queue_test.rb index 686fcdb0..d77b9281 100644 --- a/test/cli/ui/work_queue_test.rb +++ b/test/cli/ui/work_queue_test.rb @@ -2,6 +2,7 @@ require 'test_helper' require 'cli/ui/work_queue' +require 'timeout' module CLI module UI @@ -65,6 +66,105 @@ def test_future_error assert_raises(StandardError, 'Test error') { future.value } end + def test_future_non_standard_error_completes_future_and_replaces_worker + @work_queue = WorkQueue.new(1) + failed = @work_queue.enqueue { raise ScriptError, 'boom' } + replacement = @work_queue.enqueue { :ran } + + @work_queue.wait + + assert(failed.completed?, 'failing future should be completed') + error = assert_raises(ScriptError) { failed.value } + assert_equal('boom', error.message) + assert(replacement.completed?, 'subsequent task should run on a replacement worker') + assert_equal(:ran, replacement.value) + end + + def test_worker_exit_settles_its_future_and_replaces_the_worker + @work_queue = WorkQueue.new(1) + started = Queue.new + exit_now = Queue.new + abandoned = @work_queue.enqueue do + started << true + exit_now.pop + Thread.exit + end + started.pop + replacement = @work_queue.enqueue { :ran } + exit_now << true + + Timeout.timeout(10) do + @work_queue.wait + + assert(abandoned.completed?, 'a worker must never leave a future unsettled') + assert_raises(WorkQueue::AbandonedTaskError) { abandoned.value } + assert_equal(:ran, replacement.value) + end + end + + def test_non_positive_max_concurrent_runs_serially + [0, -1].each do |max_concurrent| + work_queue = WorkQueue.new(max_concurrent) + assert_equal(1, work_queue.instance_variable_get(:@max_concurrent)) + future = work_queue.enqueue { :ran } + work_queue.wait + + assert_equal(:ran, future.value) + end + end + + def test_future_settlement_is_idempotent + completed = WorkQueue::Future.new + completed.complete(:ran) + completed.fail(Interrupt.new) + assert_equal(:ran, completed.value) + + failed = WorkQueue::Future.new + error = ScriptError.new('boom') + failed.fail(error) + failed.complete(:ran) + assert_same(error, assert_raises(ScriptError) { failed.value }) + end + + def test_future_completion_defers_asynchronous_interrupt_until_settled + entered = Queue.new + release = Queue.new + mutex = Class.new do + def initialize(entered, release) + @mutex = Mutex.new + @entered = entered + @release = release + @gate = true + end + + def synchronize + @mutex.synchronize do + if @gate + @gate = false + @entered << true + @release.pop + end + yield + end + end + end.new(entered, release) + + future = WorkQueue::Future.new + future.instance_variable_set(:@mutex, mutex) + worker = Thread.new do + future.complete(:ran) + rescue Interrupt + :interrupted + end + + entered.pop + worker.raise(Interrupt) + release << true + + assert_equal(:interrupted, worker.value) + assert_equal(:ran, future.value) + end + def test_max_concurrent max_concurrent = 2 @work_queue = WorkQueue.new(max_concurrent) @@ -156,6 +256,24 @@ def test_interrupt refute(interrupted, 'Task should not complete after interrupt') assert(future.completed?, 'Future should be marked as completed after interrupt') end + + def test_interrupt_from_a_worker_interrupts_its_siblings + @work_queue = WorkQueue.new(2) + sibling_started = Queue.new + interrupter = @work_queue.enqueue do + sibling_started.pop + @work_queue.interrupt + end + sibling = @work_queue.enqueue do + sibling_started << true + sleep(5) + end + + Timeout.timeout(10) { @work_queue.wait } + + assert_raises(Interrupt) { interrupter.value } + assert_raises(Interrupt) { sibling.value } + end end end end