diff --git a/cpp/platform/impl/windows/BUILD b/cpp/platform/impl/windows/BUILD index 772c90b8..57c774de 100644 --- a/cpp/platform/impl/windows/BUILD +++ b/cpp/platform/impl/windows/BUILD @@ -58,8 +58,11 @@ cc_library( "bluetooth_classic_socket.h", "condition_variable.h", "count_down_latch.h", + "executor.h", "mutex.h", + "runner.h", "server_sync.h", + "thread_pool.h", "webrtc.h", "wifi.h", "wifi_lan.h", @@ -98,8 +101,10 @@ cc_library( "bluetooth_classic_socket.cc", "condition_variable.cc", "count_down_latch.cc", + "executor.cc", "mutex.cc", "platform.cc", + "thread_pool.cc", "utils.cc", ], hdrs = [ @@ -110,7 +115,10 @@ cc_library( "bluetooth_classic_socket.h", "condition_variable.h", "count_down_latch.h", + "executor.h", "mutex.h", + "runner.h", + "thread_pool.h", ], compatible_with = ["//buildenv/target:non_prod"], copts = ["-Ithird_party/nearby_connections/cpp/platform/impl/windows/generated"], @@ -155,6 +163,7 @@ cc_test( "condition_variable_test.cc", "count_down_latch_test.cc", "crypto_test.cc", + "executor_test.cc", "input_file_test.cc", "mutex_test.cc", "output_file_test.cc", diff --git a/cpp/platform/impl/windows/executor.cc b/cpp/platform/impl/windows/executor.cc new file mode 100644 index 00000000..1aebc85f --- /dev/null +++ b/cpp/platform/impl/windows/executor.cc @@ -0,0 +1,80 @@ +// Copyright 2020 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "platform/impl/windows/executor.h" + +#include "platform/impl/windows/generated/winrt/Windows.System.Threading.Core.h" +#include "platform/impl/windows/generated/winrt/Windows.System.Threading.h" +#include "platform/impl/windows/runner.h" +#include "platform/impl/windows/thread_pool.h" +#include "platform/public/logging.h" + +namespace location { +namespace nearby { +namespace windows { + +Executor::Executor() : Executor(1) { + // Call thread pool creator +} + +Executor::Executor(int32_t max_concurrency) + : thread_pool_(std::make_unique(max_concurrency, false)), + executor_state_(ExecutorState::NotReady), + max_concurrency_(max_concurrency) { + if (max_concurrency_ < 1) { + throw(std::invalid_argument("max_concurrency")); + } + + InitializeThreadPool(); +} + +bool Executor::InitializeThreadPool() { + if (executor_state_ != ExecutorState::NotReady) { + // To create a new pool, destroy the existing one first + return false; + } + + thread_pool_->SetPoolSize(max_concurrency_); + thread_pool_->Create(); + executor_state_ = ExecutorState::Ready; + + return true; +} + +// https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executor.html#execute-java.lang.Runnable- +void Executor::Execute(Runnable&& runnable) { + if (shut_down_) { + NEARBY_LOGS(ERROR) << "Error: " << __func__ + << "Attempt to execute on a shut down pool."; + return; + } + + if (runnable == nullptr) { + NEARBY_LOGS(ERROR) << "Error: " << __func__ << "Runnable was null."; + return; + } + + std::unique_ptr runner = std::make_unique(runnable); + thread_pool_->Run(std::move(runner)); +} + +// https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html#shutdown-- +void Executor::Shutdown() { + shut_down_ = true; + thread_pool_ = nullptr; +} + +} // namespace windows +} // namespace nearby +} // namespace location diff --git a/cpp/platform/impl/windows/executor.h b/cpp/platform/impl/windows/executor.h index cff20810..09a1f371 100644 --- a/cpp/platform/impl/windows/executor.h +++ b/cpp/platform/impl/windows/executor.h @@ -15,27 +15,42 @@ #ifndef PLATFORM_IMPL_WINDOWS_EXECUTOR_H_ #define PLATFORM_IMPL_WINDOWS_EXECUTOR_H_ +#include + #include "platform/api/executor.h" +#include "platform/impl/windows/thread_pool.h" namespace location { namespace nearby { namespace windows { +enum class ExecutorState { + Ready, // has been created and initialized + NotReady // Executor has not been initialized +}; // This abstract class is the superclass of all classes representing an // Executor. class Executor : public api::Executor { public: + Executor(); + Executor(int32_t maxConcurrency); + // Before returning from destructor, executor must wait for all pending // jobs to finish. - // TODO(b/184975123): replace with real implementation. - ~Executor() override = default; + ~Executor() override {} // https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executor.html#execute-java.lang.Runnable- - // TODO(b/184975123): replace with real implementation. - void Execute(Runnable&& runnable) override {} + void Execute(Runnable&& runnable) override; // https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html#shutdown-- - // TODO(b/184975123): replace with real implementation. - void Shutdown() override {} + void Shutdown() override; + + private: + bool InitializeThreadPool(); + std::unique_ptr thread_pool_; + + std::atomic shut_down_; + ExecutorState executor_state_; + int32_t max_concurrency_; }; } // namespace windows diff --git a/cpp/platform/impl/windows/executor_test.cc b/cpp/platform/impl/windows/executor_test.cc new file mode 100644 index 00000000..574ec2ba --- /dev/null +++ b/cpp/platform/impl/windows/executor_test.cc @@ -0,0 +1,310 @@ +// Copyright 2020 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "platform/impl/windows/executor.h" + +#include + +#include "gtest/gtest.h" + +TEST(ExecutorTests, SingleThreadedExecutorSucceeds) { + // Arrange + std::string expected("runnable 1"); + + std::unique_ptr executor = + std::make_unique(); + std::string output = std::string(); + // Container to note threads that ran + std::unique_ptr> threadIds = + std::make_unique>(); + + threadIds->push_back(GetCurrentThreadId()); + + // Act + executor->Execute([&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output.append("runnable 1"); + }); + + Sleep(1); // Yield the thread + + // Assert + // We should've run 1 time on the main thread, and 5 times on the + // workerThread + ASSERT_EQ(threadIds->size(), 2); + // We should still be on the main thread + ASSERT_EQ(GetCurrentThreadId(), threadIds->at(0)); + // We should've run all runnables on the worker thread + ASSERT_EQ(output, expected); + + executor->Shutdown(); +} + +TEST(ExecutorTests, SingleThreadedExecutorAfterShutdownFails) { + // Arrange + std::string expected(""); + + std::unique_ptr executor = + std::make_unique(); + std::unique_ptr output = std::make_unique(); + // Container to note threads that ran + std::unique_ptr> threadIds = + std::make_unique>(); + + threadIds->push_back(GetCurrentThreadId()); + executor->Shutdown(); + + // Act + executor->Execute([&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output->append("runnable 1"); + }); + + Sleep(1); // Yield the thread + + // Assert + // We should've run 1 time on the main thread, and 5 times on the + // workerThread + ASSERT_EQ(threadIds->size(), 1); + // We should still be on the main thread + ASSERT_EQ(GetCurrentThreadId(), threadIds->at(0)); + // We should've run all runnables on the worker thread + ASSERT_EQ(*output.get(), expected); +} + +TEST(ExecutorTests, SingleThreadedExecutorExecuteNullSucceeds) { + // Arrange + std::string expected("runnable 1"); + + std::unique_ptr executor = + std::make_unique(); + std::string output = std::string(); + // Container to note threads that ran + std::unique_ptr> threadIds = + std::make_unique>(); + + threadIds->push_back(GetCurrentThreadId()); + + // Act + executor->Execute(nullptr); + executor->Execute([&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output.append("runnable 1"); + }); + executor->Execute(nullptr); + + Sleep(1); // Yield the thread + + // Assert + // We should've run 1 time on the main thread, and 5 times on the + // workerThread + ASSERT_EQ(threadIds->size(), 2); + // We should still be on the main thread + ASSERT_EQ(GetCurrentThreadId(), threadIds->at(0)); + // We should've run all runnables on the worker thread + ASSERT_EQ(output, expected); + + executor->Shutdown(); +} + +TEST(ExecutorTests, SingleThreadedExecutorMultipleTasksSucceeds) { + // Arrange + std::string expected( + "runnable 1, runnable 2, runnable 3, runnable 4, runnable 5"); + + std::unique_ptr executor = + std::make_unique(); + std::string output = std::string(); + // Container to note threads that ran + std::unique_ptr> threadIds = + std::make_unique>(); + + threadIds->push_back(GetCurrentThreadId()); + + // Act + executor->Execute([&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output.append("runnable 1, "); + }); + executor->Execute([&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output.append("runnable 2, "); + }); + executor->Execute([&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output.append("runnable 3, "); + }); + executor->Execute([&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output.append("runnable 4, "); + }); + executor->Execute([&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output.append("runnable 5"); + }); + + Sleep(1); // Yield the thread + + // Assert + // We should've run 1 time on the main thread, and 5 times on the + // workerThread + ASSERT_EQ(threadIds->size(), 6); + // We should still be on the main thread + ASSERT_EQ(GetCurrentThreadId(), threadIds->at(0)); + // We should've run all runnables on the worker thread + auto workerThreadId = threadIds->at(1); + for (int index = 1; index < threadIds->size(); index++) { + ASSERT_EQ(threadIds->at(index), workerThreadId); + } + + // We should of run them in the order submitted + ASSERT_EQ(output, expected); + + executor->Shutdown(); +} + +TEST(ExecutorTests, MultiThreadedExecutorSingleTaskSucceeds) { + // Arrange + std::string expected("runnable 1"); + + std::unique_ptr executor = + std::make_unique(2); + + // Container to note threads that ran + std::unique_ptr> threadIds = + std::make_unique>(); + + std::shared_ptr output = std::make_shared(); + + threadIds->push_back(GetCurrentThreadId()); + + // Act + executor->Execute([output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output->append("runnable 1"); + }); + + Sleep(1); // Yield the processor + + // Assert + // We should've run 1 time on the main thread, and 5 times on the + // workerThread + ASSERT_EQ(threadIds->size(), 2); + // We should still be on the main thread + ASSERT_EQ(GetCurrentThreadId(), threadIds->at(0)); + // We should've run the task + ASSERT_EQ(*output.get(), expected); + + executor->Shutdown(); +} + +TEST(ExecutorTests, MultiThreadedExecutorMultipleTasksSucceeds) { + // Arrange + std::unique_ptr executor = + std::make_unique(2); + + // Container to note threads that ran + std::unique_ptr> threadIds = + std::make_unique>(); + + std::shared_ptr output = std::make_shared(); + + threadIds->push_back(GetCurrentThreadId()); + + // Act + executor->Execute([output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output->append("runnable 1, "); + }); + executor->Execute([output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output->append("runnable 2, "); + }); + executor->Execute([output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output->append("runnable 3, "); + }); + executor->Execute([output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output->append("runnable 4, "); + }); + executor->Execute([output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output->append("runnable 5"); + }); + + Sleep(1); // Yield the processor + + // Assert + // We should've run 1 time on the main thread, and 5 times on the + // workerThread + ASSERT_EQ(threadIds->size(), 6); + // We should still be on the main thread + ASSERT_EQ(GetCurrentThreadId(), threadIds->at(0)); + + executor->Shutdown(); +} + +TEST(ExecutorTests, MultiThreadedExecutorSingleTaskAfterShutdownFails) { + // Arrange + std::string expected(""); + + std::unique_ptr executor = + std::make_unique(2); + + // Container to note threads that ran + std::unique_ptr> threadIds = + std::make_unique>(); + + std::shared_ptr output = std::make_shared(); + + threadIds->push_back(GetCurrentThreadId()); + + executor->Shutdown(); + + // Act + executor->Execute([output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output->append("runnable 1"); + }); + + Sleep(1); // Yield the processor + + // Assert + // We should've run 1 time on the main thread, and 5 times on the + // workerThread + ASSERT_EQ(threadIds->size(), 1); + // We should still be on the main thread + ASSERT_EQ(GetCurrentThreadId(), threadIds->at(0)); + // We should've run the task + ASSERT_EQ(*output.get(), expected); +} + +TEST(ExecutorTests, MultiThreadedExecutorNegativeThreadsThrows) { + // Arrange + // Act + // Assert + EXPECT_THROW( + { + try { + auto result = + std::make_unique(-1); + } catch (const std::invalid_argument::exception& e) { + // and this tests that it has the correct message + EXPECT_STREQ("max_concurrency", e.what()); + throw; + } + }, + std::invalid_argument); +} diff --git a/cpp/platform/impl/windows/runner.h b/cpp/platform/impl/windows/runner.h new file mode 100644 index 00000000..2363d2b7 --- /dev/null +++ b/cpp/platform/impl/windows/runner.h @@ -0,0 +1,42 @@ +// Copyright 2020 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#ifndef PLATFORM_IMPL_WINDOWS_RUNNER_H_ +#define PLATFORM_IMPL_WINDOWS_RUNNER_H_ + +#include "platform/base/runnable.h" + +namespace location { +namespace nearby { +namespace windows { + +class ThreadPool; + +class Runner { + public: + Runner(std::function runnable) + : thread_pool_(nullptr), runnable_(runnable) {} + void Run() { runnable_(); } + ~Runner(){} + ThreadPool* thread_pool_; + + private: + std::function runnable_; +}; + +} // namespace windows +} // namespace nearby +} // namespace location + +#endif // PLATFORM_IMPL_WINDOWS_RUNNER_H_ diff --git a/cpp/platform/impl/windows/thread_pool.cc b/cpp/platform/impl/windows/thread_pool.cc new file mode 100644 index 00000000..5db4f199 --- /dev/null +++ b/cpp/platform/impl/windows/thread_pool.cc @@ -0,0 +1,478 @@ +// Copyright 2020 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "platform/impl/windows/thread_pool.h" + +#include + +#include +#include + +#include "platform/impl/windows/runner.h" +#include "platform/public/logging.h" + +namespace location { +namespace nearby { +namespace windows { + +#define POOL_NAME_BUFFER_SIZE 64 +#define EVENT_NAME_BUFFER_SIZE 64 + +__declspec(align(8)) volatile long ThreadPool::instance_ = // NOLINT + 0; // NOLINT because the Windows function takes a volatile long + +DWORD WINAPI ThreadPool::_ThreadProc(LPVOID pParam) { + DWORD wait; + ThreadPool* pool; + DWORD threadId = GetCurrentThreadId(); + HANDLE waits[2]; + std::unique_ptr runner; + + _ASSERT(pParam != NULL); + if (NULL == pParam) { + return -1; + } + + NEARBY_LOGS(VERBOSE) << "Info: " << __func__ + << ": Starting thread id: " << threadId; + + pool = static_cast(pParam); + waits[0] = pool->GetWaitHandle(threadId); + waits[1] = pool->GetShutdownHandle(); + +loop_here: + wait = WaitForMultipleObjects(2, waits, FALSE, INFINITE); + if (wait == 1) { + if (pool->CheckThreadStop()) { + if (pool->GetWorkingThreadCount() < 1) { + NEARBY_LOGS(VERBOSE) << "Info: " << __func__ + << ": Pool is being destroyed, and working thread " + "count is 0, thread exiting."; + + return 0; + } + } + } + + // a new function was added, go and get it + runner = nullptr; + + NEARBY_LOGS(VERBOSE) << "Info: " << __func__ << ": On thread id: " << threadId + << ", checking for work."; + + if (pool->GetThreadProc(threadId, std::move(runner))) { + pool->BusyNotify(threadId); + runner->Run(); + pool->FinishNotify(threadId); // tell the pool, i am now free + } + + goto loop_here; + + NEARBY_LOGS(VERBOSE) << "Info: " << __func__ + << ": Thread shutdown occurred."; + + return 0; +} + +ThreadPool::ThreadPool(int nPoolSize, bool bCreateNow) + : function_list_(std::make_unique()), + thread_map_(std::make_unique()), + thread_handles_(nullptr), + wait_for_threads_to_die_ms_(500), + notify_shutdown_(nullptr) { + pool_state_ = State::Destroyed; + pool_size_ = nPoolSize; + + InitializeCriticalSection(&critical_section_); + + if (bCreateNow) { + if (!Create()) { + throw ThreadPoolException("ThreadPool creation failed"); + } + } +} + +bool ThreadPool::Create() { + if (pool_state_ != State::Destroyed) { + // To create a new pool, destory the existing one first + return false; + } + + char buffer[POOL_NAME_BUFFER_SIZE]; + snprintf(buffer, POOL_NAME_BUFFER_SIZE, "Pool%d", + (uint32_t)(InterlockedIncrement( + &ThreadPool::instance_))); // InterlockedIncrement done here + // since there's no access to the + // instance_ var except through the + // interlocked functions + + pool_name_ = std::string(buffer); + + HANDLE thread; + DWORD threadId; + ThreadData threadData; + std::string eventName; + + // create the event which will signal the threads to stop + notify_shutdown_ = CreateEvent(NULL, TRUE, FALSE, NULL); + _ASSERT(notify_shutdown_ != NULL); + if (!notify_shutdown_) { + NEARBY_LOGS(ERROR) << "Error: " << __func__ + << ": Failed to create thread shut down event."; + + return false; + } + + thread_handles_ = new HANDLE[pool_size_]; + + // create the threads + for (int nIndex = 0; nIndex < pool_size_; nIndex++) { + char buffer[EVENT_NAME_BUFFER_SIZE]; + + snprintf(buffer, EVENT_NAME_BUFFER_SIZE, "PID:%ld IID:%d TDX:%d", + GetCurrentProcessId(), + (uint32_t)(InterlockedAdd(&ThreadPool::instance_, 0)), nIndex); + + thread = CreateThread(NULL, 0, ThreadPool::_ThreadProc, this, + CREATE_SUSPENDED, &threadId); + + _ASSERT(NULL != thread); + + if (NULL == thread) { + NEARBY_LOGS(ERROR) << "Error: " << __func__ + << ": Failed to create thread."; + return false; + } + + if (thread) { +// EnterCriticalSection(&critical_section_); + // add the entry to the map of threads + threadData.free = true; + threadData.wait_handle = + CreateEventA(NULL, TRUE, FALSE, eventName.c_str()); + + threadData.thread_handle = thread; + threadData.thread_id = threadId; + + EnterCriticalSection(&critical_section_); + + thread_map_->insert(ThreadMap::value_type(threadId, threadData)); + + LeaveCriticalSection(&critical_section_); + + ResumeThread(thread); + + NEARBY_LOGS(VERBOSE) << "Info: " << __func__ + << ": Thread created, handle: " << thread + << ", id: " << threadId; + } else { + NEARBY_LOGS(ERROR) << "Error: " << __func__ + << ": Failed to create thread."; + return false; + } + } + + pool_state_ = State::Ready; + return true; +} + +ThreadPool::~ThreadPool() { + Destroy(); + ReleaseMemory(); + DeleteCriticalSection(&critical_section_); +} + +void ThreadPool::ReleaseMemory() { + // empty all collections + EnterCriticalSection(&critical_section_); + + NEARBY_LOGS(VERBOSE) << "Info: " << __func__ + << ": Clearing the function list."; + + function_list_->clear(); + + NEARBY_LOGS(VERBOSE) << "Info: " << __func__ + << ": Clearing the thread map."; + + thread_map_->clear(); + + LeaveCriticalSection(&critical_section_); +} + +void ThreadPool::Destroy() { + if (pool_state_ == State::Destroying || pool_state_ == State::Destroyed) + return; + + NEARBY_LOGS(VERBOSE) << "Info: " << __func__ << ": Destroying thread pool."; + + pool_state_ = State::Destroying; + + EnterCriticalSection(&critical_section_); + + ThreadMap::iterator iter = thread_map_->begin(); + int index = 0; + + // Build an array of handles + while (iter != thread_map_->end()) { + thread_handles_[index] = iter->second.thread_handle; + index++; + iter++; + } + + LeaveCriticalSection(&critical_section_); + + // tell all threads to shutdown. + _ASSERT(NULL != notify_shutdown_); + SetEvent(GetShutdownHandle()); + + NEARBY_LOGS(VERBOSE) << "Info: " << __func__ + << ": Setting waits for the threads to exit."; + + if (pool_size_ == 1) { + auto wait = WaitForSingleObject(thread_handles_[0], INFINITE); + } else { + auto wait = WaitForMultipleObjects(index, thread_handles_, true, INFINITE); + } + + NEARBY_LOGS(VERBOSE) << "Info: " << __func__ + << ": All threads have exited."; + + delete[] thread_handles_; + + // close the shutdown event + CloseHandle(notify_shutdown_); + notify_shutdown_ = NULL; + + EnterCriticalSection(&critical_section_); + + ThreadMap::iterator threadMapIterator; + + // walk through the events and threads and close them all + for (threadMapIterator = thread_map_->begin(); + threadMapIterator != thread_map_->end(); threadMapIterator++) { + NEARBY_LOGS(VERBOSE) << "Info: " << __func__ << ": Closing thread handle: " + << threadMapIterator->second.thread_handle + << " thread id: " + << threadMapIterator->second.thread_id + << " wait_handle: " + << threadMapIterator->second.wait_handle; + + CloseHandle(threadMapIterator->second.wait_handle); + CloseHandle(threadMapIterator->second.thread_handle); + } + + LeaveCriticalSection(&critical_section_); + + NEARBY_LOGS(VERBOSE) << "Info: " << __func__ + << ": Sending the shutdown event."; + + ReleaseMemory(); // free any remaining UserPoolData objects + + InterlockedDecrement(&ThreadPool::instance_); + + pool_state_ = State::Destroyed; +} + +int ThreadPool::GetPoolSize() { return pool_size_; } + +void ThreadPool::SetPoolSize(int nSize) { + _ASSERT(nSize > 0); + + if (nSize <= 0) { + return; + } + + pool_size_ = nSize; +} + +HANDLE ThreadPool::GetShutdownHandle() { return notify_shutdown_; } + +bool ThreadPool::GetThreadProc(DWORD threadId, + std::unique_ptr&& runner) { + // get the first function info in the function list + FunctionList::iterator functionListIterator; + bool haveAnotherRunner = false; + + EnterCriticalSection(&critical_section_); + + functionListIterator = function_list_->begin(); + + if (functionListIterator != function_list_->end()) { + runner = std::move(*functionListIterator); + + NEARBY_LOGS(VERBOSE) << "Info: " << __func__ + << ": popping runner from the front."; + + function_list_->pop_front(); // remove the function from the list + + haveAnotherRunner = true; + } else { + thread_map_->at(threadId).free = true; + ResetEvent(thread_map_->at(threadId).wait_handle); + } + + LeaveCriticalSection(&critical_section_); + + return haveAnotherRunner; +} + +void ThreadPool::FinishNotify(DWORD threadId) { + ThreadMap::iterator threadMapIterator; + + EnterCriticalSection(&critical_section_); + + threadMapIterator = thread_map_->find(threadId); + + if (threadMapIterator == thread_map_->end()) // if search found no elements + { + _ASSERT(!"No matching thread found."); + } else { + thread_map_->at(threadId).free = true; + + if (!function_list_->empty()) { + // there are some more functions that need servicing, lets do that. + // By not doing anything here we are letting the thread go back and + // check the function list and pick up a function and execute it. + thread_map_->at(threadId).free = false; + } else { + ResetEvent(thread_map_->at(threadId).wait_handle); + } + } + + LeaveCriticalSection(&critical_section_); +} + +void ThreadPool::BusyNotify(DWORD threadId) { + ThreadMap::iterator iter; + + EnterCriticalSection(&critical_section_); + + iter = thread_map_->find(threadId); + + if (iter == thread_map_->end()) // if search found no elements + { + _ASSERT(!"No matching thread found."); + } else { + thread_map_->at(threadId).free = false; + } + + LeaveCriticalSection(&critical_section_); +} + +bool ThreadPool::Run(std::unique_ptr runner) { + if (pool_state_ == State::Destroying || pool_state_ == State::Destroyed) + return false; + + _ASSERT(runner != NULL); + + AddRunner(std::move(runner)); + + // See if any threads are free + ThreadMap::iterator iterator; + ThreadData ThreadData; + + EnterCriticalSection(&critical_section_); + + for (iterator = thread_map_->begin(); iterator != thread_map_->end(); + iterator++) { + ThreadData = (*iterator).second; + + if (ThreadData.free) { + // here is a free thread, put it to work + iterator->second.free = false; + SetEvent(ThreadData.wait_handle); + // this thread will now call GetThreadProc() and pick up the next + // function in the list. + break; + } + } + + LeaveCriticalSection(&critical_section_); + + return true; +} + +void ThreadPool::AddRunner(std::unique_ptr runner) { + // add it to the list + runner->thread_pool_ = this; + + NEARBY_LOGS(VERBOSE) << "Info: " << __func__ + << ": pushing new runner to the back."; + + EnterCriticalSection(&critical_section_); + + function_list_->push_back(std::move(runner)); + + LeaveCriticalSection(&critical_section_); +} + +HANDLE ThreadPool::GetWaitHandle(DWORD dwThreadId) { + HANDLE hWait = NULL; + ThreadMap::iterator iter; + + EnterCriticalSection(&critical_section_); + + iter = thread_map_->find(dwThreadId); + + if (iter != thread_map_->end()) // if search found no elements + { + hWait = thread_map_->at(dwThreadId).wait_handle; + } + + LeaveCriticalSection(&critical_section_); + + return hWait; +} + +bool ThreadPool::CheckThreadStop() { + EnterCriticalSection(&critical_section_); + + bool bRet = + (pool_state_ == State::Destroying || pool_state_ == State::Destroyed); + + LeaveCriticalSection(&critical_section_); + + return bRet; +} + +int ThreadPool::GetWorkingThreadCount() { + ThreadMap::iterator iter; + ThreadData threadData; + + int nCount = 0; + + EnterCriticalSection(&critical_section_); + + for (iter = thread_map_->begin(); iter != thread_map_->end(); iter++) { + threadData = (*iter).second; + + if (function_list_->empty()) { + threadData.free = true; + } + + if (!threadData.free) { + nCount++; + } + } + + LeaveCriticalSection(&critical_section_); + + return nCount; +} + +State ThreadPool::GetState() { return pool_state_; } + +} // namespace windows +} // namespace nearby +} // namespace location diff --git a/cpp/platform/impl/windows/thread_pool.h b/cpp/platform/impl/windows/thread_pool.h new file mode 100644 index 00000000..0eecee25 --- /dev/null +++ b/cpp/platform/impl/windows/thread_pool.h @@ -0,0 +1,97 @@ +// Copyright 2020 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +#ifndef PLATFORM_IMPL_WINDOWS_THREAD_POOL_H_ +#define PLATFORM_IMPL_WINDOWS_THREAD_POOL_H_ +#include + +#include +#include +#include +#include + +#include "platform/impl/windows/runner.h" +namespace location { +namespace nearby { +namespace windows { + +class ThreadPoolException : public std::runtime_error { + public: + ThreadPoolException() : std::runtime_error("") {} + ThreadPoolException(const std::string& message) + : std::runtime_error(message) {} + virtual const char* what() const throw() { + return "ThreadPool creation failed"; + } +}; + +// all functions passed in by clients will be initially stored in this list. +typedef std::list> FunctionList; +// info about threads in the pool will be saved using this struct. +typedef struct tagThreadData { + bool free; + HANDLE wait_handle; + HANDLE thread_handle; + DWORD thread_id; +} ThreadData; +// info about all threads belonging to this pool will be stored in this map +typedef std::map, + std::allocator>> + ThreadMap; +enum class State { + Ready, // has been created + Destroying, // in the process of getting destroyed, no request is processed / + // accepted + Destroyed // Destroyed, no threads are available, request can still be queued +}; +class ThreadPool { + public: + ThreadPool(int nPoolSize, bool bCreateNow); + virtual ~ThreadPool(); + bool Create(); // creates the thread pool + void Destroy(); // destroy the thread pool + int GetPoolSize(); + void SetPoolSize(int); + bool Run(std::unique_ptr runObject); + bool CheckThreadStop(); + int GetWorkingThreadCount(); + State GetState(); + + private: + static DWORD WINAPI _ThreadProc(LPVOID); + std::unique_ptr function_list_; + std::unique_ptr thread_map_; + HANDLE* thread_handles_ = nullptr; + int pool_size_; + int wait_for_threads_to_die_ms_; // In milli-seconds + std::string pool_name_; // To assist in logging and debug + HANDLE notify_shutdown_; // notifies threads that a new function + // is added + volatile State pool_state_; + static __declspec( + align(8)) volatile long instance_; // NOLINT Windows function takes + // volatile long + CRITICAL_SECTION critical_section_; + + bool GetThreadProc(DWORD dwThreadId, std::unique_ptr&& runner); + void FinishNotify(DWORD dwThreadId); + void BusyNotify(DWORD dwThreadId); + void ReleaseMemory(); + HANDLE GetWaitHandle(DWORD dwThreadId); + HANDLE GetShutdownHandle(); + void AddRunner(std::unique_ptr runner); +}; +} // namespace windows +} // namespace nearby +} // namespace location +#endif // PLATFORM_IMPL_WINDOWS_THREAD_POOL_H_