telemetry_accumulator.h

Go to the documentation of this file.

#pragma once
 
#include "bcnp/packet.h"
#include "bcnp/packet_storage.h"
#include "bcnp/static_vector.h"
 
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <functional>
#include <mutex>
#include <vector>
 
namespace bcnp {
 
struct TelemetryAccumulatorConfig {
    std::size_t flushIntervalTicks{2};
    
    std::size_t maxBufferedMessages{64};
};
 
template<typename MsgType, typename Storage = StaticVector<MsgType, 64>>
class TelemetryAccumulator {
public:
    using Clock = std::chrono::steady_clock;
 
    explicit TelemetryAccumulator(TelemetryAccumulatorConfig config = {})
        : m_config(config) {}
 
    bool Record(const MsgType& msg) {
        std::lock_guard<std::mutex> lock(m_mutex);
 
        // Ring-buffer behavior: drop oldest when full
        if (m_buffer.size() >= m_config.maxBufferedMessages) {
            DropOldestUnlocked();
            ++m_metrics.bufferOverflows;
        }
 
        m_buffer.push_back(msg);
        ++m_metrics.messagesRecorded;
        return true;
    }
 
    template<typename InputIt>
    void RecordBatch(InputIt first, InputIt last) {
        std::lock_guard<std::mutex> lock(m_mutex);
        for (auto it = first; it != last; ++it) {
            if (m_buffer.size() >= m_config.maxBufferedMessages) {
                DropOldestUnlocked();
                ++m_metrics.bufferOverflows;
            }
            m_buffer.push_back(*it);
            ++m_metrics.messagesRecorded;
        }
    }
 
    template<typename Adapter>
    bool MaybeFlush(Adapter& adapter) {
        Storage localBuffer;
        std::size_t messageCount = 0;
        
        {
            std::lock_guard<std::mutex> lock(m_mutex);
            
            ++m_tickCount;
            if (m_tickCount < m_config.flushIntervalTicks) {
                return false;
            }
            
            m_tickCount = 0;
            
            if (m_buffer.empty()) {
                return false;
            }
            
            // Swap buffer out while holding lock
            localBuffer = std::move(m_buffer);
            m_buffer = Storage{};
            messageCount = localBuffer.size();
        }
        // Lock released - now safe to do blocking I/O
        
        return SendBuffer(adapter, std::move(localBuffer), messageCount);
    }
 
    template<typename Adapter>
    bool ForceFlush(Adapter& adapter) {
        Storage localBuffer;
        std::size_t messageCount = 0;
        
        {
            std::lock_guard<std::mutex> lock(m_mutex);
            m_tickCount = 0;
            
            if (m_buffer.empty()) {
                return false;
            }
            
            // Swap buffer out while holding lock
            localBuffer = std::move(m_buffer);
            m_buffer = Storage{};
            messageCount = localBuffer.size();
        }
        // Lock released - now safe to do blocking I/O
        
        return SendBuffer(adapter, std::move(localBuffer), messageCount);
    }
 
    std::size_t BufferedCount() const {
        std::lock_guard<std::mutex> lock(m_mutex);
        return m_buffer.size();
    }
 
    void Clear() {
        std::lock_guard<std::mutex> lock(m_mutex);
        m_buffer.clear();
        m_tickCount = 0;
    }
 
    struct Metrics {
        uint64_t messagesRecorded{0};
        uint64_t messagesSent{0};
        uint64_t packetsSent{0};
        uint64_t bufferOverflows{0};
        uint64_t sendFailures{0};
    };
 
    Metrics GetMetrics() const {
        std::lock_guard<std::mutex> lock(m_mutex);
        return m_metrics;
    }
 
    void ResetMetrics() {
        std::lock_guard<std::mutex> lock(m_mutex);
        m_metrics = {};
    }
 
    void SetConfig(const TelemetryAccumulatorConfig& config) {
        std::lock_guard<std::mutex> lock(m_mutex);
        m_config = config;
    }
 
private:
    template<typename Adapter>
    bool SendBuffer(Adapter& adapter, Storage&& buffer, std::size_t messageCount) {
        // Build packet from extracted buffer
        TypedPacket<MsgType, Storage> packet;
        packet.messages = std::move(buffer);
        
        // Encode to wire format
        m_wireBuffer.clear();
        if (!EncodeTypedPacket(packet, m_wireBuffer)) {
            std::lock_guard<std::mutex> lock(m_mutex);
            ++m_metrics.sendFailures;
            return false;
        }
 
        // Send (potentially blocking syscall - but we don't hold the lock!)
        if (!adapter.SendBytes(m_wireBuffer.data(), m_wireBuffer.size())) {
            std::lock_guard<std::mutex> lock(m_mutex);
            ++m_metrics.sendFailures;
            return false;
        }
 
        // Update metrics
        {
            std::lock_guard<std::mutex> lock(m_mutex);
            m_metrics.messagesSent += messageCount;
            ++m_metrics.packetsSent;
        }
        return true;
    }
 
    TelemetryAccumulatorConfig m_config;
    Storage m_buffer{};
    std::vector<uint8_t> m_wireBuffer;  // Persistent buffer, capacity retained across flushes
    std::size_t m_tickCount{0};
    Metrics m_metrics{};
    mutable std::mutex m_mutex;
 
    void DropOldestUnlocked() {
        if (m_buffer.empty()) {
            return;
        }
 
        // Shift everything left by one; O(n) but small (default 64)
        for (std::size_t i = 1; i < m_buffer.size(); ++i) {
            m_buffer[i - 1] = std::move(m_buffer[i]);
        }
        m_buffer.pop_back();
    }
};
 
template<typename MsgType>
using DynamicTelemetryAccumulator = TelemetryAccumulator<MsgType, std::vector<MsgType>>;
 
template<typename MsgType, std::size_t Capacity = 64>
using StaticTelemetryAccumulator = TelemetryAccumulator<MsgType, StaticVector<MsgType, Capacity>>;
 
} // namespace bcnp