From 3bdeec4cec73210583454e4a8333d68ad6f34f54 Mon Sep 17 00:00:00 2001 From: "Matthias P. Braendli" Date: Sat, 12 Aug 2023 15:35:30 +0200 Subject: Update common code --- lib/ReedSolomon.cpp | 4 ++-- lib/ReedSolomon.h | 3 ++- lib/Socket.cpp | 2 +- lib/ThreadsafeQueue.h | 54 +++++++++++++++++++++++++++++++++++++++++++++++---- 4 files changed, 55 insertions(+), 8 deletions(-) (limited to 'lib') diff --git a/lib/ReedSolomon.cpp b/lib/ReedSolomon.cpp index 1bf0b24..38b3b22 100644 --- a/lib/ReedSolomon.cpp +++ b/lib/ReedSolomon.cpp @@ -29,8 +29,8 @@ #include #include #include -#include // For galois.h ... -#include // For memcpy +#include +#include // For memcpy extern "C" { #include "fec/fec.h" diff --git a/lib/ReedSolomon.h b/lib/ReedSolomon.h index abcef62..d6686ac 100644 --- a/lib/ReedSolomon.h +++ b/lib/ReedSolomon.h @@ -30,7 +30,8 @@ # include #endif -#include +#include +#include class ReedSolomon { diff --git a/lib/Socket.cpp b/lib/Socket.cpp index 10ec1ca..b71c01e 100644 --- a/lib/Socket.cpp +++ b/lib/Socket.cpp @@ -893,7 +893,7 @@ ssize_t TCPClient::recv(void *buffer, size_t length, int flags, int timeout_ms) return 0; } - return 0; + throw std::logic_error("unreachable"); } void TCPClient::reconnect() diff --git a/lib/ThreadsafeQueue.h b/lib/ThreadsafeQueue.h index 815dfe0..8b385d6 100644 --- a/lib/ThreadsafeQueue.h +++ b/lib/ThreadsafeQueue.h @@ -2,7 +2,7 @@ Copyright (C) 2007, 2008, 2009, 2010, 2011 Her Majesty the Queen in Right of Canada (Communications Research Center Canada) - Copyright (C) 2018 + Copyright (C) 2023 Matthias P. Braendli, matthias.braendli@mpb.li An implementation for a threadsafe queue, depends on C++11 @@ -32,6 +32,7 @@ #include #include #include +#include /* This queue is meant to be used by two threads. One producer * that pushes elements into the queue, and one consumer that @@ -69,7 +70,6 @@ public: } size_t queue_size = the_queue.size(); lock.unlock(); - the_rx_notification.notify_one(); return queue_size; @@ -93,11 +93,57 @@ public: return queue_size; } + struct push_overflow_result { bool overflowed; size_t new_size; }; + + /* Push one element into the queue, and if queue is + * full remove one element from the other end. + * + * max_size == 0 is not allowed. + * + * returns the new queue size and a flag if overflow occurred. + */ + push_overflow_result push_overflow(T const& val, size_t max_size) + { + assert(max_size > 0); + std::unique_lock lock(the_mutex); + + bool overflow = false; + while (the_queue.size() >= max_size) { + overflow = true; + the_queue.pop(); + } + the_queue.push(val); + const size_t queue_size = the_queue.size(); + lock.unlock(); + + the_rx_notification.notify_one(); + + return {overflow, queue_size}; + } + + push_overflow_result push_overflow(T&& val, size_t max_size) + { + assert(max_size > 0); + std::unique_lock lock(the_mutex); + + bool overflow = false; + while (the_queue.size() >= max_size) { + overflow = true; + the_queue.pop(); + } + the_queue.emplace(std::move(val)); + const size_t queue_size = the_queue.size(); + lock.unlock(); + + the_rx_notification.notify_one(); + + return {overflow, queue_size}; + } + + /* Push one element into the queue, but wait until the * queue size goes below the threshold. * - * Notify waiting thread. - * * returns the new queue size. */ size_t push_wait_if_full(T const& val, size_t threshold) -- cgit v1.2.3