P² algorithm for streaming quantile estimation. More...

#include <stream_aggregator.h>

Collaboration diagram for kcenon::monitoring::quantile_estimator:

Public Member Functions
	quantile_estimator (double p)
	Constructor.

	quantile_estimator (quantile_estimator &&other) noexcept

quantile_estimator &	operator= (quantile_estimator &&other) noexcept

	quantile_estimator (const quantile_estimator &)=delete

quantile_estimator &	operator= (const quantile_estimator &)=delete

void	add_observation (double x)
	Add an observation.

double	get_quantile () const
	Get the estimated quantile.

size_t	count () const
	Get observation count.

void	reset ()
	Reset the estimator.

Private Member Functions
void	init_markers ()

double	parabolic (int i, int sign) const

double	linear (int i, int sign) const

Private Attributes
std::shared_mutex	mutex_

double	p_

size_t	count_ = 0

double	q_ [5] = {0}

int	n_ [5] = {0}

double	n_prime_ [5] = {0}

double	dn_ [5] = {0}

Detailed Description

P² algorithm for streaming quantile estimation.

Implements the P² algorithm for estimating quantiles without storing all observations. Uses piecewise-parabolic interpolation.

Definition at line 193 of file stream_aggregator.h.

Constructor & Destructor Documentation

◆ quantile_estimator() [1/3]

kcenon::monitoring::quantile_estimator::quantile_estimator ( double p )

inlineexplicit

Constructor.

Parameters

p	The quantile to estimate (0.0 to 1.0)

Definition at line 199 of file stream_aggregator.h.

                                          : p_(p) {
        init_markers();
    }

References init_markers().

Here is the call graph for this function:

◆ quantile_estimator() [2/3]

kcenon::monitoring::quantile_estimator::quantile_estimator ( quantile_estimator && other )

inlinenoexcept

Definition at line 204 of file stream_aggregator.h.

        : p_(other.p_)
        , count_(other.count_) {
        std::copy(std::begin(other.q_), std::end(other.q_), std::begin(q_));
        std::copy(std::begin(other.n_), std::end(other.n_), std::begin(n_));
        std::copy(std::begin(other.n_prime_), std::end(other.n_prime_), std::begin(n_prime_));
        std::copy(std::begin(other.dn_), std::end(other.dn_), std::begin(dn_));
    }

References dn_, n_, n_prime_, kcenon::monitoring::other, and q_.

◆ quantile_estimator() [3/3]

kcenon::monitoring::quantile_estimator::quantile_estimator ( const quantile_estimator & )

delete

Member Function Documentation

◆ add_observation()

void kcenon::monitoring::quantile_estimator::add_observation ( double x )

inline

Add an observation.

Parameters

x	The observation value

Definition at line 259 of file stream_aggregator.h.

                                   {
        std::unique_lock<std::shared_mutex> lock(mutex_);
        count_++;
 
        if (count_ <= 5) {
            q_[count_ - 1] = x;
            if (count_ == 5) {
                std::sort(q_, q_ + 5);
            }
            return;
        }
 
        // Find cell k
        int k;
        if (x < q_[0]) {
            q_[0] = x;
            k = 0;
        } else if (x >= q_[4]) {
            q_[4] = x;
            k = 3;
        } else {
            for (k = 0; k < 4; ++k) {
                if (x < q_[k + 1]) {
                    break;
                }
            }
        }
 
        // Increment positions
        for (int i = k + 1; i < 5; ++i) {
            n_[i]++;
        }
 
        // Update desired positions
        for (int i = 0; i < 5; ++i) {
            n_prime_[i] += dn_[i];
        }
 
        // Adjust marker heights
        for (int i = 1; i < 4; ++i) {
            double d = n_prime_[i] - n_[i];
            if ((d >= 1 && n_[i + 1] - n_[i] > 1) ||
                (d <= -1 && n_[i - 1] - n_[i] < -1)) {
                int sign = (d >= 0) ? 1 : -1;
                double q_new = parabolic(i, sign);
 
                if (q_[i - 1] < q_new && q_new < q_[i + 1]) {
                    q_[i] = q_new;
                } else {
                    q_[i] = linear(i, sign);
                }
                n_[i] += sign;
            }
        }
    }

References count_, dn_, linear(), mutex_, n_, n_prime_, parabolic(), and q_.

Referenced by TEST_F(), and TEST_F().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ count()

size_t kcenon::monitoring::quantile_estimator::count ( ) const

inline

Get observation count.

Definition at line 337 of file stream_aggregator.h.

                         {
        std::shared_lock<std::shared_mutex> lock(mutex_);
        return count_;
    }

References count_, and mutex_.

◆ get_quantile()

double kcenon::monitoring::quantile_estimator::get_quantile ( ) const

inline

Get the estimated quantile.

Returns: The estimated quantile value

Definition at line 319 of file stream_aggregator.h.

                                {
        std::shared_lock<std::shared_mutex> lock(mutex_);
        if (count_ < 5) {
            if (count_ == 0) {
                return 0.0;
            }
            // For small samples, use simple interpolation
            std::vector<double> sorted(q_, q_ + count_);
            std::sort(sorted.begin(), sorted.end());
            size_t idx = static_cast<size_t>(p_ * (count_ - 1));
            return sorted[idx];
        }
        return q_[2];  // Middle marker for p-quantile
    }

References count_, mutex_, p_, and q_.

Referenced by TEST_F(), and TEST_F().

Here is the caller graph for this function:

◆ init_markers()

void kcenon::monitoring::quantile_estimator::init_markers ( )

inlineprivate

Definition at line 230 of file stream_aggregator.h.

                        {
        // Initialize marker positions
        n_[0] = 1;
        n_[1] = 2;
        n_[2] = 3;
        n_[3] = 4;
        n_[4] = 5;
 
        // Initialize desired marker positions
        n_prime_[0] = 1;
        n_prime_[1] = 1 + 2 * p_;
        n_prime_[2] = 1 + 4 * p_;
        n_prime_[3] = 3 + 2 * p_;
        n_prime_[4] = 5;
 
        // Increments
        dn_[0] = 0;
        dn_[1] = p_ / 2;
        dn_[2] = p_;
        dn_[3] = (1 + p_) / 2;
        dn_[4] = 1;
    }

References dn_, n_, n_prime_, and p_.

Referenced by quantile_estimator().

Here is the caller graph for this function:

◆ linear()

double kcenon::monitoring::quantile_estimator::linear	(	int	i,
		int	sign ) const

inlineprivate

Definition at line 372 of file stream_aggregator.h.

                                         {
        int idx = (sign < 0) ? i - 1 : i + 1;
        return q_[i] + static_cast<double>(sign) * (q_[idx] - q_[i]) /
               (n_[idx] - n_[i]);
    }

References n_, and q_.

Referenced by add_observation().

Here is the caller graph for this function:

◆ operator=() [1/2]

quantile_estimator & kcenon::monitoring::quantile_estimator::operator= ( const quantile_estimator & )

delete

◆ operator=() [2/2]

quantile_estimator & kcenon::monitoring::quantile_estimator::operator= ( quantile_estimator && other )

inlinenoexcept

Definition at line 213 of file stream_aggregator.h.

                                                                       {
        if (this != &other) {
            p_ = other.p_;
            count_ = other.count_;
            std::copy(std::begin(other.q_), std::end(other.q_), std::begin(q_));
            std::copy(std::begin(other.n_), std::end(other.n_), std::begin(n_));
            std::copy(std::begin(other.n_prime_), std::end(other.n_prime_), std::begin(n_prime_));
            std::copy(std::begin(other.dn_), std::end(other.dn_), std::begin(dn_));
        }
        return *this;
    }

References count_, dn_, n_, n_prime_, kcenon::monitoring::other, p_, and q_.

◆ parabolic()

double kcenon::monitoring::quantile_estimator::parabolic	(	int	i,
		int	sign ) const

inlineprivate

Definition at line 357 of file stream_aggregator.h.

                                            {
        double qi = q_[i];
        double qim1 = q_[i - 1];
        double qip1 = q_[i + 1];
        int ni = n_[i];
        int nim1 = n_[i - 1];
        int nip1 = n_[i + 1];
 
        double term1 = static_cast<double>(sign) / (nip1 - nim1);
        double term2 = (ni - nim1 + sign) * (qip1 - qi) / (nip1 - ni);
        double term3 = (nip1 - ni - sign) * (qi - qim1) / (ni - nim1);
 
        return qi + term1 * (term2 + term3);
    }

References n_, and q_.

Referenced by add_observation().

Here is the caller graph for this function:

◆ reset()

void kcenon::monitoring::quantile_estimator::reset ( )

inline

Reset the estimator.

Definition at line 345 of file stream_aggregator.h.

                 {
        std::unique_lock<std::shared_mutex> lock(mutex_);
        count_ = 0;
        n_[0] = 1; n_[1] = 2; n_[2] = 3; n_[3] = 4; n_[4] = 5;
        n_prime_[0] = 1;
        n_prime_[1] = 1 + 2 * p_;
        n_prime_[2] = 1 + 4 * p_;
        n_prime_[3] = 3 + 2 * p_;
        n_prime_[4] = 5;
    }