<commit_before>require 'thread'

module Sequel
  class Worker < Thread
    class WorkerStopError < RuntimeError; end
  
    attr_reader :queue
    attr_reader :errors
  
    def initialize(db = nil)
      @queue = Queue.new
      @errors = []
      t = self
      t.abort_on_exception = true
      @transaction = !db.nil?
      db ? super {db.transaction {t.work}} : super {t.work}
    end
    
    def work
      loop {next_job}
    rescue WorkerStopError # signals the worker thread to stop
    ensure
      rollback! if @should_rollback
    end
    
    def busy?
      @cur || !@queue.empty?
    end
  
    def async(proc = nil, &block)
      @queue << (proc || block)
      self
    end
    alias_method :add, :async
    alias_method :<<, :async
  
    def join
      while busy?
        sleep 0.1
      end
      self.raise WorkerStopError
      super
    end

    private
    def next_job
      @cur = @queue.pop
      @cur.call
    rescue WorkerStopError => e
      raise e
    rescue Exception => e
      @errors << e
      @should_rollback = true if @transaction
    ensure
      @cur = nil
    end
  end
end<commit_msg>Small improvement to worker code.<commit_after>require 'thread'

module Sequel
  class Worker < Thread
    class WorkerStopError < RuntimeError; end
  
    attr_reader :queue
    attr_reader :errors
  
    def initialize(db = nil)
      @queue = Queue.new
      @errors = []
      t = self
      t.abort_on_exception = true
      @transaction = !db.nil?
      db ? super {db.transaction {t.work}} : super {t.work}
    end
    
    def work
      loop {next_job}
    rescue WorkerStopError # signals the worker thread to stop
    ensure
      rollback! unless @errors.empty?
    end
    
    def busy?
      @cur || !@queue.empty?
    end
  
    def async(proc = nil, &block)
      @queue << (proc || block)
      self
    end
    alias_method :add, :async
    alias_method :<<, :async
  
    def join
      while busy?
        sleep 0.1
      end
      self.raise WorkerStopError
      super
    end

    private
    def next_job
      @cur = @queue.pop
      @cur.call
    rescue WorkerStopError => e
      raise e
    rescue Exception => e
      @errors << e
    ensure
      @cur = nil
    end
  end
end
{'fraction_non_alphanumeric': 0.08318425760286226, 'fraction_numerical': 0.0017889087656529517, 'mean_word_length': 1.8533163265306123, 'pattern_counts': {'":': 0, '<': 19, '<?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': '5799273', 'n_tokens_mistral': 711, 'n_tokens_neox': 714, 'n_words': 213}