pip/libs/main/thread/piblockingqueue.h

/*! \file piblockingqueue.h
 * \ingroup Thread
 * \~\brief
 * \~english Queue with blocking
 * \~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 <http://www.gnu.org/licenses/>.
*/

#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.
 */
template<typename T>
class PIBlockingQueue: private PIQueue<T> {
public:
	/**
	 * \brief Constructor
	 */
	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) {}

	/**
	 * \brief Copy constructor. Initialize queue with copy of other queue elements. Not thread-safe for other queue.
	 */
	explicit inline PIBlockingQueue(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 PIBlockingQueue(PIBlockingQueue<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();
	}

	~PIBlockingQueue() {
		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
	 */
	PIBlockingQueue<T> & 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();
		return *this;
	}

	PIBlockingQueue<T> & enqueue(const T & v) { return put(v); }

	/**
	 * \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
	 * @param timeout the timeout waiting for inserting if que is full, if timeout is null, then returns immediately
	 * @return true if the element was added to this queue, else false
	 */
	bool offer(const T & v, PISystemTime timeout = {}) {
		bool isOk;
		mutex.lock();
		if (timeout.isNull())
			isOk = PIDeque<T>::size() < max_size;
		else
			isOk = cond_var_rem->waitFor(mutex, timeout, [&]() { 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;
	}

	T dequeue() { return take(); }

	/**
	 * \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 timeout how long to wait before giving up
	 * @param defaultVal value, which returns if the specified waiting time elapses before an element is available
	 * @param isOk flag, which indicates result of method execution. It will be set to false if timeout, or true if
	 * return value is retrieved value
	 * @return the head of this queue, or defaultVal if the specified waiting time elapses before an element is available
	 */
	T poll(PISystemTime timeout = {}, const T & defaultVal = T(), bool * isOk = nullptr) {
		T t = defaultVal;
		bool isNotEmpty;
		mutex.lock();
		if (timeout.isNull())
			isNotEmpty = !PIDeque<T>::isEmpty();
		else
			isNotEmpty = cond_var_add->waitFor(mutex, timeout, [&]() { return !PIDeque<T>::isEmpty(); });
		if (isNotEmpty) t = PIDeque<T>::take_front();
		mutex.unlock();
		if (isNotEmpty) cond_var_rem->notifyOne();
		if (isOk) *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(PIBlockingQueue<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:
	PIMutex mutex;
	PIConditionVariable *cond_var_add, *cond_var_rem;
	size_t max_size;
};


#endif // PIBLOCKINGQUEUE_H