Added a class to schedule multiple tasks

PiperOrigin-RevId: 652871095

internal/platform/implementation/windows/BUILD

@@ -38,6 +38,7 @@ cc_library(
         "scheduled_executor.h",
         "settable_future.h",
         "submittable_executor.h",
+        "task_scheduler.h",
         "timer.h",
         "utils.h",
     ],
@@ -179,6 +180,7 @@ cc_library(
         "session_manager.cc",
         "submittable_executor.cc",
         "system_clock.cc",
+        "task_scheduler.cc",
         "thread_pool.cc",
         "utils.cc",
         "webrtc.cc",
@@ -287,6 +289,7 @@ cc_test(
         "preferences_repository_test.cc",
         "scheduled_executor_test.cc",
         "submittable_executor_test.cc",
+        "task_scheduler_test.cc",
         "thread_pool_test.cc",
         "timer_test.cc",
         "utils_test.cc",

internal/platform/implementation/windows/task_scheduler.cc

@@ -0,0 +1,148 @@
+// Copyright 2024 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/task_scheduler.h"
+
+#include <Windows.h>
+
+#include <cstdint>
+#include <memory>
+#include <utility>
+
+#include "absl/time/time.h"
+#include "internal/platform/implementation/cancelable.h"
+#include "internal/platform/logging.h"
+#include "internal/platform/mutex_lock.h"
+#include "internal/platform/runnable.h"
+
+namespace nearby::windows {
+namespace {
+void CALLBACK TimerRoutine(PVOID lpParam, BOOLEAN TimerOrWaitFired) {
+  Runnable* task = reinterpret_cast<Runnable*>(lpParam);
+  if (task != nullptr) {
+    (*task)();
+  }
+}
+}  // namespace
+
+TaskScheduler::TaskScheduler() {
+  NEARBY_LOGS(INFO) << __func__ << ": Created task scheduler: " << this;
+}
+TaskScheduler::~TaskScheduler() {
+  Shutdown();
+  NEARBY_LOGS(INFO) << __func__ << ": Destroyed task scheduler: " << this;
+}
+
+std::shared_ptr<api::Cancelable> TaskScheduler::Schedule(
+    Runnable&& runnable, absl::Duration duration) {
+  return Schedule(std::move(runnable), duration, absl::ZeroDuration());
+}
+
+std::shared_ptr<api::Cancelable> TaskScheduler::Schedule(
+    Runnable&& runnable, absl::Duration duration,
+    absl::Duration repeat_interval) {
+  MutexLock lock(&mutex_);
+  NEARBY_LOGS(INFO) << __func__
+                    << ": Scheduling task on task scheduler:" << this
+                    << ", duration: " << absl::ToInt64Milliseconds(duration)
+                    << "ms, repeat_interval: "
+                    << absl::ToInt64Milliseconds(repeat_interval) << "ms";
+
+  std::shared_ptr<ScheduledTask> task =
+      std::make_shared<ScheduledTask>(*this, std::move(runnable));
+
+  HANDLE timer_handle = nullptr;
+  if (!CreateTimerQueueTimer(&timer_handle, nullptr,
+                             static_cast<WAITORTIMERCALLBACK>(TimerRoutine),
+                             task->runnable(),
+                             absl::ToInt64Milliseconds(duration),
+                             absl::ToInt64Milliseconds(repeat_interval), 0)) {
+    NEARBY_LOGS(ERROR)
+        << __func__
+        << ": Failed to create timer queue timer in task scheduler:" << this
+        << " error: " << GetLastError();
+    return nullptr;
+  }
+
+  task->SetTimerHandle(reinterpret_cast<intptr_t>(timer_handle));
+  scheduled_tasks_.insert({reinterpret_cast<intptr_t>(timer_handle), task});
+  NEARBY_LOGS(INFO) << __func__ << ": Scheduled task " << task.get()
+                    << " on task scheduler:" << this
+                    << " timer handle: " << task->timer_handle();
+  return task;
+}
+
+void TaskScheduler::Shutdown() {
+  MutexLock lock(&mutex_);
+  NEARBY_LOGS(INFO) << __func__ << ": Shutting down task scheduler:" << this;
+  if (is_shutdown_) {
+    return;
+  }
+  for (auto& task : scheduled_tasks_) {
+    // Wait for running task to finish.
+    if (!DeleteTimerQueueTimer(
+            nullptr, reinterpret_cast<HANDLE>(task.second->timer_handle()),
+            INVALID_HANDLE_VALUE)) {
+      if (GetLastError() != ERROR_IO_PENDING) {
+        NEARBY_LOGS(ERROR) << __func__
+                           << ": Failed to delete timer queue timer: "
+                           << task.second->timer_handle()
+                           << " error: " << GetLastError();
+      }
+    }
+  }
+  scheduled_tasks_.clear();
+  is_shutdown_ = true;
+  NEARBY_LOGS(INFO) << __func__ << ": Shut down task scheduler:" << this;
+}
+
+TaskScheduler::ScheduledTask::ScheduledTask(TaskScheduler& task_scheduler,
+                                            Runnable&& runnable)
+    : task_scheduler_(&task_scheduler), runnable_(std::move(runnable)) {}
+
+bool TaskScheduler::ScheduledTask::Cancel() {
+  NEARBY_LOGS(INFO) << __func__ << ": Cancelling timer " << timer_handle_
+                    << " from task scheduler:" << this;
+  return task_scheduler_->remove_scheduled_task(timer_handle_);
+}
+
+void TaskScheduler::ScheduledTask::SetTimerHandle(intptr_t timer_handle) {
+  timer_handle_ = timer_handle;
+}
+
+Runnable* TaskScheduler::ScheduledTask::runnable() { return &runnable_; }
+
+intptr_t TaskScheduler::ScheduledTask::timer_handle() { return timer_handle_; }
+
+bool TaskScheduler::remove_scheduled_task(intptr_t timer_handle) {
+  MutexLock lock(&mutex_);
+  auto it = scheduled_tasks_.find(timer_handle);
+  if (it == scheduled_tasks_.end()) {
+    return false;
+  }
+
+  // Wait for running task to finish.
+  if (!DeleteTimerQueueTimer(nullptr, reinterpret_cast<HANDLE>(timer_handle),
+                             INVALID_HANDLE_VALUE)) {
+    if (GetLastError() != ERROR_IO_PENDING) {
+      NEARBY_LOGS(ERROR) << __func__ << ": Failed to delete timer queue timer: "
+                         << timer_handle << " error: " << GetLastError();
+    }
+  }
+
+  scheduled_tasks_.erase(it);
+  return true;
+}
+
+}  // namespace nearby::windows

internal/platform/implementation/windows/task_scheduler.h

@@ -0,0 +1,78 @@
+// Copyright 2024 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 THIRD_PARTY_NEARBY_INTERNAL_PLATFORM_IMPLEMENTATION_WINDOWS_TASK_SCHEDULER_H_
+#define THIRD_PARTY_NEARBY_INTERNAL_PLATFORM_IMPLEMENTATION_WINDOWS_TASK_SCHEDULER_H_
+
+#include <cstdint>
+#include <memory>
+
+#include "absl/base/thread_annotations.h"
+#include "absl/container/flat_hash_map.h"
+#include "absl/time/time.h"
+#include "internal/platform/implementation/cancelable.h"
+#include "internal/platform/mutex.h"
+#include "internal/platform/runnable.h"
+
+namespace nearby::windows {
+
+// TaskScheduler is a utility class to scheduled a runnable task. It is used by
+// the ScheduledExecutor and timer implementations.
+class TaskScheduler {
+ public:
+  TaskScheduler();
+  ~TaskScheduler();
+
+  std::shared_ptr<api::Cancelable> Schedule(Runnable&& runnable,
+                                            absl::Duration duration);
+
+  std::shared_ptr<api::Cancelable> Schedule(Runnable&& runnable,
+                                            absl::Duration duration,
+                                            absl::Duration repeat_interval)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+
+  void Shutdown() ABSL_LOCKS_EXCLUDED(mutex_);
+
+ private:
+  class ScheduledTask : public api::Cancelable {
+   public:
+    explicit ScheduledTask(TaskScheduler& task_scheduler, Runnable&& runnable);
+    ~ScheduledTask() override = default;
+
+    // Note: not support to cancel and shutdown a scheduled task in the
+    // callback.
+    bool Cancel() override;
+
+    void SetTimerHandle(intptr_t timer_handle);
+    intptr_t timer_handle();
+
+    Runnable* runnable();
+
+   private:
+    TaskScheduler* task_scheduler_;
+    Runnable runnable_;
+    intptr_t timer_handle_;
+  };
+
+  bool remove_scheduled_task(intptr_t timer_handle) ABSL_LOCKS_EXCLUDED(mutex_);
+
+  Mutex mutex_;
+  bool is_shutdown_ ABSL_GUARDED_BY(mutex_) = false;
+  absl::flat_hash_map<intptr_t, std::shared_ptr<ScheduledTask>> scheduled_tasks_
+      ABSL_GUARDED_BY(mutex_);
+};
+
+}  // namespace nearby::windows
+
+#endif  // THIRD_PARTY_NEARBY_INTERNAL_PLATFORM_IMPLEMENTATION_WINDOWS_TASK_SCHEDULER_H_

internal/platform/implementation/windows/task_scheduler_test.cc

@@ -0,0 +1,113 @@
+// Copyright 2024 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/task_scheduler.h"
+
+#include <memory>
+#include <vector>
+
+#include "gtest/gtest.h"
+#include "absl/synchronization/notification.h"
+#include "absl/time/clock.h"
+#include "absl/time/time.h"
+
+namespace nearby::windows {
+
+namespace {
+constexpr absl::Duration kTaskDuration = absl::Milliseconds(500);
+
+TEST(TaskScheduler, ScheduleOneTask) {
+  TaskScheduler task_scheduler;
+  int counter = 0;
+  absl::Notification notification;
+  auto task = [&counter, &notification]() {
+    counter++;
+    notification.Notify();
+  };
+  auto cancelable = task_scheduler.Schedule(task, absl::Milliseconds(100));
+  EXPECT_NE(cancelable, nullptr);
+  EXPECT_TRUE(notification.WaitForNotificationWithTimeout(kTaskDuration));
+  EXPECT_EQ(counter, 1);
+}
+
+TEST(TaskScheduler, ScheduleOneEarlierTaskAfterOneTask) {
+  TaskScheduler task_scheduler;
+  std::vector<int> result;
+  absl::Notification notification1;
+  absl::Notification notification2;
+  auto task1 = [&result, &notification1]() {
+    result.push_back(1);
+    notification1.Notify();
+  };
+
+  auto task2 = [&result, &notification2]() {
+    result.push_back(2);
+    notification2.Notify();
+  };
+  task_scheduler.Schedule(task1, absl::Milliseconds(100));
+  task_scheduler.Schedule(task2, absl::Milliseconds(50));
+  EXPECT_TRUE(notification1.WaitForNotificationWithTimeout(kTaskDuration));
+  EXPECT_TRUE(notification2.WaitForNotificationWithTimeout(kTaskDuration));
+  ASSERT_EQ(result.size(), 2);
+  EXPECT_EQ(result.at(0), 2);
+  EXPECT_EQ(result.at(1), 1);
+}
+
+TEST(TaskScheduler, CancelScheduledTask) {
+  TaskScheduler task_scheduler;
+  int counter = 0;
+  absl::Notification notification;
+  auto task = [&counter, &notification]() {
+    counter++;
+    notification.Notify();
+  };
+  auto cancelable = task_scheduler.Schedule(task, absl::Milliseconds(200));
+  cancelable->Cancel();
+  EXPECT_FALSE(notification.WaitForNotificationWithTimeout(kTaskDuration));
+  EXPECT_EQ(counter, 0);
+}
+
+TEST(TaskScheduler, ShundownShouldWaitForScheduledTaskToFinish) {
+  TaskScheduler task_scheduler;
+  int counter = 0;
+  absl::Notification notification;
+  auto task = [&counter, &notification]() {
+    absl::SleepFor(kTaskDuration);
+    counter++;
+    notification.Notify();
+  };
+  auto cancelable = task_scheduler.Schedule(task, absl::Milliseconds(100));
+  absl::SleepFor(absl::Milliseconds(200));
+  task_scheduler.Shutdown();
+  EXPECT_EQ(counter, 1);
+}
+
+TEST(TaskScheduler, CancelCancleledScheduledTask) {
+  TaskScheduler task_scheduler;
+  int counter = 0;
+  absl::Notification notification;
+  auto task = [&counter, &notification]() {
+    absl::SleepFor(kTaskDuration);
+    counter++;
+    notification.Notify();
+  };
+  auto cancelable = task_scheduler.Schedule(task, absl::Milliseconds(200));
+  EXPECT_TRUE(cancelable->Cancel());
+  EXPECT_FALSE(cancelable->Cancel());
+  task_scheduler.Shutdown();
+  EXPECT_EQ(counter, 0);
+}
+
+}  // namespace
+}  // namespace nearby::windows