224 lines
7.2 KiB
C++
224 lines
7.2 KiB
C++
/*
|
|
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 <http://www.gnu.org/licenses/>.
|
|
*/
|
|
|
|
#ifndef PIP_TESTS_PIBLOCKINGDEQUEUE_H
|
|
#define PIP_TESTS_PIBLOCKINGDEQUEUE_H
|
|
|
|
#include "pideque.h"
|
|
#include "piconditionvar.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.
|
|
*/
|
|
template <typename T>
|
|
class PIBlockingDequeue: private PIDeque<T> {
|
|
public:
|
|
|
|
/**
|
|
* @brief Constructor
|
|
*/
|
|
explicit inline PIBlockingDequeue(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) { }
|
|
|
|
/**
|
|
* @brief Copy constructor. Initialize queue with copy of other queue elements. Not thread-safe for other queue.
|
|
*/
|
|
explicit inline PIBlockingDequeue(const PIDeque<T>& other) : cond_var_add(new PIConditionVariable()), cond_var_rem(new PIConditionVariable()) {
|
|
mutex.lock();
|
|
max_size = SIZE_MAX;
|
|
PIDeque<T>::append(other);
|
|
mutex.unlock();
|
|
}
|
|
|
|
/**
|
|
* @brief Thread-safe copy constructor. Initialize queue with copy of other queue elements.
|
|
*/
|
|
inline PIBlockingDequeue(PIBlockingDequeue<T> & other) : cond_var_add(new PIConditionVariable()), cond_var_rem(new PIConditionVariable()) {
|
|
other.mutex.lock();
|
|
mutex.lock();
|
|
max_size = other.max_size;
|
|
PIDeque<T>::append(static_cast<PIDeque<T>&>(other));
|
|
mutex.unlock();
|
|
other.mutex.unlock();
|
|
}
|
|
|
|
~PIBlockingDequeue() {
|
|
delete cond_var_add;
|
|
delete cond_var_rem;
|
|
}
|
|
|
|
/**
|
|
* @brief Inserts the specified element into this queue, waiting if necessary for space to become available.
|
|
*
|
|
* @param v the element to add
|
|
*/
|
|
void put(const T & v) {
|
|
mutex.lock();
|
|
cond_var_rem->wait(mutex, [&]() { return PIDeque<T>::size() < max_size; });
|
|
PIDeque<T>::push_back(v);
|
|
mutex.unlock();
|
|
cond_var_add->notifyOne();
|
|
}
|
|
|
|
/**
|
|
* @brief Inserts the specified element at the end of this queue if it is possible to do so immediately without
|
|
* exceeding the queue's capacity, returning true upon success and false if this queue is full.
|
|
*
|
|
* @param v the element to add
|
|
* @return true if the element was added to this queue, else false
|
|
*/
|
|
bool offer(const T & v) {
|
|
mutex.lock();
|
|
if (PIDeque<T>::size() >= max_size) {
|
|
mutex.unlock();
|
|
return false;
|
|
}
|
|
PIDeque<T>::push_back(v);
|
|
mutex.unlock();
|
|
cond_var_add->notifyOne();
|
|
return true;
|
|
}
|
|
|
|
/**
|
|
* @brief Inserts the specified element into this queue, waiting up to the specified wait time if necessary for
|
|
* space to become available.
|
|
*
|
|
* @param v the element to add
|
|
* @param timeoutMs how long to wait before giving up, in milliseconds
|
|
* @return true if successful, or false if the specified waiting time elapses before space is available
|
|
*/
|
|
bool offer(const T & v, int timeoutMs) {
|
|
mutex.lock();
|
|
bool isOk = cond_var_rem->waitFor(mutex, timeoutMs, [&]() { return PIDeque<T>::size() < max_size; } );
|
|
if (isOk) PIDeque<T>::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<T>::isEmpty(); });
|
|
t = T(PIDeque<T>::take_front());
|
|
mutex.unlock();
|
|
cond_var_rem->notifyOne();
|
|
return t;
|
|
}
|
|
|
|
/**
|
|
* @brief Retrieves and removes the head of this queue, waiting up to the specified wait time if necessary for an
|
|
* element to become available.
|
|
*
|
|
* @param timeoutMs how long to wait before giving up, in milliseconds
|
|
* @param defaultVal value, which returns if the specified waiting time elapses before an element is available
|
|
* @return the head of this queue, or defaultVal if the specified waiting time elapses before an element is available
|
|
*/
|
|
T poll(int timeoutMs, const T & defaultVal, bool* isOk = nullptr) {
|
|
T t;
|
|
mutex.lock();
|
|
bool isNotEmpty = cond_var_add->waitFor(mutex, timeoutMs, [&]() { return !PIDeque<T>::isEmpty(); });
|
|
t = isNotEmpty ? T(PIDeque<T>::take_front()) : defaultVal;
|
|
mutex.unlock();
|
|
if (isNotEmpty) cond_var_rem->notifyOne();
|
|
if (isOk != nullptr) *isOk = isNotEmpty;
|
|
return t;
|
|
}
|
|
|
|
/**
|
|
* @brief Returns the number of elements that this queue can ideally (in the absence of memory or resource
|
|
* constraints) contains. This is always equal to the initial capacity of this queue less the current size of this queue.
|
|
*
|
|
* @return the capacity
|
|
*/
|
|
size_t capacity() {
|
|
size_t c;
|
|
mutex.lock();
|
|
c = max_size;
|
|
mutex.unlock();
|
|
return c;
|
|
}
|
|
|
|
/**
|
|
* @brief Returns the number of additional elements that this queue can ideally (in the absence of memory or resource
|
|
* constraints) accept. This is always equal to the initial capacity of this queue less the current size of this queue.
|
|
*
|
|
* @return the remaining capacity
|
|
*/
|
|
size_t remainingCapacity() {
|
|
mutex.lock();
|
|
size_t c = max_size - PIDeque<T>::size();
|
|
mutex.unlock();
|
|
return c;
|
|
}
|
|
|
|
/**
|
|
* @brief Returns the number of elements in this collection.
|
|
*/
|
|
size_t size() {
|
|
mutex.lock();
|
|
size_t s = PIDeque<T>::size();
|
|
mutex.unlock();
|
|
return s;
|
|
}
|
|
|
|
/**
|
|
* @brief Removes all available elements from this queue and adds them to other given queue.
|
|
*/
|
|
size_t drainTo(PIDeque<T>& other, size_t maxCount = SIZE_MAX) {
|
|
mutex.lock();
|
|
size_t count = maxCount > PIDeque<T>::size() ? PIDeque<T>::size() : maxCount;
|
|
for (size_t i = 0; i < count; ++i) other.push_back(PIDeque<T>::take_front());
|
|
mutex.unlock();
|
|
return count;
|
|
}
|
|
|
|
/**
|
|
* @brief Removes all available elements from this queue and adds them to other given queue.
|
|
*/
|
|
size_t drainTo(PIBlockingDequeue<T>& other, size_t maxCount = SIZE_MAX) {
|
|
mutex.lock();
|
|
other.mutex.lock();
|
|
size_t count = maxCount > PIDeque<T>::size() ? PIDeque<T>::size() : maxCount;
|
|
size_t otherRemainingCapacity = other.max_size - static_cast<PIDeque<T> >(other).size();
|
|
if (count > otherRemainingCapacity) count = otherRemainingCapacity;
|
|
for (size_t i = 0; i < count; ++i) other.push_back(PIDeque<T>::take_front());
|
|
other.mutex.unlock();
|
|
mutex.unlock();
|
|
return count;
|
|
}
|
|
|
|
private:
|
|
PIConditionLock mutex;
|
|
PIConditionVariable* cond_var_add;
|
|
PIConditionVariable* cond_var_rem;
|
|
size_t max_size;
|
|
};
|
|
|
|
|
|
#endif //PIP_TESTS_PIBLOCKINGDEQUEUE_H
|