<commit_before>
module Coral
module Util
class Batch < Core
  
  #-----------------------------------------------------------------------------
  # Constructor / Destructor

  def initialize(parallel = true)
    super
    
    set(:parallel, parallel)
    self.clear
    
    yield(self)
  end
  
  #-----------------------------------------------------------------------------
  # Property accessors / modifiers
  
  def parallel
    return get(:parallel, true)
  end
  
  #---
  
  def processes
    return get_hash(:processes)
  end

  #-----------------------------------------------------------------------------
  # Batch operations

  def add(name, options = {}, &code)
    processes[name] = Util::Process.new(name, options, code)
  end
  
  #---
  
  def delete(name)
    processes.delete(name)
    return self
  end
  
  #---
  
  def clear
    set(:processes, {})
    return self
  end
  
  #---
  
  #
  # This run implementation is loosely based on Vagrant's BatchAction 
  # run method which seems to work
  #
  def run
    logger.info("Running batch: Parallel: #{parallel.inspect}")

    errors = exec(get_hash(:processes))

    unless errors.empty?
      raise Errors::BatchError, :message => errors.join("\n\n")
    end
  end
  
  #-----------------------------------------------------------------------------
  # Utilities
  
  def exec(processes)
    threads = []
    
    processes.each do |name, process|
      logger.info("Executing process: #{name}")

      thread = Thread.new do
        Thread.current[:error] = nil

        start_pid = Process.pid

        begin
          process.run
          
        rescue Exception => error
          raise if ! parallel && Process.pid == start_pid
          
          Thread.current[:error] = error
        end

        if Process.pid != start_pid
          exit_status = true
          
          if Thread.current[:error]
            exit_status = false
            error       = Thread.current[:error]
            
            logger.error(error.inspect)
            logger.error(error.message)
            logger.error(error.backtrace.join("\n"))
          end

          Process.exit!(exit_status)
        end
      end

      thread[:process] = name

      thread.join if ! parallel
      threads << thread
    end
    
    return finalize(threads)  
  end
  protected :exec
  
  #---
  
  def finalize(threads)
    errors = []

    threads.each do |thread|
      thread.join
      
      if thread[:error]
        error = thread[:error]
        
        if ! thread[:error].is_a?(Errors::CoralError)
          error   = thread[:error]
          message = error.message
          message += "\n"
          message += "\n#{error.backtrace.join("\n")}"

          errors << I18n.t("coral.core.util.batch.unexpected_error",
                           :process => thread[:process],
                           :message => message)
        else
          errors << I18n.t("coral.core.util.batch.coral_error",
                           :process => thread[:process],
                           :message => thread[:error].message)
        end
      end
    end
    
    return errors  
  end
  protected :finalize
end
end
end
<commit_msg>Fixing Process class namespace issues in the batch utility class.<commit_after>
module Coral
module Util
class Batch < Core
  
  #-----------------------------------------------------------------------------
  # Constructor / Destructor

  def initialize(parallel = true)
    super
    
    set(:parallel, parallel)
    self.clear
    
    yield(self)
  end
  
  #-----------------------------------------------------------------------------
  # Property accessors / modifiers
  
  def parallel
    return get(:parallel, true)
  end
  
  #---
  
  def processes
    return get_hash(:processes)
  end

  #-----------------------------------------------------------------------------
  # Batch operations

  def add(name, options = {}, &code)
    processes[name] = Util::Process.new(name, options) do
      code.call
    end
  end
  
  #---
  
  def delete(name)
    processes.delete(name)
    return self
  end
  
  #---
  
  def clear
    set(:processes, {})
    return self
  end
  
  #---
  
  #
  # This run implementation is loosely based on Vagrant's BatchAction 
  # run method which seems to work
  #
  def run
    logger.info("Running batch: Parallel: #{parallel.inspect}")

    errors = exec(get_hash(:processes))

    unless errors.empty?
      raise Errors::BatchError, :message => errors.join("\n\n")
    end
  end
  
  #-----------------------------------------------------------------------------
  # Utilities
  
  def exec(processes)
    threads = []
    
    processes.each do |name, process|
      logger.info("Executing process: #{name}")

      thread = Thread.new do
        Thread.current[:error] = nil

        start_pid = ::Process.pid

        begin
          process.run
          
        rescue Exception => error
          raise if ! parallel && ::Process.pid == start_pid
          
          Thread.current[:error] = error
        end

        if ::Process.pid != start_pid
          exit_status = true
          
          if Thread.current[:error]
            exit_status = false
            error       = Thread.current[:error]
            
            logger.error(error.inspect)
            logger.error(error.message)
            logger.error(error.backtrace.join("\n"))
          end

          ::Process.exit!(exit_status)
        end
      end

      thread[:process] = name

      thread.join if ! parallel
      threads << thread
    end
    
    return finalize(threads)  
  end
  protected :exec
  
  #---
  
  def finalize(threads)
    errors = []

    threads.each do |thread|
      thread.join
      
      if thread[:error]
        error = thread[:error]
        
        if ! thread[:error].is_a?(Errors::CoralError)
          error   = thread[:error]
          message = error.message
          message += "\n"
          message += "\n#{error.backtrace.join("\n")}"

          errors << I18n.t("coral.core.util.batch.unexpected_error",
                           :process => thread[:process],
                           :message => message)
        else
          errors << I18n.t("coral.core.util.batch.coral_error",
                           :process => thread[:process],
                           :message => thread[:error].message)
        end
      end
    end
    
    return errors  
  end
  protected :finalize
end
end
end

{'fraction_non_alphanumeric': 0.1907478797224364, 'fraction_numerical': 0.001233616037008481, 'mean_word_length': 1.3, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 15, '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': '941974', 'n_tokens_mistral': 1756, 'n_tokens_neox': 1733, 'n_words': 399}