TimerSet#post now returns an IVar.

Author: jdantonio

lib/concurrent/executor/timer_set.rb

@@ -1,4 +1,5 @@
-require 'thread'
+require 'concurrent/errors'
+require 'concurrent/ivar'
 require 'concurrent/atomic/event'
 require 'concurrent/collection/priority_queue'
 require 'concurrent/executor/executor'
@@ -15,6 +16,57 @@ module Concurrent
   # @!macro monotonic_clock_warning
   class TimerSet < RubyExecutorService
 
+    # @!visibility private
+    class Job < Concurrent::IVar
+
+      # @!visibility private
+      def initialize(args, task)
+        super()
+        @args = args
+        @task = task
+        ensure_ivar_visibility!
+      end
+
+      # @!visibility private
+      def cancelled?
+        state == :cancelled
+      end
+
+      # @!visibility private
+      def cancel
+        if compare_and_set_state(:cancelled, :pending)
+          complete(false, nil, CancelledOperationError.new)
+          true
+        else
+          false
+        end
+      end
+
+      # @!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
+      end
+    end
+    private_constant :Task
+
     # Create a new set of timed tasks.
     #
     # @!macro [attach] executor_options
@@ -48,16 +100,19 @@ def post(delay, *args, &task)
 
       synchronize do
         return false unless running?
+        
+        job = Job.new(args, task)
 
         if (delay) <= 0.01
-          @task_executor.post(*args, &task)
+          @task_executor.post{ job.execute }
         else
-          @queue.push(Task.new(Concurrent.monotonic_time + delay, args, task))
+          @queue.push(Task.new(Concurrent.monotonic_time + delay, job))
           @timer_executor.post(&method(:process_tasks))
         end
+
         @condition.set
+        job
       end
-      true
     end
 
     # Begin an immediate shutdown. In-progress tasks will be allowed to
@@ -94,22 +149,6 @@ def self.calculate_delay!(delay)
 
     protected
 
-    # 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, :args, :op) do
-      include Comparable
-
-      def <=>(other)
-        self.time <=> other.time
-      end
-    end
-
-    private_constant :Task
-
     def ns_initialize(opts)
       @queue          = PriorityQueue.new(order: :min)
       @task_executor  = Executor.executor_from_options(opts) || Concurrent.global_io_executor
@@ -153,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.args, &task.op)
+          @task_executor.post{ task.job.execute }
         else
           @condition.wait([diff, 60].min)
         end

lib/concurrent/obligation.rb

@@ -136,23 +136,23 @@ def exception(*args)
     protected
 
     # @!visibility private
-    def get_arguments_from(opts = {}) # :nodoc:
+    def get_arguments_from(opts = {})
       [*opts.fetch(:args, [])]
     end
 
     # @!visibility private
-    def init_obligation(*args) # :nodoc:
+    def init_obligation(*args)
       init_mutex(*args)
       @event = Event.new
     end
 
     # @!visibility private
-    def event # :nodoc:
+    def event
       @event
     end
 
     # @!visibility private
-    def set_state(success, value, reason) # :nodoc:
+    def set_state(success, value, reason)
       if success
         @value = value
         @state = :fulfilled
@@ -163,7 +163,7 @@ def set_state(success, value, reason) # :nodoc:
     end
 
     # @!visibility private
-    def state=(value) # :nodoc:
+    def state=(value)
       mutex.synchronize { @state = value }
     end
 
@@ -176,7 +176,7 @@ def state=(value) # :nodoc:
     # @return [Boolean] true is state is changed, false otherwise
     #
     # @!visibility private
-    def compare_and_set_state(next_state, expected_current) # :nodoc:
+    def compare_and_set_state(next_state, expected_current)
       mutex.synchronize do
         if @state == expected_current
           @state = next_state
@@ -192,7 +192,7 @@ def compare_and_set_state(next_state, expected_current) # :nodoc:
     # @return block value if executed, false otherwise
     #
     # @!visibility private
-    def if_state(*expected_states) # :nodoc:
+    def if_state(*expected_states)
       mutex.synchronize do
         raise ArgumentError.new('no block given') unless block_given?