<commit_before>class Pigeon::Queue
  # == Constants ============================================================
  
  # == Exceptions ===========================================================

  class BlockRequired < Exception
  end
  
  class TaskNotQueued < Exception
    def initialize(task = nil)
      @task = task
    end
    
    def inspect
      "Task #{@task.inspect} not queued."
    end
    alias_method :to_s, :inspect
  end

  # == Properties ===========================================================
  
  attr_reader :processors

  # == Class Methods ========================================================
  
  # Returns the current filter configuration. This is stored as a Hash with
  # the key being the filter name, the value being the matching block. The
  # nil key is the default filter which accepts all tasks.
  def self.filters
    @filters ||= {
      nil => lambda { |task| true }
    }
  end
  
  # Defines a new filter with the given name and uses the supplied block to
  # evaluate if a task qualifies or not.
  def self.filter(name, &block)
    filters[name] = block
  end

  # == Instance Methods =====================================================

  # Creates a new queue. If a block is given, it is used to compare two tasks
  # and order them, so it should take two arguments and return the relative
  # difference (-1, 0, 1) like Array#sort would work.
  def initialize(&block)
    @filter_lock = Mutex.new
    @observer_lock = Mutex.new

    @claimable_task = { }
    @filters = self.class.filters.dup
    @observers = { }
    @processors = [ ]
    @next_task = { }
    @insert_backlog = [ ]
    
    if (block_given?)
      @sort_by = block
    else
      @sort_by = lambda { |a,b| a.priority <=> b.priority }
    end

    @tasks = Pigeon::SortedArray.new(&@sort_by)
  end
  
  # Returns the contents sorted by the given block. The block will be passed
  # a single Task and the results are sorted by the return value.
  def sort_by(&block)
    raise BlockRequired unless (block_given?)

    @sort_by = block
    @filter_lock.synchronize do
      @tasks = Pigeon::SortedArray.new(&@sort_by) + @tasks
      
      @next_task = { }
    end
  end
  
  # Sets up a callback for the queue that will execute the block if new tasks
  # are added to the queue. If filter_name is specified, this block will be
  # run for tasks matching that filtered subset.
  def observe(filter_name = nil, &block)
    raise BlockRequired unless (block_given?)
    
    @observer_lock.synchronize do
      @observers[filter_name] ||= [ ]

      @observers[filter_name] << block
    end

    task = assign_next_task(filter_name)
  end
  
  # Removes references to the callback function specified. Note that the same
  # Proc must be passed in, as a block with an identical function will not
  # be considered equivalent.
  def remove_observer(filter_name = nil, &block)
    @observer_lock.synchronize do
      set = @observers[filter_name]

      set and set.delete(block)
    end
  end
  
  # Adds a processor to the queue and adds an observer claim method.
  def add_processor(processor, &claim)
    @observer_lock.synchronize do
      @processors << processor
    end

    observe(&claim) if (claim)
  end
  
  # Removes a processor from the queue and removes an observer claim method.
  def remove_processor(processor, &claim)
    @observer_lock.synchronize do
      @processors.delete(processor)
    end

    remove_observer(&claim) if (claim)
  end
  
  # Creates a named filter for the queue using the provided block to select
  # the tasks which should match.
  def filter(filter_name, &block)
    raise BlockRequired unless (block_given?)

    @filter_lock.synchronize do
      @filters[filter_name] = block
    end
    
    assign_next_task(filter_name)
  end
  
  # Adds a task to the queue.
  def <<(task)
    # If there is an insert operation already in progress, put this task in
    # the backlog for subsequent processing.
    
    if (@observer_lock.locked?)
      @insert_backlog << task
      return task
    end
    
    active_task = task
    
    while (active_task) do
      # Set the claimable task flag for this task since it is not yet in the<commit_msg>Adding exception when adding non Task-derived objects to queues<commit_after>class Pigeon::Queue
  # == Constants ============================================================
  
  # == Exceptions ===========================================================

  class BlockRequired < Exception
  end
  
  class TaskNotQueued < Exception
    def initialize(task = nil)
      @task = task
    end
    
    def inspect
      "Task #{@task.inspect} not queued."
    end
    alias_method :to_s, :inspect
  end

  # == Properties ===========================================================
  
  attr_reader :processors

  # == Class Methods ========================================================
  
  # Returns the current filter configuration. This is stored as a Hash with
  # the key being the filter name, the value being the matching block. The
  # nil key is the default filter which accepts all tasks.
  def self.filters
    @filters ||= {
      nil => lambda { |task| true }
    }
  end
  
  # Defines a new filter with the given name and uses the supplied block to
  # evaluate if a task qualifies or not.
  def self.filter(name, &block)
    filters[name] = block
  end

  # == Instance Methods =====================================================

  # Creates a new queue. If a block is given, it is used to compare two tasks
  # and order them, so it should take two arguments and return the relative
  # difference (-1, 0, 1) like Array#sort would work.
  def initialize(&block)
    @filter_lock = Mutex.new
    @observer_lock = Mutex.new

    @claimable_task = { }
    @filters = self.class.filters.dup
    @observers = { }
    @processors = [ ]
    @next_task = { }
    @insert_backlog = [ ]
    
    if (block_given?)
      @sort_by = block
    else
      @sort_by = lambda { |a,b| a.priority <=> b.priority }
    end

    @tasks = Pigeon::SortedArray.new(&@sort_by)
  end
  
  # Returns the contents sorted by the given block. The block will be passed
  # a single Task and the results are sorted by the return value.
  def sort_by(&block)
    raise BlockRequired unless (block_given?)

    @sort_by = block
    @filter_lock.synchronize do
      @tasks = Pigeon::SortedArray.new(&@sort_by) + @tasks
      
      @next_task = { }
    end
  end
  
  # Sets up a callback for the queue that will execute the block if new tasks
  # are added to the queue. If filter_name is specified, this block will be
  # run for tasks matching that filtered subset.
  def observe(filter_name = nil, &block)
    raise BlockRequired unless (block_given?)
    
    @observer_lock.synchronize do
      @observers[filter_name] ||= [ ]

      @observers[filter_name] << block
    end

    task = assign_next_task(filter_name)
  end
  
  # Removes references to the callback function specified. Note that the same
  # Proc must be passed in, as a block with an identical function will not
  # be considered equivalent.
  def remove_observer(filter_name = nil, &block)
    @observer_lock.synchronize do
      set = @observers[filter_name]

      set and set.delete(block)
    end
  end
  
  # Adds a processor to the queue and adds an observer claim method.
  def add_processor(processor, &claim)
    @observer_lock.synchronize do
      @processors << processor
    end

    observe(&claim) if (claim)
  end
  
  # Removes a processor from the queue and removes an observer claim method.
  def remove_processor(processor, &claim)
    @observer_lock.synchronize do
      @processors.delete(processor)
    end

    remove_observer(&claim) if (claim)
  end
  
  # Creates a named filter for the queue using the provided block to select
  # the tasks which should match.
  def filter(filter_name, &block)
    raise BlockRequired unless (block_given?)

    @filter_lock.synchronize do
      @filters[filter_name] = block
    end
    
    assign_next_task(filter_name)
  end
  
  # Adds a task to the queue.
  def <<(task)
    unless (task.is_a?(Pigeon::Task))
      raise "Cannot add task of class #{task.class} to #{self.class}"
    end
    
    # If there is an insert operation already in progress, put this task in
    # the backlog for subsequent processing.
    
    if (@observer_lock.locked?)
      @insert_backlog << task
      return task
    end
    
    active_task = task
    
    while (active_task) do
      # Set the claimable task flag for this task since it is not yet in the
{'fraction_non_alphanumeric': 0.1384256559766764, 'fraction_numerical': 0.0006997084548104956, 'mean_word_length': 3.0225140712945593, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 7, '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': '5902556', 'n_tokens_mistral': 2515, 'n_tokens_neox': 2433, 'n_words': 890}