summaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/ThreadsafeQueue.h37
1 files changed, 30 insertions, 7 deletions
diff --git a/src/ThreadsafeQueue.h b/src/ThreadsafeQueue.h
index 4524559..ab287b2 100644
--- a/src/ThreadsafeQueue.h
+++ b/src/ThreadsafeQueue.h
@@ -5,7 +5,7 @@
Copyright (C) 2018
Matthias P. Braendli, matthias.braendli@mpb.li
- An implementation for a threadsafe queue using std::thread
+ An implementation for a threadsafe queue, depends on C++11
When creating a ThreadsafeQueue, one can specify the minimal number
of elements it must contain before it is possible to take one
@@ -40,9 +40,13 @@
* retrieves the elements.
*
* The queue can make the consumer block until an element
- * is available.
+ * is available, or a wakeup requested.
*/
+/* Class thrown by blocking pop to tell the consumer
+ * that there's a wakeup requested. */
+class ThreadsafeQueueWakeup {};
+
template<typename T>
class ThreadsafeQueue
{
@@ -98,6 +102,17 @@ public:
return queue_size;
}
+ /* Trigger a wakeup event on a blocking consumer, which
+ * will receive a ThreadsafeQueueWakeup exception.
+ */
+ void trigger_wakeup(void)
+ {
+ std::unique_lock<std::mutex> lock(the_mutex);
+ wakeup_requested = true;
+ lock.unlock();
+ the_rx_notification.notify_one();
+ }
+
/* Send a notification for the receiver thread */
void notify(void)
{
@@ -135,15 +150,22 @@ public:
void wait_and_pop(T& popped_value, size_t prebuffering = 1)
{
std::unique_lock<std::mutex> lock(the_mutex);
- while (the_queue.size() < prebuffering) {
+ while (the_queue.size() < prebuffering and
+ not wakeup_requested) {
the_rx_notification.wait(lock);
}
- std::swap(popped_value, the_queue.front());
- the_queue.pop();
+ if (wakeup_requested) {
+ wakeup_requested = false;
+ throw ThreadsafeQueueWakeup();
+ }
+ else {
+ std::swap(popped_value, the_queue.front());
+ the_queue.pop();
- lock.unlock();
- the_tx_notification.notify_one();
+ lock.unlock();
+ the_tx_notification.notify_one();
+ }
}
private:
@@ -151,5 +173,6 @@ private:
mutable std::mutex the_mutex;
std::condition_variable the_rx_notification;
std::condition_variable the_tx_notification;
+ bool wakeup_requested = false;
};