diff --git a/cpp/platform/impl/windows/BUILD b/cpp/platform/impl/windows/BUILD index d811540e..95a94a38 100644 --- a/cpp/platform/impl/windows/BUILD +++ b/cpp/platform/impl/windows/BUILD @@ -61,6 +61,7 @@ cc_library( "executor.h", "mutex.h", "runner.h", + "scheduled_executor.h", "server_sync.h", "submittable_executor.h", "thread_pool.h", @@ -105,6 +106,7 @@ cc_library( "executor.cc", "mutex.cc", "platform.cc", + "scheduled_executor.cc", "submittable_executor.cc", "thread_pool.cc", "utils.cc", @@ -120,6 +122,7 @@ cc_library( "executor.h", "mutex.h", "runner.h", + "scheduled_executor.h", "submittable_executor.h", "thread_pool.h", ], @@ -170,6 +173,7 @@ cc_test( "input_file_test.cc", "mutex_test.cc", "output_file_test.cc", + "scheduled_executor_test.cc", "submittable_executor_test.cc", ], copts = ["-Iplatform/impl/windows/generated"], diff --git a/cpp/platform/impl/windows/cancelable.h b/cpp/platform/impl/windows/cancelable.h index b538ea5e..1f9b2644 100644 --- a/cpp/platform/impl/windows/cancelable.h +++ b/cpp/platform/impl/windows/cancelable.h @@ -15,6 +15,8 @@ #ifndef PLATFORM_IMPL_WINDOWS_CANCELABLE_H_ #define PLATFORM_IMPL_WINDOWS_CANCELABLE_H_ +#include + #include "platform/api/cancelable.h" namespace location { @@ -25,11 +27,20 @@ namespace windows { // long-running operations. class Cancelable : public api::Cancelable { public: - // TODO(b/184975123): replace with real implementation. + Cancelable(HANDLE handle) : handle_(handle) {} ~Cancelable() override = default; - // TODO(b/184975123): replace with real implementation. - bool Cancel() override { return false; }; + bool Cancel() override { + if (CancelWaitableTimer(handle_)) { + CloseHandle(handle_); + return true; + } + + return false; + }; + + private: + HANDLE handle_; }; } // namespace windows diff --git a/cpp/platform/impl/windows/scheduled_executor.cc b/cpp/platform/impl/windows/scheduled_executor.cc new file mode 100644 index 00000000..af61a735 --- /dev/null +++ b/cpp/platform/impl/windows/scheduled_executor.cc @@ -0,0 +1,140 @@ +// 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/scheduled_executor.h" + +#include "platform/impl/windows/cancelable.h" +#include "platform/public/logging.h" + +namespace location { +namespace nearby { +namespace windows { + +class ScheduledExecutorException : public std::runtime_error { + public: + ScheduledExecutorException() : std::runtime_error("") {} + ScheduledExecutorException(const std::string& message) + : std::runtime_error(message) {} + virtual const char* what() const throw() { + return "WaitableTimer creation failed"; + } +}; + +class TimerData { + public: + TimerData(ScheduledExecutor* scheduledExecutor, + std::function runnable, HANDLE waitableTimer) + : scheduled_executor_(scheduledExecutor), + runnable_(std::move(runnable)), + waitable_timer_handle_(waitableTimer) {} + + ScheduledExecutor* GetScheduledExecutor() { return scheduled_executor_; } + std::function GetRunnable() { return runnable_; } + HANDLE GetWaitableTimerHandle() { return waitable_timer_handle_; } + + private: + ScheduledExecutor* scheduled_executor_; + std::function runnable_; + HANDLE waitable_timer_handle_; +}; + +void WINAPI ScheduledExecutor::_TimerProc(LPVOID argToCompletionRoutine, + DWORD dwTimerLowValue, + DWORD dwTimerHighValue) { + TimerData* timerData; + DWORD threadId = GetCurrentThreadId(); + + _ASSERT(argToCompletionRoutine != NULL); + if (NULL == argToCompletionRoutine) { + NEARBY_LOGS(ERROR) + << "Error: " << __func__ + << ": TimerProc argument argToCompletionRoutine was null."; + + return; + } + + timerData = static_cast(argToCompletionRoutine); + timerData->GetScheduledExecutor()->Execute(timerData->GetRunnable()); + + // Get the waitable timer and destroy it + CloseHandle(timerData->GetWaitableTimerHandle()); + free(timerData); + return; +} + +ScheduledExecutor::ScheduledExecutor() + : executor_(std::make_unique()), + shut_down_(false) {} + +// Cancelable is kept both in the executor context, and in the caller context. +// We want Cancelable to live until both caller and executor are done with it. +// Exclusive ownership model does not work for this case; +// using std:shared_ptr<> instead of std::unique_ptr<>. +std::shared_ptr ScheduledExecutor::Schedule( + Runnable&& runnable, absl::Duration duration) { + if (shut_down_) { + return nullptr; + } + + // Create the waitable timer + // Create a name for this timer + // TODO: (jfcarroll) construct a timer name based on ?? + char buffer[TIMER_NAME_BUFFER_SIZE]; + + snprintf(buffer, TIMER_NAME_BUFFER_SIZE, "PID:%ld", GetCurrentProcessId()); + + HANDLE waitableTimer = CreateWaitableTimerA(NULL, true, buffer); + + if (waitableTimer == NULL) { + throw ScheduledExecutorException("WaitableTimer creation failed"); + } + + waitable_timers_.push_back(waitableTimer); + + // Create the delay value - due time + LARGE_INTEGER dueTime; + dueTime.QuadPart = -(absl::ToChronoNanoseconds(duration).count() / 100); + + TimerData* timerData = new TimerData(this, runnable, waitableTimer); + + BOOL result = SetWaitableTimer(waitableTimer, &dueTime, 0, _TimerProc, + timerData, false); + + if (result == 0) { + NEARBY_LOGS(ERROR) << "Error: " << __func__ << ": Failed to set the timer."; + return nullptr; + } + + return std::make_shared(waitableTimer); +} + +// https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executor.html#execute-java.lang.Runnable- +void ScheduledExecutor::Execute(Runnable&& runnable) { + if (shut_down_) { + return; + } + + executor_->Execute(std::move(runnable)); +} + +// https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html#shutdown-- +void ScheduledExecutor::Shutdown() { + if (!shut_down_) { + shut_down_ = true; + executor_->Shutdown(); + } +} +} // namespace windows +} // namespace nearby +} // namespace location diff --git a/cpp/platform/impl/windows/scheduled_executor.h b/cpp/platform/impl/windows/scheduled_executor.h index 790ed70d..68db9148 100644 --- a/cpp/platform/impl/windows/scheduled_executor.h +++ b/cpp/platform/impl/windows/scheduled_executor.h @@ -15,38 +15,47 @@ #ifndef PLATFORM_IMPL_WINDOWS_SCHEDULED_EXECUTOR_H_ #define PLATFORM_IMPL_WINDOWS_SCHEDULED_EXECUTOR_H_ +#include + #include "platform/api/scheduled_executor.h" +#include "platform/impl/windows/executor.h" namespace location { namespace nearby { namespace windows { +#define TIMER_NAME_BUFFER_SIZE 64 + // An Executor that can schedule commands to run after a given delay, or to // execute periodically. // // https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ScheduledExecutorService.html class ScheduledExecutor : public api::ScheduledExecutor { public: - // TODO(b/184975123): replace with real implementation. + ScheduledExecutor(); + ~ScheduledExecutor() override = default; // Cancelable is kept both in the executor context, and in the caller context. // We want Cancelable to live until both caller and executor are done with it. // Exclusive ownership model does not work for this case; // using std:shared_ptr<> instead if std::unique_ptr<>. - // TODO(b/184975123): replace with real implementation. std::shared_ptr Schedule(Runnable&& runnable, - absl::Duration duration) override { - return nullptr; - } + absl::Duration duration) 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: + static void WINAPI _TimerProc(LPVOID lpArgToCompletionRoutine, + DWORD dwTimerLowValue, DWORD dwTimerHighValue); + + std::unique_ptr executor_; + std::vector waitable_timers_; + std::atomic_bool shut_down_; }; } // namespace windows diff --git a/cpp/platform/impl/windows/scheduled_executor_test.cc b/cpp/platform/impl/windows/scheduled_executor_test.cc new file mode 100644 index 00000000..c20fb583 --- /dev/null +++ b/cpp/platform/impl/windows/scheduled_executor_test.cc @@ -0,0 +1,174 @@ +// 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/scheduled_executor.h" + +#include + +#include "gtest/gtest.h" + +TEST(ScheduledExecutorTests, ExecuteSucceeds) { + // 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(ScheduledExecutorTests, ScheduleSucceeds) { + // 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()); + + std::chrono::system_clock::time_point timeNow = + std::chrono::system_clock::now(); + std::chrono::system_clock::time_point timeExecuted; + + // Act + submittableExecutor->Schedule( + [&output, &threadIds, &timeExecuted]() { + timeExecuted = std::chrono::system_clock::now(); + threadIds->push_back(GetCurrentThreadId()); + output.append("runnable 1"); + }, + absl::Milliseconds(50)); + + SleepEx(100, true); // Yield the thread + + submittableExecutor->Shutdown(); + + auto difference = std::chrono::duration_cast( + timeExecuted - timeNow) + .count(); + + // Assert + // We should've run 1 time on the main thread, and 1 times on the + // workerThread + ASSERT_TRUE(difference >= 50) << "difference was: " << difference; + ASSERT_TRUE(difference < 100) << "difference was: " << difference; + + 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); +} + +TEST(ScheduledExecutorTests, CancelSucceeds) { + // 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()); + + // Act + auto cancelable = submittableExecutor->Schedule( + [&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output.append("runnable 1"); + }, + absl::Milliseconds(1000)); + + SleepEx(100, true); // Yield the thread + + auto actual = cancelable->Cancel(); + + // Assert + ASSERT_TRUE(actual); + 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); + + submittableExecutor->Shutdown(); +} + +TEST(ScheduledExecutorTests, CancelAfterStartedFails) { + // 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 cancelable = submittableExecutor->Schedule( + [&output, &threadIds]() { + threadIds->push_back(GetCurrentThreadId()); + output.append("runnable 1"); + }, + absl::Milliseconds(100)); + + SleepEx(1000, true); // Yield the thread + + auto actual = cancelable->Cancel(); + + submittableExecutor->Shutdown(); + + // Assert + ASSERT_FALSE(actual); + 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); +}