serial::vita49::Vita49OutputSink Class Reference

#include "vita49_output_sink.h"

Inheritance diagram for serial::vita49::Vita49OutputSink:

Collaboration diagram for serial::vita49::Vita49OutputSink:

Public Member Functions
	Vita49OutputSink (std::unique_ptr< DatagramSender > sender=nullptr, core::ReceiverOutputTelemetryCallback telemetry_callback=nullptr)
	~Vita49OutputSink () override
	Vita49OutputSink (const Vita49OutputSink &)=delete
Vita49OutputSink &	operator= (const Vita49OutputSink &)=delete
	Vita49OutputSink (Vita49OutputSink &&)=delete
Vita49OutputSink &	operator= (Vita49OutputSink &&)=delete
void	initializeRun (const core::OutputConfig &config, std::string simulation_name) override
std::uint32_t	registerStream (const core::ReceiverStreamDescriptor &stream) override
void	openStream (std::uint32_t stream_id, RealType first_sample_time) override
void	submitBlock (const core::ReceiverSampleBlock &block) override
void	submitBlocks (std::span< const core::ReceiverSampleBlock > blocks) override
void	emitContextHeartbeat (RealType simulation_time) override
void	closeStream (std::uint32_t stream_id) override
core::OutputStats	finalize () override
core::OutputStats	snapshotStats () const override

Detailed Description

Definition at line 25 of file vita49_output_sink.h.

Constructor & Destructor Documentation

Vita49OutputSink() [1/3]

serial::vita49::Vita49OutputSink::Vita49OutputSink ( std::unique_ptr< DatagramSender >  sender = nullptr, core::ReceiverOutputTelemetryCallback  telemetry_callback = nullptr  )

explicit

Definition at line 52 of file vita49_output_sink.cpp.

                                                                                               :
        _telemetry_callback(std::move(telemetry_callback)), _provided_sender(std::move(sender))
    {
    }

~Vita49OutputSink()

serial::vita49::Vita49OutputSink::~Vita49OutputSink ( )

override

Definition at line 58 of file vita49_output_sink.cpp.

    {
        if (!_finalized)
        {
            try
            {
                (void)finalize();
            }
            catch (...)
            {
            }
        }
    }

References finalize(), and max.

Here is the call graph for this function:

Vita49OutputSink() [2/3]

serial::vita49::Vita49OutputSink::Vita49OutputSink ( const Vita49OutputSink &  )

delete

Vita49OutputSink() [3/3]

serial::vita49::Vita49OutputSink::Vita49OutputSink ( Vita49OutputSink &&  )

delete

Member Function Documentation

closeStream()

void serial::vita49::Vita49OutputSink::closeStream ( std::uint32_t  stream_id )

overridevirtual

Implements core::ReceiverOutputSink.

Definition at line 241 of file vita49_output_sink.cpp.

    {
        std::scoped_lock const lock(_mutex);
        emitTelemetry(consumeSenderDropsLocked(), false);
        auto& state = stateFor(stream_id);
        if (!state.closed)
        {
            emitContext(stream_id, state.last_context_time, false, true);
            state.closed = true;
            emitTelemetry({}, true);
        }
    }

References max.

emitContextHeartbeat()

void serial::vita49::Vita49OutputSink::emitContextHeartbeat ( RealType  simulation_time )

overridevirtual

Implements core::ReceiverOutputSink.

Definition at line 228 of file vita49_output_sink.cpp.

    {
        std::scoped_lock const lock(_mutex);
        emitTelemetry(consumeSenderDropsLocked(), false);
        for (auto& [stream_id, state] : _streams)
        {
            if (!state.closed && simulation_time - state.last_context_time >= 1.0)
            {
                emitContext(stream_id, simulation_time, false, false);
            }
        }
    }

References max.

finalize()

core::OutputStats serial::vita49::Vita49OutputSink::finalize ( )

overridevirtual

Implements core::ReceiverOutputSink.

Definition at line 254 of file vita49_output_sink.cpp.

    {
        std::scoped_lock const lock(_mutex);
        if (_finalized)
        {
            auto stats = snapshotStatsLocked();
            emitTelemetry({}, true);
            return stats;
        }
 
        if (_sender)
        {
            _sender->flush();
            emitTelemetry(consumeSenderDropsLocked(), false);
        }
 
        for (auto& [stream_id, state] : _streams)
        {
            if (!state.closed)
            {
                emitContext(stream_id, state.last_context_time, false, true);
                state.closed = true;
            }
        }
 
        if (!_pacing_started && !_pending_contexts.empty())
        {
            ensurePacketizer();
            std::vector<SerializedPacket> packets;
            std::vector<std::uint32_t> context_stream_ids;
            appendPendingContexts(packets, context_stream_ids);
            std::stable_sort(packets.begin(), packets.end(),
                             [](const SerializedPacket& lhs, const SerializedPacket& rhs)
                             { return lhs.first_sample_time < rhs.first_sample_time; });
            startPacing();
            if (enqueuePackets(std::move(packets)))
            {
                for (const auto stream_id : context_stream_ids)
                {
                    ++stateFor(stream_id).stats.context_packets;
                }
            }
        }
 
        if (_sender)
        {
            _sender->stop();
            emitTelemetry(consumeSenderDropsLocked(), false);
        }
 
        core::OutputStats stats{.mode = core::OutputMode::Vita49Udp,
                                .epoch_unix_nanoseconds = _packetizer
                                    ? std::optional<std::uint64_t>(_packetizer->epochUnixNanoseconds())
                                    : _config.vita49.epoch_unix_nanoseconds,
                                .streams = {}};
        for (auto& [stream_id, state] : _streams)
        {
            if (_sender)
            {
                state.stats.late_data_packet_count = _sender->lateDataPacketCount(stream_id);
                state.stats.late_context_packet_count = _sender->lateContextPacketCount(stream_id);
            }
            stats.streams.push_back(state.stats);
        }
        _finalized = true;
        emitTelemetry({}, true);
        return stats;
    }

References core::Vita49OutputConfig::epoch_unix_nanoseconds, max, core::OutputStats::mode, core::OutputConfig::vita49, and core::Vita49Udp.

Referenced by ~Vita49OutputSink().

Here is the caller graph for this function:

initializeRun()

void serial::vita49::Vita49OutputSink::initializeRun ( const core::OutputConfig &  config, std::string  simulation_name  )

overridevirtual

Implements core::ReceiverOutputSink.

Definition at line 72 of file vita49_output_sink.cpp.

    {
        std::scoped_lock const lock(_mutex);
        if (config.mode != core::OutputMode::Vita49Udp)
        {
            throw std::invalid_argument("Vita49OutputSink requires VITA output mode");
        }
        if (config.vita49.host.empty() || config.vita49.port == 0)
        {
            throw std::invalid_argument("VITA output requires destination host and port");
        }
        if (config.vita49.queue_depth == 0)
        {
            throw std::invalid_argument("VITA output queue depth must be positive");
        }
 
        _config = config;
        _simulation_name = std::move(simulation_name);
        _packetizer.reset();
        _sender = std::make_unique<PacedSender>(
            _provided_sender ? std::move(_provided_sender) : std::make_unique<UdpSender>(), config.vita49.queue_depth);
        _sender->open(config.vita49.host, config.vita49.port);
        _pending_contexts.clear();
        _pacing_started = false;
        _last_stats_emit = std::chrono::steady_clock::time_point::min();
        _last_packet_trace_emit = std::chrono::steady_clock::now();
        _pending_packet_traces.clear();
        _trace_sequence = 0;
        _initialized = true;
        _finalized = false;
        emitTelemetry({}, true);
    }

References max, and core::Vita49Udp.

openStream()

void serial::vita49::Vita49OutputSink::openStream ( std::uint32_t  stream_id, RealType  first_sample_time  )

overridevirtual

Implements core::ReceiverOutputSink.

Definition at line 125 of file vita49_output_sink.cpp.

    {
        std::scoped_lock const lock(_mutex);
        emitContext(stream_id, first_sample_time, true, false);
        stateFor(stream_id).opened = true;
        emitTelemetry({}, true);
    }

References max.

operator=() [1/2]

Vita49OutputSink & serial::vita49::Vita49OutputSink::operator= ( const Vita49OutputSink &  )

delete

operator=() [2/2]

Vita49OutputSink & serial::vita49::Vita49OutputSink::operator= ( Vita49OutputSink &&  )

delete

registerStream()

std::uint32_t serial::vita49::Vita49OutputSink::registerStream ( const core::ReceiverStreamDescriptor &  stream )

overridevirtual

Implements core::ReceiverOutputSink.

Definition at line 105 of file vita49_output_sink.cpp.

    {
        std::scoped_lock const lock(_mutex);
        const auto stream_id = _registry.registerStream(stream);
        if (!_streams.contains(stream_id))
        {
            StreamState state;
            state.descriptor = stream;
            state.stats.receiver_id = stream.receiver_id;
            state.stats.receiver_name = stream.receiver_name;
            state.stats.stream_id = stream_id;
            state.stats.mode = stream.mode.empty() ? "unknown" : stream.mode;
            state.stats.sample_rate = stream.sample_rate;
            state.stats.reference_frequency = stream.reference_frequency;
            _streams.emplace(stream_id, std::move(state));
            emitTelemetry({}, true);
        }
        return stream_id;
    }

References max, core::ReceiverStreamDescriptor::mode, core::ReceiverStreamDescriptor::receiver_id, core::ReceiverStreamDescriptor::receiver_name, core::ReceiverStreamDescriptor::reference_frequency, serial::vita49::StreamRegistry::registerStream(), and core::ReceiverStreamDescriptor::sample_rate.

Referenced by submitBlocks().

Here is the call graph for this function:

Here is the caller graph for this function:

snapshotStats()

core::OutputStats serial::vita49::Vita49OutputSink::snapshotStats ( ) const

overridevirtual

Reimplemented from core::ReceiverOutputSink.

Definition at line 323 of file vita49_output_sink.cpp.

    {
        std::scoped_lock const lock(_mutex);
        return snapshotStatsLocked();
    }

References max.

submitBlock()

void serial::vita49::Vita49OutputSink::submitBlock ( const core::ReceiverSampleBlock &  block )

overridevirtual

Implements core::ReceiverOutputSink.

Definition at line 133 of file vita49_output_sink.cpp.

{ submitBlocks({&block, 1}); }

References max, and submitBlocks().

Here is the call graph for this function:

submitBlocks()

void serial::vita49::Vita49OutputSink::submitBlocks ( std::span< const core::ReceiverSampleBlock >  blocks )

overridevirtual

Reimplemented from core::ReceiverOutputSink.

Definition at line 135 of file vita49_output_sink.cpp.

    {
        std::scoped_lock const lock(_mutex);
        if (!_initialized || !_sender)
        {
            throw std::logic_error("VITA output sink has not been initialized");
        }
        emitTelemetry(consumeSenderDropsLocked(), false);
        ensurePacketizer();
 
        struct PacketAccounting
        {
            std::uint32_t stream_id = 0;
            std::uint64_t sample_count = 0;
            RealType first_sample_time = 0.0;
            RealType end_sample_time = 0.0;
            core::Vita49Timestamp timestamp;
            std::optional<core::Vita49Timestamp> end_timestamp;
            bool over_range = false;
        };
 
        std::vector<SerializedPacket> packets;
        std::vector<PacketAccounting> accounting;
        std::vector<std::uint32_t> context_stream_ids;
        for (const auto& block : blocks)
        {
            const auto stream_id = registerStream(block.stream);
            auto& state = stateFor(stream_id);
            if (!state.opened)
            {
                emitContext(stream_id, block.first_sample_time, true, false);
                state.opened = true;
            }
        }
        appendPendingContexts(packets, context_stream_ids);
 
        for (const auto& block : blocks)
        {
            const auto stream_id = registerStream(block.stream);
            auto& state = stateFor(stream_id);
            auto result = _packetizer->packetize(block, stream_id, state.packet_counts, state.sample_loss_pending);
            state.sample_loss_pending = false;
            const RealType sample_rate = block.sample_rate > 0.0 ? block.sample_rate : state.stats.sample_rate;
            for (const auto& packet : result.packets)
            {
                const auto end_sample_time = sample_rate > 0.0
                    ? packet.first_sample_time + static_cast<RealType>(packet.sample_count) / sample_rate
                    : packet.first_sample_time;
                accounting.push_back(PacketAccounting{
                    .stream_id = stream_id,
                    .sample_count = packet.sample_count,
                    .first_sample_time = packet.first_sample_time,
                    .end_sample_time = end_sample_time,
                    .timestamp = toCoreTimestamp(packet.timestamp),
                    .end_timestamp = tryCoreTimestampFromEpoch(_packetizer->epochUnixNanoseconds(), end_sample_time),
                    .over_range = packet.over_range});
            }
            state.stats.over_range_count += result.over_range_count;
            packets.insert(packets.end(), std::make_move_iterator(result.packets.begin()),
                           std::make_move_iterator(result.packets.end()));
        }
 
        std::stable_sort(packets.begin(), packets.end(), [](const SerializedPacket& lhs, const SerializedPacket& rhs)
                         { return lhs.first_sample_time < rhs.first_sample_time; });
        startPacing();
        if (!enqueuePackets(std::move(packets)))
        {
            return;
        }
 
        for (const auto stream_id : context_stream_ids)
        {
            ++stateFor(stream_id).stats.context_packets;
        }
        for (const auto& packet : accounting)
        {
            auto& state = stateFor(packet.stream_id);
            state.stats.samples_emitted += packet.sample_count;
            ++state.stats.packets_emitted;
            if (!state.stats.first_sample_time.has_value())
            {
                state.stats.first_sample_time = packet.first_sample_time;
                state.stats.first_timestamp = packet.timestamp;
            }
            state.stats.end_sample_time = packet.end_sample_time;
            state.stats.end_timestamp = packet.end_timestamp;
            if (packet.over_range)
            {
                state.over_range_pending = true;
            }
        }
    }

References max, and registerStream().

Referenced by submitBlock().

Here is the call graph for this function:

Here is the caller graph for this function:

---

The documentation for this class was generated from the following files:

• packages/libfers/src/serial/vita49/vita49_output_sink.h
• packages/libfers/src/serial/vita49/vita49_output_sink.cpp