From 2b3fb246163afceec2aa93dd60e3521403a0a010 Mon Sep 17 00:00:00 2001 From: Alon Zakai Date: Thu, 25 Jan 2018 11:29:19 -0800 Subject: [PATCH 1/6] proper locking (hopefully) --- src/support/threads.cpp | 24 +++++++++++++++--------- src/support/threads.h | 9 ++++++++- 2 files changed, 23 insertions(+), 10 deletions(-) diff --git a/src/support/threads.cpp b/src/support/threads.cpp index c4f714f8e61..28d289d1ded 100644 --- a/src/support/threads.cpp +++ b/src/support/threads.cpp @@ -39,11 +39,6 @@ static std::mutex debug; namespace wasm { -// Global thread information - -static std::mutex poolMutex; -static std::unique_ptr pool; - // Thread @@ -102,13 +97,21 @@ void Thread::mainLoop(void *self_) { } } +// Global threadPool state. We have a singleton pool, which can only be +// used from one place at a time. + +static std::unique_ptr pool; + +std::mutex ThreadPool::creationMutex; +std::mutex ThreadPool::workMutex; +std::mutex ThreadPool::threadMutex; // ThreadPool void ThreadPool::initialize(size_t num) { if (num == 1) return; // no multiple cores, don't create threads DEBUG_POOL("initialize()\n"); - std::unique_lock lock(mutex); + std::unique_lock lock(threadMutex); ready.store(threads.size()); // initial state before first resetThreadsAreReady() resetThreadsAreReady(); for (size_t i = 0; i < num; i++) { @@ -143,7 +146,7 @@ ThreadPool* ThreadPool::get() { bool created = false; { // lock on the creation - std::lock_guard lock(poolMutex); + std::lock_guard poolLock(creationMutex); if (!pool) { DEBUG_POOL("::get() creating\n"); created = true; @@ -173,10 +176,13 @@ void ThreadPool::work(std::vector>& doWorkers) // run in parallel on threads // TODO: fancy work stealing DEBUG_POOL("work() on threads\n"); + // lock globally on doing work in the pool - the threadPool can only be used + // from one thread at a time, all others must wait patiently + std::lock_guard poolLock(workMutex); assert(doWorkers.size() == num); assert(!running); running = true; - std::unique_lock lock(mutex); + std::unique_lock lock(threadMutex); resetThreadsAreReady(); for (size_t i = 0; i < num; i++) { threads[i]->work(doWorkers[i]); @@ -199,7 +205,7 @@ bool ThreadPool::isRunning() { void ThreadPool::notifyThreadIsReady() { DEBUG_POOL("notify thread is ready\n";) - std::lock_guard lock(mutex); + std::lock_guard lock(threadMutex); ready.fetch_add(1); condition.notify_one(); } diff --git a/src/support/threads.h b/src/support/threads.h index 0ec109e4d66..38af448a17a 100644 --- a/src/support/threads.h +++ b/src/support/threads.h @@ -72,10 +72,17 @@ class Thread { class ThreadPool { std::vector> threads; bool running = false; - std::mutex mutex; std::condition_variable condition; std::atomic ready; + // A mutex for creating the pool safely + static std::mutex creationMutex; + // A mutex for work() so that the pool can only work on one + // thing at a time + static std::mutex workMutex; + // A mutex for communication with the worker threads + static std::mutex threadMutex; + private: void initialize(size_t num); From aa08c7528ea993d8e6c0b1be2fcf4a4660908ca7 Mon Sep 17 00:00:00 2001 From: Alon Zakai Date: Thu, 25 Jan 2018 11:42:46 -0800 Subject: [PATCH 2/6] wip [ci skip] --- auto_update_tests.py | 2 ++ check.py | 2 ++ test/example/cpp-threads.cpp | 58 ++++++++++++++++++++++++++++++++++++ 3 files changed, 62 insertions(+) create mode 100644 test/example/cpp-threads.cpp diff --git a/auto_update_tests.py b/auto_update_tests.py index 10481cab51f..050dab1d1af 100755 --- a/auto_update_tests.py +++ b/auto_update_tests.py @@ -184,6 +184,8 @@ src, '-c', '-o', 'example.o', '-Isrc', '-g', '-L' + libdir, '-pthread'] print 'build: ', ' '.join(extra) + if src.endswith('.cpp'): + extra += ['-std=c++11'] print os.getcwd() subprocess.check_call(extra) # Link against the binaryen C library DSO, using rpath diff --git a/check.py b/check.py index 5a926bf8e02..bfacdcb312f 100755 --- a/check.py +++ b/check.py @@ -511,6 +511,8 @@ def run_gcc_torture_tests(): cmd = ['example.o', '-lbinaryen'] + cmd + ['-Wl,-rpath=$ORIGIN/../lib'] else: continue + if src.endswith('.cpp'): + extra += ['-std=c++11'] print ' ', t, src, expected if os.environ.get('COMPILER_FLAGS'): for f in os.environ.get('COMPILER_FLAGS').split(' '): diff --git a/test/example/cpp-threads.cpp b/test/example/cpp-threads.cpp new file mode 100644 index 00000000000..4a41249e783 --- /dev/null +++ b/test/example/cpp-threads.cpp @@ -0,0 +1,58 @@ +// test multiple uses of the threadPool + +#include +#include +#include + +#include + +int NUM_THREADS = 33; + +void worker() { + BinaryenModuleRef module = BinaryenModuleCreate(); + + // Create a function type for i32 (i32, i32) + BinaryenType params[2] = { BinaryenTypeInt32(), BinaryenTypeInt32() }; + BinaryenFunctionTypeRef iii = BinaryenAddFunctionType(module, "iii", BinaryenTypeInt32(), params, 2); + + // Get the 0 and 1 arguments, and add them + BinaryenExpressionRef x = BinaryenGetLocal(module, 0, BinaryenTypeInt32()), + y = BinaryenGetLocal(module, 1, BinaryenTypeInt32()); + BinaryenExpressionRef add = BinaryenBinary(module, BinaryenAddInt32(), x, y); + BinaryenExpressionRef ret = BinaryenReturn(module, add); + + // Create the add function + // Note: no additional local variables + // Note: no basic blocks here, we are an AST. The function body is just an expression node. + BinaryenFunctionRef adder = BinaryenAddFunction(module, "adder", iii, NULL, 0, ret); + + // validate it + BinaryenModuleValidate(module); + + // optimize it + BinaryenModuleOptimize(module); + BinaryenModuleValidate(module); + + // Clean up the module, which owns all the objects we created above + BinaryenModuleDispose(module); +} + +int main() +{ + std::vector threads; + + std::cout << "create threads...\n"; + for (int i = 0; i < NUM_THREADS; i++) { + threads.emplace_back(worker); + } + std::cout << "threads running in parallel...\n"; + + std::cout << "waiting for threads to join...\n"; + for (auto& thread : threads) { + thread.join(); + } + + std::cout << "all done.\n"; + + return 0; +} From 65d2f6b9a2be69477b54b53036edbe8a54a188f1 Mon Sep 17 00:00:00 2001 From: Alon Zakai Date: Thu, 25 Jan 2018 12:47:57 -0800 Subject: [PATCH 3/6] fix with test --- src/support/threads.cpp | 71 ++++++++++++++++++++++-------------- test/example/cpp-threads.txt | 4 ++ 2 files changed, 48 insertions(+), 27 deletions(-) create mode 100644 test/example/cpp-threads.txt diff --git a/src/support/threads.cpp b/src/support/threads.cpp index 28d289d1ded..9411a8b023d 100644 --- a/src/support/threads.cpp +++ b/src/support/threads.cpp @@ -39,16 +39,43 @@ static std::mutex debug; namespace wasm { +// Global threadPool state. We have a singleton pool, which can only be +// used from one place at a time. + +static std::unique_ptr pool; + +std::mutex ThreadPool::creationMutex; +std::mutex ThreadPool::workMutex; +std::mutex ThreadPool::threadMutex; + +// During the creation of the singleton threadPool, the worker threads +// communicate with it using this pointer. +static ThreadPool* poolDuringCreation = nullptr; + +// Gets the threadPool from a worker. If this is during creation, we +// use poolDuringCreation +static ThreadPool* getPoolFromWorker() { + if (poolDuringCreation) { + DEBUG_THREAD("getPoolFromWorker: during creation\n"); + assert(!pool); + return poolDuringCreation; + } else { + DEBUG_THREAD("getPoolFromWorker: after creation\n"); + assert(!poolDuringCreation); + assert(pool); + return pool->get(); + } +} // Thread Thread::Thread() { - assert(!ThreadPool::isRunning()); + assert(!getPoolFromWorker()->isRunning()); thread = make_unique(mainLoop, this); } Thread::~Thread() { - assert(!ThreadPool::isRunning()); + assert(!getPoolFromWorker()->isRunning()); { std::lock_guard lock(mutex); // notify the thread that it can exit @@ -86,7 +113,7 @@ void Thread::mainLoop(void *self_) { return; } } - ThreadPool::get()->notifyThreadIsReady(); + getPoolFromWorker()->notifyThreadIsReady(); { std::unique_lock lock(self->mutex); if (!self->done && !self->doWork) { @@ -97,15 +124,6 @@ void Thread::mainLoop(void *self_) { } } -// Global threadPool state. We have a singleton pool, which can only be -// used from one place at a time. - -static std::unique_ptr pool; - -std::mutex ThreadPool::creationMutex; -std::mutex ThreadPool::workMutex; -std::mutex ThreadPool::threadMutex; - // ThreadPool void ThreadPool::initialize(size_t num) { @@ -143,21 +161,18 @@ size_t ThreadPool::getNumCores() { ThreadPool* ThreadPool::get() { DEBUG_POOL("::get()\n"); - bool created = false; - { - // lock on the creation - std::lock_guard poolLock(creationMutex); - if (!pool) { - DEBUG_POOL("::get() creating\n"); - created = true; - pool = make_unique(); - } - } - if (created) { - // if we created it here, do the initialization too. this - // is outside of the mutex, as we create child threads who - // will call ::get() themselves - pool->initialize(getNumCores()); + // lock on the creation + std::lock_guard poolLock(creationMutex); + if (!pool) { + DEBUG_POOL("::get() creating\n"); + std::unique_ptr temp = make_unique(); + // during creation, the workers need to report to it, before + // we set the global pool + poolDuringCreation = temp.get(); + temp->initialize(getNumCores()); + poolDuringCreation = nullptr; + // assign it to the global location now that it is all ready + pool.swap(temp); DEBUG_POOL("::get() created\n"); } return pool.get(); @@ -181,6 +196,7 @@ void ThreadPool::work(std::vector>& doWorkers) std::lock_guard poolLock(workMutex); assert(doWorkers.size() == num); assert(!running); + DEBUG_POOL("running = true\n"); running = true; std::unique_lock lock(threadMutex); resetThreadsAreReady(); @@ -190,6 +206,7 @@ void ThreadPool::work(std::vector>& doWorkers) DEBUG_POOL("main thread waiting\n"); condition.wait(lock, [this]() { return areThreadsReady(); }); DEBUG_POOL("main thread waiting\n"); + DEBUG_POOL("running = false\n"); running = false; DEBUG_POOL("work() is done\n"); } diff --git a/test/example/cpp-threads.txt b/test/example/cpp-threads.txt new file mode 100644 index 00000000000..2c638aaabd5 --- /dev/null +++ b/test/example/cpp-threads.txt @@ -0,0 +1,4 @@ +create threads... +threads running in parallel... +waiting for threads to join... +all done. From 19b1bf078aaa42751f1685596e50ae19a4abfab4 Mon Sep 17 00:00:00 2001 From: Alon Zakai Date: Thu, 25 Jan 2018 12:49:35 -0800 Subject: [PATCH 4/6] fix --- check.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/check.py b/check.py index bfacdcb312f..d3ed0a47891 100755 --- a/check.py +++ b/check.py @@ -501,6 +501,8 @@ def run_gcc_torture_tests(): else: src = os.path.join(options.binaryen_test, 'example', t) expected = os.path.join(options.binaryen_test, 'example', '.'.join(t.split('.')[:-1]) + '.txt') + if src.endswith('.cpp'): + extra += ['-std=c++11'] if src.endswith(('.c', '.cpp')): # build the C file separately extra = [NATIVECC, src, '-c', '-o', 'example.o', @@ -511,8 +513,6 @@ def run_gcc_torture_tests(): cmd = ['example.o', '-lbinaryen'] + cmd + ['-Wl,-rpath=$ORIGIN/../lib'] else: continue - if src.endswith('.cpp'): - extra += ['-std=c++11'] print ' ', t, src, expected if os.environ.get('COMPILER_FLAGS'): for f in os.environ.get('COMPILER_FLAGS').split(' '): From f09cd34fd28c25f6d8ef000aa93ab05611900a9a Mon Sep 17 00:00:00 2001 From: Alon Zakai Date: Thu, 25 Jan 2018 12:52:15 -0800 Subject: [PATCH 5/6] fix --- check.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/check.py b/check.py index d3ed0a47891..40092d93a17 100755 --- a/check.py +++ b/check.py @@ -501,12 +501,12 @@ def run_gcc_torture_tests(): else: src = os.path.join(options.binaryen_test, 'example', t) expected = os.path.join(options.binaryen_test, 'example', '.'.join(t.split('.')[:-1]) + '.txt') - if src.endswith('.cpp'): - extra += ['-std=c++11'] if src.endswith(('.c', '.cpp')): # build the C file separately extra = [NATIVECC, src, '-c', '-o', 'example.o', '-I' + os.path.join(options.binaryen_root, 'src'), '-g', '-L' + os.path.join(options.binaryen_bin, '..', 'lib'), '-pthread'] + if src.endswith('.cpp'): + extra += ['-std=c++11'] print 'build: ', ' '.join(extra) subprocess.check_call(extra) # Link against the binaryen C library DSO, using an executable-relative rpath From 61bf62ea67a7cf7289696d683646426d98e4a74b Mon Sep 17 00:00:00 2001 From: Alon Zakai Date: Fri, 26 Jan 2018 09:37:57 -0800 Subject: [PATCH 6/6] refactor threads code - give each worker a pointer to the parent, to avoid needing them to go through a global location (which would require some complexity to handle right) --- src/support/threads.cpp | 53 ++++++++++++----------------------------- src/support/threads.h | 5 +++- 2 files changed, 19 insertions(+), 39 deletions(-) diff --git a/src/support/threads.cpp b/src/support/threads.cpp index 9411a8b023d..ce880ac2d75 100644 --- a/src/support/threads.cpp +++ b/src/support/threads.cpp @@ -30,7 +30,7 @@ #ifdef BINARYEN_THREAD_DEBUG static std::mutex debug; #define DEBUG_THREAD(x) { std::lock_guard lock(debug); std::cerr << "[THREAD " << std::this_thread::get_id() << "] " << x; } -#define DEBUG_POOL(x) { std::lock_guard lock(debug); std::cerr << "[POOL] " << x; } +#define DEBUG_POOL(x) { std::lock_guard lock(debug); std::cerr << "[POOL " << std::this_thread::get_id() << "] " << x; } #else #define DEBUG_THREAD(x) #define DEBUG_POOL(x) @@ -39,43 +39,15 @@ static std::mutex debug; namespace wasm { -// Global threadPool state. We have a singleton pool, which can only be -// used from one place at a time. - -static std::unique_ptr pool; - -std::mutex ThreadPool::creationMutex; -std::mutex ThreadPool::workMutex; -std::mutex ThreadPool::threadMutex; - -// During the creation of the singleton threadPool, the worker threads -// communicate with it using this pointer. -static ThreadPool* poolDuringCreation = nullptr; - -// Gets the threadPool from a worker. If this is during creation, we -// use poolDuringCreation -static ThreadPool* getPoolFromWorker() { - if (poolDuringCreation) { - DEBUG_THREAD("getPoolFromWorker: during creation\n"); - assert(!pool); - return poolDuringCreation; - } else { - DEBUG_THREAD("getPoolFromWorker: after creation\n"); - assert(!poolDuringCreation); - assert(pool); - return pool->get(); - } -} - // Thread -Thread::Thread() { - assert(!getPoolFromWorker()->isRunning()); +Thread::Thread(ThreadPool* parent) : parent(parent) { + assert(!parent->isRunning()); thread = make_unique(mainLoop, this); } Thread::~Thread() { - assert(!getPoolFromWorker()->isRunning()); + assert(!parent->isRunning()); { std::lock_guard lock(mutex); // notify the thread that it can exit @@ -113,7 +85,7 @@ void Thread::mainLoop(void *self_) { return; } } - getPoolFromWorker()->notifyThreadIsReady(); + self->parent->notifyThreadIsReady(); { std::unique_lock lock(self->mutex); if (!self->done && !self->doWork) { @@ -126,6 +98,15 @@ void Thread::mainLoop(void *self_) { // ThreadPool +// Global threadPool state. We have a singleton pool, which can only be +// used from one place at a time. + +static std::unique_ptr pool; + +std::mutex ThreadPool::creationMutex; +std::mutex ThreadPool::workMutex; +std::mutex ThreadPool::threadMutex; + void ThreadPool::initialize(size_t num) { if (num == 1) return; // no multiple cores, don't create threads DEBUG_POOL("initialize()\n"); @@ -134,7 +115,7 @@ void ThreadPool::initialize(size_t num) { resetThreadsAreReady(); for (size_t i = 0; i < num; i++) { try { - threads.emplace_back(make_unique()); + threads.emplace_back(make_unique(this)); } catch (std::system_error&) { // failed to create a thread - don't use multithreading, as if num cores == 1 DEBUG_POOL("could not create thread\n"); @@ -166,11 +147,7 @@ ThreadPool* ThreadPool::get() { if (!pool) { DEBUG_POOL("::get() creating\n"); std::unique_ptr temp = make_unique(); - // during creation, the workers need to report to it, before - // we set the global pool - poolDuringCreation = temp.get(); temp->initialize(getNumCores()); - poolDuringCreation = nullptr; // assign it to the global location now that it is all ready pool.swap(temp); DEBUG_POOL("::get() created\n"); diff --git a/src/support/threads.h b/src/support/threads.h index 38af448a17a..280e1947071 100644 --- a/src/support/threads.h +++ b/src/support/threads.h @@ -38,6 +38,8 @@ enum class ThreadWorkState { Finished }; +class ThreadPool; + // // A helper thread. // @@ -45,6 +47,7 @@ enum class ThreadWorkState { // class Thread { + ThreadPool* parent; std::unique_ptr thread; std::mutex mutex; std::condition_variable condition; @@ -52,7 +55,7 @@ class Thread { std::function doWork = nullptr; public: - Thread(); + Thread(ThreadPool* parent); ~Thread(); // Start to do work, calling doWork() until