Rewritten TimerTask now an Executor instead of a Runnable.

Author: jdantonio

lib/concurrent/timer_task.rb

@@ -1,9 +1,11 @@
-require 'thread'
-
 require 'concurrent/dereferenceable'
 require 'concurrent/observable'
+require 'concurrent/atomic/atomic_fixnum'
+require 'concurrent/executor/executor'
+require 'concurrent/executor/safe_task_executor'
+
+# deprecated Updated to use `Executor` instead of `Runnable`
 require 'concurrent/runnable'
-require 'concurrent/stoppable'
 
 module Concurrent
 
@@ -144,148 +146,216 @@ module Concurrent
   # @see http://docs.oracle.com/javase/7/docs/api/java/util/TimerTask.html
   class TimerTask
     include Dereferenceable
+    include Executor
     include Concurrent::Observable
-    include Runnable
-    include Stoppable
 
-    # Default `:execution_interval`
+    # Default `:execution_interval` in seconds.
     EXECUTION_INTERVAL = 60
 
-    # Default `:timeout_interval`
+    # Default `:timeout_interval` in seconds.
     TIMEOUT_INTERVAL = 30
 
-    # Number of seconds after the task completes before the task is
-    # performed again.
-    attr_reader :execution_interval
-
-    # Number of seconds the task can run before it is considered to have failed.
-    # Failed tasks are forcibly killed.
-    attr_reader :timeout_interval
-
     # Create a new TimerTask with the given task and configuration.
     #
-    # @param [Hash] opts the options defining task execution.
-    # @option opts [Integer] :execution_interval number of seconds between
-    #   task executions (default: EXECUTION_INTERVAL)
-    # @option opts [Integer] :timeout_interval number of seconds a task can
-    #   run before it is considered to have failed (default: TIMEOUT_INTERVAL)
-    # @option opts [Boolean] :run_now Whether to run the task immediately
-    #   upon instantiation or to wait until the first #execution_interval
-    #   has passed (default: false)
-    #
-    # @raise ArgumentError when no block is given.
-    #
-    # @yield to the block after :execution_interval seconds have passed since
-    #   the last yield
-    # @yieldparam task a reference to the `TimerTask` instance so that the
-    #   block can control its own lifecycle. Necessary since `self` will
-    #   refer to the execution context of the block rather than the running
-    #   `TimerTask`.
+    # @!macro [attach] timer_task_initialize
+    #   @param [Hash] opts the options defining task execution.
+    #   @option opts [Integer] :execution_interval number of seconds between
+    #     task executions (default: EXECUTION_INTERVAL)
+    #   @option opts [Integer] :timeout_interval number of seconds a task can
+    #     run before it is considered to have failed (default: TIMEOUT_INTERVAL)
+    #   @option opts [Boolean] :run_now Whether to run the task immediately
+    #     upon instantiation or to wait until the first #  execution_interval
+    #     has passed (default: false)
+    #  
+    #   @raise ArgumentError when no block is given.
+    #  
+    #   @yield to the block after :execution_interval seconds have passed since
+    #     the last yield
+    #   @yieldparam task a reference to the `TimerTask` instance so that the
+    #     block can control its own lifecycle. Necessary since `self` will
+    #     refer to the execution context of the block rather than the running
+    #     `TimerTask`.
+    #  
+    #   @note Calls Concurrent::Dereferenceable#  set_deref_options passing `opts`.
+    #     All options supported by Concurrent::Dereferenceable can be set
+    #     during object initialization.
     #
-    # @note Calls Concurrent::Dereferenceable#set_deref_options passing `opts`.
-    #   All options supported by Concurrent::Dereferenceable can be set
-    #   during object initialization.
-    #
-    # @see Concurrent::Dereferenceable#set_deref_options
-    def initialize(opts = {}, &block)
+    #   @return [TimerTask] the new `TimerTask`
+    #  
+    #   @see Concurrent::Dereferenceable#  set_deref_options
+    def initialize(opts = {}, &task)
       raise ArgumentError.new('no block given') unless block_given?
 
+      init_executor
+      set_deref_options(opts)
+
       self.execution_interval = opts[:execution] || opts[:execution_interval] || EXECUTION_INTERVAL
       self.timeout_interval = opts[:timeout] || opts[:timeout_interval] || TIMEOUT_INTERVAL
       @run_now = opts[:now] || opts[:run_now]
+      @executor = Concurrent::SafeTaskExecutor.new(task)
+      @running = Concurrent::AtomicFixnum.new(0)
 
-      @task = block
       self.observers = CopyOnWriteObserverSet.new
-      init_mutex
-      set_deref_options(opts)
     end
 
-    # Number of seconds after the task completes before the task is
-    # performed again.
+    # Is the executor running?
+    #
+    # @return [Boolean] `true` when running, `false` when shutting down or shutdown
+    def running?
+      @running.value == 1
+    end
+
+    # Execute a previously created `TimerTask`.
     #
-    # @param [Float] value number of seconds
+    # @return [TimerTask] a reference to `self`
     #
-    # @raise ArgumentError when value is non-numeric or not greater than zero
-    def execution_interval=(value)
-      if (value = value.to_f) <= 0.0
-        raise ArgumentError.new("'execution_interval' must be non-negative number")
+    # @example Instance and execute in separate steps
+    #   task = Concurrent::TimerTask.new(execution_interval: 10){ print "Hello World\n" }
+    #   task.running? #=> false
+    #   task.execute
+    #   task.running? #=> true
+    #
+    # @example Instance and execute in one line
+    #   task = Concurrent::TimerTask.new(execution_interval: 10){ print "Hello World\n" }.execute
+    #   task.running? #=> true
+    #
+    # @since 0.6.0
+    def execute
+      mutex.synchronize do
+        if @running.value == 0
+          @running.value = 1
+          schedule_next_task(@run_now ? 0 : @execution_interval)
+        end
       end
-      @execution_interval = value
+      self
     end
 
-    # Number of seconds the task can run before it is considered to have failed.
-    # Failed tasks are forcibly killed.
+    # Create and execute a new `TimerTask`.
     #
-    # @param [Float] value number of seconds
+    # @!macro timer_task_initialize
     #
-    # @raise ArgumentError when value is non-numeric or not greater than zero
-    def timeout_interval=(value)
+    # @example 
+    #   task = Concurrent::TimerTask.execute(execution_interval: 10){ print "Hello World\n" }
+    #   task.running? #=> true
+    #
+    # @since 0.6.0
+    def self.execute(opts = {}, &task)
+      TimerTask.new(opts, &task).execute
+    end
+
+    # @!attribute [rw] execution_interval
+    # @return [Fixnum] Number of seconds after the task completes before the
+    #   task is performed again.
+    def execution_interval
+      mutex.synchronize{ @execution_interval }
+    end
+
+    # @!attribute [rw] execution_interval
+    # @return [Fixnum] Number of seconds after the task completes before the
+    #   task is performed again.
+    def execution_interval=(value)
       if (value = value.to_f) <= 0.0
-        raise ArgumentError.new("'timeout_interval' must be non-negative number")
+        raise ArgumentError.new('must be greater than zero')
+      else
+        mutex.synchronize{ @execution_interval = value }
       end
-      @timeout_interval = value
     end
 
-    # Terminate with extreme prejudice. Useful in cases where `#stop` doesn't
-    # work because one of the threads becomes unresponsive.
-    #
-    # @return [Boolean] indicating whether or not the `TimerTask` was killed
-    #
-    # @note Do not use this method unless `#stop` has failed.
-    def kill
-      return true unless running?
-      mutex.synchronize do
-        @running = false
-        Thread.kill(@worker) unless @worker.nil?
-        Thread.kill(@monitor) unless @monitor.nil?
+    # @!attribute [rw] timeout_interval
+    # @return [Fixnum] Number of seconds the task can run before it is
+    #   considered to have failed.
+    def timeout_interval
+      mutex.synchronize{ @timeout_interval }
+    end
+
+    # @!attribute [rw] timeout_interval
+    # @return [Fixnum] Number of seconds the task can run before it is
+    #   considered to have failed.
+    def timeout_interval=(value)
+      if (value = value.to_f) <= 0.0
+        raise ArgumentError.new('must be greater than zero')
+      else
+        mutex.synchronize{ @timeout_interval = value }
       end
-      return true
-    rescue
-      return false
-    ensure
-      @worker = @monitor = nil
     end
-    alias_method :terminate, :kill
 
-    alias_method :cancel, :stop
+    # @deprecated Updated to use `Executor` instead of `Runnable`
+    def terminate(*args) deprecated(:terminate, :kill, *args); end
+
+    # @deprecated Updated to use `Executor` instead of `Runnable`
+    def stop(*args) deprecated(:stop, :shutdown, *args); end
+
+    # @deprecated Updated to use `Executor` instead of `Runnable`
+    def cancel(*args) deprecated(:cancel, :shutdown, *args); end
+
+    # @deprecated Updated to use `Executor` instead of `Runnable`
+    def run!(*args) deprecated(:run!, :execute); end
+
+    # @deprecated Updated to use `Executor` instead of `Runnable`
+    def self.run!(*args, &block)
+      warn "[DEPRECATED] `run!` is deprecated, please use `execute` instead."
+      Concurrent::Runnable::Context.new(TimerTask.new(*args, &block))
+    end
+
+    # @deprecated Updated to use `Executor` instead of `Runnable`
+    def run
+      raise Concurrent::Runnable::LifecycleError.new('already running') if @running.value == 1
+      self.execute
+      self.wait_for_termination
+      true
+    end
+
+    private :post, :<<
 
     protected
 
-    def on_run # :nodoc:
-      @monitor = Thread.current
+    # @!visibility private
+    def shutdown_execution
+      @running.value = 0
+      super
     end
 
-    def on_stop # :nodoc:
-      before_stop_proc.call if before_stop_proc
-      @monitor.wakeup if @monitor.alive?
-      Thread.pass
+    # @!visibility private
+    def kill_execution
+      @running.value = 0
+      super
     end
 
-    def on_task # :nodoc:
-      if @run_now
-        @run_now = false
-      else
-        sleep(@execution_interval)
-      end
-      execute_task
+    # @!visibility private
+    def schedule_next_task(interval = execution_interval)
+      Concurrent::timer(interval, Concurrent::Event.new, &method(:execute_task))
     end
 
-    def execute_task # :nodoc:
-      @value = ex = nil
-      @worker = Thread.new do
-        Thread.current.abort_on_exception = false
-        Thread.current[:result] = @task.call(self)
+    # @!visibility private
+    def execute_task(completion)
+      return unless @running.value == 1
+      Concurrent::timer(timeout_interval, completion, &method(:timeout_task))
+      success, value, reason = @executor.execute(self)
+      if completion.try?
+        self.value = value
+        schedule_next_task
+        time = Time.now
+        observers.notify_observers do
+          [time, self.value, reason]
+        end
       end
-      raise TimeoutError if @worker.join(@timeout_interval).nil?
-      mutex.synchronize { @value = @worker[:result] }
-    rescue Exception => e
-      ex = e
-    ensure
-      observers.notify_observers(Time.now, self.value, ex)
-      unless @worker.nil?
-        Thread.kill(@worker)
-        @worker = nil
+    end
+
+    # @!visibility private
+    def timeout_task(completion)
+      return unless @running.value == 1
+      if completion.try?
+        self.value = value
+        schedule_next_task
+        observers.notify_observers(Time.now, nil, Concurrent::TimeoutError.new)
       end
     end
+
+    # @deprecated Updated to use `Executor` instead of `Runnable`
+    # @!visibility private
+    def deprecated(old, new, *args)
+      warn "[DEPRECATED] `#{old}` is deprecated, please use `#{new}` instead."
+      self.send(new, *args)
+    end
   end
 end