From 1d7eca3825ffb61b6da8c3bf8b63dba852a39f24 Mon Sep 17 00:00:00 2001 From: jfcarroll Date: Wed, 8 Sep 2021 15:53:33 -0700 Subject: [PATCH] Initial implementation of SubmittableExecutor PiperOrigin-RevId: 395576173 --- cpp/platform/impl/windows/BUILD | 4 + .../impl/windows/submittable_executor.cc | 59 ++++ .../impl/windows/submittable_executor.h | 17 +- .../impl/windows/submittable_executor_test.cc | 275 ++++++++++++++++++ 4 files changed, 348 insertions(+), 7 deletions(-) create mode 100644 cpp/platform/impl/windows/submittable_executor.cc create mode 100644 cpp/platform/impl/windows/submittable_executor_test.cc diff --git a/cpp/platform/impl/windows/BUILD b/cpp/platform/impl/windows/BUILD index 57c774de..d811540e 100644 --- a/cpp/platform/impl/windows/BUILD +++ b/cpp/platform/impl/windows/BUILD @@ -62,6 +62,7 @@ cc_library( "mutex.h", "runner.h", "server_sync.h", + "submittable_executor.h", "thread_pool.h", "webrtc.h", "wifi.h", @@ -104,6 +105,7 @@ cc_library( "executor.cc", "mutex.cc", "platform.cc", + "submittable_executor.cc", "thread_pool.cc", "utils.cc", ], @@ -118,6 +120,7 @@ cc_library( "executor.h", "mutex.h", "runner.h", + "submittable_executor.h", "thread_pool.h", ], compatible_with = ["//buildenv/target:non_prod"], @@ -167,6 +170,7 @@ cc_test( "input_file_test.cc", "mutex_test.cc", "output_file_test.cc", + "submittable_executor_test.cc", ], copts = ["-Iplatform/impl/windows/generated"], deps = [ diff --git a/cpp/platform/impl/windows/submittable_executor.cc b/cpp/platform/impl/windows/submittable_executor.cc new file mode 100644 index 00000000..4dd1a4e9 --- /dev/null +++ b/cpp/platform/impl/windows/submittable_executor.cc @@ -0,0 +1,59 @@ +// Copyright 2021 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "platform/impl/windows/submittable_executor.h" + +#include "platform/impl/windows/executor.h" +#include "platform/public/logging.h" + +namespace location { +namespace nearby { +namespace windows { + +SubmittableExecutor::SubmittableExecutor() : SubmittableExecutor(1) {} + +SubmittableExecutor::SubmittableExecutor(int32_t max_concurrancy) + : executor_(std::make_unique(max_concurrancy)), + shut_down_(false) {} + +bool SubmittableExecutor::DoSubmit(Runnable&& wrapped_callable) { + if (!shut_down_) { + executor_->Execute(std::move(wrapped_callable)); + return true; + } + + return false; +} + +// https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executor.html#execute-java.lang.Runnable- +void SubmittableExecutor::Execute(Runnable&& runnable) { + if (!shut_down_) { + executor_->Execute(std::move(runnable)); + } else { + NEARBY_LOGS(ERROR) << "Error: " << __func__ + << ": Attempt to Execute on a shutdown executor."; + } +} + +// https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html#shutdown-- +void SubmittableExecutor::Shutdown() { + if (!shut_down_) { + executor_->Shutdown(); + shut_down_ = true; + } +} + +} // namespace windows +} // namespace nearby +} // namespace location diff --git a/cpp/platform/impl/windows/submittable_executor.h b/cpp/platform/impl/windows/submittable_executor.h index 0fdc0c88..dad778c4 100644 --- a/cpp/platform/impl/windows/submittable_executor.h +++ b/cpp/platform/impl/windows/submittable_executor.h @@ -16,6 +16,7 @@ #define PLATFORM_IMPL_WINDOWS_SUBMITTABLE_EXECUTOR_H_ #include "platform/api/submittable_executor.h" +#include "platform/impl/windows/executor.h" namespace location { namespace nearby { @@ -27,22 +28,24 @@ namespace windows { // Platform must override bool submit(std::function) method. class SubmittableExecutor : public api::SubmittableExecutor { public: - // TODO(b/184975123): replace with real implementation. + SubmittableExecutor(); + SubmittableExecutor(int32_t maxConcurrancy); ~SubmittableExecutor() override = default; // Submit a callable (with no delay). // Returns true, if callable was submitted, false otherwise. // Callable is not submitted if shutdown is in progress. - // TODO(b/184975123): replace with real implementation. - bool DoSubmit(Runnable&& wrapped_callable) override { return false; } + bool DoSubmit(Runnable&& wrapped_callable) override; // https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executor.html#execute-java.lang.Runnable- - // TODO(b/184975123): replace with real implementation. - void Execute(Runnable&& runnable) override {} + void Execute(Runnable&& runnable) override; // https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html#shutdown-- - // TODO(b/184975123): replace with real implementation. - void Shutdown() override {} + void Shutdown() override; + + private: + std::unique_ptr executor_; + std::atomic_bool shut_down_; }; } // namespace windows diff --git a/cpp/platform/impl/windows/submittable_executor_test.cc b/cpp/platform/impl/windows/submittable_executor_test.cc new file mode 100644 index 00000000..c9497628 --- /dev/null +++ b/cpp/platform/impl/windows/submittable_executor_test.cc @@ -0,0 +1,275 @@ +// Copyright 2021 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +#include "platform/impl/windows/submittable_executor.h" + +#include + +#include "gtest/gtest.h" + +TEST(SubmittableExecutorTests, SingleThreadedExecuteSucceeds) { + // Arrange + std::string expected("runnable 1"); + + std::unique_ptr + submittableExecutor = + std::make_unique(); + std::string output = std::string(); + // Container to note threads that ran + std::unique_ptr> threadIds = + std::make_unique>(); + + threadIds->push_back(GetCurrentThreadId()); + + // Act + submittableExecutor->Execute([&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output.append("runnable 1"); + }); + + Sleep(1); // Yield the thread + + // Assert + // We should've run 1 time on the main thread, and 1 times on the + // workerThread + ASSERT_EQ(threadIds->size(), 2); + // We should still be on the main thread + ASSERT_EQ(GetCurrentThreadId(), threadIds->at(0)); + // We should've run all runnables on the worker thread + ASSERT_EQ(output, expected); + + submittableExecutor->Shutdown(); +} + +TEST(SubmittableExecutorTests, SingleThreadedExecuteAfterShutdownFails) { + // Arrange + std::string expected(""); + + std::unique_ptr + submittableExecutor = + std::make_unique(); + std::string output = std::string(); + // Container to note threads that ran + std::unique_ptr> threadIds = + std::make_unique>(); + + threadIds->push_back(GetCurrentThreadId()); + + submittableExecutor->Shutdown(); + + // Act + submittableExecutor->Execute([&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output.append("runnable 1"); + }); + + Sleep(1); // Yield the thread + + // Assert + // We should've run 1 time on the main thread, and 0 times on the + // workerThread + ASSERT_EQ(threadIds->size(), 1); + // We should still be on the main thread + ASSERT_EQ(GetCurrentThreadId(), threadIds->at(0)); + // We should've run all runnables on the worker thread + ASSERT_EQ(output, expected); +} + +TEST(SubmittableExecutorTests, SingleThreadedDoSubmitSucceeds) { + // Arrange + std::string expected("runnable 1"); + + std::unique_ptr + submittableExecutor = + std::make_unique(); + std::string output = std::string(); + // Container to note threads that ran + std::unique_ptr> threadIds = + std::make_unique>(); + + threadIds->push_back(GetCurrentThreadId()); + + // Act + auto result = submittableExecutor->DoSubmit([&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output.append("runnable 1"); + }); + + Sleep(1); // Yield the thread + + // Assert + // We should've said we were going to run this one + ASSERT_TRUE(result); + // We should've run 1 time on the main thread, and 1 times on the + // workerThread + ASSERT_EQ(threadIds->size(), 2); + // We should still be on the main thread + ASSERT_EQ(GetCurrentThreadId(), threadIds->at(0)); + // We should've run all runnables on the worker thread + ASSERT_EQ(output, expected); + + submittableExecutor->Shutdown(); +} + +TEST(SubmittableExecutorTests, + SingleThreadedDoSubmitAfterShutdownReturnsFalse) { + // Arrange + std::string expected(""); + + std::unique_ptr + submittableExecutor = + std::make_unique(); + std::unique_ptr output = std::make_unique(); + // Container to note threads that ran + std::unique_ptr> threadIds = + std::make_unique>(); + + threadIds->push_back(GetCurrentThreadId()); + + submittableExecutor->Shutdown(); + + // Act + auto result = submittableExecutor->DoSubmit([&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output->append("runnable 1"); + }); + + Sleep(1); // Yield the thread + + // Assert + // We should've said we were going to run this one + ASSERT_FALSE(result); + // We should've run 1 time on the main thread, and 1 times on the + // workerThread + ASSERT_EQ(threadIds->size(), 1); + // We should still be on the main thread + ASSERT_EQ(GetCurrentThreadId(), threadIds->at(0)); + // We should've run all runnables on the worker thread + ASSERT_EQ(*output.get(), expected); +} + +TEST(SubmittableExecutorTests, SingleThreadedExecuteMultipleTasksSucceeds) { + // Arrange + std::string expected( + "runnable 1, runnable 2, runnable 3, runnable 4, runnable 5"); + + std::unique_ptr + submittableExecutor = + std::make_unique(); + std::unique_ptr output = std::make_unique(); + // Container to note threads that ran + std::unique_ptr> threadIds = + std::make_unique>(); + + threadIds->push_back(GetCurrentThreadId()); + + // Act + submittableExecutor->Execute([&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output->append("runnable 1, "); + }); + submittableExecutor->Execute([&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output->append("runnable 2, "); + }); + submittableExecutor->Execute([&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output->append("runnable 3, "); + }); + submittableExecutor->Execute([&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output->append("runnable 4, "); + }); + submittableExecutor->Execute([&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output->append("runnable 5"); + }); + + Sleep(1); // Yield the thread + + // Assert + // We should've run 1 time on the main thread, and 5 times on the + // workerThread + ASSERT_EQ(threadIds->size(), 6); + // We should still be on the main thread + ASSERT_EQ(GetCurrentThreadId(), threadIds->at(0)); + // We should've run all runnables on the worker thread + auto workerThreadId = threadIds->at(1); + for (int index = 1; index < threadIds->size(); index++) { + ASSERT_EQ(threadIds->at(index), workerThreadId); + } + + // We should of run them in the order submitted + ASSERT_EQ(*output.get(), expected); + + submittableExecutor->Shutdown(); +} + +TEST(SubmittableExecutorTests, SingleThreadedDoSubmitMultipleTasksSucceeds) { + // Arrange + std::string expected( + "runnable 1, runnable 2, runnable 3, runnable 4, runnable 5"); + + std::unique_ptr + submittableExecutor = + std::make_unique(); + std::unique_ptr output = std::make_unique(); + // Container to note threads that ran + std::unique_ptr> threadIds = + std::make_unique>(); + + threadIds->push_back(GetCurrentThreadId()); + + // Act + auto result = submittableExecutor->DoSubmit([&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output->append("runnable 1, "); + }); + result |= submittableExecutor->DoSubmit([&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output->append("runnable 2, "); + }); + result |= submittableExecutor->DoSubmit([&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output->append("runnable 3, "); + }); + result |= submittableExecutor->DoSubmit([&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output->append("runnable 4, "); + }); + result |= submittableExecutor->DoSubmit([&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output->append("runnable 5"); + }); + + Sleep(1); // Yield the thread + + // Assert + // All of these should have submitted + ASSERT_TRUE(result); + // We should've run 1 time on the main thread, and 5 times on the + // workerThread + ASSERT_EQ(threadIds->size(), 6); + // We should still be on the main thread + ASSERT_EQ(GetCurrentThreadId(), threadIds->at(0)); + // We should've run all runnables on the worker thread + auto workerThreadId = threadIds->at(1); + for (int index = 1; index < threadIds->size(); index++) { + ASSERT_EQ(threadIds->at(index), workerThreadId); + } + + // We should of run them in the order submitted + ASSERT_EQ(*output.get(), expected); + + submittableExecutor->Shutdown(); +}