Extracted methods into IVar from subclasses.

Author: jdantonio

lib/concurrent/executor/timer_set.rb

@@ -16,15 +16,32 @@ module Concurrent
   # @!macro monotonic_clock_warning
   class TimerSet < RubyExecutorService
 
+    # A class for encapsulating a task and its intended execution time.
+    # It facilitates proper prioritization by overriding the comparison
+    # (spaceship) operator as a comparison of the intended execution
+    # times.
+    #
     # @!visibility private
-    class Job < Concurrent::IVar
+    class Task < Concurrent::IVar
+      include Comparable
 
       # @!visibility private
-      def initialize(args, task)
+      def initialize(time, args, task)
         super()
-        @args = args
-        @task = task
-        ensure_ivar_visibility!
+        synchronize do
+          @time = time
+          @args = args
+          @task = task
+        end
+      end
+
+      def time
+        synchronize { @time }
+      end
+
+      # @!visibility private
+      def <=>(other)
+        self.time <=> other.time
       end
 
       # @!visibility private
@@ -44,25 +61,7 @@ def cancel
 
       # @!visibility private
       def execute
-        if compare_and_set_state(:processing, :pending)
-          success, val, reason = SafeTaskExecutor.new(@task, rescue_exception: true).execute(*@args)
-          complete(success, val, reason)
-        end
-      end
-    end
-    private_constant :Job
-
-    # A struct for encapsulating a task and its intended execution time.
-    # It facilitates proper prioritization by overriding the comparison
-    # (spaceship) operator as a comparison of the intended execution
-    # times.
-    #
-    # @!visibility private
-    Task = Struct.new(:time, :job) do
-      include Comparable
-
-      def <=>(other)
-        self.time <=> other.time
+        safe_execute(@task, @args)
       end
     end
     private_constant :Task
@@ -100,18 +99,19 @@ def post(delay, *args, &task)
 
       synchronize do
         return false unless running?
-        
-        job = Job.new(args, task)
+
+        time = Concurrent.monotonic_time + delay
+        task = Task.new(time, args, task)
 
         if (delay) <= 0.01
-          @task_executor.post{ job.execute }
+          @task_executor.post{ task.execute }
         else
-          @queue.push(Task.new(Concurrent.monotonic_time + delay, job))
+          @queue.push(task)
           @timer_executor.post(&method(:process_tasks))
         end
 
         @condition.set
-        job
+        task
       end
     end
 
@@ -192,7 +192,7 @@ def process_tasks
           # queue now must have the same pop time, or a closer one, as
           # when we peeked).
           task = synchronize { @queue.pop }
-          @task_executor.post{ task.job.execute }
+          @task_executor.post{ task.execute }
         else
           @condition.wait([diff, 60].min)
         end

lib/concurrent/future.rb

@@ -45,7 +45,7 @@ def initialize(opts = {}, &block)
     #   future.state #=> :pending
     def execute
       if compare_and_set_state(:pending, :unscheduled)
-        @executor.post(@args){ work }
+        @executor.post{ safe_execute(@task, @args) }
         self
       end
     end
@@ -90,7 +90,12 @@ def set(value = IVar::NO_VALUE, &block)
     #
     # @return [Boolean] was the operation successfully cancelled.
     def cancel
-      compare_and_set_state(:cancelled, :pending)
+      if compare_and_set_state(:cancelled, :pending)
+        complete(false, nil, CancelledOperationError.new)
+        true
+      else
+        false
+      end
     end
 
     # Has the operation been successfully cancelled?
@@ -125,17 +130,5 @@ def ns_initialize(value, opts)
       @executor = Executor.executor_from_options(opts) || Concurrent.global_io_executor
       @args = get_arguments_from(opts)
     end
-
-    private
-
-    # @!visibility private
-    def work
-      if compare_and_set_state(:processing, :pending)
-        success, val, reason = SafeTaskExecutor.new(@task, rescue_exception: true).execute(*@args)
-        complete(success, val, reason)
-      else
-        complete(false, nil, CancelledOperationError.new)
-      end
-    end
   end
 end

lib/concurrent/ivar.rb

@@ -148,6 +148,7 @@ def try_set(value = NO_VALUE, &block)
 
     protected
 
+    # @!visibility private
     def ns_initialize(value, opts)
       init_obligation(self)
       self.observers = CopyOnWriteObserverSet.new
@@ -160,21 +161,34 @@ def ns_initialize(value, opts)
       end
     end
 
+    # @!visibility private
+    def safe_execute(task, args = [])
+      if compare_and_set_state(:processing, :pending)
+        success, val, reason = SafeTaskExecutor.new(task, rescue_exception: true).execute(*@args)
+        complete(success, val, reason)
+        yield(success, val, reason) if block_given?
+      end
+    end
+
+    # @!visibility private
     def complete(success, value, reason)
       complete_without_notification(success, value, reason)
       notify_observers(self.value, reason)
       self
     end
 
+    # @!visibility private
     def complete_without_notification(success, value, reason)
       synchronize { ns_complete_without_notification(success, value, reason) }
       self
     end
 
+    # @!visibility private
     def notify_observers(value, reason)
       observers.notify_and_delete_observers{ [Time.now, value, reason] }
     end
 
+    # @!visibility private
     def ns_complete_without_notification(success, value, reason)
       raise MultipleAssignmentError if [:fulfilled, :rejected].include? @state
       set_state(success, value, reason)

lib/concurrent/scheduled_task.rb

@@ -29,13 +29,13 @@ module Concurrent
   # `ScheduledTask` mixes in the  [Obligation](Obligation) module thus giving it
   # "future" behavior. This includes the expected lifecycle states. `ScheduledTask`
   # has one additional state, however. While the task (block) is being executed the
-  # state of the object will be `:in_progress`. This additional state is necessary
+  # state of the object will be `:processing`. This additional state is necessary
   # because it has implications for task cancellation. 
   # 
   # **Cancellation**
   # 
   # A `:pending` task can be cancelled using the `#cancel` method. A task in any
-  # other state, including `:in_progress`, cannot be cancelled. The `#cancel`
+  # other state, including `:processing`, cannot be cancelled. The `#cancel`
   # method returns a boolean indicating the success of the cancellation attempt.
   # A cancelled `ScheduledTask` cannot be restarted. It is immutable. 
   # 
@@ -220,8 +220,18 @@ def cancelled?
     # In the task execution in progress?
     #
     # @return [Boolean] true if the task is in the given state else false
+    def processing?
+      state == :processing
+    end
+
+    # In the task execution in progress?
+    #
+    # @return [Boolean] true if the task is in the given state else false
+    #
+    # @deprecated
     def in_progress?
-      state == :in_progress
+      warn '[DEPRECATED] use #processing? instead'
+      processing?
     end
 
     # Cancel this task and prevent it from executing. A task can only be
@@ -244,14 +254,8 @@ def cancel
 
     # @!visibility private
     def process_task
-      if compare_and_set_state(:in_progress, :pending)
-        success, val, reason = SafeTaskExecutor.new(@task).execute
-
-        mutex.synchronize do
-          set_state(success, val, reason)
-          event.set
-        end
-
+      safe_execute(@task) do |success, val, reason|
+        event.set
         time = Time.now
         observers.notify_and_delete_observers { [time, self.value, reason] }
       end