Merge pull request #54 from jdantonio/safe_timer_set

Safe timer set

Author: jdantonio

lib/concurrent/executor/timer_set.rb

@@ -25,6 +25,7 @@ def initialize(opts = {})
       @queue = PriorityQueue.new(order: :min)
       @executor = get_executor_from(opts)
       @thread = nil
+      @condition = Condition.new
       init_executor
     end
 
@@ -43,10 +44,21 @@ def initialize(opts = {})
     # @raise [ArgumentError] if no block is given
     def post(intended_time, &task)
       time = TimerSet.calculate_schedule_time(intended_time).to_f
-      if super(time, &task)
-        check_processing_thread!
+      raise ArgumentError.new('no block given') unless block_given?
+
+      mutex.synchronize do
+        return false unless running?
+
+        if (time - Time.now.to_f) <= 0.01
+          @executor.post(&task)
+        else
+          @queue.push(Task.new(time, task))
+          check_processing_thread!
+        end
+
         true
       end
+
     end
 
     alias_method :kill, :shutdown
@@ -84,19 +96,13 @@ def self.calculate_schedule_time(intended_time, now = Time.now)
     # @!visibility private
     Task = Struct.new(:time, :op) do
       include Comparable
+
       def <=>(other)
         self.time <=> other.time
       end
     end
 
-    # @!visibility private
-    def execute(time, &task)
-      if (time - Time.now.to_f) <= 0.01
-        @executor.post(&task)
-      else
-        @queue.push(Task.new(time, task))
-      end
-    end
+    private_constant :Task
 
     # @!visibility private
     def shutdown_execution
@@ -114,50 +120,14 @@ def shutdown_execution
     #
     # @!visibility private
     def check_processing_thread!
-      mutex.synchronize do
-        return if shutdown? || @queue.empty?
-        if @thread && @thread.status == 'sleep'
-          @thread.wakeup
-        elsif @thread.nil? || ! @thread.alive?
-          @thread = Thread.new do
-            Thread.current.abort_on_exception = true
-            process_tasks
-          end
-        end
-      end
-    end
+      return if shutdown? || @queue.empty?
 
-    # Check the head of the internal task queue for a ready task.
-    #
-    # @return [Task] the next task to be executed or nil if none are ready
-    #
-    # @!visibility private
-    def next_task
-      mutex.synchronize do
-        unless @queue.empty? || @queue.peek.time > Time.now.to_f
-          @queue.pop
-        else
-          nil
-        end
+      @thread ||= Thread.new do
+        Thread.current.abort_on_exception = true
+        process_tasks
       end
     end
 
-    # Calculate the time difference, in seconds and milliseconds, between
-    # now and the intended execution time of the next task to be ececuted.
-    #
-    # @return [Integer] the number of seconds and milliseconds to sleep
-    #   or nil if the task queue is empty
-    #
-    # @!visibility private
-    def next_sleep_interval
-      mutex.synchronize do
-        if @queue.empty?
-          nil
-        else
-          @queue.peek.time - Time.now.to_f
-        end
-      end
-    end
 
     # Run a loop and execute tasks in the scheduled order and at the approximate
     # scheduled time. If no tasks remain the thread will exit gracefully so that
@@ -167,13 +137,22 @@ def next_sleep_interval
     # @!visibility private
     def process_tasks
       loop do
-        while task = next_task do
-          @executor.post(&task.op)
-        end
-        if (interval = next_sleep_interval).nil?
-          break
-        else
-          sleep([interval, 60].min)
+
+        mutex.synchronize do
+          if @queue.empty?
+            @thread = nil
+            break
+          end
+
+          task = @queue.peek
+          interval = task.time - Time.now.to_f
+
+          if interval <= 0
+            @executor.post(&task.op)
+            @queue.pop
+          else
+            @condition.wait(mutex, [interval, 60].min)
+          end
         end
       end
     end

spec/concurrent/executor/timer_set_spec.rb

@@ -24,31 +24,28 @@ module Concurrent
     end
 
     it 'executes a given task when given a Time' do
-      expected = false
-      subject.post(Time.now + 0.1){ expected = true }
-      sleep(0.2)
-      expected.should be_true
+      latch = CountDownLatch.new(1)
+      subject.post(Time.now + 0.1){ latch.count_down }
+      latch.wait(0.2).should be_true
     end
 
     it 'executes a given task when given an interval in seconds' do
-      expected = false
-      subject.post(0.1){ expected = true }
-      sleep(0.2)
+      latch = CountDownLatch.new(1)
+      subject.post(0.1){ latch.count_down }
+      latch.wait(0.2).should be_true
     end
 
     it 'immediately posts a task when the delay is zero' do
       Thread.should_not_receive(:new).with(any_args)
-      expected = false
-      subject.post(0){ expected = true }
+      subject.post(0){ true }
     end
 
     it 'does not execute tasks early' do
-      pending('intermittently failing on Travis CI')
       expected = AtomicFixnum.new(0)
       subject.post(0.2){ expected.increment }
-      sleep(0.1)
+      sleep(0.15)
       expected.value.should eq 0
-      sleep(0.1)
+      sleep(0.10)
       expected.value.should eq 1
     end
 
@@ -71,11 +68,9 @@ module Concurrent
     end
 
     it 'executes all tasks scheduled for the same time' do
-      pending('intermittently failing on Travis CI')
-      expected = AtomicFixnum.new(0)
-      5.times{ subject.post(0.1){ expected.increment } }
-      sleep(0.2)
-      expected.value.should eq 5
+      latch = CountDownLatch.new(5)
+      5.times{ subject.post(0.1){ latch.count_down } }
+      latch.wait(0.2).should be_true
     end
 
     it 'executes tasks with different times in schedule order' do
@@ -104,21 +99,23 @@ module Concurrent
     end
 
     it 'stops the monitor thread on #shutdown' do
-      subject.post(0.1){ nil } # start the monitor thread
-      sleep(0.2)
-      subject.instance_variable_get(:@thread).should_not be_nil
+      subject.post(0.2){ nil } # start the monitor thread
+      sleep(0.1)
+      thread = subject.instance_variable_get(:@thread)
+      thread.should_not be_nil
       subject.shutdown
       sleep(0.1)
-      subject.instance_variable_get(:@thread).should_not be_alive
+      thread.should_not be_alive
     end
 
     it 'kills the monitor thread on #kill' do
-      subject.post(0.1){ nil } # start the monitor thread
-      sleep(0.2)
-      subject.instance_variable_get(:@thread).should_not be_nil
+      subject.post(0.2){ nil } # start the monitor thread
+      sleep(0.1)
+      thread = subject.instance_variable_get(:@thread)
+      thread.should_not be_nil
       subject.kill
       sleep(0.1)
-      subject.instance_variable_get(:@thread).should_not be_alive
+      thread.should_not be_alive
     end
 
     it 'rejects tasks once shutdown' do