<commit_before>module TaskTempest
  class Bookkeeper
    attr_reader :logger
    
    def initialize(options)
      options.each{ |k, v| instance_variable_set("@#{k}", v) }
    end
    
    def report(executions)
      @timer      ||= Time.now
      @executions ||= []
      @executions += executions
      
      if Time.now - @timer > @interval
        logger.info "[STATS] " + make_book.inspect
        @executions.clear
        @timer = Time.now
      end
    end
    
    def make_book
      
      # Do some setup.
      ObjectSpace.garbage_collect
      queue = @queue_factory.call
      book = {}
      
      # Reset memoized objects.
      @memory = nil
      @files = nil
      
      # Task success/error counts.
      book[:tasks] = {}
      book[:tasks][:counts] = task_counts
      book[:tasks][:per_thread] = tasks_per_thread
      book[:tasks][:durations] = task_durations
      book[:tasks][:throughput] = task_throughput
      
      # Thread (worker) info.
      book[:threads] = {}
      book[:threads][:busy] = @storm.busy_workers.length
      book[:threads][:idle] = @storm.size - book[:threads][:busy]
      book[:threads][:saturation] = (book[:threads][:busy] / @storm.size.to_f * 100).round(2)
      
      # Memory, Object, GC info.
      book[:memory] = {}
      book[:memory][:live_objects] = ObjectSpace.live_objects rescue nil
      book[:memory][:resident] = format_memory(get_memory(:resident))
      book[:memory][:virtual] = format_memory(get_memory(:virtual))
      
      # Open file counts.
      book[:files] = {}
      book[:files][:total_count] = get_files(:total)
      book[:files][:tcp_count] = get_files(:tcp)
      
      # Queue info.
      book[:queue] = {}
      book[:queue][:size] = queue.size if queue.respond_to?(:size)
      book[:queue][:backlog] = @storm.executions.inject(0){ |memo, e| memo += 1 unless e.started?; memo }
      
      book
    end
    
    def task_counts
      tot = @executions.length
      err = @executions.sum{ |e| e.exception ? 1 : 0 }
      pct = begin
        if tot > 0
          (err.to_f / tot)
        else
          0.0
        end
      end
      { :tot => tot, :err => err, :pct => pct.round(2) }
    end
    
    def task_throughput
      duration = Time.now - @timer
      per_sec = @executions.length.to_f / duration
      per_min = (per_sec * 60).round(2)
      "#{per_min}/m"
    end
    
    def tasks_per_thread
      counts_by_thread = @storm.threads.inject({}) do |memo, thread|
        memo[thread] = 0
        memo
      end
      @executions.each{ |e| counts_by_thread[e.thread] += 1 }
      counts = counts_by_thread.values<commit_msg>Higher precision on pct stat.<commit_after>module TaskTempest
  class Bookkeeper
    attr_reader :logger
    
    def initialize(options)
      options.each{ |k, v| instance_variable_set("@#{k}", v) }
    end
    
    def report(executions)
      @timer      ||= Time.now
      @executions ||= []
      @executions += executions
      
      if Time.now - @timer > @interval
        logger.info "[STATS] " + make_book.inspect
        @executions.clear
        @timer = Time.now
      end
    end
    
    def make_book
      
      # Do some setup.
      ObjectSpace.garbage_collect
      queue = @queue_factory.call
      book = {}
      
      # Reset memoized objects.
      @memory = nil
      @files = nil
      
      # Task success/error counts.
      book[:tasks] = {}
      book[:tasks][:counts] = task_counts
      book[:tasks][:per_thread] = tasks_per_thread
      book[:tasks][:durations] = task_durations
      book[:tasks][:throughput] = task_throughput
      
      # Thread (worker) info.
      book[:threads] = {}
      book[:threads][:busy] = @storm.busy_workers.length
      book[:threads][:idle] = @storm.size - book[:threads][:busy]
      book[:threads][:saturation] = (book[:threads][:busy] / @storm.size.to_f * 100).round(2)
      
      # Memory, Object, GC info.
      book[:memory] = {}
      book[:memory][:live_objects] = ObjectSpace.live_objects rescue nil
      book[:memory][:resident] = format_memory(get_memory(:resident))
      book[:memory][:virtual] = format_memory(get_memory(:virtual))
      
      # Open file counts.
      book[:files] = {}
      book[:files][:total_count] = get_files(:total)
      book[:files][:tcp_count] = get_files(:tcp)
      
      # Queue info.
      book[:queue] = {}
      book[:queue][:size] = queue.size if queue.respond_to?(:size)
      book[:queue][:backlog] = @storm.executions.inject(0){ |memo, e| memo += 1 unless e.started?; memo }
      
      book
    end
    
    def task_counts
      tot = @executions.length
      err = @executions.sum{ |e| e.exception ? 1 : 0 }
      pct = begin
        if tot > 0
          (err.to_f / tot)
        else
          0.0
        end
      end
      { :tot => tot, :err => err, :pct => pct.round(3) }
    end
    
    def task_throughput
      duration = Time.now - @timer
      per_sec = @executions.length.to_f / duration
      per_min = (per_sec * 60).round(2)
      "#{per_min}/m"
    end
    
    def tasks_per_thread
      counts_by_thread = @storm.threads.inject({}) do |memo, thread|
        memo[thread] = 0
        memo
      end
      @executions.each{ |e| counts_by_thread[e.thread] += 1 }
      counts = counts_by_thread.values
{'fraction_non_alphanumeric': 0.13742911153119092, 'fraction_numerical': 0.006427221172022685, 'mean_word_length': 2.1606929510155317, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 13, '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': '6790495', 'n_tokens_mistral': 1756, 'n_tokens_neox': 1736, 'n_words': 345}