diff options
author | Matthias P. Braendli <matthias.braendli@mpb.li> | 2023-02-21 18:13:26 +0100 |
---|---|---|
committer | Matthias P. Braendli <matthias.braendli@mpb.li> | 2023-02-21 18:13:26 +0100 |
commit | 05d3e26409a8f62c7f55851390d61d953f59489a (patch) | |
tree | 45c9428132e4f02f799ca81fce434fcf33fd1120 /lib | |
parent | ba0c32703ee1af770deabdf864c806d974ea8206 (diff) | |
download | dabmod-05d3e26409a8f62c7f55851390d61d953f59489a.tar.gz dabmod-05d3e26409a8f62c7f55851390d61d953f59489a.tar.bz2 dabmod-05d3e26409a8f62c7f55851390d61d953f59489a.zip |
Change mod to output queue behaviour, fix SFN dexter + B200
Diffstat (limited to 'lib')
-rw-r--r-- | lib/ThreadsafeQueue.h | 54 |
1 files changed, 50 insertions, 4 deletions
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 <condition_variable> #include <queue> #include <utility> +#include <cassert> /* 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<std::mutex> 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<std::mutex> 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) |