SHS/pip
4
3
Fork 0
Code Issues 67 Pull Requests 1 Actions Releases Wiki Activity

rename PIBlockingDequeue -> PIBlockingQueue

Browse Source
This commit is contained in:
Бычков Андрей
2020-08-11 19:05:47 +03:00
parent 8c3349d84a
commit 3ba6a7b0e8
3 changed files with 43 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
lib/main/thread/piblockingdequeue.h → lib/main/thread/piblockingqueue.h
View File

@@ -17,8 +17,8 @@
along with this program. If not, see <http://www.gnu.org/licenses/>. along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#ifndef PIBLOCKINGDEQUEUE_H #ifndef PIBLOCKINGQUEUE_H
#define PIBLOCKINGDEQUEUE_H #define PIBLOCKINGQUEUE_H
#include "pideque.h" #include "pideque.h"
#include "piconditionvar.h" #include "piconditionvar.h"
@@ -28,13 +28,13 @@
* wait for space to become available in the queue when storing an element. * wait for space to become available in the queue when storing an element.
*/ */
template <typename T> template <typename T>
class PIBlockingDequeue: private PIDeque<T> { class PIBlockingQueue: private PIQueue<T> {
public: public:
/** /**
* @brief Constructor * @brief Constructor
*/ */
explicit inline PIBlockingDequeue(size_t capacity = SIZE_MAX, explicit inline PIBlockingQueue(size_t capacity = SIZE_MAX,
PIConditionVariable* cond_var_add = new PIConditionVariable(), PIConditionVariable* cond_var_add = new PIConditionVariable(),
PIConditionVariable* cond_var_rem = new PIConditionVariable()) PIConditionVariable* cond_var_rem = new PIConditionVariable())
: cond_var_add(cond_var_add), cond_var_rem(cond_var_rem), max_size(capacity) { } : cond_var_add(cond_var_add), cond_var_rem(cond_var_rem), max_size(capacity) { }
@@ -42,7 +42,7 @@ public:
/** /**
* @brief Copy constructor. Initialize queue with copy of other queue elements. Not thread-safe for other queue. * @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()) { explicit inline PIBlockingQueue(const PIDeque<T>& other) : cond_var_add(new PIConditionVariable()), cond_var_rem(new PIConditionVariable()) {
mutex.lock(); mutex.lock();
max_size = SIZE_MAX; max_size = SIZE_MAX;
PIDeque<T>::append(other); PIDeque<T>::append(other);
@@ -52,7 +52,7 @@ public:
/** /**
* @brief Thread-safe copy constructor. Initialize queue with copy of other queue elements. * @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()) { inline PIBlockingQueue(PIBlockingQueue<T> & other) : cond_var_add(new PIConditionVariable()), cond_var_rem(new PIConditionVariable()) {
other.mutex.lock(); other.mutex.lock();
mutex.lock(); mutex.lock();
max_size = other.max_size; max_size = other.max_size;
@@ -61,7 +61,7 @@ public:
other.mutex.unlock(); other.mutex.unlock();
} }
~PIBlockingDequeue() { ~PIBlockingQueue() {
delete cond_var_add; delete cond_var_add;
delete cond_var_rem; delete cond_var_rem;
} }
@@ -71,14 +71,17 @@ public:
* *
* @param v the element to add * @param v the element to add
*/ */
void put(const T & v) { PIBlockingQueue<T> & put(const T & v) {
mutex.lock(); mutex.lock();
cond_var_rem->wait(mutex, [&]() { return PIDeque<T>::size() < max_size; }); cond_var_rem->wait(mutex, [&]() { return PIDeque<T>::size() < max_size; });
PIDeque<T>::push_back(v); PIDeque<T>::push_back(v);
mutex.unlock(); mutex.unlock();
cond_var_add->notifyOne(); 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 * @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. * exceeding the queue's capacity, returning true upon success and false if this queue is full.
@@ -114,6 +117,8 @@ public:
return t; 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 * @brief Retrieves and removes the head of this queue, waiting up to the specified wait time if necessary for an
* element to become available. * element to become available.
@@ -190,7 +195,7 @@ public:
/** /**
* @brief Removes all available elements from this queue and adds them to other given queue. * @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) { size_t drainTo(PIBlockingQueue<T>& other, size_t maxCount = SIZE_MAX) {
mutex.lock(); mutex.lock();
other.mutex.lock(); other.mutex.lock();
size_t count = maxCount > PIDeque<T>::size() ? PIDeque<T>::size() : maxCount; size_t count = maxCount > PIDeque<T>::size() ? PIDeque<T>::size() : maxCount;
@@ -210,4 +215,4 @@ private:
}; };
#endif // PIBLOCKINGDEQUEUE_H #endif // PIBLOCKINGQUEUE_H

4
lib/main/thread/pithreadpoolexecutor.h
View File

@@ -20,7 +20,7 @@
#ifndef PITHREADPOOLEXECUTOR_H #ifndef PITHREADPOOLEXECUTOR_H
#define PITHREADPOOLEXECUTOR_H #define PITHREADPOOLEXECUTOR_H
#include "piblockingdequeue.h" #include "piblockingqueue.h"
#include <atomic> #include <atomic>
@@ -54,7 +54,7 @@ public:
private: private:
std::atomic_bool isShutdown_; std::atomic_bool isShutdown_;
PIBlockingDequeue<std::function<void()> > taskQueue; PIBlockingQueue<std::function<void()> > taskQueue;
PIVector<PIThread*> threadPool; PIVector<PIThread*> threadPool;
bool queue_own; bool queue_own;

52
tests/concurrent/BlockingDequeueUnitTest.cpp
View File

@@ -1,5 +1,5 @@
#include "gtest/gtest.h" #include "gtest/gtest.h"
#include "piblockingdequeue.h" #include "piblockingqueue.h"
class MockConditionVar: public PIConditionVariable { class MockConditionVar: public PIConditionVariable {
public: public:
@@ -39,7 +39,7 @@ TEST(BlockingDequeueUnitTest, put_is_block_when_capacity_reach) {
size_t capacity = 0; size_t capacity = 0;
auto conditionVarAdd = new MockConditionVar(); auto conditionVarAdd = new MockConditionVar();
auto conditionVarRem = new MockConditionVar(); auto conditionVarRem = new MockConditionVar();
PIBlockingDequeue<int> dequeue(capacity, conditionVarAdd, conditionVarRem); PIBlockingQueue<int> dequeue(capacity, conditionVarAdd, conditionVarRem);
dequeue.put(11); dequeue.put(11);
ASSERT_TRUE(conditionVarRem->isWaitCalled); ASSERT_TRUE(conditionVarRem->isWaitCalled);
ASSERT_FALSE(conditionVarRem->isTrueCondition); ASSERT_FALSE(conditionVarRem->isTrueCondition);
@@ -50,7 +50,7 @@ TEST(BlockingDequeueUnitTest, offer_timedout_is_false_when_capacity_reach) {
int timeout = 11; int timeout = 11;
auto conditionVarAdd = new MockConditionVar(); auto conditionVarAdd = new MockConditionVar();
auto conditionVarRem = new MockConditionVar(); auto conditionVarRem = new MockConditionVar();
PIBlockingDequeue<int> dequeue(capacity, conditionVarAdd, conditionVarRem); PIBlockingQueue<int> dequeue(capacity, conditionVarAdd, conditionVarRem);
ASSERT_FALSE(dequeue.offer(11, timeout)); ASSERT_FALSE(dequeue.offer(11, timeout));
} }
@@ -59,7 +59,7 @@ TEST(BlockingDequeueUnitTest, offer_timedout_is_block_when_capacity_reach) {
int timeout = 11; int timeout = 11;
auto conditionVarAdd = new MockConditionVar(); auto conditionVarAdd = new MockConditionVar();
auto conditionVarRem = new MockConditionVar(); auto conditionVarRem = new MockConditionVar();
PIBlockingDequeue<int> dequeue(capacity, conditionVarAdd, conditionVarRem); PIBlockingQueue<int> dequeue(capacity, conditionVarAdd, conditionVarRem);
dequeue.offer(11, timeout); dequeue.offer(11, timeout);
EXPECT_TRUE(conditionVarRem->isWaitForCalled); EXPECT_TRUE(conditionVarRem->isWaitForCalled);
EXPECT_EQ(timeout, conditionVarRem->timeout); EXPECT_EQ(timeout, conditionVarRem->timeout);
@@ -68,13 +68,13 @@ TEST(BlockingDequeueUnitTest, offer_timedout_is_block_when_capacity_reach) {
TEST(BlockingDequeueUnitTest, offer_is_true_before_capacity_reach) { TEST(BlockingDequeueUnitTest, offer_is_true_before_capacity_reach) {
size_t capacity = 1; size_t capacity = 1;
PIBlockingDequeue<int> dequeue(capacity); PIBlockingQueue<int> dequeue(capacity);
ASSERT_TRUE(dequeue.offer(10)); ASSERT_TRUE(dequeue.offer(10));
} }
TEST(BlockingDequeueUnitTest, offer_is_false_when_capacity_reach) { TEST(BlockingDequeueUnitTest, offer_is_false_when_capacity_reach) {
size_t capacity = 1; size_t capacity = 1;
PIBlockingDequeue<int> dequeue(capacity); PIBlockingQueue<int> dequeue(capacity);
dequeue.offer(11); dequeue.offer(11);
ASSERT_FALSE(dequeue.offer(10)); ASSERT_FALSE(dequeue.offer(10));
} }
@@ -83,7 +83,7 @@ TEST(BlockingDequeueUnitTest, offer_is_false_when_capacity_reach) {
TEST(DISABLED_BlockingDequeueUnitTest, take_is_block_when_empty) { TEST(DISABLED_BlockingDequeueUnitTest, take_is_block_when_empty) {
size_t capacity = 1; size_t capacity = 1;
auto conditionVar = new MockConditionVar(); auto conditionVar = new MockConditionVar();
PIBlockingDequeue<int> dequeue(capacity, conditionVar); PIBlockingQueue<int> dequeue(capacity, conditionVar);
// May cause segfault because take front of empty queue // May cause segfault because take front of empty queue
dequeue.take(); dequeue.take();
EXPECT_TRUE(conditionVar->isWaitCalled); EXPECT_TRUE(conditionVar->isWaitCalled);
@@ -93,7 +93,7 @@ TEST(DISABLED_BlockingDequeueUnitTest, take_is_block_when_empty) {
TEST(BlockingDequeueUnitTest, take_is_not_block_when_not_empty) { TEST(BlockingDequeueUnitTest, take_is_not_block_when_not_empty) {
size_t capacity = 1; size_t capacity = 1;
auto conditionVar = new MockConditionVar(); auto conditionVar = new MockConditionVar();
PIBlockingDequeue<int> dequeue(capacity, conditionVar); PIBlockingQueue<int> dequeue(capacity, conditionVar);
dequeue.offer(111); dequeue.offer(111);
dequeue.take(); dequeue.take();
@@ -104,7 +104,7 @@ TEST(BlockingDequeueUnitTest, take_is_not_block_when_not_empty) {
TEST(BlockingDequeueUnitTest, take_is_value_eq_to_offer_value) { TEST(BlockingDequeueUnitTest, take_is_value_eq_to_offer_value) {
size_t capacity = 1; size_t capacity = 1;
auto conditionVar = new MockConditionVar(); auto conditionVar = new MockConditionVar();
PIBlockingDequeue<int> dequeue(capacity, conditionVar); PIBlockingQueue<int> dequeue(capacity, conditionVar);
dequeue.offer(111); dequeue.offer(111);
ASSERT_EQ(dequeue.take(), 111); ASSERT_EQ(dequeue.take(), 111);
@@ -113,7 +113,7 @@ TEST(BlockingDequeueUnitTest, take_is_value_eq_to_offer_value) {
TEST(BlockingDequeueUnitTest, take_is_last) { TEST(BlockingDequeueUnitTest, take_is_last) {
size_t capacity = 10; size_t capacity = 10;
auto conditionVar = new MockConditionVar(); auto conditionVar = new MockConditionVar();
PIBlockingDequeue<int> dequeue(capacity, conditionVar); PIBlockingQueue<int> dequeue(capacity, conditionVar);
EXPECT_TRUE(dequeue.offer(111)); EXPECT_TRUE(dequeue.offer(111));
EXPECT_TRUE(dequeue.offer(222)); EXPECT_TRUE(dequeue.offer(222));
ASSERT_EQ(dequeue.take(), 111); ASSERT_EQ(dequeue.take(), 111);
@@ -124,7 +124,7 @@ TEST(BlockingDequeueUnitTest, poll_is_not_block_when_empty) {
size_t capacity = 1; size_t capacity = 1;
bool isOk; bool isOk;
auto conditionVar = new MockConditionVar(); auto conditionVar = new MockConditionVar();
PIBlockingDequeue<int> dequeue(capacity, conditionVar); PIBlockingQueue<int> dequeue(capacity, conditionVar);
dequeue.poll(0, 111, &isOk); dequeue.poll(0, 111, &isOk);
EXPECT_FALSE(conditionVar->isWaitForCalled); EXPECT_FALSE(conditionVar->isWaitForCalled);
} }
@@ -133,7 +133,7 @@ TEST(BlockingDequeueUnitTest, poll_is_default_value_when_empty) {
size_t capacity = 1; size_t capacity = 1;
bool isOk; bool isOk;
auto conditionVar = new MockConditionVar(); auto conditionVar = new MockConditionVar();
PIBlockingDequeue<int> dequeue(capacity, conditionVar); PIBlockingQueue<int> dequeue(capacity, conditionVar);
ASSERT_EQ(dequeue.poll(0, 111, &isOk), 111); ASSERT_EQ(dequeue.poll(0, 111, &isOk), 111);
} }
@@ -141,7 +141,7 @@ TEST(BlockingDequeueUnitTest, poll_is_offer_value_when_not_empty) {
size_t capacity = 1; size_t capacity = 1;
bool isOk; bool isOk;
auto conditionVar = new MockConditionVar(); auto conditionVar = new MockConditionVar();
PIBlockingDequeue<int> dequeue(capacity, conditionVar); PIBlockingQueue<int> dequeue(capacity, conditionVar);
dequeue.offer(111); dequeue.offer(111);
ASSERT_EQ(dequeue.poll(0, -1, &isOk), 111); ASSERT_EQ(dequeue.poll(0, -1, &isOk), 111);
} }
@@ -150,7 +150,7 @@ TEST(BlockingDequeueUnitTest, poll_timeouted_is_block_when_empty) {
size_t capacity = 1; size_t capacity = 1;
int timeout = 11; int timeout = 11;
auto conditionVar = new MockConditionVar(); auto conditionVar = new MockConditionVar();
PIBlockingDequeue<int> dequeue(capacity, conditionVar); PIBlockingQueue<int> dequeue(capacity, conditionVar);
dequeue.poll(timeout, 111); dequeue.poll(timeout, 111);
EXPECT_TRUE(conditionVar->isWaitForCalled); EXPECT_TRUE(conditionVar->isWaitForCalled);
EXPECT_EQ(timeout, conditionVar->timeout); EXPECT_EQ(timeout, conditionVar->timeout);
@@ -161,7 +161,7 @@ TEST(BlockingDequeueUnitTest, poll_timeouted_is_default_value_when_empty) {
size_t capacity = 1; size_t capacity = 1;
int timeout = 11; int timeout = 11;
auto conditionVar = new MockConditionVar(); auto conditionVar = new MockConditionVar();
PIBlockingDequeue<int> dequeue(capacity, conditionVar); PIBlockingQueue<int> dequeue(capacity, conditionVar);
ASSERT_EQ(dequeue.poll(timeout, 111), 111); ASSERT_EQ(dequeue.poll(timeout, 111), 111);
} }
@@ -169,7 +169,7 @@ TEST(BlockingDequeueUnitTest, poll_timeouted_is_not_block_when_not_empty) {
size_t capacity = 1; size_t capacity = 1;
int timeout = 11; int timeout = 11;
auto conditionVar = new MockConditionVar(); auto conditionVar = new MockConditionVar();
PIBlockingDequeue<int> dequeue(capacity, conditionVar); PIBlockingQueue<int> dequeue(capacity, conditionVar);
dequeue.offer(111); dequeue.offer(111);
dequeue.poll(timeout, -1); dequeue.poll(timeout, -1);
@@ -181,7 +181,7 @@ TEST(BlockingDequeueUnitTest, poll_timeouted_is_offer_value_when_not_empty) {
size_t capacity = 1; size_t capacity = 1;
int timeout = 11; int timeout = 11;
auto conditionVar = new MockConditionVar(); auto conditionVar = new MockConditionVar();
PIBlockingDequeue<int> dequeue(capacity, conditionVar); PIBlockingQueue<int> dequeue(capacity, conditionVar);
dequeue.offer(111); dequeue.offer(111);
ASSERT_EQ(dequeue.poll(timeout, -1), 111); ASSERT_EQ(dequeue.poll(timeout, -1), 111);
} }
@@ -189,7 +189,7 @@ TEST(BlockingDequeueUnitTest, poll_timeouted_is_offer_value_when_not_empty) {
TEST(BlockingDequeueUnitTest, poll_timeouted_is_last) { TEST(BlockingDequeueUnitTest, poll_timeouted_is_last) {
size_t capacity = 10; size_t capacity = 10;
auto conditionVar = new MockConditionVar(); auto conditionVar = new MockConditionVar();
PIBlockingDequeue<int> dequeue(capacity, conditionVar); PIBlockingQueue<int> dequeue(capacity, conditionVar);
dequeue.offer(111); dequeue.offer(111);
dequeue.offer(222); dequeue.offer(222);
ASSERT_EQ(dequeue.poll(10, -1), 111); ASSERT_EQ(dequeue.poll(10, -1), 111);
@@ -198,13 +198,13 @@ TEST(BlockingDequeueUnitTest, poll_timeouted_is_last) {
TEST(BlockingDequeueUnitTest, capacity_is_eq_constructor_capacity) { TEST(BlockingDequeueUnitTest, capacity_is_eq_constructor_capacity) {
size_t capacity = 10; size_t capacity = 10;
PIBlockingDequeue<int> dequeue(capacity); PIBlockingQueue<int> dequeue(capacity);
ASSERT_EQ(dequeue.capacity(), capacity); ASSERT_EQ(dequeue.capacity(), capacity);
} }
TEST(BlockingDequeueUnitTest, remainingCapacity_is_dif_of_capacity_and_size) { TEST(BlockingDequeueUnitTest, remainingCapacity_is_dif_of_capacity_and_size) {
size_t capacity = 2; size_t capacity = 2;
PIBlockingDequeue<int> dequeue(capacity); PIBlockingQueue<int> dequeue(capacity);
ASSERT_EQ(dequeue.remainingCapacity(), capacity); ASSERT_EQ(dequeue.remainingCapacity(), capacity);
dequeue.offer(111); dequeue.offer(111);
ASSERT_EQ(dequeue.remainingCapacity(), capacity - 1); ASSERT_EQ(dequeue.remainingCapacity(), capacity - 1);
@@ -212,7 +212,7 @@ TEST(BlockingDequeueUnitTest, remainingCapacity_is_dif_of_capacity_and_size) {
TEST(BlockingDequeueUnitTest, remainingCapacity_is_zero_when_capacity_reach) { TEST(BlockingDequeueUnitTest, remainingCapacity_is_zero_when_capacity_reach) {
size_t capacity = 1; size_t capacity = 1;
PIBlockingDequeue<int> dequeue(capacity); PIBlockingQueue<int> dequeue(capacity);
dequeue.offer(111); dequeue.offer(111);
dequeue.offer(111); dequeue.offer(111);
ASSERT_EQ(dequeue.remainingCapacity(), 0); ASSERT_EQ(dequeue.remainingCapacity(), 0);
@@ -220,7 +220,7 @@ TEST(BlockingDequeueUnitTest, remainingCapacity_is_zero_when_capacity_reach) {
TEST(BlockingDequeueUnitTest, size_is_eq_to_num_of_elements) { TEST(BlockingDequeueUnitTest, size_is_eq_to_num_of_elements) {
size_t capacity = 1; size_t capacity = 1;
PIBlockingDequeue<int> dequeue(capacity); PIBlockingQueue<int> dequeue(capacity);
ASSERT_EQ(dequeue.size(), 0); ASSERT_EQ(dequeue.size(), 0);
dequeue.offer(111); dequeue.offer(111);
ASSERT_EQ(dequeue.size(), 1); ASSERT_EQ(dequeue.size(), 1);
@@ -228,7 +228,7 @@ TEST(BlockingDequeueUnitTest, size_is_eq_to_num_of_elements) {
TEST(BlockingDequeueUnitTest, size_is_eq_to_capacity_when_capacity_reach) { TEST(BlockingDequeueUnitTest, size_is_eq_to_capacity_when_capacity_reach) {
size_t capacity = 1; size_t capacity = 1;
PIBlockingDequeue<int> dequeue(capacity); PIBlockingQueue<int> dequeue(capacity);
dequeue.offer(111); dequeue.offer(111);
dequeue.offer(111); dequeue.offer(111);
ASSERT_EQ(dequeue.size(), capacity); ASSERT_EQ(dequeue.size(), capacity);
@@ -238,7 +238,7 @@ TEST(BlockingDequeueUnitTest, drainTo_is_elements_moved) {
size_t capacity = 10; size_t capacity = 10;
PIDeque<int> refDeque; PIDeque<int> refDeque;
for (size_t i = 0; i < capacity / 2; ++i) refDeque.push_back(i * 10); for (size_t i = 0; i < capacity / 2; ++i) refDeque.push_back(i * 10);
PIBlockingDequeue<int> blockingDequeue(refDeque); PIBlockingQueue<int> blockingDequeue(refDeque);
PIDeque<int> deque; PIDeque<int> deque;
blockingDequeue.drainTo(deque); blockingDequeue.drainTo(deque);
ASSERT_EQ(blockingDequeue.size(), 0); ASSERT_EQ(blockingDequeue.size(), 0);
@@ -249,7 +249,7 @@ TEST(BlockingDequeueUnitTest, drainTo_is_ret_eq_to_size_when_all_moved) {
size_t capacity = 10; size_t capacity = 10;
PIDeque<int> refDeque; PIDeque<int> refDeque;
for (size_t i = 0; i < capacity / 2; ++i) refDeque.push_back(i * 10); for (size_t i = 0; i < capacity / 2; ++i) refDeque.push_back(i * 10);
PIBlockingDequeue<int> blockingDequeue(refDeque); PIBlockingQueue<int> blockingDequeue(refDeque);
PIDeque<int> deque; PIDeque<int> deque;
ASSERT_EQ(blockingDequeue.drainTo(deque), refDeque.size()); ASSERT_EQ(blockingDequeue.drainTo(deque), refDeque.size());
} }
@@ -258,7 +258,7 @@ TEST(BlockingDequeueUnitTest, drainTo_is_ret_eq_to_maxCount) {
size_t capacity = 10; size_t capacity = 10;
PIDeque<int> refDeque; PIDeque<int> refDeque;
for (size_t i = 0; i < capacity / 2; ++i) refDeque.push_back(i * 10); for (size_t i = 0; i < capacity / 2; ++i) refDeque.push_back(i * 10);
PIBlockingDequeue<int> blockingDequeue(refDeque); PIBlockingQueue<int> blockingDequeue(refDeque);
PIDeque<int> deque; PIDeque<int> deque;
ASSERT_EQ(blockingDequeue.drainTo(deque, refDeque.size() - 1), refDeque.size() - 1); ASSERT_EQ(blockingDequeue.drainTo(deque, refDeque.size() - 1), refDeque.size() - 1);
} }