<commit_before>
module Flor

  class UnitExecutor < Flor::Executor

    attr_reader :exid
    attr_reader :consumed

    def initialize(unit, messages)

      @exid = messages.first[:exid]

      super(
        unit,
        unit.loader.load_hooks(@exid),
        unit.storage.fetch_traps(@exid),
        unit.storage.load_execution(@exid))

      @messages =
        messages.collect { |m|
          Flor::Storage
            .from_blob(m[:content])
            .tap { |mm| mm['mid'] = m[:id].to_i }
        }
      @consumed = []

      @alive = true
      @shutdown = false

      @thread = nil
    end

    def alive?; @alive; end

    def run

      @thread = Thread.new { do_run }
#p [ :unit_executor, :thread, @thread.object_id ]

      self
    end

    def shutdown

      @shutdown = true
      @thread.join
    end

    protected

    def do_run

      @unit.logger.log_run_start(self)

      counter_next('runs')

      t0 = Time.now

      (@unit.conf['exe_max_messages'] || 77).times do |i|

        m = @messages.shift
        break unless m
        break if @shutdown

        if m['point'] == 'terminated' && @messages.any?
          #
          # try to handle 'terminated' last
          #
          @messages << m
          m = @messages.shift
        end

        point = m['point']

        ms = process(m)

        @consumed << m

        ims, oms = ms.partition { |m| m['exid'] == @exid }
          # qui est "in", qui est "out"?

        counter_add('omsgs', oms.size)
          # keep track of "out" messages, messages to other executions

        @messages.concat(ims)
        @unit.storage.put_messages(oms)
      end

      @unit.storage.consume(@consumed)

      @alive = false

      @unit.storage.put_execution(@execution)
      @unit.storage.put_messages(@messages)

      du = Time.now - t0
      t0 = Flor.tstamp(t0)

      @unit.logger.log_run_end(self, t0, du)
      @unit.hooker.notify(self, make_end_message(t0, du))

      @consumed.clear

    rescue Exception => exc

# TODO eventually, have a dump dir
      fn =
        [
          'flor',
          @unit.conf['env'], @unit.identifier, @exid,
          'r' + counter('runs').to_s
        ].collect(&:to_s).join('_') + '.dump'<commit_msg>Flag messages as consumed after execution storage<commit_after>
module Flor

  class UnitExecutor < Flor::Executor

    attr_reader :exid
    attr_reader :consumed

    def initialize(unit, messages)

      @exid = messages.first[:exid]

      super(
        unit,
        unit.loader.load_hooks(@exid),
        unit.storage.fetch_traps(@exid),
        unit.storage.load_execution(@exid))

      @messages =
        messages.collect { |m|
          Flor::Storage
            .from_blob(m[:content])
            .tap { |mm| mm['mid'] = m[:id].to_i }
        }
      @consumed = []

      @alive = true
      @shutdown = false

      @thread = nil
    end

    def alive?; @alive; end

    def run

      @thread = Thread.new { do_run }
#p [ :unit_executor, :thread, @thread.object_id ]

      self
    end

    def shutdown

      @shutdown = true
      @thread.join
    end

    protected

    def do_run

      @unit.logger.log_run_start(self)

      counter_next('runs')

      t0 = Time.now

      (@unit.conf['exe_max_messages'] || 77).times do |i|

        m = @messages.shift
        break unless m
        break if @shutdown

        if m['point'] == 'terminated' && @messages.any?
          #
          # try to handle 'terminated' last
          #
          @messages << m
          m = @messages.shift
        end

        point = m['point']

        ms = process(m)

        @consumed << m

        ims, oms = ms.partition { |m| m['exid'] == @exid }
          # qui est "in", qui est "out"?

        counter_add('omsgs', oms.size)
          # keep track of "out" messages, messages to other executions

        @messages.concat(ims)
        @unit.storage.put_messages(oms)
      end

      @alive = false

      @unit.storage.put_execution(@execution)
      @unit.storage.consume(@consumed)

      @unit.storage.put_messages(@messages)

      du = Time.now - t0
      t0 = Flor.tstamp(t0)

      @unit.logger.log_run_end(self, t0, du)
      @unit.hooker.notify(self, make_end_message(t0, du))

      @consumed.clear

    rescue Exception => exc

# TODO eventually, have a dump dir
      fn =
        [
          'flor',
          @unit.conf['env'], @unit.identifier, @exid,
          'r' + counter('runs').to_s
        ].collect(&:to_s).join('_') + '.dump'
{'fraction_non_alphanumeric': 0.12894619831036017, 'fraction_numerical': 0.0035571365051133837, 'mean_word_length': 1.8420720151610865, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 5, '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': '6168620', 'n_tokens_mistral': 1517, 'n_tokens_neox': 1434, 'n_words': 321}