<commit_before>class Tableflip::Executor
  def initialize(strategy)
    @strategy = strategy

    @time_format = '%Y-%m-%d %H:%M:%S'
  end

  def log(message)
    puts "[%s] %s" % [ Time.now.strftime(@time_format), message ]
  end

  def await
    @await ||= Hash.new { |h, k| h[k] = [ ] }

    fibers = @await[Fiber.current]

    fibers << Fiber.current

    yield if (block_given?)

    fibers.delete(Fiber.current)

    while (fibers.any?)
      Fiber.yield
    end
  end

  def defer
    parent_fiber = Fiber.current

    fibers = @await[parent_fiber]

    fiber = Fiber.new do
      yield if (block_given?)

      fibers.delete(Fiber.current)

      parent_fiber.resume
    end

    fibers << fiber

    EventMachine.next_tick do
      fiber.resume
    end
  end

  def execute!
    require 'eventmachine'
    require 'em-synchrony'

    if (@strategy.message)
      puts @strategy.message
      exit(0)
    end

    tables = { }

    EventMachine.synchrony do
      timer = EventMachine::PeriodicTimer.new(1) do
        if (@strategy.complete?)
          EventMachine.stop_event_loop
        end
      end

      @strategy.complete = true

      await do
        @strategy.tables.each do |table|
          defer do
            queue = @strategy.actions.dup
            source_db = Tableflip::DatabaseHandle.connect(@strategy.source_env)

            table_config = tables[table] = {
              :table => table,
              :queue => queue
            }
          
            while (action = queue.shift)
              log("#{table} [#{action}]")

              case (action)
              when :tracking_add
                tracking_add(source_db, table_config)
              when :tracking_remove
                tracking_remove(source_db, table_config)
              when :table_migrate
                @strategy.complete = false

                target_db = Tableflip::DatabaseHandle.connect(@strategy.target_env)
                table_migrate(source_db, target_db, table_config)
              when :table_report_status
                target_db = Tableflip::DatabaseHandle.connect(@strategy.target_env)
                table_report_status(source_db, target_db, table_config)
              when :table_count
                table_count(source_db, target_db, table_config)<commit_msg>Failsafe to let report produce output<commit_after>class Tableflip::Executor
  def initialize(strategy)
    @strategy = strategy

    @time_format = '%Y-%m-%d %H:%M:%S'
  end

  def log(message)
    puts "[%s] %s" % [ Time.now.strftime(@time_format), message ]
  end

  def await
    @await ||= Hash.new { |h, k| h[k] = [ ] }

    fibers = @await[Fiber.current]

    fibers << Fiber.current

    yield if (block_given?)

    fibers.delete(Fiber.current)

    while (fibers.any?)
      Fiber.yield
    end
  end

  def defer
    parent_fiber = Fiber.current

    fibers = @await[parent_fiber]

    fiber = Fiber.new do
      yield if (block_given?)

      fibers.delete(Fiber.current)

      parent_fiber.resume
    end

    fibers << fiber

    EventMachine.next_tick do
      fiber.resume
    end
  end

  def execute!
    require 'eventmachine'
    require 'em-synchrony'

    if (@strategy.message)
      puts @strategy.message
      exit(0)
    end

    tables = { }

    EventMachine.synchrony do
      timer = EventMachine::PeriodicTimer.new(1) do
        if (@strategy.complete?)
          EventMachine.stop_event_loop
        end
      end

      @strategy.complete = true

      await do
        @strategy.tables.each do |table|
          defer do
            queue = @strategy.actions.dup
            source_db = Tableflip::DatabaseHandle.connect(@strategy.source_env)

            table_config = tables[table] = {
              :table => table,
              :queue => queue
            }
          
            while (action = queue.shift)
              log("#{table} [#{action}]")

              case (action)
              when :tracking_add
                tracking_add(source_db, table_config)
              when :tracking_remove
                tracking_remove(source_db, table_config)
              when :table_migrate
                @strategy.complete = false

                target_db = Tableflip::DatabaseHandle.connect(@strategy.target_env)
                table_migrate(source_db, target_db, table_config)
              when :table_report_status
                @strategy.complete = false

                target_db = Tableflip::DatabaseHandle.connect(@strategy.target_env)
                table_report_status(source_db, target_db, table_config)

                @strategy.complete = true
              when :table_count
                table_count(source_db, target_db, table_config)
{'fraction_non_alphanumeric': 0.09390269810919906, 'fraction_numerical': 0.0008497981729339282, 'mean_word_length': 1.9351620947630923, 'pattern_counts': {'":': 0, '<': 11, '<?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': '1846897', 'n_tokens_mistral': 1461, 'n_tokens_neox': 1382, 'n_words': 284}