Implemented a TimerQueue class and a Timer class

The Windows implementation of the Timer class (in timer.h) used Timer
Queues. This brings in implementation for a mostly Windows complient
TimerQueue class to work with the Timer class.

internal/platform/implementation/linux/timer.cc

@@ -0,0 +1,120 @@
+// 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 "internal/platform/implementation/linux/timer.h"
+#include "internal/platform/implementation/linux/timer_queue.h"
+
+#include "absl/synchronization/mutex.h"
+#include "internal/platform/logging.h"
+
+namespace nearby {
+namespace linux {
+
+Timer::~Timer() { Stop(); }
+
+bool Timer::Create(int delay, int interval,
+                   absl::AnyInvocable<void()> callback) {
+  absl::MutexLock lock(&mutex_);
+
+  if ((delay < 0) || (interval < 0)) {
+    NEARBY_LOGS(WARNING) << "Delay and interval shouldn\'t be negative value.";
+    return false;
+  }
+
+  if (timer_queue_handle_) {
+    return false;
+  }
+
+  timer_queue_handle_ = TimerQueue::CreateTimerQueue();
+  if (!timer_queue_handle_) {
+    NEARBY_LOGS(ERROR) << "Failed to create timer queue.";
+    return false;
+  }
+
+  delay_ = delay;
+  interval_ = interval;
+  callback_ = std::move(callback);
+
+  absl::StatusOr<uint16_t> createStatus = timer_queue_handle_->CreateTimerQueueTimer(TimerRoutine,
+                             &callback_, std::chrono::milliseconds(delay), std::chrono::milliseconds(interval), TimerQueue::WT_EXECUTEDEFAULT);
+
+  if (!createStatus.ok()) {
+    if (!timer_queue_handle_->DeleteTimerQueueEx(TimerQueue::CE_IMEDIATERETURN).ok()) {
+      NEARBY_LOGS(ERROR) << "Failed to create timer in timer queue.";
+    }
+    delete timer_queue_handle_.release();
+    return false;
+  }
+  handle_ = createStatus.value();
+  return true;
+}
+
+bool Timer::Stop() {
+  absl::MutexLock lock(&mutex_);
+
+  if (!timer_queue_handle_) {
+    return true;
+  }
+
+  absl::Status deleteStatus = timer_queue_handle_->DeleteTimerQueueTimer(handle_, TimerQueue::CE_IMEDIATERETURN);
+
+  if (!deleteStatus.ok()) {
+    NEARBY_LOGS(ERROR) << "Failed to delete timer from queue: " << deleteStatus.message();
+  }
+
+  handle_ = 0;
+
+  deleteStatus = timer_queue_handle_->DeleteTimerQueueEx(TimerQueue::CE_IMEDIATERETURN);
+
+  if (!deleteStatus.ok()) {
+    NEARBY_LOGS(ERROR) << "Failed to delete timer queue: " << deleteStatus.message();
+    return false;
+  }
+
+  timer_queue_handle_ = nullptr;
+  return true;
+}
+
+bool Timer::FireNow() {
+  absl::MutexLock lock(&mutex_);
+
+  if (!timer_queue_handle_ || !callback_) {
+    return false;
+  }
+
+  if (task_executor_ == nullptr) {
+    task_executor_ = std::make_unique<SubmittableExecutor>();
+  }
+
+  if (task_executor_ == nullptr) {
+    NEARBY_LOGS(ERROR)
+        << "Failed to fire the task due to cannot create executor.";
+    return false;
+  }
+
+  task_executor_->Execute([&]() { callback_(); });
+
+  return true;
+}
+
+void Timer::TimerRoutine(void *lpParam) {
+  absl::AnyInvocable<void()>* callback =
+      reinterpret_cast<absl::AnyInvocable<void()>*>(lpParam);
+  if (*callback != nullptr) {
+    (*callback)();
+  }
+}
+
+}  // namespace linux
+}  // namespace nearby

internal/platform/implementation/linux/timer.h

@@ -0,0 +1,56 @@
+// 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.
+
+#ifndef PLATFORM_IMPL_LINUX_TIMER_H_
+#define PLATFORM_IMPL_LINUX_TIMER_H_
+
+#include <memory>
+
+#include "absl/base/thread_annotations.h"
+#include "absl/synchronization/mutex.h"
+#include "internal/platform/implementation/timer.h"
+#include "internal/platform/implementation/linux/submittable_executor.h"
+#include "internal/platform/implementation/linux/timer_queue.h"
+
+namespace nearby {
+namespace linux {
+
+class Timer : public api::Timer {
+ public:
+  Timer() = default;
+  ~Timer() override;
+
+  bool Create(int delay, int interval,
+              absl::AnyInvocable<void()> callback) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
+  bool Stop() override ABSL_LOCKS_EXCLUDED(mutex_);
+  bool FireNow() override ABSL_LOCKS_EXCLUDED(mutex_);
+
+ private:
+  static void TimerRoutine(void *lpParam);
+
+  mutable absl::Mutex mutex_;
+  int delay_ ABSL_GUARDED_BY(mutex_);
+  int interval_ ABSL_GUARDED_BY(mutex_);
+  absl::AnyInvocable<void()> callback_;
+  uint16_t handle_ ABSL_GUARDED_BY(mutex_) = 0;
+  std::unique_ptr<TimerQueue> timer_queue_handle_;
+  std::unique_ptr<SubmittableExecutor> task_executor_ ABSL_GUARDED_BY(mutex_) =
+      nullptr;
+};
+
+}  // namespace linux
+}  // namespace nearby
+
+#endif  // PLATFORM_IMPL_LINUX_TIMER_H_

internal/platform/implementation/linux/timer_queue.cc

@@ -0,0 +1,157 @@
+// 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 "internal/platform/implementation/linux/timer_queue.h"
+
+#include "internal/platform/implementation/linux/utils.h"
+
+#include "internal/platform/logging.h"
+
+namespace nearby {
+namespace linux {
+
+std::unique_ptr<TimerQueue> TimerQueue::CreateTimerQueue() {
+  return std::unique_ptr<TimerQueue>(new TimerQueue);
+}
+
+TimerQueue::TimerQueue() {
+  thread_ = std::thread([this]() {Run();});
+}
+
+TimerQueue::~TimerQueue() {
+  DeleteTimerQueueEx(CE_WAITFORCALLBACKS); // NOLINT
+  cancelled_ = true;
+  thread_.join();
+}
+
+void TimerQueue::Run() {
+  while(!cancelled_ || !work_.empty()) {
+    absl::MutexLock lk(&mutex_);
+    empty_ = (work_.empty() ? true : false);
+    for (auto &workItem : work_) {
+      if (workItem.Finished) {
+        if (workItem.Due < std::chrono::steady_clock::now()) {
+          work_.erase(workItem);
+          break;
+        }
+      }
+      if (workItem.Cancelled) {
+        if (workItem.Worker.valid()) {
+          if (workItem.Worker.wait_for(std::chrono_literals::operator""ms(0)) == std::future_status::ready) {
+            work_.erase(workItem);
+            break;
+          }
+        }
+      }
+      if (workItem.Due <= std::chrono::steady_clock::now()) {
+        if ((workItem.Flags & WT_EXECUTEDEFAULT) == WT_EXECUTEDEFAULT) {
+          if ((workItem.Flags & WT_EXECUTEONLYONCE) == WT_EXECUTEONLYONCE) {
+            workItem.Worker = std::async(workItem.Callback, workItem.Parameter);
+            workItem.Finished = true;
+          }
+          else {
+            if (workItem.Period > std::chrono_literals::operator""ms(0)) {
+              workItem.Worker = std::async(workItem.Callback, workItem.Parameter);
+              workItem.Finished = true;
+              workItem.Due = std::chrono::steady_clock::now() + workItem.Period;
+            }
+            else {
+              workItem.Worker = std::async(workItem.Callback, workItem.Parameter);
+              workItem.Finished = true;
+            }
+          }
+        }
+        else if ((workItem.Flags & WT_EXECUTEINTIMERTHREAD) == WT_EXECUTEINTIMERTHREAD) {
+          if ((workItem.Flags & WT_EXECUTEONLYONCE) == WT_EXECUTEONLYONCE) {
+            workItem.Callback(workItem.Parameter);
+            workItem.Finished = true;
+          }
+          else {
+            if (workItem.Period > std::chrono_literals::operator""ms(0)) {
+              workItem.Callback(workItem.Parameter);
+              workItem.Finished = true;
+              workItem.Due = std::chrono::steady_clock::now() + workItem.Period;
+            }
+            else {
+              workItem.Callback(workItem.Parameter);
+              workItem.Finished = true;
+            }
+          }
+        }
+      }
+    }
+  }
+}
+
+absl::StatusOr<uint16_t> TimerQueue::CreateTimerQueueTimer(absl::AnyInvocable<void(void *lpParameter)> Callback, void *Parameter, std::chrono::milliseconds DueTime, std::chrono::milliseconds Period, unsigned long int Flags) {
+  TimerWork work;
+
+  work.Callback = std::move(Callback);
+  work.WorkId = next_id_;
+  work.Flags = Flags;
+  next_id_++;
+  work.Due = std::chrono::steady_clock::now() + DueTime;
+  work.Period = Period;
+  work.Parameter = Parameter;
+  absl::MutexLock lk(&mutex_);
+  work_.insert(std::move(work));
+  return next_id_ - 1;
+}
+
+absl::Status TimerQueue::DeleteTimerQueueTimer(uint16_t WorkId, unsigned long int CompletionEvent) {
+  for (auto &workItem : work_) {
+    absl::MutexLock lk(&mutex_);
+    if (workItem.WorkId == WorkId) {
+      workItem.Cancelled = true;
+    }
+  }
+  return absl::OkStatus();
+}
+
+namespace {
+  bool check(bool *arg) {
+    return *arg;
+  }
+}  // namespace
+
+absl::Status TimerQueue::DeleteTimerQueueEx(unsigned int long CompletionEvent) {
+  switch (CompletionEvent) {
+    case CE_IMEDIATERETURN: {
+      absl::MutexLock lk(&mutex_);
+      for (auto workItem = work_.begin(); workItem != work_.end();) {
+        if (workItem->Worker.valid()) {
+          workItem->Worker.wait();
+        }
+        workItem = work_.erase(workItem);
+      }
+      return absl::OkStatus();
+    }
+    case CE_WAITFORCALLBACKS: {
+      mutex_.Lock();
+      for (auto &workItem : work_) {
+        workItem.Cancelled = true;
+      }
+      clearing_ = true;
+      mutex_.Await(absl::Condition(check, &empty_));
+      mutex_.Unlock();
+      return absl::OkStatus();
+    }
+    default: {
+      return absl::InvalidArgumentError("Invalid Completion Event value.");
+    }
+  }
+}
+
+}  // namespace linux
+}  // namespace nearby

internal/platform/implementation/linux/timer_queue.h

@@ -0,0 +1,124 @@
+// 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.
+
+#ifndef PLATFORM_IMPL_LINUX_TIMER_QUEUE_H_
+#define PLATFORM_IMPL_LINUX_TIMER_QUEUE_H_
+
+#include <memory>
+#include <functional>
+#include <thread>
+#include <chrono>
+#include <set>
+#include <future>
+
+#include "absl/status/status.h"
+#include "absl/status/statusor.h"
+#include "absl/functional/any_invocable.h"
+#include "absl/functional/function_ref.h"
+#include "absl/synchronization/mutex.h"
+#include "absl/base/thread_annotations.h"
+
+namespace nearby {
+namespace linux {
+
+/*
+ * This class represents a a que for timers. This class is built to try to replicate the Windows Timer-Queue functionality.
+ * https://learn.microsoft.com/en-us/windows/win32/api/threadpoollegacyapiset/nf-threadpoollegacyapiset-createtimerqueue
+ */
+class TimerQueue {
+public:
+  /*
+   * Parameters to be used with the Timer Queue Timer.
+   * See https://learn.microsoft.com/en-us/windows/win32/api/threadpoollegacyapiset/nf-threadpoollegacyapiset-createtimerqueuetimer
+   */
+  static constexpr unsigned long int WT_EXECUTEDEFAULT = 1 << 0;
+  static constexpr unsigned long int WT_EXECUTEINTIMERTHREAD = 1 << 1;
+  static constexpr unsigned long int WT_EXECUTEINIOTHREAD = 1 << 2; // Unused
+  static constexpr unsigned long int WT_EXECUTEINPERSISTENTTHREAD = 1 << 3; // Unused
+  static constexpr unsigned long int WT_EXECUTELONGFUNCTION = 1 << 4; // Unused
+  static constexpr unsigned long int WT_EXECUTEONLYONCE = 1 << 5;
+  static constexpr unsigned long int WT_TRANSFER_IMPERSONATION = 1 << 6; // Unused
+  static constexpr unsigned long int CE_WAITFORCALLBACKS = 1 << 0; // Equivalent to `INVALID_HANDLE_VALUE` when passing for a CompletionEvent
+  static constexpr unsigned long int CE_IMEDIATERETURN = 1 << 1; //  Equivalent to `NULL` when passing for a CompletionEvent
+  // Creates a new timer queue for the user to use.
+  static std::unique_ptr<TimerQueue> CreateTimerQueue();
+
+  // Returns a thread unique identifier that the timer is running on if it is successful. Status if not.
+  absl::StatusOr<uint16_t> CreateTimerQueueTimer(absl::AnyInvocable<void(void *lpParameter)> Callback, void *Parameter, std::chrono::milliseconds DueTime, std::chrono::milliseconds Period, unsigned long int Flags = WT_EXECUTEDEFAULT);
+
+  // Removes a timer based on the thread id the timer is running on.
+  absl::Status DeleteTimerQueueTimer(uint16_t WorkId, unsigned long int CompletionEvent = CE_WAITFORCALLBACKS);
+
+  // Removes all timers in the timer queue
+  absl::Status DeleteTimerQueueEx(unsigned long int CompletionEvent = CE_WAITFORCALLBACKS);
+
+  ~TimerQueue();
+
+private:
+  TimerQueue();
+
+  void Run();
+
+  uint16_t next_id_;
+
+  // A struct that represents a timer of work that needs to be done.
+  struct TimerWork {
+    mutable std::chrono::time_point<std::chrono::steady_clock> Due;
+    mutable absl::AnyInvocable<void(void *lpParameter)> Callback;
+    mutable std::chrono::milliseconds Period;
+    mutable void *Parameter = nullptr;
+    mutable long int Flags;
+    mutable std::future<void> Worker;
+    mutable std::atomic_bool Cancelled = false;
+    mutable std::atomic_bool Finished = false;
+    uint16_t WorkId = 0;
+
+    // Useful for time comparisons on other work items
+    bool operator<(const TimerWork &other) const {
+      return Due < other.Due;
+    }
+    bool operator>(const TimerWork &other) const {
+      return Due > other.Due;
+    }
+    bool operator==(const TimerWork &other) const {
+      return Due == other.Due;
+    }
+    bool operator!=(const TimerWork &other) const {
+      return !operator==(other);
+    }
+    bool operator<=(const TimerWork &other) const {
+      return (operator<(other) || operator==(other));
+    }
+    bool operator>=(const TimerWork &other) const {
+      return (operator>(other) || operator==(other));
+    }
+  };
+
+  std::set<TimerWork> work_;
+  std::vector<std::thread> threads_;
+
+  std::thread thread_;
+
+  bool cancelled_;
+  bool clearing_;
+  bool empty_;
+
+  absl::CondVar condvar_;
+  absl::Mutex mutex_;
+};
+
+}  // namespace linux
+}  // namespace nearby
+
+#endif  // PLATFORM_IMPL_LINUX_TIMER_QUEUE_H_