diff --git a/src_concurrent/executor.cpp b/src_concurrent/executor.cpp index 82df5b72..6b537c32 100644 --- a/src_concurrent/executor.cpp +++ b/src_concurrent/executor.cpp @@ -11,7 +11,8 @@ PIThreadPoolExecutor::PIThreadPoolExecutor(size_t corePoolSize, PIBlockingDequeu auto runnable = taskQueue->poll(100, std::function()); if (runnable) { runnable(); - } else if (isShutdown_) threadPool[i]->stop(); + } + if (isShutdown_ && taskQueue->size() == 0) threadPool[i]->stop(); }); threadPool.push_back(thread); thread->start(); @@ -35,7 +36,6 @@ void PIThreadPoolExecutor::shutdownNow() { PIThreadPoolExecutor::~PIThreadPoolExecutor() { shutdownNow(); - taskQueue->getConditionVar()->notifyAll(); while (threadPool.size() > 0) delete threadPool.take_back(); delete threadFactory; delete taskQueue; @@ -48,3 +48,7 @@ void PIThreadPoolExecutor::execute(const std::function &runnable) { volatile bool PIThreadPoolExecutor::isShutdown() const { return isShutdown_; } + +void PIThreadPoolExecutor::shutdown() { + isShutdown_ = true; +} diff --git a/src_concurrent/piconditionvar.cpp b/src_concurrent/piconditionvar.cpp index 450eb460..f11f57b6 100644 --- a/src_concurrent/piconditionvar.cpp +++ b/src_concurrent/piconditionvar.cpp @@ -3,13 +3,12 @@ // #include "piplatform.h" + #ifdef WINDOWS #define _WIN32_WINNT 0x0600 #include "synchapi.h" #include #include -#else - #endif #include "piconditionvar.h" @@ -68,15 +67,15 @@ void PIConditionVariable::wait(PIConditionLock& lk, const std::function& } bool PIConditionVariable::waitFor(PIConditionLock &lk, int timeoutMs) { - bool isTimeout; + bool isNotTimeout; #ifdef WINDOWS - isTimeout = SleepConditionVariableCS(&PRIVATE->nativeHandle, (PCRITICAL_SECTION)lk.handle(), timeoutMs) != 0; + isNotTimeout = SleepConditionVariableCS(&PRIVATE->nativeHandle, (PCRITICAL_SECTION)lk.handle(), timeoutMs) != 0; #else timespec abstime = {.tv_sec = timeoutMs / 1000, .tv_nsec = timeoutMs * 1000 * 1000}; - isTimeout = pthread_cond_timedwait(&PRIVATE->nativeHandle, (pthread_mutex_t*)lk.handle(), &abstime) == 0; + isNotTimeout = pthread_cond_timedwait(&PRIVATE->nativeHandle, (pthread_mutex_t*)lk.handle(), &abstime) == 0; #endif if (PRIVATE->isDestroying) return false; - return isTimeout; + return isNotTimeout; } bool PIConditionVariable::waitFor(PIConditionLock& lk, int timeoutMs, const std::function &condition) { diff --git a/src_concurrent/test/ExecutorIntegrationTest.cpp b/src_concurrent/test/ExecutorIntegrationTest.cpp index 843a5fe9..6fadad1e 100644 --- a/src_concurrent/test/ExecutorIntegrationTest.cpp +++ b/src_concurrent/test/ExecutorIntegrationTest.cpp @@ -31,3 +31,28 @@ TEST(ExcutorIntegrationTest, execute_is_not_execute_after_shutdown) { piMSleep(WAIT_THREAD_TIME_MS); ASSERT_FALSE(isRunnableInvoke); } + +TEST(ExcutorIntegrationTest, execute_is_execute_before_shutdown) { + bool isRunnableInvoke = false; + PIThreadPoolExecutor executorService(1); + executorService.execute([&]() { + piMSleep(WAIT_THREAD_TIME_MS); + isRunnableInvoke = true; + }); + executorService.shutdown(); + piMSleep(2 * WAIT_THREAD_TIME_MS); + ASSERT_TRUE(isRunnableInvoke); +} + +TEST(ExcutorIntegrationTest, execute_is_awaitTermination_wait) { + PIThreadPoolExecutor executorService(1); + executorService.execute([&]() { + piMSleep(2 * WAIT_THREAD_TIME_MS); + }); + executorService.shutdown(); + PITimeMeasurer measurer; + ASSERT_TRUE(executorService.awaitTermination(4 * WAIT_THREAD_TIME_MS)); + double waitTime = measurer.elapsed_m(); + ASSERT_GE(waitTime, WAIT_THREAD_TIME_MS); + ASSERT_LE(waitTime, 3 * WAIT_THREAD_TIME_MS); +} \ No newline at end of file diff --git a/src_main/concurrent/executor.h b/src_main/concurrent/executor.h index 71742144..7095172f 100644 --- a/src_main/concurrent/executor.h +++ b/src_main/concurrent/executor.h @@ -72,9 +72,7 @@ public: * accepted. Invocation has no additional effect if already shut down. This method does not wait for previously * submitted tasks to complete execution. Use awaitTermination to do that. */ - void shutdown() { - isShutdown_ = true; - } + void shutdown(); volatile bool isShutdown() const; diff --git a/src_main/concurrent/piblockingdequeue.h b/src_main/concurrent/piblockingdequeue.h index c1b50a35..e0611e2b 100644 --- a/src_main/concurrent/piblockingdequeue.h +++ b/src_main/concurrent/piblockingdequeue.h @@ -15,19 +15,38 @@ template class PIBlockingDequeue: private PIDeque { 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& other) : condition_var(new PIConditionVariable()) { + 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& other) : cond_var_add(new PIConditionVariable()), cond_var_rem(new PIConditionVariable()) { max_size = SIZE_MAX; PIDeque::append(other); } - inline PIBlockingDequeue(PIBlockingDequeue & other) : condition_var(new PIConditionVariable()) { + inline PIBlockingDequeue(PIBlockingDequeue & other) : cond_var_add(new PIConditionVariable()), cond_var_rem(new PIConditionVariable()) { other.mutex.lock(); max_size = other.max_size; PIDeque::append(static_cast&>(other)); other.mutex.unlock(); } virtual ~PIBlockingDequeue() { - delete condition_var; + 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::size() < max_size; }); + PIDeque::push_back(v); + mutex.unlock(); + cond_var_add->notifyOne(); } /** @@ -45,7 +64,7 @@ public: } PIDeque::push_back(v); mutex.unlock(); - condition_var->notifyOne(); + cond_var_add->notifyOne(); return true; } @@ -57,9 +76,10 @@ public: virtual T take() { T t; mutex.lock(); - condition_var->wait(mutex, [&]() { return !PIDeque::isEmpty(); }); + cond_var_add->wait(mutex, [&]() { return !PIDeque::isEmpty(); }); t = T(PIDeque::take_front()); mutex.unlock(); + cond_var_rem->notifyOne(); return t; } @@ -74,9 +94,10 @@ public: virtual T poll(int timeoutMs, const T & defaultVal) { T t; mutex.lock(); - bool isOk = condition_var->waitFor(mutex, timeoutMs, [&]() { return !PIDeque::isEmpty(); }); + bool isOk = cond_var_add->waitFor(mutex, timeoutMs, [&]() { return !PIDeque::isEmpty(); }); t = isOk ? T(PIDeque::take_front()) : defaultVal; mutex.unlock(); + if (isOk) cond_var_rem->notifyOne(); return t; } @@ -134,13 +155,10 @@ public: return count; } - PIConditionVariable *getConditionVar() const { - return condition_var; - } - private: PIConditionLock mutex; - PIConditionVariable* condition_var; + PIConditionVariable* cond_var_add; + PIConditionVariable* cond_var_rem; size_t max_size; }; diff --git a/src_main/concurrent/piconditionlock.h b/src_main/concurrent/piconditionlock.h index 722885d7..158a81cb 100644 --- a/src_main/concurrent/piconditionlock.h +++ b/src_main/concurrent/piconditionlock.h @@ -8,10 +8,13 @@ #include #include +/** + * @brief Continued + */ class PIP_EXPORT PIConditionLock { public: explicit PIConditionLock(); - ~PIConditionLock(); + virtual ~PIConditionLock(); void lock(); void unlock(); diff --git a/src_main/concurrent/piconditionvar.h b/src_main/concurrent/piconditionvar.h index b87aacc0..58fc5e46 100644 --- a/src_main/concurrent/piconditionvar.h +++ b/src_main/concurrent/piconditionvar.h @@ -9,15 +9,29 @@ #include #include "piinit.h" -#define PICONDITION_RELIABLE true - +/** + * @brief A condition variable is an object able to block the calling thread until notified to resume. + * + * It uses a PIConditionLock to lock the thread when one of its wait functions is called. The thread remains + * blocked until woken up by another thread that calls a notification function on the same PIConditionVariable object. + */ class PIP_EXPORT PIConditionVariable { public: explicit PIConditionVariable(); virtual ~PIConditionVariable(); + /** + * @brief Unblocks one of the threads currently waiting for this condition. If no threads are waiting, the function + * does nothing. If more than one, it is unspecified which of the threads is selected. + */ virtual void notifyOne(); + + /** + * @brief Unblocks all threads currently waiting for this condition. If no threads are waiting, the function does + * nothing. + */ virtual void notifyAll(); + virtual void wait(PIConditionLock& lk); virtual void wait(PIConditionLock& lk, const std::function& condition); virtual bool waitFor(PIConditionLock& lk, int timeoutMs);