pip/src_main/concurrent/piblockingdequeue.h

//
// Created by fomenko on 20.09.2019.
//

#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:
    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) { }
    explicit inline PIBlockingDequeue(const PIDeque<T>& other) : cond_var_add(new PIConditionVariable()), cond_var_rem(new PIConditionVariable()) {
        max_size = SIZE_MAX;
        PIDeque<T>::append(other);
    }
    inline PIBlockingDequeue(PIBlockingDequeue<T> & other) : cond_var_add(new PIConditionVariable()), cond_var_rem(new PIConditionVariable()) {
        other.mutex.lock();
        max_size = other.max_size;
        PIDeque<T>::append(static_cast<PIDeque<T>&>(other));
        other.mutex.unlock();
    }
    virtual ~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.
     *
     * @todo write tests
     *
     * @param v the element to add
     */
    virtual 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
     */
    virtual 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 Retrieves and removes the head of this queue, waiting if necessary until an element becomes available.
     *
     * @return the head of this queue
     */
    virtual 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
     */
    virtual T poll(int timeoutMs, const T & defaultVal) {
        T t;
        mutex.lock();
        bool isOk = cond_var_add->waitFor(mutex, timeoutMs, [&]() { return !PIDeque<T>::isEmpty(); });
        t = isOk ? T(PIDeque<T>::take_front()) : defaultVal;
        mutex.unlock();
        if (isOk) cond_var_rem->notifyOne();
        return t;
    }

    virtual 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
     */
    virtual size_t remainingCapacity() {
        mutex.lock();
        size_t c = max_size - PIDeque<T>::size();
        mutex.unlock();
        return c;
    }

    virtual 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.
     */
    virtual 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.
     */
    virtual 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