ruby-concurrency / concurrent-ruby

Modern concurrency tools including agents, futures, promises, thread pools, supervisors, and more. Inspired by Erlang, Clojure, Scala, Go, Java, JavaScript, and classic concurrency patterns.
https://ruby-concurrency.github.io/concurrent-ruby/
Other
5.68k stars 418 forks source link

ScheduledTask triggering ThreadPoolExecutor's `fallback_policy: :abort` breaks ScheduledTasks #889

Closed bensheldon closed 10 months ago

bensheldon commented 4 years ago

When ScheduledTasks are being run on a ThreadPoolExecutor with fallback_policy: :abort, triggering the abort breaks all subsequent usage of ScheduledTasks, they never execute, even when running on a different executor.

Apologies that I don't quite understand the internals of ScheduledTasks, but I'm hypothesizing that when their delay expires, they attempt to execute, and then the executor raises the abort exception, and the ScheduledTask internal is not recovering from the exception.

executor_options = {
  name: 'test',
  min_threads: 0,
  max_threads: 2,
  auto_terminate: true,
  idletime: 60,
  max_queue: -1,
  fallback_policy: :abort, # <= this configuration is the problem; :discard works fine
}.freeze

executor = Concurrent::ThreadPoolExecutor.new(executor_options)

(1..4).to_a.each do |i|
  # Schedule more tasks to execute at the same time than there are available threads
  4.times do
    task = Concurrent::ScheduledTask.new(i, executor: executor) { puts Time.now }
    task.add_observer(proc { |time, output, thread_error| puts "Finished: #{time}, #{output}, #{thread_error}" }, :call)
    task.execute
  end
end

# No scheduled tasks work from now on, not even ones of the global pool
task = Concurrent::ScheduledTask.execute(1) { puts Time.now } # <= never executes
task.value # <= never completes
* Operating system:                mac
* Ruby implementation:             MRI Ruby
* `concurrent-ruby` version:       v1.1.7
* `concurrent-ruby-ext` installed: no
* `concurrent-ruby-edge` used:     no
pitr-ch commented 3 years ago

I think it is a combination of the abstraction and executor combination nobody was thinking about. Could you open a PR which would raise an ArgumentError if the ScheduledTask is given an executor with fallback_policy: :abort?