message_collector.hpp

Go to the documentation of this file.

#pragma once

 
#include <pulp/midi/buffer.hpp>
#include <pulp/midi/message.hpp>
#include <pulp/runtime/spsc_queue.hpp>
 
#include <algorithm>
#include <array>
#include <atomic>
#include <cmath>
#include <cstddef>
#include <cstdint>
#include <optional>
 
namespace pulp::midi {

struct TimestampedShortMessage {
    choc::midi::ShortMessage message{};
    double timestamp_seconds = 0.0;
};
 
struct MidiMessageCollectorTelemetry {
    std::size_t queue_size_approx = 0;
    std::size_t queue_capacity = 0;
    std::uint64_t queue_overflow_count = 0;
    std::size_t pending_size_approx = 0;
    std::size_t pending_capacity = 0;
    std::size_t dropped_future = 0;
};
 
template<std::size_t Capacity = 256>
class MidiMessageCollector {
public:
    MidiMessageCollector() = default;

    bool push_now(const MidiEvent& event, double now_seconds) {
        TimestampedShortMessage entry{event.message, now_seconds};
        return queue_.try_push(entry);
    }

    bool push_now(const MidiEvent& event) {
        return push_now(event, event.timestamp);
    }

    std::size_t drain_into(MidiBuffer& out,
                           double block_start_seconds,
                           int block_size_samples,
                           double sample_rate) {
        if (block_size_samples <= 0 || sample_rate <= 0.0) return 0;
        const double block_duration = block_size_samples / sample_rate;
        const double block_end_seconds = block_start_seconds + block_duration;
 
        std::size_t drained = 0;
 
        auto deliver = [&](const TimestampedShortMessage& entry) {
            int sample_offset = 0;
            if (entry.timestamp_seconds > block_start_seconds) {
                const double offset_seconds = entry.timestamp_seconds - block_start_seconds;
                // Round to nearest sample — truncation would lose
                // ~1 sample on every tick whose product is not exactly
                // representable in binary float.
                sample_offset = static_cast<int>(std::lround(offset_seconds * sample_rate));
                sample_offset = std::clamp(sample_offset, 0, block_size_samples - 1);
            }
            MidiEvent event{entry.message, sample_offset, entry.timestamp_seconds};
            out.add(event);
        };
 
        // Walk the consumer-owned `pending_` ring first: any deferred-
        // future entry whose timestamp now fits the current block is
        // delivered. Entries that are still in the future stay put so
        // a later block can consume them. Keeping these events out of
        // the producer queue ensures the consumer thread never writes
        // back to the SPSC FIFO it is draining.
        for (auto& slot : pending_) {
            if (slot.has_value() && slot->timestamp_seconds < block_end_seconds) {
                deliver(*slot);
                slot.reset();
                ++drained;
            }
        }
 
        // Drain the queue. Events that fit the current block are
        // delivered immediately; future events are stashed in the
        // consumer-owned pending ring. Draining continues after a
        // future event lands so later in-block events still get
        // delivered when producer timestamps arrive out of order.
        //
        // The pending ring is sized (`kPendingSlots`) to absorb the
        // realistic Pulp workloads in one drain: UI clicks, scripting
        // schedulers with look-ahead, MIDI file playback. When the
        // ring is genuinely saturated, additional FUTURE events have
        // nowhere to live and are dropped — observable via the atomic
        // `dropped_future_` counter so callers can detect pathological
        // back-pressure.
        while (auto next = queue_.try_pop()) {
            const auto& entry = *next;
            if (entry.timestamp_seconds < block_end_seconds) {
                deliver(entry);
                ++drained;
                continue;
            }
            // Future event: try to stash in the pending ring.
            if (stash_pending(entry)) {
                continue;
            }
            // Ring genuinely saturated. Drop with atomic counter.
            dropped_future_.fetch_add(1, std::memory_order_relaxed);
        }
        return drained;
    }

    std::size_t dropped_future() const {
        return dropped_future_.load(std::memory_order_relaxed);
    }

    std::uint64_t queue_overflow_count() const {
        return queue_.overflow_count();
    }

    void reset_queue_overflow_count() {
        queue_.reset_overflow_count();
    }

    MidiMessageCollectorTelemetry telemetry() const {
        const auto queue = queue_.telemetry();
        return {
            .queue_size_approx = queue.size_approx,
            .queue_capacity = queue.capacity,
            .queue_overflow_count = queue.overflow_count,
            .pending_size_approx = pending_size_approx(),
            .pending_capacity = pending_capacity(),
            .dropped_future = dropped_future(),
        };
    }

    std::size_t size_approx() const { return queue_.size_approx(); }

    static constexpr std::size_t capacity() { return Capacity; }

    static constexpr std::size_t pending_capacity() { return kPendingSlots; }
 
private:
    static constexpr std::size_t kPendingSlots = 64;
 
    pulp::runtime::SpscQueue<TimestampedShortMessage, Capacity> queue_;
    std::array<std::optional<TimestampedShortMessage>, kPendingSlots> pending_{};
    std::atomic<std::size_t> dropped_future_{0};
 
    std::size_t pending_size_approx() const {
        std::size_t count = 0;
        for (const auto& slot : pending_) {
            if (slot.has_value()) ++count;
        }
        return count;
    }

    bool stash_pending(const TimestampedShortMessage& entry) {
        for (auto& slot : pending_) {
            if (!slot.has_value()) {
                slot = entry;
                return true;
            }
        }
        return false; // all slots full
    }
};
 
} // namespace pulp::midi