buffer.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 "common.hpp"
#include "context.hpp"
#include "errors.hpp"
#include "introspect.hpp"
#include "processor.hpp"
#include "vector_queue.hpp"
 
#include <algorithm>
#include <chrono>
#include <concepts>
#include <condition_variable>
#include <cstddef>
#include <functional>
#include <mutex>
#include <stdexcept>
#include <type_traits>
#include <utility>
 
namespace tcspc {
 
namespace internal {
 
// Avoid std::hardware_destructive_interference_size, because it suffers from
// ABI compatibility requirements and therefore may not have the best value
// (for example, it seems to be 256 on Linux/aarch64). Instead, just default to
// 64 (correct for most x86_64 and many ARM and RISC-V processors) except in
// known cases where a larger value is appropriate.
#if (defined(__APPLE__) && defined(__arm64__)) ||                             \
    (defined(__powerpc64__) || defined(__ppc64__) || defined(__PPC64__))
inline constexpr std::size_t destructive_interference_size = 128;
#else
inline constexpr std::size_t destructive_interference_size = 64;
#endif
 
} // namespace internal

class buffer_accessor {
    std::function<void()> halt_fn;
    std::function<void()> pump_fn;
 
  public:
    template <typename HaltFunc, typename PumpFunc>
    explicit buffer_accessor(HaltFunc halt_func, PumpFunc pump_func)
        : halt_fn(halt_func), pump_fn(pump_func) {}

    void halt() noexcept { halt_fn(); } // NOLINT(bugprone-exception-escape)

    void pump() { pump_fn(); }
};
 
namespace internal {
 
template <typename Event, bool LatencyLimited, typename Downstream>
    requires processor<Downstream, Event>
class buffer {
    static_assert(std::move_constructible<Event>,
                  "buffer requires Event to be move-constructible (events are "
                  "stored in an internal queue)");
 
    using clock_type = std::chrono::steady_clock;
    using queue_type = vector_queue<Event>;
 
    std::size_t threshold;
    clock_type::duration max_latency;
 
    std::mutex mutex;
    std::condition_variable has_data_condition;
    queue_type shared_queue;
    clock_type::time_point oldest_enqueued_time;
    bool upstream_flushed = false;
    bool upstream_halted = false;
    bool downstream_threw = false;
 
#ifdef _MSC_VER
#pragma warning(push)
#pragma warning(disable : 4324) // Structure padded due to alignment specifier
#endif
 
    // To reduce lock contention on the shared_queue, we use a second queue
    // that is accessed only by the emitting thread and is not protected by the
    // mutex. Events in the shared_queue are transferred in bulk to the
    // emit_queue while the mutex is held.
    // This means that the mutex does not need to be acquired between every
    // event emitted, so the producer will be less likely to block when the
    // data rate is momentarily high, and the consumer will be less likely to
    // block while catching up on buffered events.
    // Furthermore, we ensure that the emit_queue and downstream do not share a
    // CPU cache line with the shared_queue, to prevent false sharing.
    alignas(destructive_interference_size) queue_type emit_queue;
 
#ifdef _MSC_VER
#pragma warning(pop)
#endif
 
    Downstream downstream;
 
    // Cold data after downstream.
    bool pumped = false;
    access_tracker<buffer_accessor> trk;
 
    void halt() noexcept {
        {
            auto const lock = std::lock_guard(mutex);
            upstream_halted = true;
        }
        has_data_condition.notify_one();
    }
 
    void pump() {
        try {
            auto lock = std::unique_lock(mutex);
            if (pumped) {
                throw std::logic_error(
                    "buffer may not be pumped a second time");
            }
            pumped = true;
 
            for (;;) {
                if constexpr (LatencyLimited) {
                    has_data_condition.wait(lock, [&] {
                        return not shared_queue.empty() || upstream_flushed ||
                               upstream_halted;
                    });
                    // Won't overflow due to 24 h limit on max_latency:
                    auto const deadline = oldest_enqueued_time + max_latency;
                    has_data_condition.wait_until(lock, deadline, [&] {
                        return shared_queue.size() >= threshold ||
                               upstream_flushed || upstream_halted;
                    });
                } else {
                    has_data_condition.wait(lock, [&] {
                        return shared_queue.size() >= threshold ||
                               upstream_flushed || upstream_halted;
                    });
                }
 
                if (not upstream_flushed && upstream_halted)
                    throw source_halted();
                if (shared_queue.empty() && upstream_flushed) {
                    lock.unlock();
                    return downstream.flush();
                }
 
                emit_queue.swap(shared_queue);
                lock.unlock();
                while (!emit_queue.empty()) {
                    downstream.handle(std::move(emit_queue.front()));
                    emit_queue.pop();
                }
                lock.lock();
            }
        } catch (source_halted const &) {
            throw;
        } catch (...) {
            auto const lock = std::lock_guard(mutex);
            downstream_threw = true;
            throw;
        }
    }
 
  public:
    template <typename Rep, typename Period>
    explicit buffer(arg::threshold<std::size_t> threshold,
                    std::chrono::duration<Rep, Period> latency_limit,
                    access_tracker<buffer_accessor> &&tracker,
                    Downstream downstream)
        : threshold(threshold.value >= 0 ? threshold.value : 1),
          max_latency(
              std::chrono::duration_cast<clock_type::duration>(latency_limit)),
          downstream(std::move(downstream)), trk(std::move(tracker)) {
        // Limit to avoid integer overflow.
        if (max_latency > std::chrono::hours(24)) {
            throw std::invalid_argument(
                "buffer latency limit must not be greater than 24 h");
        }
 
        trk.register_accessor_factory([](auto &tracker) {
            auto *self = LIBTCSPC_OBJECT_FROM_TRACKER(buffer, trk, tracker);
            return buffer_accessor([self] { self->halt(); },
                                   [self] { self->pump(); });
        });
    }
 
    // NOLINTBEGIN(cppcoreguidelines-pro-type-member-init)
    explicit buffer(arg::threshold<std::size_t> threshold,
                    access_tracker<buffer_accessor> &&tracker,
                    Downstream downstream)
        : buffer(threshold, std::chrono::hours(24), std::move(tracker),
                 std::move(downstream)) {}
    // NOLINTEND(cppcoreguidelines-pro-type-member-init)
 
    // Custom move ctor because we have a mutex. Move only works when not
    // running.
    ~buffer() = default;
 
    buffer(buffer const &) = delete;
    auto operator=(buffer const &) = delete;
 
    buffer(buffer &&other) noexcept
        : threshold(other.threshold), max_latency(other.max_latency),
          shared_queue(std::move(other.shared_queue)),
          oldest_enqueued_time(other.oldest_enqueued_time),
          upstream_flushed(other.upstream_flushed),
          upstream_halted(other.upstream_halted),
          downstream_threw(other.downstream_threw),
          emit_queue(std::move(other.emit_queue)),
          downstream(std::move(other.downstream)), pumped(other.pumped),
          trk(std::move(other.trk)) {}
 
    auto operator=(buffer &&) = delete;
 
    [[nodiscard]] auto introspect_node() const -> processor_info {
        return processor_info(this, "buffer");
    }
 
    [[nodiscard]] auto introspect_graph() const -> processor_graph {
        return downstream.introspect_graph().push_entry_point(this);
    }
 
    template <typename E>
        requires std::convertible_to<std::remove_cvref_t<E>, Event>
    void handle(E &&event) {
        bool should_notify{};
        {
            auto const lock = std::lock_guard(mutex);
            if (downstream_threw)
                throw end_of_processing(
                    "ending upstream of buffer upon end of downstream processing");
 
            shared_queue.push(std::forward<E>(event));
            should_notify = shared_queue.size() == threshold;
            if constexpr (LatencyLimited) {
                if (shared_queue.size() == 1) {
                    oldest_enqueued_time = clock_type::now();
                    should_notify = true; // Wake up once to set deadline.
                }
            }
        }
        if (should_notify)
            has_data_condition.notify_one();
    }
 
    void flush() {
        {
            auto const lock = std::lock_guard(mutex);
            if (downstream_threw)
                throw end_of_processing(
                    "ending upstream of buffer upon end of downstream processing");
            upstream_flushed = true;
        }
        has_data_condition.notify_one();
    }
};
 
} // namespace internal

template <typename Event, typename Downstream>
auto buffer(arg::threshold<std::size_t> threshold,
            access_tracker<buffer_accessor> &&tracker, Downstream downstream) {
    return internal::buffer<Event, false, Downstream>(
        threshold, std::move(tracker), std::move(downstream));
}

template <typename Event, typename Rep, typename Period, typename Downstream>
auto real_time_buffer(arg::threshold<std::size_t> threshold,
                      std::chrono::duration<Rep, Period> latency_limit,
                      access_tracker<buffer_accessor> &&tracker,
                      Downstream downstream) {
    return internal::buffer<Event, true, Downstream>(
        threshold, latency_limit, std::move(tracker), std::move(downstream));
}
 
} // namespace tcspc