diff --git a/internal/platform/implementation/windows/BUILD b/internal/platform/implementation/windows/BUILD index 8f6969b0..0e90af77 100644 --- a/internal/platform/implementation/windows/BUILD +++ b/internal/platform/implementation/windows/BUILD @@ -58,7 +58,6 @@ cc_library( "condition_variable.h", "executor.h", "mutex.h", - "runner.h", "scheduled_executor.h", "server_sync.h", "submittable_executor.h", @@ -174,6 +173,7 @@ cc_test( "platform_test.cc", "scheduled_executor_test.cc", "submittable_executor_test.cc", + "thread_pool_test.cc", "utils_test.cc", ], copts = ["-Ithird_party/nearby/internal/platform/implementation/windows/generated -DCORE_ADAPTER_DLL"], diff --git a/internal/platform/implementation/windows/crypto_test.cc b/internal/platform/implementation/windows/crypto_test.cc index 55f65dc3..5d182810 100644 --- a/internal/platform/implementation/windows/crypto_test.cc +++ b/internal/platform/implementation/windows/crypto_test.cc @@ -14,6 +14,8 @@ #include "internal/platform/implementation/crypto.h" +#include + #include "gtest/gtest.h" namespace location { diff --git a/internal/platform/implementation/windows/executor.cc b/internal/platform/implementation/windows/executor.cc index e5b3e046..2a26e8a0 100644 --- a/internal/platform/implementation/windows/executor.cc +++ b/internal/platform/implementation/windows/executor.cc @@ -14,64 +14,41 @@ #include "internal/platform/implementation/windows/executor.h" -#include "internal/platform/implementation/windows/generated/winrt/Windows.System.Threading.Core.h" -#include "internal/platform/implementation/windows/generated/winrt/Windows.System.Threading.h" -#include "internal/platform/implementation/windows/runner.h" -#include "internal/platform/implementation/windows/thread_pool.h" +#include + #include "internal/platform/logging.h" namespace location { namespace nearby { namespace windows { -Executor::Executor() : Executor(1) { - // Call thread pool creator -} +Executor::Executor() : Executor(1) {} 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(); + : max_concurrency_(max_concurrency) { + assert(max_concurrency_ >= 1); + thread_pool_ = ThreadPool::Create(max_concurrency); + assert(thread_pool_ != nullptr); } -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(VERBOSE) << "Warning: " << __func__ - << ": Attempt to execute on a shut down pool."; + << ": Attempt to execute on a shut down pool."; return; } if (runnable == nullptr) { - NEARBY_LOGS(VERBOSE) << "Error: " << __func__ << "Runnable was null."; + NEARBY_LOGS(ERROR) << __func__ << ": Runnable was null."; return; } - std::unique_ptr runner = std::make_unique(runnable); - thread_pool_->Run(std::move(runner)); + thread_pool_->Run(std::move(runnable)); } -// https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html#shutdown-- void Executor::Shutdown() { shut_down_ = true; + thread_pool_->ShutDown(); thread_pool_ = nullptr; } diff --git a/internal/platform/implementation/windows/executor.h b/internal/platform/implementation/windows/executor.h index 14472eec..9a87550e 100644 --- a/internal/platform/implementation/windows/executor.h +++ b/internal/platform/implementation/windows/executor.h @@ -23,33 +23,24 @@ 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); + explicit Executor(int max_concurrency); // Before returning from destructor, executor must wait for all pending // jobs to finish. ~Executor() override {} - // https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executor.html#execute-java.lang.Runnable- - void Execute(Runnable&& runnable) override; - // https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html#shutdown-- + void Execute(Runnable&& runnable) override; void Shutdown() override; private: - bool InitializeThreadPool(); - std::unique_ptr thread_pool_; - + std::unique_ptr thread_pool_ = nullptr; std::atomic shut_down_; - ExecutorState executor_state_; int32_t max_concurrency_; }; diff --git a/internal/platform/implementation/windows/executor_test.cc b/internal/platform/implementation/windows/executor_test.cc index 585e5a93..b5db8a74 100644 --- a/internal/platform/implementation/windows/executor_test.cc +++ b/internal/platform/implementation/windows/executor_test.cc @@ -17,16 +17,26 @@ #include #include +#include "gtest/gtest.h" +#include "absl/synchronization/blocking_counter.h" +#include "absl/synchronization/mutex.h" +#include "absl/synchronization/notification.h" +#include "absl/time/time.h" #include "internal/platform/implementation/windows/test_data.h" -#include "gtest/gtest.h" +namespace location { +namespace nearby { +namespace windows { +namespace { + +constexpr absl::Duration kWaitTimeout = absl::Milliseconds(200); TEST(ExecutorTests, SingleThreadedExecutorSucceeds) { + absl::Notification notification; // Arrange std::string expected(RUNNABLE_0_TEXT.c_str()); - std::unique_ptr executor = - std::make_unique(); + auto executor = std::make_unique(); std::string output = std::string(); // Container to note threads that ran std::unique_ptr> threadIds = @@ -35,11 +45,13 @@ TEST(ExecutorTests, SingleThreadedExecutorSucceeds) { threadIds->push_back(GetCurrentThreadId()); // Act - executor->Execute([&output, &threadIds]() { + executor->Execute([&]() { threadIds->push_back(GetCurrentThreadId()); output.append(RUNNABLE_0_TEXT.c_str()); + notification.Notify(); }); + ASSERT_TRUE(notification.WaitForNotificationWithTimeout(kWaitTimeout)); executor->Shutdown(); // Assert @@ -56,8 +68,7 @@ TEST(ExecutorTests, SingleThreadedExecutorAfterShutdownFails) { // Arrange std::string expected(""); - std::unique_ptr executor = - std::make_unique(); + std::unique_ptr executor = std::make_unique(); std::unique_ptr output = std::make_unique(); // Container to note threads that ran std::unique_ptr> threadIds = @@ -83,11 +94,11 @@ TEST(ExecutorTests, SingleThreadedExecutorAfterShutdownFails) { } TEST(ExecutorTests, SingleThreadedExecutorExecuteNullSucceeds) { + absl::Notification notification; // Arrange std::string expected(RUNNABLE_0_TEXT.c_str()); - std::unique_ptr executor = - std::make_unique(); + auto executor = std::make_unique(); std::string output = std::string(); // Container to note threads that ran std::unique_ptr> threadIds = @@ -97,12 +108,14 @@ TEST(ExecutorTests, SingleThreadedExecutorExecuteNullSucceeds) { // Act executor->Execute(nullptr); - executor->Execute([&output, &threadIds]() { + executor->Execute([&]() { threadIds->push_back(GetCurrentThreadId()); output.append(RUNNABLE_0_TEXT.c_str()); + notification.Notify(); }); executor->Execute(nullptr); + ASSERT_TRUE(notification.WaitForNotificationWithTimeout(kWaitTimeout)); executor->Shutdown(); // Assert @@ -116,11 +129,12 @@ TEST(ExecutorTests, SingleThreadedExecutorExecuteNullSucceeds) { } TEST(ExecutorTests, SingleThreadedExecutorMultipleTasksSucceeds) { + absl::BlockingCounter block_count(5); + // Arrange std::string expected(RUNNABLE_ALL_TEXT.c_str()); - std::unique_ptr executor = - std::make_unique(); + auto executor = std::make_unique(); std::string output = std::string(); // Container to note threads that ran std::unique_ptr> threadIds = @@ -130,14 +144,16 @@ TEST(ExecutorTests, SingleThreadedExecutorMultipleTasksSucceeds) { // Act for (int index = 0; index < 5; index++) { - executor->Execute([&output, &threadIds, index]() { + executor->Execute([&, index]() { threadIds->push_back(GetCurrentThreadId()); char buffer[128]; snprintf(buffer, sizeof(buffer), "%s%d, ", RUNNABLE_TEXT.c_str(), index); output.append(std::string(buffer)); + block_count.DecrementCount(); }); } + block_count.Wait(); executor->Shutdown(); // Assert @@ -157,11 +173,12 @@ TEST(ExecutorTests, SingleThreadedExecutorMultipleTasksSucceeds) { } TEST(ExecutorTests, MultiThreadedExecutorSingleTaskSucceeds) { + absl::Notification notification; + // Arrange std::string expected(RUNNABLE_0_TEXT.c_str()); - std::unique_ptr executor = - std::make_unique(2); + auto executor = std::make_unique(2); // Container to note threads that ran std::unique_ptr> threadIds = @@ -172,11 +189,13 @@ TEST(ExecutorTests, MultiThreadedExecutorSingleTaskSucceeds) { threadIds->push_back(GetCurrentThreadId()); // Act - executor->Execute([output, &threadIds]() { + executor->Execute([&, output]() { threadIds->push_back(GetCurrentThreadId()); output->append(RUNNABLE_0_TEXT.c_str()); + notification.Notify(); }); + ASSERT_TRUE(notification.WaitForNotificationWithTimeout(kWaitTimeout)); executor->Shutdown(); // Assert @@ -190,9 +209,10 @@ TEST(ExecutorTests, MultiThreadedExecutorSingleTaskSucceeds) { } TEST(ExecutorTests, MultiThreadedExecutorMultipleTasksSucceeds) { + absl::BlockingCounter block_count(5); + // Arrange - std::unique_ptr executor = - std::make_unique(2); + auto executor = std::make_unique(2); // Container to note threads that ran std::unique_ptr> threadIds = @@ -204,14 +224,16 @@ TEST(ExecutorTests, MultiThreadedExecutorMultipleTasksSucceeds) { // Act for (int index = 0; index < 5; index++) { - executor->Execute([&output, &threadIds, index]() { + executor->Execute([&, index]() { threadIds->push_back(GetCurrentThreadId()); char buffer[128]; snprintf(buffer, sizeof(buffer), "%s %d, ", RUNNABLE_TEXT.c_str(), index); output->append(std::string(buffer)); + block_count.DecrementCount(); }); } + block_count.Wait(); executor->Shutdown(); // Assert @@ -226,8 +248,7 @@ TEST(ExecutorTests, MultiThreadedExecutorSingleTaskAfterShutdownFails) { // Arrange std::string expected(""); - std::unique_ptr executor = - std::make_unique(2); + auto executor = std::make_unique(2); // Container to note threads that ran std::unique_ptr> threadIds = @@ -255,83 +276,38 @@ TEST(ExecutorTests, MultiThreadedExecutorSingleTaskAfterShutdownFails) { 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(INVALID_ARGUMENT_TEXT, e.what()); - throw; - } - }, - std::invalid_argument); -} - -TEST(ExecutorTests, MultiThreadedExecutorTooManyThreadsThrows) { - // Arrange - // Act - // Assert - EXPECT_THROW( - { - try { - auto result = - std::make_unique(65); - } catch (const location::nearby::windows::ThreadPoolException& e) { - // and this tests that it has the correct message - EXPECT_STREQ(THREADPOOL_MAX_SIZE_TEXT, e.what()); - throw; - } - }, - location::nearby::windows::ThreadPoolException); -} - TEST(ExecutorTests, MultiThreadedExecutorMultipleTasksLargeNumberOfThreadsSucceeds) { + absl::BlockingCounter block_count(250); + // Arrange - std::unique_ptr executor = - std::make_unique( - MAXIMUM_WAIT_OBJECTS - 1); + auto executor = std::make_unique(32); // Container to note threads that ran std::vector threadIds = std::vector(); - std::shared_ptr output = std::make_shared(); - threadIds.push_back(GetCurrentThreadId()); - - CRITICAL_SECTION testCriticalSection; - InitializeCriticalSection(&testCriticalSection); - + absl::Mutex mutex; // Act for (int index = 0; index < 250; index++) { - executor->Execute( - [output, &threadIds, index, &testCriticalSection]() mutable { - DWORD id = GetCurrentThreadId(); + executor->Execute([&]() mutable { + DWORD id = GetCurrentThreadId(); + { + absl::MutexLock lock(&mutex); + threadIds.push_back(id); + } - EnterCriticalSection(&testCriticalSection); + // Using rand since this is in a critical section + // and windows doesn't have a rand_r anyway + auto sleepTime = (std::rand() % 101) + 1; // NOLINT - threadIds.push_back(id); - output->append(RUNNABLE_TEXT); - output->append(std::to_string(index)); - output->append(RUNNABLE_SEPARATOR_TEXT); - - LeaveCriticalSection(&testCriticalSection); - // Using rand since this is in a critical section - // and windows doesn't have a rand_r anyway - auto sleepTime = (std::rand() % 101) + 1; // NOLINT - - Sleep(sleepTime); - }); + Sleep(sleepTime); + block_count.DecrementCount(); + }); } + block_count.Wait(); executor->Shutdown(); - DeleteCriticalSection(&testCriticalSection); // Assert // We should still be on the main thread @@ -341,8 +317,13 @@ TEST(ExecutorTests, int64_t uniqueIds = std::unique(threadIds.begin(), threadIds.end()) - threadIds.begin(); - ASSERT_EQ(uniqueIds, 64); + ASSERT_EQ(uniqueIds, 33); // We should've run 1 time on the main thread, and 200 times on the // workerThreads ASSERT_EQ(threadIds.size(), 251); } + +} // namespace +} // namespace windows +} // namespace nearby +} // namespace location diff --git a/internal/platform/implementation/windows/runner.h b/internal/platform/implementation/windows/runner.h deleted file mode 100644 index c6d414c0..00000000 --- a/internal/platform/implementation/windows/runner.h +++ /dev/null @@ -1,42 +0,0 @@ -// 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 "internal/platform/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/internal/platform/implementation/windows/scheduled_executor_test.cc b/internal/platform/implementation/windows/scheduled_executor_test.cc index d9126d3a..0176eff8 100644 --- a/internal/platform/implementation/windows/scheduled_executor_test.cc +++ b/internal/platform/implementation/windows/scheduled_executor_test.cc @@ -11,15 +11,17 @@ // 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 "internal/platform/implementation/windows/scheduled_executor.h" #include +#include "gtest/gtest.h" +#include "absl/synchronization/notification.h" #include "internal/platform/implementation/windows/test_data.h" -#include "gtest/gtest.h" - TEST(ScheduledExecutorTests, ExecuteSucceeds) { + absl::Notification notification; // Arrange std::string expected(RUNNABLE_0_TEXT.c_str()); @@ -34,11 +36,14 @@ TEST(ScheduledExecutorTests, ExecuteSucceeds) { threadIds->push_back(GetCurrentThreadId()); // Act - submittableExecutor->Execute([&output, &threadIds]() { + submittableExecutor->Execute([&]() { threadIds->push_back(GetCurrentThreadId()); output.append(RUNNABLE_0_TEXT.c_str()); + notification.Notify(); }); + ASSERT_TRUE( + notification.WaitForNotificationWithTimeout(absl::Milliseconds(200))); submittableExecutor->Shutdown(); // Assert @@ -52,6 +57,7 @@ TEST(ScheduledExecutorTests, ExecuteSucceeds) { } TEST(ScheduledExecutorTests, ScheduleSucceeds) { + absl::Notification notification; // Arrange std::string expected(RUNNABLE_0_TEXT.c_str()); @@ -71,15 +77,17 @@ TEST(ScheduledExecutorTests, ScheduleSucceeds) { // Act submittableExecutor->Schedule( - [&output, &threadIds, &timeExecuted]() { + [&]() { timeExecuted = std::chrono::system_clock::now(); threadIds->push_back(GetCurrentThreadId()); output.append(RUNNABLE_0_TEXT.c_str()); + notification.Notify(); }, absl::Milliseconds(50)); SleepEx(100, true); // Yield the thread - + ASSERT_TRUE( + notification.WaitForNotificationWithTimeout(absl::Milliseconds(200))); submittableExecutor->Shutdown(); auto difference = std::chrono::duration_cast( @@ -100,6 +108,7 @@ TEST(ScheduledExecutorTests, ScheduleSucceeds) { } TEST(ScheduledExecutorTests, CancelSucceeds) { + absl::Notification notification; // Arrange std::string expected(""); @@ -115,9 +124,10 @@ TEST(ScheduledExecutorTests, CancelSucceeds) { // Act auto cancelable = submittableExecutor->Schedule( - [&output, &threadIds]() { + [&]() { threadIds->push_back(GetCurrentThreadId()); output.append(RUNNABLE_0_TEXT.c_str()); + notification.Notify(); }, absl::Milliseconds(1000)); @@ -125,6 +135,8 @@ TEST(ScheduledExecutorTests, CancelSucceeds) { auto actual = cancelable->Cancel(); + EXPECT_FALSE( + notification.WaitForNotificationWithTimeout(absl::Milliseconds(2000))); submittableExecutor->Shutdown(); // Assert @@ -137,6 +149,7 @@ TEST(ScheduledExecutorTests, CancelSucceeds) { } TEST(ScheduledExecutorTests, CancelAfterStartedFails) { + absl::Notification notification; // Arrange std::string expected(RUNNABLE_0_TEXT.c_str()); @@ -152,9 +165,10 @@ TEST(ScheduledExecutorTests, CancelAfterStartedFails) { // Act auto cancelable = submittableExecutor->Schedule( - [&output, &threadIds]() { + [&]() { threadIds->push_back(GetCurrentThreadId()); output.append(RUNNABLE_0_TEXT.c_str()); + notification.Notify(); }, absl::Milliseconds(100)); @@ -162,6 +176,8 @@ TEST(ScheduledExecutorTests, CancelAfterStartedFails) { auto actual = cancelable->Cancel(); + ASSERT_TRUE( + notification.WaitForNotificationWithTimeout(absl::Milliseconds(2000))); submittableExecutor->Shutdown(); // Assert diff --git a/internal/platform/implementation/windows/submittable_executor_test.cc b/internal/platform/implementation/windows/submittable_executor_test.cc index 99bc6bb5..041c7fc2 100644 --- a/internal/platform/implementation/windows/submittable_executor_test.cc +++ b/internal/platform/implementation/windows/submittable_executor_test.cc @@ -11,34 +11,44 @@ // 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 "internal/platform/implementation/windows/submittable_executor.h" #include +#include "gtest/gtest.h" +#include "absl/synchronization/blocking_counter.h" +#include "absl/synchronization/notification.h" +#include "absl/time/time.h" #include "internal/platform/implementation/windows/test_data.h" -#include "gtest/gtest.h" +namespace location { +namespace nearby { +namespace windows { +namespace { + +constexpr absl::Duration kWaitTimeout = absl::Milliseconds(200); TEST(SubmittableExecutorTests, SingleThreadedExecuteSucceeds) { + absl::Notification notification; // Arrange std::string expected(RUNNABLE_0_TEXT.c_str()); - std::unique_ptr - submittableExecutor = - std::make_unique(); + auto submittableExecutor = std::make_unique(); std::string output = std::string(); // Container to note threads that ran - std::unique_ptr> threadIds = - std::make_unique>(); + auto threadIds = std::make_unique>(); threadIds->push_back(GetCurrentThreadId()); // Act - submittableExecutor->Execute([&output, &threadIds]() { + submittableExecutor->Execute([&]() { threadIds->push_back(GetCurrentThreadId()); output.append(RUNNABLE_0_TEXT.c_str()); + notification.Notify(); }); + ASSERT_TRUE(notification.WaitForNotificationWithTimeout(kWaitTimeout)); submittableExecutor->Shutdown(); // Assert @@ -55,13 +65,10 @@ TEST(SubmittableExecutorTests, SingleThreadedExecuteAfterShutdownFails) { // Arrange std::string expected(""); - std::unique_ptr - submittableExecutor = - std::make_unique(); + auto submittableExecutor = std::make_unique(); std::string output = std::string(); // Container to note threads that ran - std::unique_ptr> threadIds = - std::make_unique>(); + auto threadIds = std::make_unique>(); threadIds->push_back(GetCurrentThreadId()); @@ -84,25 +91,25 @@ TEST(SubmittableExecutorTests, SingleThreadedExecuteAfterShutdownFails) { } TEST(SubmittableExecutorTests, SingleThreadedDoSubmitSucceeds) { + absl::Notification notification; // Arrange std::string expected(RUNNABLE_0_TEXT.c_str()); - std::unique_ptr - submittableExecutor = - std::make_unique(); + auto submittableExecutor = std::make_unique(); std::string output = std::string(); // Container to note threads that ran - std::unique_ptr> threadIds = - std::make_unique>(); + auto threadIds = std::make_unique>(); threadIds->push_back(GetCurrentThreadId()); // Act - auto result = submittableExecutor->DoSubmit([&output, &threadIds]() { + auto result = submittableExecutor->DoSubmit([&]() { threadIds->push_back(GetCurrentThreadId()); output.append(RUNNABLE_0_TEXT.c_str()); + notification.Notify(); }); + ASSERT_TRUE(notification.WaitForNotificationWithTimeout(kWaitTimeout)); submittableExecutor->Shutdown(); // Assert @@ -122,13 +129,10 @@ TEST(SubmittableExecutorTests, // Arrange std::string expected(""); - std::unique_ptr - submittableExecutor = - std::make_unique(); + auto submittableExecutor = std::make_unique(); std::unique_ptr output = std::make_unique(); // Container to note threads that ran - std::unique_ptr> threadIds = - std::make_unique>(); + auto threadIds = std::make_unique>(); threadIds->push_back(GetCurrentThreadId()); @@ -153,29 +157,30 @@ TEST(SubmittableExecutorTests, } TEST(SubmittableExecutorTests, SingleThreadedExecuteMultipleTasksSucceeds) { + absl::BlockingCounter blocking_counter(5); + // Arrange std::string expected(RUNNABLE_ALL_TEXT.c_str()); - std::unique_ptr - submittableExecutor = - std::make_unique(); + auto submittableExecutor = std::make_unique(); std::unique_ptr output = std::make_unique(); // Container to note threads that ran - std::unique_ptr> threadIds = - std::make_unique>(); + auto threadIds = std::make_unique>(); threadIds->push_back(GetCurrentThreadId()); // Act for (int index = 0; index < 5; index++) { - submittableExecutor->Execute([&output, &threadIds, index]() { + submittableExecutor->Execute([&, index]() { threadIds->push_back(GetCurrentThreadId()); char buffer[128]; snprintf(buffer, sizeof(buffer), "%s%d, ", RUNNABLE_TEXT.c_str(), index); output->append(std::string(buffer)); + blocking_counter.DecrementCount(); }); } + blocking_counter.Wait(); submittableExecutor->Shutdown(); // Assert @@ -195,30 +200,31 @@ TEST(SubmittableExecutorTests, SingleThreadedExecuteMultipleTasksSucceeds) { } TEST(SubmittableExecutorTests, SingleThreadedDoSubmitMultipleTasksSucceeds) { + absl::BlockingCounter blocking_counter(5); + // Arrange std::string expected(RUNNABLE_ALL_TEXT.c_str()); - std::unique_ptr - submittableExecutor = - std::make_unique(); + auto submittableExecutor = std::make_unique(); std::unique_ptr output = std::make_unique(); // Container to note threads that ran - std::unique_ptr> threadIds = - std::make_unique>(); + auto threadIds = std::make_unique>(); threadIds->push_back(GetCurrentThreadId()); // Act bool result = true; for (int index = 0; index < 5; index++) { - result &= submittableExecutor->DoSubmit([&output, &threadIds, index]() { + result &= submittableExecutor->DoSubmit([&, index]() { threadIds->push_back(GetCurrentThreadId()); char buffer[128]; snprintf(buffer, sizeof(buffer), "%s%d, ", RUNNABLE_TEXT.c_str(), index); output->append(std::string(buffer)); + blocking_counter.DecrementCount(); }); } + blocking_counter.Wait(); submittableExecutor->Shutdown(); // Assert @@ -238,3 +244,8 @@ TEST(SubmittableExecutorTests, SingleThreadedDoSubmitMultipleTasksSucceeds) { // We should of run them in the order submitted ASSERT_EQ(*output.get(), expected); } + +} // namespace +} // namespace windows +} // namespace nearby +} // namespace location diff --git a/internal/platform/implementation/windows/thread_pool.cc b/internal/platform/implementation/windows/thread_pool.cc index 7e92c06d..305d0403 100644 --- a/internal/platform/implementation/windows/thread_pool.cc +++ b/internal/platform/implementation/windows/thread_pool.cc @@ -14,518 +14,153 @@ #include "internal/platform/implementation/windows/thread_pool.h" -#include +#include -#include -#include +#include +#include -#include "internal/platform/implementation/windows/runner.h" +#include "absl/memory/memory.h" +#include "absl/synchronization/mutex.h" #include "internal/platform/logging.h" +#include "internal/platform/runnable.h" namespace location { namespace nearby { namespace windows { -#define POOL_NAME_BUFFER_SIZE 64 -#define EVENT_NAME_BUFFER_SIZE 64 +VOID CALLBACK WorkCallback(PTP_CALLBACK_INSTANCE instance, PVOID parameter, + PTP_WORK work) { + // Instance is not used in thread pool. + UNREFERENCED_PARAMETER(instance); -__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) { - NEARBY_LOGS(ERROR) << __func__ << ": pParam must not be null."; - 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* thread_pool = reinterpret_cast(parameter); + thread_pool->RunNextTask(); + CloseThreadpoolWork(work); } -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) { - // The MAXIMUM_WAIT_OBJECTS is the limiting factor, currently - // windows has a max of 64. This means we can only wait on up - // to 64 threads, anything more gives undesirable results. - if (nPoolSize > 63) { - NEARBY_LOGS(ERROR) << __func__ << ": Thread pool max size exceeded."; - throw ThreadPoolException("Thread pool max size exceeded."); - } +std::unique_ptr ThreadPool::Create(int max_pool_size) { + NEARBY_LOGS(VERBOSE) << __func__ << ": Create thread pool with maximum size(" + << max_pool_size << ")."; - pool_state_ = State::Destroyed; - pool_size_ = nPoolSize; + PTP_POOL thread_pool = nullptr; + TP_CALLBACK_ENVIRON thread_pool_environ; + InitializeThreadpoolEnvironment(&thread_pool_environ); - InitializeCriticalSection(&critical_section_); - - if (bCreateNow) { - if (!Create()) { - NEARBY_LOGS(ERROR) << __func__ << ": Thread pool creation failed."; - throw ThreadPoolException("Thread pool creation failed."); - } - } -} - -bool ThreadPool::Create() { - if (pool_state_ != State::Destroyed) { - // To create a new pool, destroy the existing one first + if (max_pool_size <= 0) { NEARBY_LOGS(ERROR) << __func__ - << ": Attempt to create a new thread pool before " - "destroying the old one."; - return false; + << ": Maximum pool size must be positive integer value."; + return nullptr; } - 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); - - // create the event which will signal the threads to stop - std::string eventName; - 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_pool = CreateThreadpool(NULL); + if (thread_pool == nullptr) { + NEARBY_LOGS(ERROR) << __func__ + << ": failed to create thread pool. LastError: " + << GetLastError(); + return nullptr; } - SYSTEM_INFO sysinfo; - GetSystemInfo(&sysinfo); - int numCPU = sysinfo.dwNumberOfProcessors; - - int threadsToCreate = 0; - - // We are going to initially allocate the first n - // threads based on the number of logical cores - if (pool_size_ > numCPU) { - threadsToCreate = STARTUP_THREAD_COUNT; - } else { - threadsToCreate = pool_size_; + // Sets thread pool maximum value. In order to release all threads, + // will keep at least one thread. + SetThreadpoolThreadMaximum(thread_pool, max_pool_size); + if (!SetThreadpoolThreadMinimum(thread_pool, 1)) { + NEARBY_LOGS(ERROR) + << __func__ << ": failed to set minimum thread pool size. LastError: " + << GetLastError(); + CloseThreadpool(thread_pool); + return nullptr; } - thread_handles_ = new HANDLE[pool_size_]; + // + // Associate the callback environment with our thread pool. + // + SetThreadpoolCallbackPool(&thread_pool_environ, thread_pool); - // create the threads - for (int index = 0; index < threadsToCreate; index++) { - CreateThreadPoolThread(&thread_handles_[index]); - } + return absl::WrapUnique( + new ThreadPool(thread_pool, thread_pool_environ, max_pool_size)); +} - pool_state_ = State::Ready; - return true; +ThreadPool::ThreadPool(PTP_POOL thread_pool, + TP_CALLBACK_ENVIRON thread_pool_environ, + int max_pool_size) + : thread_pool_(thread_pool), + thread_pool_environ_(thread_pool_environ), + max_pool_size_(max_pool_size) { + NEARBY_LOGS(VERBOSE) << __func__ << ": Thread pool(" << this + << ") is created."; } ThreadPool::~ThreadPool() { - Destroy(); - ReleaseMemory(); - DeleteCriticalSection(&critical_section_); + NEARBY_LOGS(VERBOSE) << __func__ << ": Thread pool(" << this + << ") is released."; + + ShutDown(); } -DWORD ThreadPool::CreateThreadPoolThread(HANDLE* handles) { - HANDLE thread; - DWORD threadId; - std::unique_ptr threadData = std::make_unique(); - - 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)), - (int)thread_map_->size()); - - 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 NULL; - } - - if (thread) { - // add the entry to the map of threads - EnterCriticalSection(&critical_section_); - - threadData->free = true; - threadData->wait_handle = CreateEventA(NULL, TRUE, FALSE, buffer); - - threadData->thread_handle = thread; - threadData->thread_id = threadId; - - thread_map_->insert(ThreadMap::value_type(threadId, std::move(threadData))); - *handles = thread; - LeaveCriticalSection(&critical_section_); - - ResumeThread(thread); - - NEARBY_LOGS(VERBOSE) << "Info: " << __func__ - << ": Thread created, handle: " << thread - << ", id: " << threadId; - - return threadId; - } else { - NEARBY_LOGS(ERROR) << "Error: " << __func__ << ": Failed to create thread."; - return NULL; - } -} - -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; - - bool notDone = true; - - 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) { - // Waits until the specified object is in the signaled state or the time-out - // interval elapses. - // https://docs.microsoft.com/en-us/windows/win32/api/synchapi/nf-synchapi-waitforsingleobject - auto wait = WaitForSingleObject( - thread_handles_[0], // A handle to the object - INFINITE // The time-out interval, in milliseconds. - ); - } else { - auto wait = - // Waits until one or all of the specified objects are in the signaled - // state or the time-out interval elapses. - // https://docs.microsoft.com/en-us/windows/win32/api/synchapi/nf-synchapi-waitformultipleobjects - WaitForMultipleObjects( - index, // The number of object handles in the array. - thread_handles_, // An array of object handles. - true, // If this parameter is TRUE, the function returns when the - // state of all objects in the handles array are signaled. - INFINITE // The time-out interval, in milliseconds. - ); - } - - 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) { - NEARBY_LOGS(ERROR) - << __func__ << ": 0 or negative value is not a valid thread pool size."; - 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."); - NEARBY_LOGS(ERROR) << __func__ << ": 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) +bool ThreadPool::Run(Runnable task) { + if (thread_pool_ == nullptr) { return false; - - _ASSERT(runner != NULL); - - AddRunner(std::move(runner)); - - // See if any threads are free - ThreadMap::iterator iterator; - std::unique_ptr threadData; - - bool freeThreadFound = false; - - EnterCriticalSection(&critical_section_); - - for (iterator = thread_map_->begin(); iterator != thread_map_->end(); - iterator++) { - if (iterator->second->free) { - // here is a free thread, put it to work - iterator->second->free = false; - SetEvent(iterator->second->wait_handle); - // this thread will now call GetThreadProc() and pick up the next - // function in the list. - freeThreadFound = true; - break; - } } - if (!freeThreadFound && thread_map_->size() < pool_size_) { - // We haven't used up all of our threads, go ahead and spin up another one - DWORD threadId = - CreateThreadPoolThread(&thread_handles_[thread_map_->size()]); - thread_map_->at(threadId)->free = false; - SetEvent(thread_map_->at(threadId)->wait_handle); + absl::MutexLock lock(&mutex_); + PTP_WORK work; + tasks_.push(std::move(task)); + NEARBY_LOGS(VERBOSE) << __func__ << ": Scheduled to run task(" + << &tasks_.back() << ")."; + + work = CreateThreadpoolWork(WorkCallback, this, &thread_pool_environ_); + if (work == nullptr) { + NEARBY_LOGS(ERROR) << __func__ + << ": failed to create thread pool work. LastError: " + << GetLastError(); + return false; } - LeaveCriticalSection(&critical_section_); - + // + // Submit the work to the pool. Because this was a pre-allocated + // work item (using CreateThreadpoolWork), it is guaranteed to execute. + // + SubmitThreadpoolWork(work); return true; } -void ThreadPool::AddRunner(std::unique_ptr runner) { - // add it to the list - runner->thread_pool_ = this; +void ThreadPool::ShutDown() { + NEARBY_LOGS(VERBOSE) << __func__ << ": Shutdown thread pool(" << this << ")."; + if (thread_pool_ == nullptr) { + NEARBY_LOGS(WARNING) << __func__ << ": Shutdown on closed thread pool(" + << this << ")."; + return; + } - NEARBY_LOGS(VERBOSE) << "Info: " << __func__ - << ": pushing new runner to the back."; - - EnterCriticalSection(&critical_section_); - - function_list_->push_back(std::move(runner)); - - LeaveCriticalSection(&critical_section_); + CloseThreadpool(thread_pool_); + thread_pool_ = nullptr; } -HANDLE ThreadPool::GetWaitHandle(DWORD dwThreadId) { - HANDLE hWait = NULL; - ThreadMap::iterator iter; +void ThreadPool::RunNextTask() { + if (thread_pool_ == nullptr) { + return; + } - EnterCriticalSection(&critical_section_); - - iter = thread_map_->find(dwThreadId); - - if (iter != thread_map_->end()) // if search found no elements + Runnable task = nullptr; { - hWait = thread_map_->at(dwThreadId)->wait_handle; - } + absl::MutexLock lock(&mutex_); + if (!tasks_.empty()) { + NEARBY_LOGS(VERBOSE) << __func__ << ": Run task(" << &tasks_.front() + << ")."; - 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; - - int nCount = 0; - - EnterCriticalSection(&critical_section_); - - for (iter = thread_map_->begin(); iter != thread_map_->end(); iter++) { - if (function_list_->empty()) { - iter->second->free = true; - } - - if (!iter->second->free) { - nCount++; + task = tasks_.front(); + tasks_.pop(); } } + if (task == nullptr) { + NEARBY_LOGS(WARNING) << __func__ + << ": Tried to run task in an empty thread pool."; + return; + } - LeaveCriticalSection(&critical_section_); - - return nCount; + task(); } -State ThreadPool::GetState() { return pool_state_; } - } // namespace windows } // namespace nearby } // namespace location diff --git a/internal/platform/implementation/windows/thread_pool.h b/internal/platform/implementation/windows/thread_pool.h index 7aedf157..014dc10e 100644 --- a/internal/platform/implementation/windows/thread_pool.h +++ b/internal/platform/implementation/windows/thread_pool.h @@ -11,96 +11,65 @@ // 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 +#include + +#include "absl/synchronization/mutex.h" +#include "internal/platform/runnable.h" -#include "internal/platform/implementation/windows/runner.h" namespace location { namespace nearby { namespace windows { -// This is the number of threads that will be started initially if the pool -// size is greater than 4, or if the pool size is greater than the number -// of cores present, including virtual cores -#define STARTUP_THREAD_COUNT 4 - -class ThreadPoolException : public std::runtime_error { - public: - ThreadPoolException() : std::runtime_error("") {} - ThreadPoolException(const std::string& message) - : std::runtime_error(message), message_(message) {} - virtual const char* what() const throw() { - return message_.c_str(); - } - - private: - const std::string message_; -}; - -// 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> - 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(); + static std::unique_ptr Create(int max_pool_size); + + // Runs a task on thread pool. The result indicates whether the task is put + // into the thread pool. + bool Run(Runnable task); + + // The thread pool is closed immediately if there are no outstanding work, + // I/O, timer, or wait objects that are bound to the pool; otherwise, the + // thread pool is released asynchronously after the outstanding objects are + // freed. + void ShutDown(); 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_; + ThreadPool(PTP_POOL thread_pool, TP_CALLBACK_ENVIRON thread_pool_environ, + int max_pool_size); + void RunNextTask(); - 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); - DWORD CreateThreadPoolThread(HANDLE* handles); + // Protects the access to tasks of the thread pool. + mutable absl::Mutex mutex_; + + // The task queue of the thread pool. Thread pool will pick up task to run + // when it is idle. + std::queue tasks_ ABSL_GUARDED_BY(mutex_); + + // Keeps the pointer of the thread pool. It is created when constructing the + // thread pool. + PTP_POOL thread_pool_ = nullptr; + + // Keeps the environment of the thread pool. + TP_CALLBACK_ENVIRON thread_pool_environ_; + + // The maximum thread count in the thread pool + int max_pool_size_ = 0; + + friend VOID CALLBACK WorkCallback(PTP_CALLBACK_INSTANCE instance, + PVOID parameter, PTP_WORK work); }; } // namespace windows } // namespace nearby } // namespace location + #endif // PLATFORM_IMPL_WINDOWS_THREAD_POOL_H_ diff --git a/internal/platform/implementation/windows/thread_pool_test.cc b/internal/platform/implementation/windows/thread_pool_test.cc new file mode 100644 index 00000000..456ed438 --- /dev/null +++ b/internal/platform/implementation/windows/thread_pool_test.cc @@ -0,0 +1,74 @@ +// Copyright 2022 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 "internal/platform/implementation/windows/thread_pool.h" + +#include +#include + +#include "gtest/gtest.h" +#include "absl/synchronization/blocking_counter.h" +#include "absl/synchronization/notification.h" +#include "absl/time/clock.h" +#include "absl/time/time.h" + +namespace location { +namespace nearby { +namespace windows { +namespace { + +constexpr int kTaskCount = 10; + +TEST(ThreadPool, TasksInSingleThreadRunInSequence) { + absl::BlockingCounter blocking_counter(kTaskCount); + auto pool = ThreadPool::Create(1); + std::vector completed_tasks; + std::vector expected_tasks; + + for (int i = 0; i < kTaskCount; ++i) { + expected_tasks.push_back(i); + pool->Run([&, i]() { + absl::SleepFor(absl::Milliseconds(200)); + completed_tasks.push_back(i); + blocking_counter.DecrementCount(); + }); + } + + blocking_counter.Wait(); + EXPECT_EQ(completed_tasks, expected_tasks); + pool->ShutDown(); +} + +TEST(ThreadPool, TasksInMultipleThreadsRunInParallel) { + absl::BlockingCounter blocking_counter(kTaskCount); + absl::Time start_time = absl::Now(); + + auto pool = ThreadPool::Create(2); + + for (int i = 0; i < kTaskCount; ++i) { + pool->Run([&]() { + absl::SleepFor(absl::Milliseconds(200)); + blocking_counter.DecrementCount(); + }); + } + + blocking_counter.Wait(); + EXPECT_TRUE(absl::Now() - start_time < absl::Milliseconds(1500)); + pool->ShutDown(); +} + +} // namespace +} // namespace windows +} // namespace nearby +} // namespace location