changed timer set executor strategy

Author: mighe

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
 
@@ -120,47 +121,13 @@ def shutdown_execution
     # @!visibility private
     def check_processing_thread!
       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
 
-    # 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
@@ -170,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

@@ -41,12 +41,11 @@ module Concurrent
     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
 
@@ -69,10 +68,9 @@ module Concurrent
     end
 
     it 'executes all tasks scheduled for the same time' do
-      pending('intermittently failing on Travis CI')
       latch = CountDownLatch.new(5)
       5.times{ subject.post(0.1){ latch.count_down } }
-      latch.wait(0.2).should eq 5
+      latch.wait(0.2).should be_true
     end
 
     it 'executes tasks with different times in schedule order' do
@@ -101,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