From b7748b46cdba04b062d2ee6c69d8e57437ba3fae Mon Sep 17 00:00:00 2001 From: Gord Pearson Date: Fri, 14 Aug 2026 14:01:45 -0400 Subject: [PATCH] Settle futures and recover WorkQueue workers Ensure every dequeued task settles its future, replace workers after fatal exits, and keep workers alive for recoverable task errors. Mask queue-pop handoff and worker-retirement accounting against asynchronous exceptions. Add scoped cancellation for SpinGroups using shared queues, make cancellation handoffs race-safe, and normalize zero/negative concurrency consistently. Co-authored-by: River Assisted-By: devx/67971778-53ac-43ed-9659-234fe50f4cd9 --- lib/cli/ui/spinner/spin_group.rb | 131 +++++--- lib/cli/ui/work_queue.rb | 233 +++++++++++--- test/cli/ui/spinner/spin_group_test.rb | 333 ++++++++++++++++++++ test/cli/ui/work_queue_test.rb | 404 +++++++++++++++++++++++++ 4 files changed, 1017 insertions(+), 84 deletions(-) diff --git a/lib/cli/ui/spinner/spin_group.rb b/lib/cli/ui/spinner/spin_group.rb index ab459921..896d0ddc 100644 --- a/lib/cli/ui/spinner/spin_group.rb +++ b/lib/cli/ui/spinner/spin_group.rb @@ -44,7 +44,8 @@ def pause_spinners(&block) # * +:auto_debrief+ - Automatically debrief exceptions or through success_debrief? Default to true # * +:interrupt_debrief+ - Automatically debrief on interrupt. Default to false # * +:max_concurrent+ - Maximum number of concurrent tasks. Default is 0 (effectively unlimited) - # * +:work_queue+ - Custom WorkQueue instance. If not provided, a new one will be created + # * +:work_queue+ - Custom WorkQueue instance. If not provided, a new one will be created. + # Stopping this group cancels only this group's futures on a shared queue. # * +:to+ - Target stream, like $stdout or $stderr. Can be anything with print and puts methods, # or under Sorbet, IO or StringIO. Defaults to $stdout # @@ -69,7 +70,7 @@ def initialize(auto_debrief: true, interrupt_debrief: false, max_concurrent: 0, @start = Time.new @stopped = false @internal_work_queue = work_queue.nil? - @work_queue = work_queue || WorkQueue.new(max_concurrent.zero? ? 1024 : max_concurrent) #: WorkQueue + @work_queue = work_queue || WorkQueue.new(max_concurrent) #: WorkQueue if block_given? yield self wait(to: to) @@ -92,6 +93,11 @@ class Task #: Integer? attr_reader :progress_percentage + # Internal: SpinGroup#stop cancels these on a shared queue. + # Not part of the public API. + #: WorkQueue::Future + attr_reader :future # :nodoc: + # Initializes a new Task # This is managed entirely internally by +SpinGroup+ # @@ -143,16 +149,28 @@ 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 - @on_done&.call(self) - @done end + # Runs the on_done callback at most once. SpinGroup calls this outside + # its render and pause mutexes so callbacks may safely reenter it. + #: -> void + def notify_done # :nodoc: + callback = @on_done + @on_done = nil + callback&.call(self) + end + # Re-renders the task if required: # # We try to be as lazy as possible in re-rendering the full line. The @@ -322,9 +340,11 @@ def add( #: -> void def stop - # If we already own the mutex (called from within another synchronized block), - # set stopped directly to avoid deadlock - if @m.owned? + # A render-time callback may already own @m. Mark the one-way state + # without relocking in that case, but otherwise preserve synchronized + # visibility and ordering with task additions. + mutex_owned = @m.owned? + if mutex_owned return if @stopped @stopped = true @@ -335,8 +355,13 @@ def stop @stopped = true end end - # Interrupt is thread-safe on its own, so we can call it outside the mutex - @work_queue.interrupt + + # Closing plus scoped cancellation lets render-time callers return + # without joining under @m. Ordinary callers retain the historical + # synchronous stop behavior. + @work_queue.close if @internal_work_queue + @work_queue.cancel(task_futures) + @work_queue.wait if @internal_work_queue && !mutex_owned end #: -> bool @@ -385,6 +410,8 @@ def wait(to: $stdout) # Update progress mode based on task states current_mode = update_progress_mode(reporter, current_mode, first_render) + newly_done = [] #: Array[Task] + self.class.pause_mutex.synchronize do next if self.class.paused? @@ -394,7 +421,7 @@ def wait(to: $stdout) force_full_render = render_puts_above(to, consumed_lines) # Render all tasks - done_count, consumed_lines = render_tasks( + done_count, consumed_lines, newly_done = render_tasks( to: to, tasks_seen: tasks_seen, tasks_seen_done: tasks_seen_done, @@ -407,6 +434,8 @@ def wait(to: $stdout) end end + newly_done.each(&:notify_done) + break if done_count == @tasks.size # After first render, start the progress reporter in indeterminate mode @@ -434,11 +463,15 @@ def wait(to: $stdout) end end + @work_queue.wait if @internal_work_queue result rescue Interrupt - @work_queue.interrupt + interrupt_work_queue debrief(to: to) if @interrupt_debrief stopped? ? false : raise + rescue SystemExit + interrupt_work_queue + raise end #: (String message) -> void @@ -469,6 +502,22 @@ def all_succeeded? private + #: -> void + def interrupt_work_queue + if @internal_work_queue + @work_queue.interrupt + else + @work_queue.cancel(task_futures) + end + end + + # @tasks needs @m unless a render-time callback already owns it. + #: -> Array[WorkQueue::Future] + def task_futures + tasks = @m.owned? ? @tasks.dup : @m.synchronize { @tasks.dup } + tasks.map(&:future) + end + # Update progress reporter mode based on task progress states #: (CLI::UI::ProgressReporter::Reporter reporter, Symbol current_mode, bool first_render) -> Symbol def update_progress_mode(reporter, current_mode, first_render) @@ -522,14 +571,16 @@ def render_puts_above(to, consumed_lines) end # Render all tasks - #: (to: io_like, tasks_seen: Array[bool], tasks_seen_done: Array[bool], consumed_lines: Integer, idx: Integer, force_full_render: bool, width: Integer) -> [Integer, Integer] + #: (to: io_like, tasks_seen: Array[bool], tasks_seen_done: Array[bool], consumed_lines: Integer, idx: Integer, force_full_render: bool, width: Integer) -> [Integer, Integer, Array[Task]] def render_tasks(to:, tasks_seen:, tasks_seen_done:, consumed_lines:, idx:, force_full_render:, width:) done_count = 0 + newly_done = [] #: Array[Task] @tasks.each.with_index do |task, int_index| nat_index = int_index + 1 task_done = task.check done_count += 1 if task_done + newly_done << task if task_done && !tasks_seen_done[int_index] if CLI::UI.enable_cursor? if nat_index > consumed_lines @@ -550,7 +601,7 @@ def render_tasks(to:, tasks_seen:, tasks_seen_done:, consumed_lines:, idx:, forc tasks_seen_done[int_index] ||= task_done end - [done_count, consumed_lines] + [done_count, consumed_lines, newly_done] end # Debriefs failed tasks is +auto_debrief+ is true @@ -562,39 +613,41 @@ def render_tasks(to:, tasks_seen:, tasks_seen_done:, consumed_lines:, idx:, forc # #: (?to: io_like) -> bool def debrief(to: $stdout) - @m.synchronize do - @tasks.each do |task| - next unless task.done - - title = task.title - out = task.stdout - err = task.stderr + # Debrief callbacks are caller code and may reenter the group. + tasks = @m.synchronize { @tasks.dup } - if task.success - next @success_debrief&.call(title, out, err) - end + tasks.each do |task| + next unless task.done - # exception will not be set if the wait loop is stopped before the task is checked - e = task.exception - next @failure_debrief.call(title, e, out, err) if @failure_debrief + title = task.title + out = task.stdout + err = task.stderr - CLI::UI::Frame.open('Task Failed: ' + title, color: :red, timing: Time.new - @start) do - if e - to.puts("#{e.class}: #{e.message}") - to.puts("\tfrom #{e.backtrace.join("\n\tfrom ")}") - end + if task.success + next @success_debrief&.call(title, out, err) + end - CLI::UI::Frame.divider('STDOUT') - out = '(empty)' if out.nil? || out.strip.empty? - to.puts(out) + # exception will not be set if the wait loop is stopped before the task is checked + e = task.exception + next @failure_debrief.call(title, e, out, err) if @failure_debrief - CLI::UI::Frame.divider('STDERR') - err = '(empty)' if err.nil? || err.strip.empty? - to.puts(err) + CLI::UI::Frame.open('Task Failed: ' + title, color: :red, timing: Time.new - @start) do + if e + to.puts("#{e.class}: #{e.message}") + to.puts("\tfrom #{e.backtrace.join("\n\tfrom ")}") end + + CLI::UI::Frame.divider('STDOUT') + out = '(empty)' if out.nil? || out.strip.empty? + to.puts(out) + + CLI::UI::Frame.divider('STDERR') + err = '(empty)' if err.nil? || err.strip.empty? + to.puts(err) end - @tasks.all?(&:success) end + + tasks.all?(&:success) end end end diff --git a/lib/cli/ui/work_queue.rb b/lib/cli/ui/work_queue.rb index 47650e17..9448ef83 100644 --- a/lib/cli/ui/work_queue.rb +++ b/lib/cli/ui/work_queue.rb @@ -4,6 +4,20 @@ module CLI module UI class WorkQueue + # Locking invariants: + # - @mutex protects @workers, @active_workers, and queue lifecycle changes. + # - #start_worker and #prune_workers are called only while @mutex is held. + # - Worker threads are never joined while @mutex is held. + # - Queue code may acquire a Future mutex, but Future code never acquires + # the queue mutex. + + # Concurrency used when a caller explicitly requests no limit. + UNLIMITED_CONCURRENCY = 1024 #: Integer + + # 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 @@ -13,25 +27,34 @@ def initialize @started = false #: bool @result = nil #: untyped @error = nil #: Exception? + @worker = nil #: Thread? end #: (untyped result) -> void - def complete(result) - @mutex.synchronize do - @completed = true - @result = result - @condition.broadcast + def complete(result) # :nodoc: + Thread.handle_interrupt(Exception => :never) do + @mutex.synchronize do + return if @completed + + @result = result + @worker = nil + @completed = true + @condition.broadcast + end end end #: (Exception error) -> void - def fail(error) - @mutex.synchronize do - return if @completed + def fail(error) # :nodoc: + Thread.handle_interrupt(Exception => :never) do + @mutex.synchronize do + return if @completed - @completed = true - @error = error - @condition.broadcast + @error = error + @worker = nil + @completed = true + @condition.broadcast + end end end @@ -55,87 +78,207 @@ def started? @mutex.synchronize { @started } end - #: -> void - def start - @mutex.synchronize do - @started = true - @condition.broadcast + #: (?Thread? worker) -> bool + def start(worker = nil) # :nodoc: + Thread.handle_interrupt(Exception => :never) do + @mutex.synchronize do + return false if @completed + + @worker = worker + @started = true + @condition.broadcast + true + end + end + end + + #: (Thread interrupting_thread) -> Thread? + def interrupt(interrupting_thread) # :nodoc: + Thread.handle_interrupt(Exception => :never) do + @mutex.synchronize do + return if @completed + + worker = @worker + error = Interrupt.new + @error = error + @worker = nil + @completed = true + @condition.broadcast + + # Deliver while holding the future lock. The worker cannot + # complete this future and advance to unrelated work between + # selecting it for interruption and receiving the exception. + if worker && worker != interrupting_thread + begin + worker.raise(error) if worker.alive? + rescue ThreadError + # The worker exited between alive? and raise. + end + end + + worker + end end end end #: (Integer max_concurrent) -> void def initialize(max_concurrent) - @max_concurrent = max_concurrent + raise ArgumentError, 'max_concurrent must be non-negative' if max_concurrent.negative? + + @max_concurrent = (max_concurrent.zero? ? UNLIMITED_CONCURRENCY : max_concurrent) #: 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 + prune_workers + @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 + + loop do + workers = @mutex.synchronize { @workers.dup } + break if workers.empty? + + # Do not rescue Interrupt here: a caller interrupting #wait should + # observe it. Worker exceptions are contained by their lifecycle. + workers.each(&:join) + @mutex.synchronize { prune_workers } + end + end + + # Cancels selected futures without closing the queue or waiting for their + # workers to unwind. + #: (Array[Future] futures) -> void + def cancel(futures) + workers = futures.filter_map { |future| future.interrupt(Thread.current) }.uniq + @mutex.synchronize { prune_workers } + raise Interrupt if workers.include?(Thread.current) 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.clear + @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 Exception # rubocop:disable Lint/RescueException + # Backstop: workers retire without propagating, and a late exception + # must not replace what this thread is already propagating. end + @mutex.synchronize { prune_workers } + + # Keep the worker history so #wait can still join the current worker + # while it unwinds from the Interrupt raised below. + raise Interrupt if current_worker end private + #: -> void + def prune_workers + @workers.select!(&:alive?) + end + #: -> 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 + raise ThreadError, 'start_worker requires the queue mutex' unless @mutex.owned? + + worker = Thread.new do + # Only task bodies are interruptible. Masking the dequeue handoff + # prevents an asynchronous exception from removing work before its + # future is known. Idle workers still wake when the queue is closed. + Thread.handle_interrupt(Exception => :never) do + loop do + work = @queue.pop + break if work.nil? + + future, block = work + run_task(future, block) 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; retirement replaces it if needed. + ensure + retire_worker + end + + @workers << worker + @active_workers += 1 + end + + # Runs one dequeued task to settlement on this worker. + #: (Future future, ^() -> untyped block) -> void + def run_task(future, block) + return unless future.start(Thread.current) + + Thread.handle_interrupt(Exception => :immediate) do + future.complete(block.call) + end + rescue StandardError, ScriptError => e + # Recoverable task failures do not poison the worker. + future.fail(e) + 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 + + # Releases this worker's slot, replacing it while work remains. + #: -> void + def retire_worker + # An asynchronous exception here would strand this worker's slot. + Thread.handle_interrupt(Exception => :never) do + @mutex.synchronize do + @active_workers -= 1 + start_worker if @active_workers < @max_concurrent && !@queue.empty? + end end + rescue Exception # rubocop:disable Lint/RescueException + # A worker never propagates: #wait joins on behalf of a caller who was + # never interrupted. 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..6013a4db 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,245 @@ 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 + unwound = 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 + ensure + unwound << true + 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') + refute(unwound.empty?, 'remaining workers should unwind 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) + work_queue = sg.instance_variable_get(:@work_queue) + assert_equal( + WorkQueue::UNLIMITED_CONCURRENCY, + work_queue.instance_variable_get(:@max_concurrent), + ) + sg.add('s') { true } + + assert(Timeout.timeout(10) { sg.wait }) + end + end + + def test_spin_group_negative_max_concurrent_raises + capture_io do + CLI::UI::StdoutRouter.ensure_activated + + error = assert_raises(ArgumentError) { SpinGroup.new(max_concurrent: -1) } + assert_equal('max_concurrent must be non-negative', error.message) + end + end + + def test_spin_group_system_exit_interrupts_its_tasks_without_closing_a_shared_work_queue + capture_io do + CLI::UI::StdoutRouter.ensure_activated + + work_queue = WorkQueue.new(3) + unrelated_started = Queue.new + release_unrelated = Queue.new + unrelated = work_queue.enqueue do + unrelated_started << true + release_unrelated.pop + :unrelated + end + + slow_started = Queue.new + interrupted = Queue.new + sg = SpinGroup.new(auto_debrief: false, work_queue: work_queue) + 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 + + unrelated_started.pop + assert_raises(SystemExit) { sg.wait } + assert(Timeout.timeout(1) { interrupted.pop }, 'remaining group tasks should observe Interrupt') + + release_unrelated << true + followup = work_queue.enqueue { :ran } + work_queue.wait + assert_equal(:unrelated, unrelated.value) + assert_equal(:ran, followup.value) + end + end + + def test_stop_from_on_done_does_not_deadlock_a_shared_work_queue + capture_io do + CLI::UI::StdoutRouter.ensure_activated + + work_queue = WorkQueue.new(2) + cleanup_started = Queue.new + cleanup_unwound = Queue.new + sg = SpinGroup.new(auto_debrief: false, work_queue: work_queue) + sg.add('stopper') do |task| + task.on_done { sg.stop } + cleanup_started.pop + true + end + sg.add('cleanup') do + cleanup_started << true + sleep(5) + ensure + sg.puts_above('cleanup finished') + cleanup_unwound << true + end + + refute(Timeout.timeout(10) { sg.wait }) + assert(Timeout.timeout(1) { cleanup_unwound.pop }) + + followup = work_queue.enqueue { :ran } + work_queue.wait + assert_equal(:ran, followup.value) + end + end + + def test_stop_from_on_done_does_not_deadlock_an_internal_work_queue + capture_io do + CLI::UI::StdoutRouter.ensure_activated + + cleanup_started = Queue.new + cleanup_unwound = Queue.new + sg = SpinGroup.new(auto_debrief: false, max_concurrent: 2) + sg.add('stopper') do |task| + task.on_done { sg.stop } + cleanup_started.pop + true + end + sg.add('cleanup') do + cleanup_started << true + sleep(5) + ensure + sg.puts_above('cleanup finished') + cleanup_unwound << true + end + + refute(Timeout.timeout(10) { sg.wait }) + assert(Timeout.timeout(1) { cleanup_unwound.pop }) + end + end + + def test_spin_group_stop_interrupts_its_tasks_without_closing_a_shared_work_queue + capture_io do + CLI::UI::StdoutRouter.ensure_activated + + work_queue = WorkQueue.new(1) + started = Queue.new + interrupted = Queue.new + completed = false + sg = SpinGroup.new(auto_debrief: false, work_queue: work_queue) + sg.add('slow') do + started << true + sleep(5) + completed = true + rescue Interrupt + interrupted << true + raise + end + + waiter = Thread.new { sg.wait } + started.pop + sg.stop + + refute(Timeout.timeout(10) { waiter.value }) + assert(Timeout.timeout(1) { interrupted.pop }, 'the group task should observe Interrupt') + followup = work_queue.enqueue { :ran } + work_queue.wait + refute(completed, 'the stopped group task should not complete') + assert_equal(:ran, followup.value) + end + end + + def test_spin_group_interrupt_interrupts_its_tasks_without_closing_a_shared_work_queue + capture_io do + CLI::UI::StdoutRouter.ensure_activated + + work_queue = WorkQueue.new(1) + started = Queue.new + interrupted = Queue.new + completed = false + sg = SpinGroup.new(auto_debrief: false, work_queue: work_queue) + sg.add('slow') do + started << true + sleep(5) + completed = true + rescue Interrupt + interrupted << true + raise + end + + waiter = Thread.new { sg.wait } + waiter.report_on_exception = false + started.pop + waiter.raise(Interrupt) + + assert_raises(Interrupt) { waiter.join } + assert(Timeout.timeout(1) { interrupted.pop }, 'the group task should observe Interrupt') + followup = work_queue.enqueue { :ran } + work_queue.wait + refute(completed, 'the interrupted group task should not complete') + assert_equal(:ran, followup.value) + end + end + def test_spin_group_success_debrief capture_io do CLI::UI::StdoutRouter.ensure_activated @@ -187,6 +427,39 @@ def test_spin_group_stop end end + def test_external_stop_waits_for_internal_workers_to_unwind + capture_io do + CLI::UI::StdoutRouter.ensure_activated + + started = Queue.new + cleanup_started = Queue.new + release_cleanup = Queue.new + sg = SpinGroup.new(auto_debrief: false) + sg.add('cleanup') do + started << true + sleep(5) + ensure + cleanup_started << true + release_cleanup.pop + end + + waiter = Thread.new { sg.wait } + Timeout.timeout(10) { started.pop } + stopper = Thread.new { sg.stop } + Timeout.timeout(10) { cleanup_started.pop } + + begin + assert_nil(stopper.join(0.1), 'external stop should wait for worker cleanup') + ensure + release_cleanup << true + Timeout.timeout(10) do + stopper.join + waiter.join + end + end + end + end + def test_spin_group_nested_stop capture_io do CLI::UI::StdoutRouter.ensure_activated @@ -305,6 +578,66 @@ def test_task_on_done_callback assert(callback_executed, 'on_done callback should have been executed') end end + + def test_task_on_done_callback_may_reenter_the_group + capture_io do + CLI::UI::StdoutRouter.ensure_activated + + sg = SpinGroup.new(auto_debrief: false) + sg.add('reenters') do |task| + task.on_done do + SpinGroup.pause_spinners { nil } + sg.puts_above('from the callback') + end + true + end + + assert(Timeout.timeout(10) { sg.wait }) + end + end + + def test_success_debrief_callback_may_reenter_the_group + capture_io do + CLI::UI::StdoutRouter.ensure_activated + + sg = SpinGroup.new + sg.success_debrief do + SpinGroup.pause_spinners { nil } + sg.puts_above('from the debrief') + end + sg.add('succeeds') { true } + + assert(Timeout.timeout(10) { sg.wait }) + end + end + + def test_stop_from_a_final_glyph_defers_join_until_outside_the_render_mutex + capture_io do + CLI::UI::StdoutRouter.ensure_activated + + cleanup_started = Queue.new + cleanup_unwound = Queue.new + sg = SpinGroup.new(auto_debrief: false, max_concurrent: 2) + stopping_glyph = lambda do |_success| + sg.stop + CLI::UI::Glyph::CHECK + end + sg.add('stopper', final_glyph: stopping_glyph) do + cleanup_started.pop + true + end + sg.add('cleanup') do + cleanup_started << true + sleep(5) + ensure + sg.puts_above('cleanup finished') + cleanup_unwound << true + end + + refute(Timeout.timeout(10) { sg.wait }) + assert(Timeout.timeout(1) { cleanup_unwound.pop }) + end + end end end end diff --git a/test/cli/ui/work_queue_test.rb b/test/cli/ui/work_queue_test.rb index 686fcdb0..92770565 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,226 @@ def test_future_error assert_raises(StandardError, 'Test error') { future.value } end + def test_script_error_completes_future_without_replacing_worker + @work_queue = WorkQueue.new(1) + failed_worker = nil + failed = @work_queue.enqueue do + failed_worker = Thread.current + raise ScriptError, 'boom' + end + followup = @work_queue.enqueue { Thread.current } + + @work_queue.wait + + assert(failed.completed?, 'failing future should be completed') + error = assert_raises(ScriptError) { failed.value } + assert_equal('boom', error.message) + assert(followup.completed?, 'subsequent task should run on the same worker') + assert_same(failed_worker, followup.value) + assert_empty(@work_queue.instance_variable_get(:@workers)) + end + + def test_script_errors_do_not_accumulate_dead_workers + @work_queue = WorkQueue.new(4) + + begin + futures = 200.times.map { @work_queue.enqueue { raise LoadError, 'boom' } } + futures.each { |future| assert_raises(LoadError) { future.value } } + + workers = @work_queue.instance_variable_get(:@workers) + assert_equal(4, workers.size) + assert(workers.all?(&:alive?)) + assert_equal(4, @work_queue.instance_variable_get(:@active_workers)) + ensure + @work_queue.interrupt + end + end + + def test_interrupt_cannot_abandon_work_returned_by_queue_pop + popped = Queue.new + release_pop = Queue.new + queue = Class.new do + def initialize(popped, release_pop) + @queue = Queue.new + @popped = popped + @release_pop = release_pop + @gate_return = true + end + + def push(work) + @queue.push(work) + end + + def pop(non_block = false) + work = @queue.pop(non_block) + if @gate_return + @gate_return = false + @popped << true + @release_pop.pop + end + work + end + + def close + @queue.close + end + + def empty? + @queue.empty? + end + end.new(popped, release_pop) + @work_queue.instance_variable_set(:@queue, queue) + + future = @work_queue.enqueue { :ran } + popped.pop + interrupter = Thread.new { @work_queue.interrupt } + interrupter.report_on_exception = false + + begin + assert_nil( + interrupter.join(0.1), + 'interrupt should wait until the popped work is associated with its future', + ) + ensure + release_pop << true + Timeout.timeout(10) { interrupter.join } + end + + assert(future.completed?, 'the popped future must be settled') + assert_raises(Interrupt) { future.value } + end + + def test_worker_retirement_survives_asynchronous_termination + @work_queue = WorkQueue.new(1) + started = Queue.new + release_task = Queue.new + future = @work_queue.enqueue do + started << true + release_task.pop + :ran + end + started.pop + + mutex = @work_queue.instance_variable_get(:@mutex) + queue = @work_queue.instance_variable_get(:@queue) + worker = @work_queue.instance_variable_get(:@workers).first + mutex.lock + begin + queue.close + release_task << true + Timeout.timeout(10) do + sleep(0.001) until future.completed? && worker.status == 'sleep' + end + worker.raise(Interrupt) + ensure + mutex.unlock + end + + Timeout.timeout(10) { @work_queue.wait } + + assert_equal(:ran, future.value) + assert_equal( + 0, + @work_queue.instance_variable_get(:@active_workers), + 'a terminated worker must still release its slot', + ) + 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_zero_max_concurrent_uses_the_unlimited_default + work_queue = WorkQueue.new(0) + assert_equal(WorkQueue::UNLIMITED_CONCURRENCY, work_queue.instance_variable_get(:@max_concurrent)) + future = work_queue.enqueue { :ran } + work_queue.wait + + assert_equal(:ran, future.value) + end + + def test_negative_max_concurrent_raises + error = assert_raises(ArgumentError) { WorkQueue.new(-1) } + assert_equal('max_concurrent must be non-negative', error.message) + 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_start_remains_compatible_without_a_worker_argument + future = WorkQueue::Future.new + + assert(future.start) + assert(future.started?) + 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 +377,189 @@ 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_interrupting_selected_futures_keeps_the_queue_usable + @work_queue = WorkQueue.new(2) + selected_started = Queue.new + unrelated_started = Queue.new + release_unrelated = Queue.new + selected = @work_queue.enqueue do + selected_started << true + sleep(5) + end + unrelated = @work_queue.enqueue do + unrelated_started << true + release_unrelated.pop + :unrelated + end + + selected_started.pop + unrelated_started.pop + @work_queue.cancel([selected]) + + assert_raises(Interrupt) { selected.value } + release_unrelated << true + followup = @work_queue.enqueue { :followup } + @work_queue.wait + assert_equal(:unrelated, unrelated.value) + assert_equal(:followup, followup.value) + end + + def test_interrupting_a_queued_future_prevents_it_from_running + @work_queue = WorkQueue.new(1) + blocker_started = Queue.new + release_blocker = Queue.new + ran = false + blocker = @work_queue.enqueue do + blocker_started << true + release_blocker.pop + :blocker + end + cancelled = @work_queue.enqueue { ran = true } + + blocker_started.pop + @work_queue.cancel([cancelled]) + release_blocker << true + followup = @work_queue.enqueue { :followup } + @work_queue.wait + + assert_equal(:blocker, blocker.value) + assert_raises(Interrupt) { cancelled.value } + refute(ran) + assert_equal(:followup, followup.value) + end + + def test_scoped_interrupts_prune_retired_worker_history + @work_queue = WorkQueue.new(1) + + 20.times do + started = Queue.new + future = @work_queue.enqueue do + started << true + sleep(5) + end + started.pop + @work_queue.cancel([future]) + + Timeout.timeout(10) do + loop do + active_workers = @work_queue.instance_variable_get(:@active_workers) + workers = @work_queue.instance_variable_get(:@workers) + break if active_workers.zero? && workers.none?(&:alive?) + + sleep(0.001) + end + end + end + + assert_equal(1, @work_queue.instance_variable_get(:@workers).size) + followup = @work_queue.enqueue { :followup } + assert_equal(1, @work_queue.instance_variable_get(:@workers).size) + + @work_queue.wait + + assert_equal(:followup, followup.value) + assert_empty(@work_queue.instance_variable_get(:@workers)) + end + + def test_interrupting_a_future_cannot_land_on_the_next_task + @work_queue = WorkQueue.new(1) + selected_started = Queue.new + release_selected = Queue.new + interrupt_recorded = Queue.new + finish_interrupt = Queue.new + next_started = Queue.new + release_next = Queue.new + + selected = @work_queue.enqueue do + selected_started << true + release_selected.pop + end + unrelated = @work_queue.enqueue do + next_started << true + release_next.pop + :unrelated + end + + selected.singleton_class.prepend(Module.new do + define_method(:interrupt) do |interrupting_thread| + worker = super(interrupting_thread) + interrupt_recorded << true + finish_interrupt.pop + worker + end + end) + + selected_started.pop + interrupter = Thread.new { @work_queue.cancel([selected]) } + interrupt_recorded.pop + release_selected << true + next_started.pop + finish_interrupt << true + interrupter.join + release_next << true + @work_queue.wait + + assert_raises(Interrupt) { selected.value } + assert_equal(:unrelated, unrelated.value) + end + + def test_worker_can_interrupt_its_own_future_without_closing_the_queue + @work_queue = WorkQueue.new(1) + current_future = Queue.new + selected = @work_queue.enqueue do + @work_queue.cancel([current_future.pop]) + end + unrelated = @work_queue.enqueue { :unrelated } + current_future << selected + + Timeout.timeout(10) { @work_queue.wait } + + assert_raises(Interrupt) { selected.value } + assert_equal(:unrelated, unrelated.value) + 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 + + def test_wait_joins_a_worker_that_interrupts_the_queue_until_it_unwinds + @work_queue = WorkQueue.new(1) + unwinding = Queue.new + release = Queue.new + future = @work_queue.enqueue do + @work_queue.interrupt + ensure + unwinding << true + release.pop + end + + unwinding.pop + waiter = Thread.new { @work_queue.wait } + + begin + assert_nil(waiter.join(0.1), 'wait should not return while the current worker is still unwinding') + ensure + release << true + Timeout.timeout(10) { waiter.join } + end + + assert_raises(Interrupt) { future.value } + end end end end