mirror of
https://github.com/kidfromjupiter/nearby.git
synced 2026-09-15 07:06:11 -04:00
Applied thread pool implementation from Windows platform
PiperOrigin-RevId: 449385710
This commit is contained in:
guogang
committed by
Copybara-Service
Copybara-Service
parent
412f9e0752
commit
4aee0b681c
@@ -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"],
|
||||
|
||||
@@ -14,6 +14,8 @@
|
||||
|
||||
#include "internal/platform/implementation/crypto.h"
|
||||
|
||||
#include <string>
|
||||
|
||||
#include "gtest/gtest.h"
|
||||
|
||||
namespace location {
|
||||
|
||||
@@ -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 <cassert>
|
||||
|
||||
#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<ThreadPool>(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> runner = std::make_unique<Runner>(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;
|
||||
}
|
||||
|
||||
|
||||
@@ -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<ThreadPool> thread_pool_;
|
||||
|
||||
std::unique_ptr<ThreadPool> thread_pool_ = nullptr;
|
||||
std::atomic<bool> shut_down_;
|
||||
ExecutorState executor_state_;
|
||||
int32_t max_concurrency_;
|
||||
};
|
||||
|
||||
|
||||
@@ -17,16 +17,26 @@
|
||||
#include <algorithm>
|
||||
#include <utility>
|
||||
|
||||
#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<location::nearby::windows::Executor> executor =
|
||||
std::make_unique<location::nearby::windows::Executor>();
|
||||
auto executor = std::make_unique<Executor>();
|
||||
std::string output = std::string();
|
||||
// Container to note threads that ran
|
||||
std::unique_ptr<std::vector<DWORD>> 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<location::nearby::windows::Executor> executor =
|
||||
std::make_unique<location::nearby::windows::Executor>();
|
||||
std::unique_ptr<Executor> executor = std::make_unique<Executor>();
|
||||
std::unique_ptr<std::string> output = std::make_unique<std::string>();
|
||||
// Container to note threads that ran
|
||||
std::unique_ptr<std::vector<DWORD>> 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<location::nearby::windows::Executor> executor =
|
||||
std::make_unique<location::nearby::windows::Executor>();
|
||||
auto executor = std::make_unique<Executor>();
|
||||
std::string output = std::string();
|
||||
// Container to note threads that ran
|
||||
std::unique_ptr<std::vector<DWORD>> 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<location::nearby::windows::Executor> executor =
|
||||
std::make_unique<location::nearby::windows::Executor>();
|
||||
auto executor = std::make_unique<Executor>();
|
||||
std::string output = std::string();
|
||||
// Container to note threads that ran
|
||||
std::unique_ptr<std::vector<DWORD>> 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<location::nearby::windows::Executor> executor =
|
||||
std::make_unique<location::nearby::windows::Executor>(2);
|
||||
auto executor = std::make_unique<Executor>(2);
|
||||
|
||||
// Container to note threads that ran
|
||||
std::unique_ptr<std::vector<DWORD>> 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<location::nearby::windows::Executor> executor =
|
||||
std::make_unique<location::nearby::windows::Executor>(2);
|
||||
auto executor = std::make_unique<Executor>(2);
|
||||
|
||||
// Container to note threads that ran
|
||||
std::unique_ptr<std::vector<DWORD>> 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<location::nearby::windows::Executor> executor =
|
||||
std::make_unique<location::nearby::windows::Executor>(2);
|
||||
auto executor = std::make_unique<Executor>(2);
|
||||
|
||||
// Container to note threads that ran
|
||||
std::unique_ptr<std::vector<DWORD>> 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<location::nearby::windows::Executor>(-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<location::nearby::windows::Executor>(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<location::nearby::windows::Executor> executor =
|
||||
std::make_unique<location::nearby::windows::Executor>(
|
||||
MAXIMUM_WAIT_OBJECTS - 1);
|
||||
auto executor = std::make_unique<Executor>(32);
|
||||
|
||||
// Container to note threads that ran
|
||||
std::vector<DWORD> threadIds = std::vector<DWORD>();
|
||||
|
||||
std::shared_ptr<std::string> output = std::make_shared<std::string>();
|
||||
|
||||
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
|
||||
|
||||
@@ -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<void()> runnable)
|
||||
: thread_pool_(nullptr), runnable_(runnable) {}
|
||||
void Run() { runnable_(); }
|
||||
~Runner(){}
|
||||
ThreadPool* thread_pool_;
|
||||
|
||||
private:
|
||||
std::function<void()> runnable_;
|
||||
};
|
||||
|
||||
} // namespace windows
|
||||
} // namespace nearby
|
||||
} // namespace location
|
||||
|
||||
#endif // PLATFORM_IMPL_WINDOWS_RUNNER_H_
|
||||
@@ -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 <utility>
|
||||
|
||||
#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<std::chrono::milliseconds>(
|
||||
@@ -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
|
||||
|
||||
@@ -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 <utility>
|
||||
|
||||
#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<location::nearby::windows::SubmittableExecutor>
|
||||
submittableExecutor =
|
||||
std::make_unique<location::nearby::windows::SubmittableExecutor>();
|
||||
auto submittableExecutor = std::make_unique<SubmittableExecutor>();
|
||||
std::string output = std::string();
|
||||
// Container to note threads that ran
|
||||
std::unique_ptr<std::vector<DWORD>> threadIds =
|
||||
std::make_unique<std::vector<DWORD>>();
|
||||
auto threadIds = std::make_unique<std::vector<DWORD>>();
|
||||
|
||||
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<location::nearby::windows::SubmittableExecutor>
|
||||
submittableExecutor =
|
||||
std::make_unique<location::nearby::windows::SubmittableExecutor>();
|
||||
auto submittableExecutor = std::make_unique<SubmittableExecutor>();
|
||||
std::string output = std::string();
|
||||
// Container to note threads that ran
|
||||
std::unique_ptr<std::vector<DWORD>> threadIds =
|
||||
std::make_unique<std::vector<DWORD>>();
|
||||
auto threadIds = std::make_unique<std::vector<DWORD>>();
|
||||
|
||||
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<location::nearby::windows::SubmittableExecutor>
|
||||
submittableExecutor =
|
||||
std::make_unique<location::nearby::windows::SubmittableExecutor>();
|
||||
auto submittableExecutor = std::make_unique<SubmittableExecutor>();
|
||||
std::string output = std::string();
|
||||
// Container to note threads that ran
|
||||
std::unique_ptr<std::vector<DWORD>> threadIds =
|
||||
std::make_unique<std::vector<DWORD>>();
|
||||
auto threadIds = std::make_unique<std::vector<DWORD>>();
|
||||
|
||||
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<location::nearby::windows::SubmittableExecutor>
|
||||
submittableExecutor =
|
||||
std::make_unique<location::nearby::windows::SubmittableExecutor>();
|
||||
auto submittableExecutor = std::make_unique<SubmittableExecutor>();
|
||||
std::unique_ptr<std::string> output = std::make_unique<std::string>();
|
||||
// Container to note threads that ran
|
||||
std::unique_ptr<std::vector<DWORD>> threadIds =
|
||||
std::make_unique<std::vector<DWORD>>();
|
||||
auto threadIds = std::make_unique<std::vector<DWORD>>();
|
||||
|
||||
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<location::nearby::windows::SubmittableExecutor>
|
||||
submittableExecutor =
|
||||
std::make_unique<location::nearby::windows::SubmittableExecutor>();
|
||||
auto submittableExecutor = std::make_unique<SubmittableExecutor>();
|
||||
std::unique_ptr<std::string> output = std::make_unique<std::string>();
|
||||
// Container to note threads that ran
|
||||
std::unique_ptr<std::vector<DWORD>> threadIds =
|
||||
std::make_unique<std::vector<DWORD>>();
|
||||
auto threadIds = std::make_unique<std::vector<DWORD>>();
|
||||
|
||||
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<location::nearby::windows::SubmittableExecutor>
|
||||
submittableExecutor =
|
||||
std::make_unique<location::nearby::windows::SubmittableExecutor>();
|
||||
auto submittableExecutor = std::make_unique<SubmittableExecutor>();
|
||||
std::unique_ptr<std::string> output = std::make_unique<std::string>();
|
||||
// Container to note threads that ran
|
||||
std::unique_ptr<std::vector<DWORD>> threadIds =
|
||||
std::make_unique<std::vector<DWORD>>();
|
||||
auto threadIds = std::make_unique<std::vector<DWORD>>();
|
||||
|
||||
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
|
||||
|
||||
@@ -14,518 +14,153 @@
|
||||
|
||||
#include "internal/platform/implementation/windows/thread_pool.h"
|
||||
|
||||
#include <stdio.h>
|
||||
#include <windows.h>
|
||||
|
||||
#include <iomanip>
|
||||
#include <iostream>
|
||||
#include <queue>
|
||||
#include <utility>
|
||||
|
||||
#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> 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<ThreadPool*>(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<ThreadPool*>(parameter);
|
||||
thread_pool->RunNextTask();
|
||||
CloseThreadpoolWork(work);
|
||||
}
|
||||
|
||||
ThreadPool::ThreadPool(int nPoolSize, bool bCreateNow)
|
||||
: function_list_(std::make_unique<FunctionList>()),
|
||||
thread_map_(std::make_unique<ThreadMap>()),
|
||||
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> 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> threadData = std::make_unique<ThreadData>();
|
||||
|
||||
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>&& 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> 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> 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> 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
|
||||
|
||||
@@ -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 <windows.h>
|
||||
|
||||
#include <functional>
|
||||
#include <list>
|
||||
#include <map>
|
||||
#include <stdexcept>
|
||||
#include <queue>
|
||||
#include <utility>
|
||||
|
||||
#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<std::unique_ptr<Runner>> 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<DWORD, std::unique_ptr<ThreadData>>
|
||||
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<Runner> runObject);
|
||||
bool CheckThreadStop();
|
||||
int GetWorkingThreadCount();
|
||||
State GetState();
|
||||
static std::unique_ptr<ThreadPool> 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<FunctionList> function_list_;
|
||||
std::unique_ptr<ThreadMap> 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>&& runner);
|
||||
void FinishNotify(DWORD dwThreadId);
|
||||
void BusyNotify(DWORD dwThreadId);
|
||||
void ReleaseMemory();
|
||||
HANDLE GetWaitHandle(DWORD dwThreadId);
|
||||
HANDLE GetShutdownHandle();
|
||||
void AddRunner(std::unique_ptr<Runner> 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<Runnable> 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_
|
||||
|
||||
@@ -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 <functional>
|
||||
#include <vector>
|
||||
|
||||
#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<int> completed_tasks;
|
||||
std::vector<int> 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
|
||||
Reference in New Issue
Block a user