histogram.hpp

/*
 * This file is part of libtcspc
 * Copyright 2019-2026 Board of Regents of the University of Wisconsin System
 * SPDX-License-Identifier: MIT
 */
 
#pragma once
 
#include "arg_wrappers.hpp"
#include "bucket.hpp"
#include "common.hpp"
#include "core.hpp"
#include "data_types.hpp"
#include "errors.hpp"
#include "histogram_events.hpp"
#include "histogram_impl.hpp"
#include "histogram_policy.hpp"
#include "introspect.hpp"
#include "processor.hpp"
 
#include <algorithm>
#include <cstddef>
#include <memory>
#include <span>
#include <stdexcept>
#include <type_traits>
#include <utility>
 
namespace tcspc {
 
namespace internal {
 
template <histogram_policy Policy, typename ResetEvent, typename DataTypes,
          typename Downstream>
class histogram {
    static constexpr histogram_policy overflow_policy =
        Policy & histogram_policy::overflow_mask;
    static constexpr bool emit_concluding =
        (Policy & histogram_policy::emit_concluding_events) !=
        histogram_policy::default_policy;
 
    static_assert(processor<Downstream, histogram_event<DataTypes>>);
    static_assert(std::is_same_v<ResetEvent, never_event> ||
                  handler_for<Downstream, ResetEvent>);
    static_assert(overflow_policy != histogram_policy::saturate_on_overflow ||
                  handler_for<Downstream, warning_event>);
    static_assert(
        (Policy & histogram_policy::emit_concluding_events) ==
            histogram_policy::default_policy ||
        handler_for<Downstream, concluding_histogram_event<DataTypes>>);
 
    using internal_overflow_policy = std::conditional_t<
        overflow_policy == histogram_policy::saturate_on_overflow,
        saturate_on_internal_overflow, stop_on_internal_overflow>;
 
    using bin_index_type = typename DataTypes::bin_index_type;
    using bin_type = typename DataTypes::bin_type;
 
    std::shared_ptr<bucket_source<bin_type>> bsource;
    bucket<bin_type> hist_bucket;
    single_histogram<bin_index_type, bin_type, internal_overflow_policy> shist;
    bool saturate_warning_issued = false;
 
    Downstream downstream;
 
    LIBTCSPC_NOINLINE void start_new_round() {
        hist_bucket = bsource->bucket_of_size(shist.num_bins());
        shist = decltype(shist){hist_bucket, shist};
        shist.clear();
    }
 
    void reset() {
        if (hist_bucket.empty())
            start_new_round();
        if constexpr (emit_concluding) {
            downstream.handle(
                concluding_histogram_event<DataTypes>{std::move(hist_bucket)});
        }
        hist_bucket = {};
        if constexpr (overflow_policy ==
                      histogram_policy::saturate_on_overflow)
            saturate_warning_issued = false;
    }
 
    [[noreturn]] LIBTCSPC_NOINLINE void overflow_error() {
        throw histogram_overflow_error("histogram bin overflowed");
    }
 
    [[noreturn]] LIBTCSPC_NOINLINE void overflow_stop() {
        reset();
        downstream.flush();
        throw end_of_processing("histogram bin overflowed");
    }
 
    LIBTCSPC_NOINLINE void saturated_warning() {
        downstream.handle(warning_event{"histogram bin saturated"});
        saturate_warning_issued = true;
    }
 
    template <typename DT>
    LIBTCSPC_NOINLINE void
    overflow_reset(bin_increment_event<DT> const &event) {
        if (shist.max_per_bin() == 0)
            overflow_error();
        reset();
        // Recurse at most once, because overflow on single increment (when
        // max_per_bin == 0) is error.
        return handle(event);
    }
 
  public:
    explicit histogram(
        arg::num_bins<std::size_t> num_bins,
        arg::max_per_bin<bin_type> max_per_bin,
        std::shared_ptr<bucket_source<bin_type>> buffer_provider,
        Downstream downstream)
        : bsource(std::move(buffer_provider)),
          shist(hist_bucket, max_per_bin, num_bins),
          downstream(std::move(downstream)) {
        if (num_bins.value == 0)
            throw std::invalid_argument("histogram must have at least 1 bin");
        if (max_per_bin.value < 0)
            throw std::invalid_argument(
                "histogram max_per_bin must not be negative");
    }
 
    [[nodiscard]] auto introspect_node() const -> processor_info {
        return processor_info(this, "histogram");
    }
 
    [[nodiscard]] auto introspect_graph() const -> processor_graph {
        return downstream.introspect_graph().push_entry_point(this);
    }
 
    template <typename DT> void handle(bin_increment_event<DT> const &event) {
        static_assert(std::is_same_v<typename DT::bin_index_type,
                                     typename DataTypes::bin_index_type>);
        if (hist_bucket.empty())
            start_new_round();
        if (not shist.apply_increments({&event.bin_index, 1})) {
            if constexpr (overflow_policy ==
                          histogram_policy::error_on_overflow) {
                overflow_error(); // noreturn
            } else if constexpr (overflow_policy ==
                                 histogram_policy::stop_on_overflow) {
                overflow_stop(); // noreturn
            } else if constexpr (overflow_policy ==
                                 histogram_policy::saturate_on_overflow) {
                if (not saturate_warning_issued)
                    saturated_warning();
            } else if constexpr (overflow_policy ==
                                 histogram_policy::reset_on_overflow) {
                return overflow_reset(event);
            }
        }
 
        auto const hist_event =
            histogram_event<DataTypes>{ad_hoc_bucket(std::span(hist_bucket))};
        downstream.handle(hist_event);
    }
 
    // NOLINTNEXTLINE(cppcoreguidelines-rvalue-reference-param-not-moved)
    template <typename DT> void handle(bin_increment_event<DT> &&event) {
        handle(static_cast<bin_increment_event<DT> const &>(event));
    }
 
    template <typename E>
        requires handler_for<Downstream, std::remove_cvref_t<E>>
    void handle(E &&event) {
        if constexpr (std::is_convertible_v<std::remove_cvref_t<E>,
                                            ResetEvent>)
            reset();
        downstream.handle(std::forward<E>(event));
    }
 
    void flush() { downstream.flush(); }
};
 
} // namespace internal

template <histogram_policy Policy = histogram_policy::default_policy,
          typename ResetEvent = never_event,
          typename DataTypes = default_data_types, typename Downstream>
auto histogram(arg::num_bins<std::size_t> num_bins,
               arg::max_per_bin<typename DataTypes::bin_type> max_per_bin,
               std::shared_ptr<bucket_source<typename DataTypes::bin_type>>
                   buffer_provider,
               Downstream downstream) {
    return internal::histogram<Policy, ResetEvent, DataTypes, Downstream>(
        num_bins, max_per_bin, std::move(buffer_provider),
        std::move(downstream));
}
 
} // namespace tcspc