Initial implementation of ScheduledExecutor

PiperOrigin-RevId: 396036314

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"],

cpp/platform/impl/windows/cancelable.h

@@ -15,6 +15,8 @@
 #ifndef PLATFORM_IMPL_WINDOWS_CANCELABLE_H_
 #define PLATFORM_IMPL_WINDOWS_CANCELABLE_H_
 
+#include <windows.h>
+
 #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

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<void()> runnable, HANDLE waitableTimer)
+      : scheduled_executor_(scheduledExecutor),
+        runnable_(std::move(runnable)),
+        waitable_timer_handle_(waitableTimer) {}
+
+  ScheduledExecutor* GetScheduledExecutor() { return scheduled_executor_; }
+  std::function<void()> GetRunnable() { return runnable_; }
+  HANDLE GetWaitableTimerHandle() { return waitable_timer_handle_; }
+
+ private:
+  ScheduledExecutor* scheduled_executor_;
+  std::function<void()> 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<TimerData*>(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<nearby::windows::Executor>()),
+      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<api::Cancelable> 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<nearby::windows::Cancelable>(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

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 <windows.h>
+
 #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<api::Cancelable> 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<nearby::windows::Executor> executor_;
+  std::vector<HANDLE> waitable_timers_;
+  std::atomic_bool shut_down_;
 };
 
 }  // namespace windows

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 <utility>
+
+#include "gtest/gtest.h"
+
+TEST(ScheduledExecutorTests, ExecuteSucceeds) {
+  // Arrange
+  std::string expected("runnable 1");
+
+  std::unique_ptr<location::nearby::windows::ScheduledExecutor>
+      submittableExecutor =
+          std::make_unique<location::nearby::windows::ScheduledExecutor>();
+  std::string output = std::string();
+  // Container to note threads that ran
+  std::unique_ptr<std::vector<DWORD>> threadIds =
+      std::make_unique<std::vector<DWORD>>();
+
+  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<location::nearby::windows::ScheduledExecutor>
+      submittableExecutor =
+          std::make_unique<location::nearby::windows::ScheduledExecutor>();
+  std::string output = std::string();
+  // Container to note threads that ran
+  std::unique_ptr<std::vector<DWORD>> threadIds =
+      std::make_unique<std::vector<DWORD>>();
+
+  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<std::chrono::milliseconds>(
+                        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<location::nearby::windows::ScheduledExecutor>
+      submittableExecutor =
+          std::make_unique<location::nearby::windows::ScheduledExecutor>();
+  std::string output = std::string();
+  // Container to note threads that ran
+  std::unique_ptr<std::vector<DWORD>> threadIds =
+      std::make_unique<std::vector<DWORD>>();
+
+  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<location::nearby::windows::ScheduledExecutor>
+      submittableExecutor =
+          std::make_unique<location::nearby::windows::ScheduledExecutor>();
+  std::string output = std::string();
+  // Container to note threads that ran
+  std::unique_ptr<std::vector<DWORD>> threadIds =
+      std::make_unique<std::vector<DWORD>>();
+
+  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);
+}