<commit_before>require 'atomic'
require 'red_black_tree'
require 'metriks/snapshot'

module Metriks
  class ExponentiallyDecayingSample
    RESCALE_THRESHOLD = 60 * 60 # 1 hour

    def initialize(reservoir_size, alpha, values = nil)
      @values = values || RedBlackTree.new
      @count = Atomic.new(0)
      @next_scale_time = Atomic.new(0)
      @alpha = alpha
      @reservoir_size = reservoir_size
      @mutex = Mutex.new
      clear
    end

    def clear
      @mutex.synchronize do
        @values.clear
        @count.value = 0
        @next_scale_time.value = Time.now + RESCALE_THRESHOLD
        @start_time = Time.now
      end
    end

    def size
      count = @count.value
      count < @reservoir_size ? count : @reservoir_size
    end

    def snapshot
      @mutex.synchronize do
        Snapshot.new(@values.values)
      end
    end

    def update(value, timestamp = Time.now)
      @mutex.synchronize do
        priority = weight(timestamp - @start_time) / rand
        new_count = @count.update { |v| v + 1 }

        if priority.nan?
          warn "ExponentiallyDecayingSample found priority of NaN. timestamp: #{timestamp.to_f} start_time: #{@start_time.to_f}"
          return
        end

        if new_count <= @reservoir_size
          @values[priority] = value
        else
          first_priority = @values.first[0]
          if first_priority < priority
            unless @values[priority]
              @values[priority] = value

              until @values.delete(first_priority)
                first_priority = @values.first[0]
              end
            end<commit_msg>Fix exception raised when fetching an empty snapshot<commit_after>require 'atomic'
require 'red_black_tree'
require 'metriks/snapshot'

module Metriks
  class ExponentiallyDecayingSample
    RESCALE_THRESHOLD = 60 * 60 # 1 hour

    def initialize(reservoir_size, alpha, values = nil)
      @values = values || RedBlackTree.new
      @count = Atomic.new(0)
      @next_scale_time = Atomic.new(0)
      @alpha = alpha
      @reservoir_size = reservoir_size
      @mutex = Mutex.new
      clear
    end

    def clear
      @mutex.synchronize do
        @values.clear
        @count.value = 0
        @next_scale_time.value = Time.now + RESCALE_THRESHOLD
        @start_time = Time.now
      end
    end

    def size
      count = @count.value
      count < @reservoir_size ? count : @reservoir_size
    end

    def snapshot
      @mutex.synchronize do
        # Work around a bug in avl_tree right now -- see Issue #6 for details
        if @values.empty?
          Snapshot.new([])
        else
          Snapshot.new(@values.values)
        end
      end
    end

    def update(value, timestamp = Time.now)
      @mutex.synchronize do
        priority = weight(timestamp - @start_time) / rand
        new_count = @count.update { |v| v + 1 }

        if priority.nan?
          warn "ExponentiallyDecayingSample found priority of NaN. timestamp: #{timestamp.to_f} start_time: #{@start_time.to_f}"
          return
        end

        if new_count <= @reservoir_size
          @values[priority] = value
        else
          first_priority = @values.first[0]
          if first_priority < priority
            unless @values[priority]
              @values[priority] = value

              until @values.delete(first_priority)
                first_priority = @values.first[0]
              end
            end
{'fraction_non_alphanumeric': 0.08277470125327893, 'fraction_numerical': 0.006703584960652871, 'mean_word_length': 2.0264550264550265, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 3, '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': '6603862', 'n_tokens_mistral': 1056, 'n_tokens_neox': 1010, 'n_words': 254}