diff --git a/internal/platform/implementation/linux/timer.cc b/internal/platform/implementation/linux/timer.cc new file mode 100644 index 00000000..2468a009 --- /dev/null +++ b/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 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 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(); + } + + 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* callback = + reinterpret_cast*>(lpParam); + if (*callback != nullptr) { + (*callback)(); + } +} + +} // namespace linux +} // namespace nearby diff --git a/internal/platform/implementation/linux/timer.h b/internal/platform/implementation/linux/timer.h new file mode 100644 index 00000000..8a75163c --- /dev/null +++ b/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 + +#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 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 callback_; + uint16_t handle_ ABSL_GUARDED_BY(mutex_) = 0; + std::unique_ptr timer_queue_handle_; + std::unique_ptr task_executor_ ABSL_GUARDED_BY(mutex_) = + nullptr; +}; + +} // namespace linux +} // namespace nearby + +#endif // PLATFORM_IMPL_LINUX_TIMER_H_ diff --git a/internal/platform/implementation/linux/timer_queue.cc b/internal/platform/implementation/linux/timer_queue.cc new file mode 100644 index 00000000..5048008c --- /dev/null +++ b/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::CreateTimerQueue() { + return std::unique_ptr(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 TimerQueue::CreateTimerQueueTimer(absl::AnyInvocable 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 diff --git a/internal/platform/implementation/linux/timer_queue.h b/internal/platform/implementation/linux/timer_queue.h new file mode 100644 index 00000000..db745324 --- /dev/null +++ b/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 +#include +#include +#include +#include +#include + +#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 CreateTimerQueue(); + + // Returns a thread unique identifier that the timer is running on if it is successful. Status if not. + absl::StatusOr CreateTimerQueueTimer(absl::AnyInvocable 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 Due; + mutable absl::AnyInvocable Callback; + mutable std::chrono::milliseconds Period; + mutable void *Parameter = nullptr; + mutable long int Flags; + mutable std::future 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 work_; + std::vector 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_