serial::vita49::PacedSender Class Reference

#include "paced_sender.h"

Public Member Functions
	PacedSender (std::unique_ptr< DatagramSender > sender, std::size_t queue_depth)
	~PacedSender ()
	PacedSender (const PacedSender &)=delete
PacedSender &	operator= (const PacedSender &)=delete
void	open (const std::string &host, std::uint16_t port)
void	start (RealType simulation_epoch_time=0.0)
EnqueueResult	enqueue (SerializedPacket packet)
bool	enqueueBatch (std::vector< SerializedPacket > packets)
void	flush ()
void	stop ()
std::uint64_t	lateDataPacketCount (std::uint32_t stream_id) const
std::uint64_t	lateContextPacketCount (std::uint32_t stream_id) const
std::uint64_t	sentPacketCount (std::uint32_t stream_id) const
std::uint64_t	sendFailureCount (std::uint32_t stream_id) const
std::uint64_t	droppedDataPacketCount (std::uint32_t stream_id) const
std::uint64_t	droppedContextPacketCount (std::uint32_t stream_id) const
std::uint64_t	droppedSampleCount (std::uint32_t stream_id) const
std::vector< DroppedDatagram >	consumeDroppedDatagrams ()

Detailed Description

Definition at line 38 of file paced_sender.h.

Constructor & Destructor Documentation

PacedSender() [1/2]

serial::vita49::PacedSender::PacedSender	(	std::unique_ptr< DatagramSender >	sender,
		std::size_t	queue_depth
	)

Definition at line 37 of file paced_sender.cpp.

                                                                                              :
        _sender(std::move(sender)), _queue_depth(queue_depth)
    {
        if (!_sender)
        {
            throw std::invalid_argument("PacedSender requires a datagram sender");
        }
        if (queue_depth == 0)
        {
            throw std::invalid_argument("PacedSender queue depth must be positive");
        }
    }

~PacedSender()

serial::vita49::PacedSender::~PacedSender

(

)

Definition at line 50 of file paced_sender.cpp.

{ stop(); }

References stop().

Here is the call graph for this function:

PacedSender() [2/2]

serial::vita49::PacedSender::PacedSender ( const PacedSender &  )

delete

Member Function Documentation

consumeDroppedDatagrams()

std::vector< DroppedDatagram > serial::vita49::PacedSender::consumeDroppedDatagrams

(

)

Definition at line 221 of file paced_sender.cpp.

    {
        std::scoped_lock const lock(_mutex);
        auto result = std::move(_pending_dropped_datagrams);
        _pending_dropped_datagrams.clear();
        return result;
    }

References max.

droppedContextPacketCount()

std::uint64_t serial::vita49::PacedSender::droppedContextPacketCount

(

std::uint32_t

stream_id

)

const

Definition at line 207 of file paced_sender.cpp.

    {
        std::scoped_lock const lock(_mutex);
        const auto found = _dropped_context_packets.find(stream_id);
        return found == _dropped_context_packets.end() ? 0 : found->second;
    }

References max.

droppedDataPacketCount()

std::uint64_t serial::vita49::PacedSender::droppedDataPacketCount

(

std::uint32_t

stream_id

)

const

Definition at line 200 of file paced_sender.cpp.

    {
        std::scoped_lock const lock(_mutex);
        const auto found = _dropped_data_packets.find(stream_id);
        return found == _dropped_data_packets.end() ? 0 : found->second;
    }

References max.

droppedSampleCount()

std::uint64_t serial::vita49::PacedSender::droppedSampleCount

(

std::uint32_t

stream_id

)

const

Definition at line 214 of file paced_sender.cpp.

    {
        std::scoped_lock const lock(_mutex);
        const auto found = _dropped_samples.find(stream_id);
        return found == _dropped_samples.end() ? 0 : found->second;
    }

References max.

enqueue()

EnqueueResult serial::vita49::PacedSender::enqueue

(

SerializedPacket

packet

)

Definition at line 68 of file paced_sender.cpp.

    {
        std::unique_lock lock(_mutex);
        if (!_started)
        {
            throw std::logic_error("PacedSender must be started before enqueue");
        }
        if (_stopping)
        {
            return EnqueueResult{
                .enqueued = false,
                .dropped = std::nullopt,
            };
        }
 
        while (queuedOrSendingCount() >= _queue_depth)
        {
            const bool precedes_latest = !_queue.empty() && packet.first_sample_time < _queue.back().first_sample_time;
            if (!_priority_overflow_in_progress && precedes_latest)
            {
                _priority_overflow_in_progress = true;
                insertByDeadlineUnlocked(std::move(packet));
                _cv.notify_all();
                _cv.wait(lock, [this] { return _stopping || queuedOrSendingCount() <= _queue_depth; });
                _priority_overflow_in_progress = false;
                _cv.notify_all();
                return EnqueueResult{.enqueued = true, .dropped = std::nullopt};
            }
 
            _cv.wait(lock);
            if (_stopping)
            {
                return EnqueueResult{
                    .enqueued = false,
                    .dropped = std::nullopt,
                };
            }
        }
 
        insertByDeadlineUnlocked(std::move(packet));
        _cv.notify_all();
        return EnqueueResult{.enqueued = true, .dropped = std::nullopt};
    }

References serial::vita49::EnqueueResult::enqueued, and max.

enqueueBatch()

bool serial::vita49::PacedSender::enqueueBatch

(

std::vector< SerializedPacket >

packets

)

Definition at line 112 of file paced_sender.cpp.

    {
        if (packets.empty())
        {
            return true;
        }
 
        std::unique_lock lock(_mutex);
        if (!_started)
        {
            throw std::logic_error("PacedSender must be started before enqueue");
        }
        if (_stopping)
        {
            return false;
        }
 
        for (auto& packet : packets)
        {
            insertByDeadlineUnlocked(std::move(packet));
        }
        _cv.notify_all();
        _cv.wait(lock, [this] { return _stopping || queuedOrSendingCount() <= _queue_depth; });
        return true;
    }

References max.

flush()

void serial::vita49::PacedSender::flush

(

)

Definition at line 138 of file paced_sender.cpp.

    {
        std::unique_lock lock(_mutex);
        _cv.wait(lock, [this] { return _queue.empty() && !_send_in_progress; });
    }

References max.

lateContextPacketCount()

std::uint64_t serial::vita49::PacedSender::lateContextPacketCount

(

std::uint32_t

stream_id

)

const

Definition at line 179 of file paced_sender.cpp.

    {
        std::scoped_lock const lock(_mutex);
        const auto found = _late_context_packets.find(stream_id);
        return found == _late_context_packets.end() ? 0 : found->second;
    }

References max.

lateDataPacketCount()

std::uint64_t serial::vita49::PacedSender::lateDataPacketCount

(

std::uint32_t

stream_id

)

const

Definition at line 172 of file paced_sender.cpp.

    {
        std::scoped_lock const lock(_mutex);
        const auto found = _late_data_packets.find(stream_id);
        return found == _late_data_packets.end() ? 0 : found->second;
    }

References max.

open()

void serial::vita49::PacedSender::open	(	const std::string &	host,
		std::uint16_t	port
	)

Definition at line 52 of file paced_sender.cpp.

{ _sender->open(host, port); }

operator=()

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

delete

sendFailureCount()

std::uint64_t serial::vita49::PacedSender::sendFailureCount

(

std::uint32_t

stream_id

)

const

Definition at line 193 of file paced_sender.cpp.

    {
        std::scoped_lock const lock(_mutex);
        const auto found = _send_failures.find(stream_id);
        return found == _send_failures.end() ? 0 : found->second;
    }

References max.

sentPacketCount()

std::uint64_t serial::vita49::PacedSender::sentPacketCount

(

std::uint32_t

stream_id

)

const

Definition at line 186 of file paced_sender.cpp.

    {
        std::scoped_lock const lock(_mutex);
        const auto found = _sent_packets.find(stream_id);
        return found == _sent_packets.end() ? 0 : found->second;
    }

References max.

start()

void serial::vita49::PacedSender::start

(

RealType

simulation_epoch_time = 0.0

)

Definition at line 54 of file paced_sender.cpp.

    {
        std::scoped_lock const lock(_mutex);
        if (_started)
        {
            return;
        }
        _simulation_epoch_time = simulation_epoch_time;
        _steady_epoch = std::chrono::steady_clock::now();
        _stopping = false;
        _started = true;
        _thread = std::thread([this] { run(); });
    }

References max.

stop()

void serial::vita49::PacedSender::stop

(

)

Definition at line 144 of file paced_sender.cpp.

    {
        {
            std::scoped_lock const lock(_mutex);
            if (!_started && !_thread.joinable())
            {
                _sender->close();
                return;
            }
            _stopping = true;
            _cv.notify_all();
        }
 
        if (_thread.joinable())
        {
            _thread.join();
        }
 
        {
            std::scoped_lock const lock(_mutex);
            _started = false;
            _stopping = false;
            _send_in_progress = false;
            _cv.notify_all();
        }
        _sender->close();
    }

References max.

Referenced by ~PacedSender().

Here is the caller graph for this function:

---

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

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