<commit_before>require 'monitor'

module Chore
  class Batcher
    attr_accessor :callback
    attr_accessor :batch

    def initialize(size)
      @size = size
      @batch = []
      @mutex = Monitor.new
      @last_message = nil
      @callback = nil
    end

    def schedule(batch_timeout=20)
      Thread.new(batch_timeout) do |timeout|
        Chore.logger.info "Batching timeout thread starting"
        loop do
          begin 
            Chore.logger.debug "Last message added to batch: #{@last_message}: #{@batch.size}"
            if @last_message && Time.now > (@last_message + timeout)
              Chore.logger.debug "Batching timeout reached (#{@last_message + timeout}), current size: #{@batch.size}"
              self.execute
              @last_message = nil
            end
            sleep(1) 
          rescue => e
            Chore.logger.error "Batcher#schedule raised an exception: #{e.inspect}"
          end
        end
      end
    end

    def add(item)
      @mutex.synchronize do
        @batch << item
        @last_message = Time.now
        if @batch.size >= @size
          execute
        end
      end
    end

    def execute
      @mutex.synchronize do
        @callback.call(@batch)
        @batch.clear
      end
    end
  end

  class ThreadPerConsumerStrategy
    attr_accessor :batcher

    def initialize(fetcher)
      @fetcher = fetcher
      @batcher = Batcher.new(Chore.config.batch_size)
      @batcher.callback = lambda { |batch| @fetcher.manager.assign(batch) }
      @batcher.schedule<commit_msg>Bubble configuration requirements up from the strategy.<commit_after>require 'monitor'

module Chore
  class Batcher
    attr_accessor :callback
    attr_accessor :batch

    def initialize(size)
      @size = size
      @batch = []
      @mutex = Monitor.new
      @last_message = nil
      @callback = nil
    end

    def schedule(batch_timeout=20)
      Thread.new(batch_timeout) do |timeout|
        Chore.logger.info "Batching timeout thread starting"
        loop do
          begin 
            Chore.logger.debug "Last message added to batch: #{@last_message}: #{@batch.size}"
            if @last_message && Time.now > (@last_message + timeout)
              Chore.logger.debug "Batching timeout reached (#{@last_message + timeout}), current size: #{@batch.size}"
              self.execute
              @last_message = nil
            end
            sleep(1) 
          rescue => e
            Chore.logger.error "Batcher#schedule raised an exception: #{e.inspect}"
          end
        end
      end
    end

    def add(item)
      @mutex.synchronize do
        @batch << item
        @last_message = Time.now
        if @batch.size >= @size
          execute
        end
      end
    end

    def execute
      @mutex.synchronize do
        @callback.call(@batch)
        @batch.clear
      end
    end
  end

  class ThreadPerConsumerStrategy
    attr_accessor :batcher

    Chore::CLI.register_option 'batch_size', '--batch-size', 'Number of items to collect for a single worker to process'
    Chore::CLI.register_option 'threads_per_queue', '--threads-per-queue', 'Number of threads to create for each named queue'

    def initialize(fetcher)
      @fetcher = fetcher
      @batcher = Batcher.new(Chore.config.batch_size)
      @batcher.callback = lambda { |batch| @fetcher.manager.assign(batch) }
      @batcher.schedule
{'fraction_non_alphanumeric': 0.09299587992937022, 'fraction_numerical': 0.0017657445556209534, 'mean_word_length': 2.073236889692586, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 9, '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': '136487', 'n_tokens_mistral': 987, 'n_tokens_neox': 965, 'n_words': 257}