<commit_before>require 'thread'
require_relative 'executor'
require 'concurrent/options_parser'
require 'concurrent/atomic/event'
require 'concurrent/collection/priority_queue'
require 'concurrent/executor/single_thread_executor'

module Concurrent

  # Executes a collection of tasks at the specified times. A master thread
  # monitors the set and schedules each task for execution at the appropriate
  # time. Tasks are run on the global task pool or on the supplied executor.
  class TimerSet
    include RubyExecutor

    # Create a new set of timed tasks.
    #
    # @param [Hash] opts the options controlling how the future will be processed
    # @option opts [Boolean] :operation (false) when `true` will execute the future on the global
    #   operation pool (for long-running operations), when `false` will execute the future on the
    #   global task pool (for short-running tasks)
    # @option opts [object] :executor when provided will run all operations on
    #   this executor rather than the global thread pool (overrides :operation)
    def initialize(opts = {})
      @queue = PriorityQueue.new(order: :min)
      @task_executor = OptionsParser::get_executor_from(opts)
      @timer_executor = SingleThreadExecutor.new
      @condition = Condition.new
      init_executor
    end

    # Post a task to be execute at the specified time. The given time may be either
    # a `Time` object or the number of seconds to wait. If the intended execution
    # time is within 1/100th of a second of the current time the task will be
    # immediately post to the executor.
    #
    # @param [Object] intended_time the time to schedule the task for execution
    #
    # @yield the task to be performed
    #
    # @return [Boolean] true if the message is post, false after shutdown
    #
    # @raise [ArgumentError] if the intended execution time is not in the future
    # @raise [ArgumentError] if no block is given
    def post(intended_time, *args, &task)
      time = TimerSet.calculate_schedule_time(intended_time).to_f
      raise ArgumentError.new('no block given') unless block_given?

      mutex.synchronize do
        return false unless running?

        if (time - Time.now.to_f) <= 0.01
          @task_executor.post(*args, &task)
        else
          @queue.push(Task.new(time, args, task))
          @timer_executor.post(&method(:process_tasks))
        end

        true
      end

    end

    alias_method :kill, :shutdown

    # Calculate an Epoch time with milliseconds at which to execute a
    # task. If the given time is a `Time` object it will be converted<commit_msg>Redefine kill, which was relying on incorrect @queue.clear in shutdown<commit_after>require 'thread'
require_relative 'executor'
require 'concurrent/options_parser'
require 'concurrent/atomic/event'
require 'concurrent/collection/priority_queue'
require 'concurrent/executor/single_thread_executor'

module Concurrent

  # Executes a collection of tasks at the specified times. A master thread
  # monitors the set and schedules each task for execution at the appropriate
  # time. Tasks are run on the global task pool or on the supplied executor.
  class TimerSet
    include RubyExecutor

    # Create a new set of timed tasks.
    #
    # @param [Hash] opts the options controlling how the future will be processed
    # @option opts [Boolean] :operation (false) when `true` will execute the future on the global
    #   operation pool (for long-running operations), when `false` will execute the future on the
    #   global task pool (for short-running tasks)
    # @option opts [object] :executor when provided will run all operations on
    #   this executor rather than the global thread pool (overrides :operation)
    def initialize(opts = {})
      @queue = PriorityQueue.new(order: :min)
      @task_executor = OptionsParser::get_executor_from(opts)
      @timer_executor = SingleThreadExecutor.new
      @condition = Condition.new
      init_executor
    end

    # Post a task to be execute at the specified time. The given time may be either
    # a `Time` object or the number of seconds to wait. If the intended execution
    # time is within 1/100th of a second of the current time the task will be
    # immediately post to the executor.
    #
    # @param [Object] intended_time the time to schedule the task for execution
    #
    # @yield the task to be performed
    #
    # @return [Boolean] true if the message is post, false after shutdown
    #
    # @raise [ArgumentError] if the intended execution time is not in the future
    # @raise [ArgumentError] if no block is given
    def post(intended_time, *args, &task)
      time = TimerSet.calculate_schedule_time(intended_time).to_f
      raise ArgumentError.new('no block given') unless block_given?

      mutex.synchronize do
        return false unless running?

        if (time - Time.now.to_f) <= 0.01
          @task_executor.post(*args, &task)
        else
          @queue.push(Task.new(time, args, task))
          @timer_executor.post(&method(:process_tasks))
        end

        true
      end

    end

    # For a timer, #kill is like an orderly shutdown, except we need to manually
    # (and destructively) clear the queue first
    def kill
      @queue.clear
      shutdown
    end

    # Calculate an Epoch time with milliseconds at which to execute a
    # task. If the given time is a `Time` object it will be converted
{'fraction_non_alphanumeric': 0.06786830972963032, 'fraction_numerical': 0.0025749494206363803, 'mean_word_length': 3.1196969696969696, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5378526', 'n_tokens_mistral': 1502, 'n_tokens_neox': 1449, 'n_words': 628}