Concurrent::timer and global timer pool now use TimerSet.

Author: jdantonio

lib/concurrent/configuration.rb

@@ -66,7 +66,7 @@ def global_operation_pool
     #
     # @see Concurrent::timer
     def global_timer_pool
-      @global_timer_pool ||= Concurrent::CachedThreadPool.new
+      @global_timer_pool ||= Concurrent::TimerSet.new
     end
 
     # Global thread pool optimized for short *tasks*.

lib/concurrent/executor.rb

@@ -5,3 +5,4 @@
 require 'concurrent/executor/safe_task_executor'
 require 'concurrent/executor/single_thread_executor'
 require 'concurrent/executor/thread_pool_executor'
+require 'concurrent/executor/timer_set'

lib/concurrent/executor/timer_set.rb

@@ -0,0 +1,86 @@
+require 'thread'
+require 'concurrent/options_parser'
+require 'concurrent/collection/priority_queue'
+
+module Concurrent
+
+  class TimerSet
+
+    def initialize(opts = {})
+      @mutex = Mutex.new
+      @queue = PriorityQueue.new(order: :min)
+      @executor = OptionsParser::get_executor_from(opts)
+      @thread = nil
+    end
+
+    def post(intended_time, &block)
+      raise ArgumentError.new('no block given') unless block_given?
+      time = calculate_schedule_time(intended_time)
+      @mutex.synchronize{ @queue.push(Task.new(time, block)) }
+      check_processing_thread
+    end
+
+    private
+
+    Task = Struct.new(:time, :op) do
+      include Comparable
+      def <=>(other)
+        self.time <=> other.time
+      end
+    end
+
+    def calculate_schedule_time(intended_time, now = Time.now)
+      if intended_time.is_a?(Time)
+        raise SchedulingError.new('schedule time must be in the future') if intended_time <= now
+        intended_time.to_f
+      else
+        raise SchedulingError.new('seconds must be greater than zero') if intended_time.to_f <= 0.0
+        now.to_f + intended_time.to_f
+      end
+    end
+
+    def check_processing_thread
+      if @thread && @thread.status == 'sleep'
+        @thread.wakeup
+      elsif @thread.nil? || ! @thread.alive?
+        @thread = Thread.new do
+          Thread.current.abort_on_exception = false
+          process_tasks
+        end
+      end
+    end
+
+    def next_task
+      @mutex.synchronize do
+        unless @queue.empty? || @queue.peek.time > Time.now.to_f
+          @queue.pop
+        else
+          nil
+        end
+      end
+    end
+
+    def next_sleep_interval
+      @mutex.synchronize do
+        if @queue.empty?
+          nil
+        else
+          @queue.peek.time - Time.now.to_f
+        end
+      end
+    end
+
+    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)
+        end
+      end
+    end
+  end
+end

lib/concurrent/options_parser.rb

@@ -20,5 +20,6 @@ def get_executor_from(opts = {})
         Concurrent.configuration.global_task_pool
       end
     end
+    module_function :get_executor_from
   end
 end

lib/concurrent/utilities/timer.rb

@@ -1,4 +1,4 @@
-require 'rbconfig'
+require 'concurrent/configuration'
 require 'thread'
 
 module Concurrent
@@ -8,18 +8,11 @@ module Concurrent
   # @param [Fixnum] seconds the interval in seconds to wait before executing the task
   # @yield the task to execute
   # @return [Boolean] true
-  def timer(seconds)
+  def timer(seconds, &block)
     raise ArgumentError.new('no block given') unless block_given?
     raise ArgumentError.new('interval must be greater than or equal to zero') if seconds < 0
 
-    Concurrent.configuration.global_timer_pool.post do
-      begin
-        sleep(seconds)
-        yield
-      rescue Exception
-        # suppress
-      end
-    end
+    Concurrent.configuration.global_timer_pool.post(seconds, &block)
     true
   end
   module_function :timer

spec/concurrent/utilities/timer_spec.rb

@@ -34,7 +34,7 @@ module Concurrent
     end
 
     it 'runs the task on the global timer pool' do
-      Concurrent.configuration.global_timer_pool.should_receive(:post).with(no_args)
+      Concurrent.configuration.global_timer_pool.should_receive(:post).with(0.1)
       Concurrent::timer(0.1){ :foo }
     end
   end