//! \file piblockingqueue.h
//! \ingroup Thread
//! \brief
//! \~english Queue with blocking
//! \~russian Блокирующая очередь
//!
//! \details
//! \~english Thread-safe queue that supports blocking operations - waits for space when storing and waits for element when retrieving.
//! \~russian Потокобезопасная очередь с поддержкой блокирующих операций - ожидает место при добавлении и ожидает элемент при получении.
/*
PIP - Platform Independent Primitives
Stephan Fomenko
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see .
*/
#ifndef PIBLOCKINGQUEUE_H
#define PIBLOCKINGQUEUE_H
#include "piconditionvar.h"
#include "piqueue.h"
/**
* \brief A Queue that supports operations that wait for the queue to become non-empty when retrieving an element, and
* wait for space to become available in the queue when storing an element.
*/
//! \~english Thread-safe blocking queue template class
//! \~russian Шаблонный класс потокобезопасной блокирующей очереди
template
class PIBlockingQueue: private PIQueue {
public:
//! \~english Constructs queue with specified capacity
//! \~russian Создает очередь с указанной емкостью
explicit inline PIBlockingQueue(size_t capacity = SIZE_MAX,
PIConditionVariable * cond_var_add = new PIConditionVariable(),
PIConditionVariable * cond_var_rem = new PIConditionVariable())
: cond_var_add(cond_var_add)
, cond_var_rem(cond_var_rem)
, max_size(capacity) {}
//! \~english Copy constructor from PIDeque
//! \~russian Конструктор копирования из PIDeque
explicit inline PIBlockingQueue(const PIDeque & other)
: cond_var_add(new PIConditionVariable())
, cond_var_rem(new PIConditionVariable()) {
mutex.lock();
max_size = SIZE_MAX;
PIDeque::append(other);
mutex.unlock();
}
//! \~english Thread-safe copy constructor from another PIBlockingQueue
//! \~russian Потокобезопасный конструктор копирования из другой PIBlockingQueue
inline PIBlockingQueue(PIBlockingQueue & other): cond_var_add(new PIConditionVariable()), cond_var_rem(new PIConditionVariable()) {
other.mutex.lock();
mutex.lock();
max_size = other.max_size;
PIDeque::append(static_cast &>(other));
mutex.unlock();
other.mutex.unlock();
}
~PIBlockingQueue() {
delete cond_var_add;
delete cond_var_rem;
}
//! \~english Inserts element waiting for space to become available
//! \~russian Вставляет элемент, ожидая освобождения места
PIBlockingQueue & put(const T & v) {
mutex.lock();
cond_var_rem->wait(mutex, [&]() { return PIDeque::size() < max_size; });
PIDeque::push_back(v);
mutex.unlock();
cond_var_add->notifyOne();
return *this;
}
PIBlockingQueue & enqueue(const T & v) { return put(v); }
//! \~english Inserts element if possible without exceeding capacity
//! \~russian Вставляет элемент если возможно без превышения емкости
bool offer(const T & v, PISystemTime timeout = {}) {
bool isOk;
mutex.lock();
if (timeout.isNull())
isOk = PIDeque::size() < max_size;
else
isOk = cond_var_rem->waitFor(mutex, timeout, [&]() { return PIDeque::size() < max_size; });
if (isOk) PIDeque::push_back(v);
mutex.unlock();
if (isOk) cond_var_add->notifyOne();
return isOk;
}
/**
* \brief Retrieves and removes the head of this queue, waiting if necessary until an element becomes available.
*
* @return the head of this queue
*/
T take() {
T t;
mutex.lock();
cond_var_add->wait(mutex, [&]() { return !PIDeque::isEmpty(); });
t = T(PIDeque::take_front());
mutex.unlock();
cond_var_rem->notifyOne();
return t;
}
T dequeue() { return take(); }
//! \~english Retrieves and removes head, waiting until element becomes available
//! \~russian Извлекает и удаляет голову очереди, ожидая появления элемента
T poll(PISystemTime timeout = {}, const T & defaultVal = T(), bool * isOk = nullptr) {
T t = defaultVal;
bool isNotEmpty;
mutex.lock();
if (timeout.isNull())
isNotEmpty = !PIDeque::isEmpty();
else
isNotEmpty = cond_var_add->waitFor(mutex, timeout, [&]() { return !PIDeque::isEmpty(); });
if (isNotEmpty) t = PIDeque::take_front();
mutex.unlock();
if (isNotEmpty) cond_var_rem->notifyOne();
if (isOk) *isOk = isNotEmpty;
return t;
}
//! \~english Returns queue capacity
//! \~russian Возвращает емкость очереди
size_t capacity() {
size_t c;
mutex.lock();
c = max_size;
mutex.unlock();
return c;
}
//! \~english Returns remaining capacity
//! \~russian Возвращает оставшуюся емкость
size_t remainingCapacity() {
mutex.lock();
size_t c = max_size - PIDeque::size();
mutex.unlock();
return c;
}
//! \~english Returns number of elements in queue
//! \~russian Возвращает количество элементов в очереди
size_t size() {
mutex.lock();
size_t s = PIDeque::size();
mutex.unlock();
return s;
}
//! \~english Removes all available elements and adds them to another queue
//! \~russian Удаляет все доступные элементы и добавляет их в другую очередь
size_t drainTo(PIDeque & other, size_t maxCount = SIZE_MAX) {
mutex.lock();
size_t count = ((maxCount > PIDeque::size()) ? PIDeque::size() : maxCount);
for (size_t i = 0; i < count; ++i)
other.push_back(PIDeque::take_front());
mutex.unlock();
return count;
}
//! \~english Removes all available elements and adds them to another blocking queue
//! \~russian Удаляет все доступные элементы и добавляет их в другую блокирующую очередь
size_t drainTo(PIBlockingQueue & other, size_t maxCount = SIZE_MAX) {
mutex.lock();
other.mutex.lock();
size_t count = maxCount > PIDeque::size() ? PIDeque::size() : maxCount;
size_t otherRemainingCapacity = other.max_size - static_cast>(other).size();
if (count > otherRemainingCapacity) count = otherRemainingCapacity;
for (size_t i = 0; i < count; ++i)
other.push_back(PIDeque::take_front());
other.mutex.unlock();
mutex.unlock();
return count;
}
private:
PIMutex mutex;
PIConditionVariable *cond_var_add, *cond_var_rem;
size_t max_size;
};
#endif // PIBLOCKINGQUEUE_H