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

merged concurrent to main library

Browse Source 
removed PIConditionLock, use PIMutex instead
This commit is contained in:
Пелипенко Иван
2020-07-30 18:50:42 +03:00
parent 4dd59132d5
commit 2ffc457566
31 changed files with 558 additions and 2176 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
lib/concurrent/executor.cpp
View File

@@ -1,74 +0,0 @@
/*
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/>.
*/
#include "executor.h"
PIThreadPoolExecutor::PIThreadPoolExecutor(size_t corePoolSize, PIBlockingDequeue<std::function<void()>> *taskQueue_) : isShutdown_(false), taskQueue(taskQueue_) {
for (size_t i = 0; i < corePoolSize; ++i) {
PIThread * thread = new PIThread([&, i](){
auto runnable = taskQueue->poll(100, std::function<void()>());
if (runnable) {
runnable();
}
if (isShutdown_ && taskQueue->size() == 0) threadPool[i]->stop();
});
threadPool.push_back(thread);
thread->start();
}
}
bool PIThreadPoolExecutor::awaitTermination(int timeoutMs) {
PITimeMeasurer measurer;
for (size_t i = 0; i < threadPool.size(); ++i) {
int dif = timeoutMs - (int)measurer.elapsed_m();
if (dif < 0) return false;
if (!threadPool[i]->waitForFinish(dif)) return false;
}
return true;
}
void PIThreadPoolExecutor::shutdownNow() {
isShutdown_ = true;
for (size_t i = 0; i < threadPool.size(); ++i) threadPool[i]->stop();
}
PIThreadPoolExecutor::~PIThreadPoolExecutor() {
shutdownNow();
while (threadPool.size() > 0) delete threadPool.take_back();
delete taskQueue;
}
void PIThreadPoolExecutor::execute(const std::function<void()> &runnable) {
if (!isShutdown_) taskQueue->offer(runnable);
}
bool PIThreadPoolExecutor::isShutdown() const {
return isShutdown_;
}
void PIThreadPoolExecutor::shutdown() {
isShutdown_ = true;
}

95
lib/concurrent/piconditionlock.cpp
View File

@@ -1,95 +0,0 @@
/*
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/>.
*/
#include "piconditionlock.h"
#ifdef WINDOWS
# include <synchapi.h>
#else
# include <pthread.h>
#endif
PRIVATE_DEFINITION_START(PIConditionLock)
#ifdef WINDOWS
CRITICAL_SECTION
#else
pthread_mutex_t
#endif
nativeHandle;
PRIVATE_DEFINITION_END(PIConditionLock)
PIConditionLock::PIConditionLock() {
#ifdef WINDOWS
InitializeCriticalSection(&PRIVATE->nativeHandle);
#else
pthread_mutexattr_t attr;
memset(&attr, 0, sizeof(attr));
pthread_mutexattr_init(&attr);
pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE);
memset(&(PRIVATE->nativeHandle), 0, sizeof(PRIVATE->nativeHandle));
pthread_mutex_init(&(PRIVATE->nativeHandle), &attr);
pthread_mutexattr_destroy(&attr);
#endif
}
PIConditionLock::~PIConditionLock() {
#ifdef WINDOWS
DeleteCriticalSection(&PRIVATE->nativeHandle);
#else
pthread_mutex_destroy(&(PRIVATE->nativeHandle));
#endif
}
void PIConditionLock::lock() {
#ifdef WINDOWS
EnterCriticalSection(&PRIVATE->nativeHandle);
#else
pthread_mutex_lock(&(PRIVATE->nativeHandle));
#endif
}
void PIConditionLock::unlock() {
#ifdef WINDOWS
LeaveCriticalSection(&PRIVATE->nativeHandle);
#else
pthread_mutex_unlock(&(PRIVATE->nativeHandle));
#endif
}
void *PIConditionLock::handle() {
#ifdef WINDOWS
return &PRIVATE->nativeHandle;
#else
return &PRIVATE->nativeHandle;
#endif
}
bool PIConditionLock::tryLock() {
#ifdef WINDOWS
return TryEnterCriticalSection(&PRIVATE->nativeHandle) != 0;
#else
return (pthread_mutex_trylock(&(PRIVATE->nativeHandle)) == 0);
#endif
}

141
lib/concurrent/piconditionvar.cpp
View File

@@ -1,141 +0,0 @@
/*
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/>.
*/
#include "piconditionvar.h"
#include "pithread.h"
#include "pitime.h"
#ifdef WINDOWS
# undef _WIN32_WINNT
# define _WIN32_WINNT 0x0600
# include <synchapi.h>
# include <windef.h>
# include <winbase.h>
#endif
PRIVATE_DEFINITION_START(PIConditionVariable)
#ifdef WINDOWS
CONDITION_VARIABLE nativeHandle;
#else
pthread_cond_t nativeHandle;
PIConditionLock* currentLock;
#endif
bool isDestroying;
PRIVATE_DEFINITION_END(PIConditionVariable)
PIConditionVariable::PIConditionVariable() {
#ifdef WINDOWS
InitializeConditionVariable(&PRIVATE->nativeHandle);
#else
PRIVATE->isDestroying = false;
PRIVATE->currentLock = nullptr;
memset(&(PRIVATE->nativeHandle), 0, sizeof(PRIVATE->nativeHandle));
pthread_cond_init(&PRIVATE->nativeHandle, NULL);
#endif
}
PIConditionVariable::~PIConditionVariable() {
#ifdef WINDOWS
#else
pthread_cond_destroy(&PRIVATE->nativeHandle);
#endif
}
void PIConditionVariable::wait(PIConditionLock& lk) {
#ifdef WINDOWS
SleepConditionVariableCS(&PRIVATE->nativeHandle, (PCRITICAL_SECTION)lk.handle(), INFINITE);
#else
pthread_cond_wait(&PRIVATE->nativeHandle, (pthread_mutex_t*)lk.handle());
#endif
}
void PIConditionVariable::wait(PIConditionLock& lk, const std::function<bool()>& condition) {
bool isCondition;
while (true) {
isCondition = condition();
if (isCondition) break;
#ifdef WINDOWS
SleepConditionVariableCS(&PRIVATE->nativeHandle, (PCRITICAL_SECTION)lk.handle(), INFINITE);
#else
pthread_cond_wait(&PRIVATE->nativeHandle, (pthread_mutex_t*)lk.handle());
#endif
if (PRIVATE->isDestroying) return;
}
}
bool PIConditionVariable::waitFor(PIConditionLock &lk, int timeoutMs) {
bool isNotTimeout;
#ifdef WINDOWS
isNotTimeout = SleepConditionVariableCS(&PRIVATE->nativeHandle, (PCRITICAL_SECTION)lk.handle(), timeoutMs) != 0;
#else
timespec abstime = {.tv_sec = timeoutMs / 1000, .tv_nsec = timeoutMs * 1000 * 1000};
isNotTimeout = pthread_cond_timedwait(&PRIVATE->nativeHandle, (pthread_mutex_t*)lk.handle(), &abstime) == 0;
#endif
if (PRIVATE->isDestroying) return false;
return isNotTimeout;
}
bool PIConditionVariable::waitFor(PIConditionLock& lk, int timeoutMs, const std::function<bool()> &condition) {
bool isCondition;
PITimeMeasurer measurer;
while (true) {
isCondition = condition();
if (isCondition) break;
#ifdef WINDOWS
WINBOOL isTimeout = SleepConditionVariableCS(
&PRIVATE->nativeHandle,
(PCRITICAL_SECTION)lk.handle(),
timeoutMs - (int)measurer.elapsed_m());
if (isTimeout == 0) return false;
#else
int timeoutCurr = timeoutMs - (int)measurer.elapsed_m();
timespec abstime = {.tv_sec = timeoutCurr / 1000, .tv_nsec = timeoutCurr * 1000 * 1000};
bool isTimeout = pthread_cond_timedwait(&PRIVATE->nativeHandle, (pthread_mutex_t*)lk.handle(), &abstime) == 0;
if (isTimeout) return false;
#endif
if (PRIVATE->isDestroying) return false;
}
return true;
}
void PIConditionVariable::notifyOne() {
#ifdef WINDOWS
WakeConditionVariable(&PRIVATE->nativeHandle);
#else
pthread_cond_signal(&PRIVATE->nativeHandle);
#endif
}
void PIConditionVariable::notifyAll() {
#ifdef WINDOWS
WakeAllConditionVariable(&PRIVATE->nativeHandle);
#else
pthread_cond_broadcast(&PRIVATE->nativeHandle);
#endif
}

8
lib/concurrent/test/BlockingDequeueUnitTest.cpp
View File

@@ -8,24 +8,24 @@ public:
bool isTrueCondition = false;
int timeout = -1;
void wait(PIConditionLock& lk) override {
void wait(PIMutex& lk) override {
isWaitCalled = true;
}
void wait(PIConditionLock& lk, const std::function<bool()>& condition) override {
void wait(PIMutex& lk, const std::function<bool()>& condition) override {
isWaitCalled = true;
lk.lock();
isTrueCondition = condition();
lk.unlock();
}
bool waitFor(PIConditionLock& lk, int timeoutMs) override {
bool waitFor(PIMutex& lk, int timeoutMs) override {
isWaitForCalled = true;
timeout = timeoutMs;
return false;
}
bool waitFor(PIConditionLock& lk, int timeoutMs, const std::function<bool()>& condition) override {
bool waitFor(PIMutex& lk, int timeoutMs, const std::function<bool()>& condition) override {
isWaitForCalled = true;
lk.lock();
isTrueCondition = condition();

6
lib/concurrent/test/ConditionLockIntegrationTest.cpp
View File

@@ -2,12 +2,12 @@
#include "gmock/gmock.h"
#include "piconditionvar.h"
#include <pithread.h>
#include "pithread.h"
#include "testutil.h"
class ConditionLock : public ::testing::Test, public TestUtil {
public:
PIConditionLock* m = new PIConditionLock();
PIMutex* m = new PIMutex();
};
TEST_F(ConditionLock, lock_is_protect) {
@@ -50,4 +50,4 @@ TEST_F(ConditionLock, tryLock_is_true_when_unlocked) {
TEST_F(ConditionLock, tryLock_is_recursive_lock_enable) {
m->lock();
ASSERT_TRUE(m->tryLock());
}
}

4
lib/concurrent/test/ConditionVariableIntegrationTest.cpp
View File

@@ -5,7 +5,7 @@
class ConditionVariable : public ::testing::Test, public TestUtil {
public:
PIConditionLock m;
PIMutex m;
PIConditionVariable* variable;
protected:
@@ -197,4 +197,4 @@ TEST_F(ConditionVariable, waitFor_is_unblock_when_condition_and_notifyOne) {
variable->notifyOne();
msleep(WAIT_THREAD_TIME_MS);
ASSERT_FALSE(thread->isRunning());
}
}

4
lib/concurrent/test/ExecutorIntegrationTest.cpp
View File

@@ -1,5 +1,5 @@
#include "gtest/gtest.h"
#include "executor.h"
#include "pithreadpoolexecutor.h"
#include "pimutex.h"
const int WAIT_THREAD_TIME_MS = 30;
@@ -51,4 +51,4 @@ TEST(ExcutorIntegrationTest, execute_is_awaitTermination_wait) {
double waitTime = measurer.elapsed_m();
ASSERT_GE(waitTime, WAIT_THREAD_TIME_MS);
ASSERT_LE(waitTime, 4 * WAIT_THREAD_TIME_MS);
}
}