Added inter thread communication lib

git-svn-id: svn://db.shs.com.ru/pip@858 12ceb7fc-bf1f-11e4-8940-5bc7170c53b5

src_main/concurrent/piblockingdequeue.h

@@ -0,0 +1,148 @@
+//
+// 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 = new PIConditionVariable()) : condition_var(cond_var), max_size(capacity) { }
+    explicit inline PIBlockingDequeue(const PIDeque<T>& other) : condition_var(new PIConditionVariable()) {
+        max_size = SIZE_MAX;
+        PIDeque<T>::append(other);
+    }
+    inline PIBlockingDequeue(PIBlockingDequeue<T> & other) : condition_var(new PIConditionVariable()) {
+        other.mutex.lock();
+        max_size = other.max_size;
+        PIDeque<T>::append(static_cast<PIDeque<T>&>(other));
+        other.mutex.unlock();
+    }
+    virtual ~PIBlockingDequeue() {
+        delete condition_var;
+    }
+
+    /**
+     * @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();
+        condition_var->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();
+        condition_var->wait(mutex, [&]() { return !PIDeque<T>::isEmpty(); });
+        t = T(PIDeque<T>::take_front());
+        mutex.unlock();
+        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 = condition_var->waitFor(mutex, timeoutMs, [&]() { return !PIDeque<T>::isEmpty(); });
+        t = isOk ? T(PIDeque<T>::take_front()) : defaultVal;
+        mutex.unlock();
+        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;
+    }
+
+    PIConditionVariable *getConditionVar() const {
+        return condition_var;
+    }
+
+private:
+    PIConditionLock mutex;
+    PIConditionVariable* condition_var;
+    size_t max_size;
+};
+
+
+#endif //PIP_TESTS_PIBLOCKINGDEQUEUE_H