Class: Concurrent::TimerSet

# File 'lib/concurrent-ruby/concurrent/executor/timer_set.rb', line 74

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
end

# File 'lib/concurrent-ruby/concurrent/executor/timer_set.rb', line 94

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

# File 'lib/concurrent-ruby/concurrent/executor/timer_set.rb', line 129

def ns_reset_if_forked
  if $$ != @ruby_pid
    @queue.clear
    @condition.reset
    @ruby_pid = $$
  end
end

# File 'lib/concurrent-ruby/concurrent/executor/timer_set.rb', line 122

def ns_shutdown_execution
  ns_reset_if_forked
  @queue.clear
  @timer_executor.kill
  stopped_event.set
end

# File 'lib/concurrent-ruby/concurrent/executor/timer_set.rb', line 48

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

# File 'lib/concurrent-ruby/concurrent/executor/timer_set.rb', line 89

def post_task(task)
  synchronize { ns_post_task(task) }
end

# File 'lib/concurrent-ruby/concurrent/executor/timer_set.rb', line 143

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 }
      begin
        task.executor.post { task.process_task }
      rescue RejectedExecutionError
        # ignore and continue
      end
    else
      @condition.wait([diff, 60].min)
    end
  end
end

# File 'lib/concurrent-ruby/concurrent/executor/timer_set.rb', line 115

def remove_task(task)
  synchronize { @queue.delete(task) }
end