Gecko [ lorencats.com ]

Name	Size	Permission
abstract_executor_service.rb	3.29 KB	-rw-r--r--
cached_thread_pool.rb	2.58 KB	-rw-r--r--
executor_service.rb	5.65 KB	-rw-r--r--
fixed_thread_pool.rb	9.57 KB	-rw-r--r--
immediate_executor.rb	1.8 KB	-rw-r--r--
indirect_immediate_executor.rb	1.54 KB	-rw-r--r--
java_executor_service.rb	2.5 KB	-rw-r--r--
java_single_thread_executor.rb	919 B	-rw-r--r--
java_thread_pool_executor.rb	3.99 KB	-rw-r--r--
ruby_executor_service.rb	1.49 KB	-rw-r--r--
ruby_single_thread_executor.rb	607 B	-rw-r--r--
ruby_thread_pool_executor.rb	9.98 KB	-rw-r--r--
safe_task_executor.rb	1.01 KB	-rw-r--r--
serial_executor_service.rb	991 B	-rw-r--r--
serialized_execution.rb	2.73 KB	-rw-r--r--
serialized_execution_delegator...	859 B	-rw-r--r--
simple_executor_service.rb	2.57 KB	-rw-r--r--
single_thread_executor.rb	2.47 KB	-rw-r--r--
thread_pool_executor.rb	4.19 KB	-rw-r--r--
timer_set.rb	5.92 KB	-rw-r--r--

Code Editor : timer_set.rb

require 'concurrent/scheduled_task'
require 'concurrent/atomic/event'
require 'concurrent/collection/non_concurrent_priority_queue'
require 'concurrent/executor/executor_service'
require 'concurrent/executor/single_thread_executor'

require 'concurrent/options'

module Concurrent

# Executes a collection of tasks, each after a given delay. A master task
  # monitors the set and schedules each task for execution at the appropriate
  # time. Tasks are run on the global thread pool or on the supplied executor.
  # Each task is represented as a `ScheduledTask`.
  #
  # @see Concurrent::ScheduledTask
  #
  # @!macro monotonic_clock_warning
  class TimerSet < RubyExecutorService

# Create a new set of timed tasks.
    #
    # @!macro [attach] executor_options
    #
    #   @param [Hash] opts the options used to specify the executor on which to perform actions
    #   @option opts [Executor] :executor when set use the given `Executor` instance.
    #     Three special values are also supported: `:task` returns the global task pool,
    #     `:operation` returns the global operation pool, and `:immediate` returns a new
    #     `ImmediateExecutor` object.
    def initialize(opts = {})
      super(opts)
    end

# Post a task to be execute run after a given delay (in seconds). If the
    # delay is less than 1/100th of a second the task will be immediately post
    # to the executor.
    #
    # @param [Float] delay the number of seconds to wait for before executing the task.
    # @param [Array<Object>] args the arguments passed to the task on execution.
    #
    # @yield the task to be performed.
    #
    # @return [Concurrent::ScheduledTask, false] IVar representing the task if the post
    #   is successful; false after shutdown.
    #
    # @raise [ArgumentError] if the intended execution time is not in the future.
    # @raise [ArgumentError] if no block is given.
    def post(delay, *args, &task)
      raise ArgumentError.new('no block given') unless block_given?
      return false unless running?
      opts = {
        executor: @task_executor,
        args: args,
        timer_set: self
      }
      task = ScheduledTask.execute(delay, opts, &task) # may raise exception
      task.unscheduled? ? false : task
    end

# Begin an immediate shutdown. In-progress tasks will be allowed to
    # complete but enqueued tasks will be dismissed and no new tasks
    # will be accepted. Has no additional effect if the thread pool is
    # not running.
    def kill
      shutdown
    end

private :<<

private

# Initialize the object.
    #
    # @param [Hash] opts the options to create the object with.
    # @!visibility private
    def ns_initialize(opts)
      @queue          = Collection::NonConcurrentPriorityQueue.new(order: :min)
      @task_executor  = Options.executor_from_options(opts) || Concurrent.global_io_executor
      @timer_executor = SingleThreadExecutor.new
      @condition      = Event.new
      @ruby_pid       = $$ # detects if Ruby has forked
      self.auto_terminate = opts.fetch(:auto_terminate, true)
    end

# Post the task to the internal queue.
    #
    # @note This is intended as a callback method from ScheduledTask
    #   only. It is not intended to be used directly. Post a task
    #   by using the `SchedulesTask#execute` method.
    #
    # @!visibility private
    def post_task(task)
      synchronize{ ns_post_task(task) }
    end

# @!visibility private
    def ns_post_task(task)
      return false unless ns_running?
      ns_reset_if_forked
      if (task.initial_delay) <= 0.01
        task.executor.post{ task.process_task }
      else
        @queue.push(task)
        # only post the process method when the queue is empty
        @timer_executor.post(&method(:process_tasks)) if @queue.length == 1
        @condition.set
      end
      true
    end

# Remove the given task from the queue.
    #
    # @note This is intended as a callback method from `ScheduledTask`
    #   only. It is not intended to be used directly. Cancel a task
    #   by using the `ScheduledTask#cancel` method.
    #
    # @!visibility private
    def remove_task(task)
      synchronize{ @queue.delete(task) }
    end

# `ExecutorService` callback called during shutdown.
    #
    # @!visibility private
    def ns_shutdown_execution
      ns_reset_if_forked
      @queue.clear
      @timer_executor.kill
      stopped_event.set
    end

def ns_reset_if_forked
      if $$ != @ruby_pid
        @queue.clear
        @condition.reset
        @ruby_pid = $$
      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
    # garbage collection can occur. If there are no ready tasks it will sleep
    # for up to 60 seconds waiting for the next scheduled task.
    #
    # @!visibility private
    def process_tasks
      loop do
        task = synchronize { @condition.reset; @queue.peek }
        break unless task

now = Concurrent.monotonic_time
        diff = task.schedule_time - now

if diff <= 0
          # We need to remove the task from the queue before passing
          # it to the executor, to avoid race conditions where we pass
          # the peek'ed task to the executor and then pop a different
          # one that's been added in the meantime.
          #
          # Note that there's no race condition between the peek and
          # this pop - this pop could retrieve a different task from
          # the peek, but that task would be due to fire now anyway
          # (because @queue is a priority queue, and this thread is
          # the only reader, so whatever timer is at the head of the
          # queue now must have the same pop time, or a closer one, as
          # when we peeked).
          task = synchronize { @queue.pop }
          task.executor.post{ task.process_task }
        else
          @condition.wait([diff, 60].min)
        end
      end
    end
  end
end

${this.title}

Code Editor : timer_set.rb