scan_histograms.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 scan_histograms {
    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 constexpr bool reset_after_scan =
        (Policy & histogram_policy::reset_after_scan) !=
        histogram_policy::default_policy;
    static constexpr bool clear_every_scan =
        (Policy & histogram_policy::clear_every_scan) !=
        histogram_policy::default_policy;
    static constexpr bool clear_new_bucket =
        (Policy & histogram_policy::no_clear_new_bucket) ==
        histogram_policy::default_policy;
 
    static_assert(
        processor<Downstream, histogram_array_progress_event<DataTypes>,
                  histogram_array_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_array_event<DataTypes>>);
 
    // There is no way to roll back a partial scan in saturate mode, so
    // concluding events cannot be emitted.
    static_assert(overflow_policy != histogram_policy::saturate_on_overflow ||
                  (Policy & histogram_policy::emit_concluding_events) ==
                      histogram_policy::default_policy);
 
    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;
 
    using journal_type =
        std::conditional_t<emit_concluding ||
                               overflow_policy ==
                                   histogram_policy::reset_on_overflow,
                           bin_increment_cluster_journal<bin_index_type>,
                           null_journal<bin_index_type>>;
 
    std::shared_ptr<bucket_source<bin_type>> bsource;
    bucket<bin_type> hist_bucket;
    multi_histogram_accumulation<bin_index_type, bin_type,
                                 internal_overflow_policy>
        mhista;
    bool saturate_warning_issued = false;
    journal_type journal;
 
    Downstream downstream;
 
    LIBTCSPC_NOINLINE void start_new_round() {
        auto const size = mhista.num_elements() * mhista.num_bins();
        hist_bucket = bsource->bucket_of_size(size);
        if constexpr (clear_new_bucket)
            std::fill(hist_bucket.begin(), hist_bucket.end(), bin_type{0});
        mhista = decltype(mhista)(hist_bucket, mhista, not clear_new_bucket);
    }
 
    void reset_without_replay() {
        if (hist_bucket.empty())
            start_new_round();
        if constexpr (emit_concluding) {
            mhista.roll_back_current_scan(journal);
            downstream.handle(concluding_histogram_array_event<DataTypes>{
                std::move(hist_bucket)});
        }
        hist_bucket = {};
        journal.clear();
        if constexpr (overflow_policy ==
                      histogram_policy::saturate_on_overflow)
            saturate_warning_issued = false;
    }
 
    void end_of_scan() {
        mhista.new_scan(journal, clear_every_scan);
        auto const array_event = histogram_array_event<DataTypes>{
            ad_hoc_bucket(std::span(hist_bucket))};
        downstream.handle(array_event);
        if constexpr (reset_after_scan)
            reset_without_replay();
    }
 
    [[noreturn]] LIBTCSPC_NOINLINE void overflow_error() {
        throw histogram_overflow_error("histogram array bin overflowed");
    }
 
    [[noreturn]] LIBTCSPC_NOINLINE void overflow_stop() {
        if constexpr (emit_concluding) {
            mhista.roll_back_current_scan(journal);
            downstream.handle(concluding_histogram_array_event<DataTypes>{
                std::move(hist_bucket)});
        }
        downstream.flush();
        throw end_of_processing("histogram array bin overflowed");
    }
 
    LIBTCSPC_NOINLINE void saturated_warning() {
        downstream.handle(warning_event{"histogram array bin saturated"});
        saturate_warning_issued = true;
    }
 
    template <typename DT>
    LIBTCSPC_NOINLINE void
    overflow_reset(bin_increment_cluster_event<DT> const &event) {
        if (mhista.scan_index() == 0)
            throw histogram_overflow_error(
                "histogram array bin overflowed on first scan");
        mhista.roll_back_current_scan(journal);
        if constexpr (emit_concluding)
            downstream.handle(concluding_histogram_array_event<DataTypes>{
                std::move(hist_bucket)});
        start_new_round();
        mhista.replay(journal);
        // Recurse at most once, because overflow on first scan will be error.
        return handle(event);
    }
 
  public:
    explicit scan_histograms(
        arg::num_elements<std::size_t> num_elements,
        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)
        : bsource(std::move(buffer_provider)),
          mhista(hist_bucket, max_per_bin, num_bins, num_elements, true),
          downstream(std::move(downstream)) {
        if (num_elements.value == 0)
            throw std::invalid_argument(
                "scan_histograms must have at least 1 element");
        if (num_bins.value == 0)
            throw std::invalid_argument(
                "scan_histograms must have at least 1 bin per element");
        if (max_per_bin.value < 0)
            throw std::invalid_argument(
                "scan_histograms max_per_bin must not be negative");
    }
 
    [[nodiscard]] auto introspect_node() const -> processor_info {
        return processor_info(this, "scan_histograms");
    }
 
    [[nodiscard]] auto introspect_graph() const -> processor_graph {
        return downstream.introspect_graph().push_entry_point(this);
    }
 
    template <typename DT>
    void handle(bin_increment_cluster_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();
        auto const element_index = mhista.next_element_index();
        if (not mhista.apply_increment_cluster(event.bin_indices, journal)) {
            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 progress = histogram_array_progress_event<DataTypes>{
            (element_index + 1) * mhista.num_bins(),
            ad_hoc_bucket(std::span(hist_bucket))};
        downstream.handle(progress);
 
        if (mhista.is_scan_complete())
            end_of_scan();
    }
 
    // NOLINTBEGIN(cppcoreguidelines-rvalue-reference-param-not-moved)
    template <typename DT>
    void handle(bin_increment_cluster_event<DT> &&event) {
        handle(static_cast<bin_increment_cluster_event<DT> const &>(event));
    }
    // NOLINTEND(cppcoreguidelines-rvalue-reference-param-not-moved)
 
    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_without_replay();
        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 scan_histograms(
    arg::num_elements<std::size_t> num_elements,
    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::scan_histograms<Policy, ResetEvent, DataTypes,
                                     Downstream>(
        num_elements, num_bins, max_per_bin, std::move(buffer_provider),
        std::move(downstream));
}
 
} // namespace tcspc