Shutting down TimerTask is stopping already executing tasks

[stravid]

Shutting down a timer task is not waiting for already executing tasks to finish. I talked to @headius and he encouraged me to open an issue since it is not clear if this is a feature or bug.

Example code:

require "rubygems"
require "bundler/setup"
require "concurrent"

tasks = 10.times.map do |s|
  Concurrent::TimerTask.new(execution_interval: 1) do
    sleep s
    puts "s: #{s}\n"
  end
end

begin
  tasks.each { |t| t.execute }
  sleep
rescue Interrupt
  puts "Received INT"
  puts "Shutting down tasks"
  tasks.each { |t| t.shutdown }
ensure
  puts "Releasing resources"
end

[headius]

It seems like a bug if shutting down a TimerTask kills a timed execution that has already started. Seems like it should wait for completion, or there should be another way to say "shutdown_after_current_task_has_completed".

[pitr-ch]

Thank you, I'll have a look.

[ajlai]

The TimerTask seems to complete if you wait around long enough (not getting killed):

require 'concurrent'

t = Concurrent::TimerTask.new(execution_interval: 1) do
  puts "#{Time.now}: start task"
  sleep 5
  puts "#{Time.now}: finish task"
end

begin
  t.execute
  sleep
rescue Interrupt => e
  puts "#{Time.now}: interrupt rescued, shutting down"
  t.shutdown
  sleep 5 # Waiting 'long enough' here
end

#=> 2019-01-24 15:54:16 -0800: start task
#=> 2019-01-24 15:54:21 -0800: finish task
#=> 2019-01-24 15:54:22 -0800: start task
#=> ^C2019-01-24 15:54:24 -0800: interrupt rescued, shutting down
#=> 2019-01-24 15:54:27 -0800: finish task

there should be another way to say "shutdown_after_current_task_has_completed".

There is a method t.wait_for_termination which should block until the shutdown is totally done if it behaves the same way as threadpool#wait_for_termination (http://ruby-concurrency.github.io/concurrent-ruby/master/file.thread_pools.html#Thread_Pool_Status_and_Shutdown).

Unfortunately, it doesn't seem to be working for TimerTask (Task is still running even after wait_for_termination returns.)

[jamesarosen]

I would have expected Concurrent.global_timer_set.wait_for_termination to be the implementation of "wait around long enough," but it resolves as soon as the stopped_event is synchronized across the threads. That's because TimerTask#ns_shutdown_execution only blocks on synchronizing the @running variable. #execute_task doesn't record the current state in any way, so #ns_shutdown_execution can't watch for it.

[jamesarosen]

This is my current workaround:

module ShutdownConcurrentTasks
  def self.all(tasks, timeout)
    observers = tasks.map { |t| t.add_observer(ShutdownConcurrentTasks::Observer.new) }
    stop_at = Time.now
    Concurrent.global_timer_set.shutdown
    tasks.each(&:shutdown)
    timeout.times do
      break if observers.all? { |o| o.task_run_since?(stop_at) }
      sleep 1
    end
  end

  class Observer
    def initialize
      @last_run = Time.now
    end

    def update(time, *args)
      @last_run = time
    end

    def task_run_since?(time)
      @last_run > time
    end
  end
end


tasks = build_some_tasks

begin
  tasks.each(&:execute)
  sleep
rescue Interrupt
  ShutdownConcurrentTasks.all(tasks, 20)
ensure
  puts "Shut down all tasks"
end

That will give the tasks up to 20 seconds to finish their current run. By calling task.shutdown on each task, task.running? will be correct in the task block, which lets you build well-behaved tasks:

Concurrent::TimerTask do |task|
  10.times do
    break if !task.running?
    puts 'still alive'
    sleep 1
  end
end

[bensheldon]

TimerTask executes the task as a ScheduledTask on the default global executor (Concurrent.global_io_executor):

concurrent-ruby/lib/concurrent-ruby/concurrent/timer_task.rb

Line 292 in 9f40827

ScheduledTask.execute(interval, args: [Concurrent::Event.new], &method(:execute_task))

I'm not sure if that's a mistake when it was introduced and instead it should have the argument executor: self; though going back farther in git history I don't see a TimerTask ever making use of itself inheritting from RubyExecutorService.

So as it is today, to safely shutdown, you'll want to shutdown and wait on Concurrent.global_io_executor when an interrupt signal is trapped.

I think this could also imply:

• A fix to make the ScheduledTask have set executor: self
• Allow passing in an explicit executor: when TimerTask is initialized so someone can set their own Executor.