An accumulator for computing median of a stream

median_accumulator.cpp

#include <queue>
#include <cassert>
#include <vector>
#include <iostream>

template <typename T>
class MedianAccumulator {
public:
  MedianAccumulator();
  void operator()(T val);
  T get() const;
private:
  void rebalance_();
  bool invariant_();
  std::priority_queue<T, std::vector<T>, std::less<T>> lower_;
  std::priority_queue<T, std::vector<T>, std::greater<T>> upper_;
};

template <typename T>
MedianAccumulator<T>::MedianAccumulator()
{
}

template <typename T>
bool MedianAccumulator<T>::invariant_() {
  return (lower_.size() == upper_.size()) || (lower_.size() == upper_.size() + 1);
}

template <typename T>
void MedianAccumulator<T>::operator()(T val)
{
  if ((lower_.size() == 0) || (lower_.top() >= val)) {
    lower_.push(val);
  } else {
    upper_.push(val);
  }
  rebalance_();
}

template <typename T>
T MedianAccumulator<T>::get() const
{
  assert(invariant_());
  if (lower_.size() > upper_.size()) {
    return lower_.top();
  } else {
    return (lower_.top() + upper_.top()) / static_cast<T>(2);
  }
}

template <typename T>
void MedianAccumulator<T>::rebalance_()
{
  if (lower_.size()  > upper_.size() + 1) {
    upper_.push(lower_.top());
    lower_.pop();
  } else if (lower_.size() < upper_.size()) {
    lower_.push(upper_.top());
    upper_.pop();
  }
  assert(invariant_());
}

int main()
{
  std::vector<int> v1{2, 4, 6, 8, 10, 12};
  std::vector<int> medians1{2, 3, 4, 5, 6, 7};

  std::vector<int> v2{6, 2, 4, 10, 8, 12};
  std::vector<int> medians2{6, 4, 4, 5, 6, 7};

  MedianAccumulator<int> ma;
  for (int i = 0; i < v1.size(); ++i) {
    ma(v1[i]);
    std::cout << medians1[i] << ' ' << ma.get() << '\n';
  }
}