<commit_before>﻿using System;
using System.Collections.Generic;
using Nohros.Collections;
using Nohros.Concurrent;

namespace Nohros.Metrics
{
  /// <summary>
  /// An exponentially-decaying random sample of longs. Uses Cormode et al's
  /// forward-decaying priority resevoir sampling method to produce a
  /// statistically representative sample, exponentially biased towards newer
  /// entries.
  /// </summary>
  /// <remarks>
  /// Cormode et al. Forward Decay: A Practical Time Decay Model for
  /// Streaming Systems. ICDE '09: Proceedings of the 2009 IEEE International
  /// Conference on Data Engineering (2009).
  /// http://www.research.att.com/people/Cormode_Graham/library/publications/CormodeShkapenyukSrivastavaXu09.pdf
  /// </remarks>
  public class ExponentiallyDecayingResevoir : IResevoir
  {
    const long kRescaleThreshold = 36000000000; // 1 hour ticks
    const int kDefaultSampleSize = 1028;
    const double kDefaultAlpha = 0.015;

    readonly double alpha_;
    readonly Clock clock_;
    readonly AndersonTree<double, long> priorities_;
    readonly Random rand_;
    //readonly long[] resevoir_;
    readonly int resevoir_size_;
    int count_;
    long next_scale_time_;
    long start_time_;

    /// <summary>
    /// Initializes a new instance of the
    /// <see cref="ExponentiallyDecayingResevoir"/> of 1028 elements, which
    /// offers 99.9% confidence level with a 5% margin of error assuming a
    /// normal distribution, and an alpha factor of 0.015, which heavily
    /// biases the resevoir to the past 5 minutes of measurements.
    /// </summary>
    public ExponentiallyDecayingResevoir()
      : this(kDefaultSampleSize, kDefaultAlpha) {
    }

    /// <summary>
    /// Initializes a new instance of the
    /// <see cref="ExponentiallyDecayingResevoir"/> class by using the specified
    /// resevoir size and exponential decay factor.
    /// </summary>
    /// <param name="resevoir_size">
    /// The number of samples to keep in the sampling resevoir.
    /// </param>
    /// <param name="alpha">
    /// The exponential decay factor; the higher this is, the more biased the
    /// sample will be towards newer values.
    /// </param>
    /// <remarks>
    /// The use of the executor returned by the method
    /// <see cref="Executors.SameThreadExecutor"/> is not encouraged, because
    /// the executor does not returns until the execution list is empty and,
    /// this can cause significant pauses in the thread that is executing the
    /// sample update.
    /// </remarks>
    public ExponentiallyDecayingResevoir(int resevoir_size, double alpha)
      : this(resevoir_size, alpha, new StopwatchClock()) {
    }

    /// <summary>
    /// Initializes a new instance of the
    /// <see cref="ExponentiallyDecayingResevoir"/> class by using the specified
    /// resevoir size and exponential decay factor.
    /// </summary>
    /// <param name="resevoir_size">
    /// The number of samples to keep in the sampling resevoir.
    /// </param>
    /// <param name="alpha">
    /// The exponential decay factor; the higher this is, the more biased the
    /// sample will be towards newer values.
    /// </param>
    /// <param name="clock">
    /// A <see cref="Clock"/> that can be used to mark the passage of time.
    /// </param>
    /// <remarks>
    /// The use of the executor returned by the method
    /// <see cref="Executors.SameThreadExecutor"/> is not encouraged, because
    /// the executor does not returns until the execution list is empty and,
    /// this can cause significant pauses in the thread that is executing the
    /// sample update.
    /// </remarks>
    internal ExponentiallyDecayingResevoir(int resevoir_size, double alpha,
      Clock clock) {
      count_ = 0;
      clock_ = clock;
      rand_ = new Random();
      next_scale_time_ = 0;
      alpha_ = alpha;
      resevoir_size_ = resevoir_size;
      priorities_ = new AndersonTree<double, long>();
      //resevoir_ = new long[resevoir_size];
      start_time_ = TimeInSeconds(clock_.Tick);
      next_scale_time_ = clock_.Tick + kRescaleThreshold;
    }

    /// <inheritdoc/>
    public void Update(long value) {
      Update(value, clock_.Tick);
    }

    public Snapshot Snapshot {
      get {
        int size = Size;
        var resevoir = new List<long>(size);
        priorities_.InOrderTreeWalk(node => {
          resevoir.Add(node.Value);
          return true;
        });
        return new Snapshot(resevoir);
      }
    }

    /// <inheritdoc/>
    public int Size {
      get { return Math.Min(resevoir_size_, count_); }
    }

    /// <summary>
    /// Adds an old value with fixed timestamp to the sample.
    /// </summary>
    /// <param name="value">The value to be added.</param>
    /// <param name="timestamp">The epoch timestamp of <paramref name="value"/>
    /// in seconds.</param>
    public void Update(long value, long timestamp) {
      RescaleIfNeeded();

      double priority = Priority(timestamp);

      // Fills the resevoir with the first "m" values and keep elements
      // with the greatest priorities in the resvoir.
      if (++count_ <= resevoir_size_) {
        priorities_[priority] = value;
        //resevoir_[count_ - 1] = value;
      } else {
        KeyValuePair<double, long> first = priorities_.First;
        if (first.Key < priority) {
          // replace the element associated with the smallest key by the
          // sampled value.
          priorities_.Remove(first.Key);
          priorities_[priority] = first.Value;
          //resevoir_[first.Value] = value;
        }
      }
    }

    /// <inheritdoc/>
    public long Timestamp {<commit_msg>Fix a bug in the ExponentiallyDecayingResevoir that causes the value to be incorrectly registered.<commit_after>﻿using System;
using System.Collections.Generic;
using Nohros.Collections;
using Nohros.Concurrent;

namespace Nohros.Metrics
{
  /// <summary>
  /// An exponentially-decaying random sample of longs. Uses Cormode et al's
  /// forward-decaying priority resevoir sampling method to produce a
  /// statistically representative sample, exponentially biased towards newer
  /// entries.
  /// </summary>
  /// <remarks>
  /// Cormode et al. Forward Decay: A Practical Time Decay Model for
  /// Streaming Systems. ICDE '09: Proceedings of the 2009 IEEE International
  /// Conference on Data Engineering (2009).
  /// http://www.research.att.com/people/Cormode_Graham/library/publications/CormodeShkapenyukSrivastavaXu09.pdf
  /// </remarks>
  public class ExponentiallyDecayingResevoir : IResevoir
  {
    const long kRescaleThreshold = 36000000000; // 1 hour ticks
    const int kDefaultSampleSize = 1028;
    const double kDefaultAlpha = 0.015;

    readonly double alpha_;
    readonly Clock clock_;
    readonly AndersonTree<double, long> priorities_;
    readonly Random rand_;
    //readonly long[] resevoir_;
    readonly int resevoir_size_;
    int count_;
    long next_scale_time_;
    long start_time_;

    /// <summary>
    /// Initializes a new instance of the
    /// <see cref="ExponentiallyDecayingResevoir"/> of 1028 elements, which
    /// offers 99.9% confidence level with a 5% margin of error assuming a
    /// normal distribution, and an alpha factor of 0.015, which heavily
    /// biases the resevoir to the past 5 minutes of measurements.
    /// </summary>
    public ExponentiallyDecayingResevoir()
      : this(kDefaultSampleSize, kDefaultAlpha) {
    }

    /// <summary>
    /// Initializes a new instance of the
    /// <see cref="ExponentiallyDecayingResevoir"/> class by using the specified
    /// resevoir size and exponential decay factor.
    /// </summary>
    /// <param name="resevoir_size">
    /// The number of samples to keep in the sampling resevoir.
    /// </param>
    /// <param name="alpha">
    /// The exponential decay factor; the higher this is, the more biased the
    /// sample will be towards newer values.
    /// </param>
    /// <remarks>
    /// The use of the executor returned by the method
    /// <see cref="Executors.SameThreadExecutor"/> is not encouraged, because
    /// the executor does not returns until the execution list is empty and,
    /// this can cause significant pauses in the thread that is executing the
    /// sample update.
    /// </remarks>
    public ExponentiallyDecayingResevoir(int resevoir_size, double alpha)
      : this(resevoir_size, alpha, new StopwatchClock()) {
    }

    /// <summary>
    /// Initializes a new instance of the
    /// <see cref="ExponentiallyDecayingResevoir"/> class by using the specified
    /// resevoir size and exponential decay factor.
    /// </summary>
    /// <param name="resevoir_size">
    /// The number of samples to keep in the sampling resevoir.
    /// </param>
    /// <param name="alpha">
    /// The exponential decay factor; the higher this is, the more biased the
    /// sample will be towards newer values.
    /// </param>
    /// <param name="clock">
    /// A <see cref="Clock"/> that can be used to mark the passage of time.
    /// </param>
    /// <remarks>
    /// The use of the executor returned by the method
    /// <see cref="Executors.SameThreadExecutor"/> is not encouraged, because
    /// the executor does not returns until the execution list is empty and,
    /// this can cause significant pauses in the thread that is executing the
    /// sample update.
    /// </remarks>
    internal ExponentiallyDecayingResevoir(int resevoir_size, double alpha,
      Clock clock) {
      count_ = 0;
      clock_ = clock;
      rand_ = new Random();
      next_scale_time_ = 0;
      alpha_ = alpha;
      resevoir_size_ = resevoir_size;
      priorities_ = new AndersonTree<double, long>();
      //resevoir_ = new long[resevoir_size];
      start_time_ = TimeInSeconds(clock_.Tick);
      next_scale_time_ = clock_.Tick + kRescaleThreshold;
    }

    /// <inheritdoc/>
    public void Update(long value) {
      Update(value, clock_.Tick);
    }

    public Snapshot Snapshot {
      get {
        int size = Size;
        var resevoir = new List<long>(size);
        priorities_.InOrderTreeWalk(node => {
          resevoir.Add(node.Value);
          return true;
        });
        return new Snapshot(resevoir);
      }
    }

    /// <inheritdoc/>
    public int Size {
      get { return Math.Min(resevoir_size_, count_); }
    }

    /// <summary>
    /// Adds an old value with fixed timestamp to the sample.
    /// </summary>
    /// <param name="value">The value to be added.</param>
    /// <param name="timestamp">The epoch timestamp of <paramref name="value"/>
    /// in seconds.</param>
    public void Update(long value, long timestamp) {
      RescaleIfNeeded();

      double priority = Priority(timestamp);

      // Fills the resevoir with the first "m" values and keep elements
      // with the greatest priorities in the resvoir.
      if (++count_ <= resevoir_size_) {
        priorities_[priority] = value;
        //resevoir_[count_ - 1] = value;
      } else {
        KeyValuePair<double, long> first = priorities_.First;
        if (first.Key < priority) {
          // replace the element associated with the smallest key by the
          // sampled value.
          priorities_.Remove(first.Key);
          priorities_[priority] = value;
          //resevoir_[first.Value] = value;
        }
      }
    }

    /// <inheritdoc/>
    public long Timestamp {
{'fraction_non_alphanumeric': 0.11106252732837779, 'fraction_numerical': 0.00839527765631832, 'mean_word_length': 2.457073760580411, 'pattern_counts': {'":': 0, '<': 95, '<?xml version=': 0, '>': 93, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, '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': '6649665', 'n_tokens_mistral': 3315, 'n_tokens_neox': 3114, 'n_words': 1091}