TySS-Dev/nearby
1
0
Fork 0
mirror of https://github.com/kidfromjupiter/nearby.git synced 2026-09-14 14:46:12 -04:00
Code Issues Packages Projects Releases Wiki Activity

Rewrite timer to use POSIX timers.

Browse Source
This commit is contained in:
Vibhav Pant
2023-08-27 20:50:26 +05:30
parent 61b4e1d74d
commit 4978e6bea9
5 changed files with 88 additions and 354 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/platform/implementation/linux/BUILD
+5 -2
View File
@@ -17,7 +17,6 @@ cc_library(
"scheduled_executor.h",
"submittable_executor.h",
"timer.h",
"timer_queue.h",
"thread_pool.h",
"log_message.h",
"org_freedesktop_logcontrol_server_glue.h",
@@ -29,8 +28,9 @@ cc_library(
srcs = [
"device_info.cc",
"log_message.cc",
"timer.cc"
"timer.cc",
],
copts = ["-lrt"],
visibility = ["//third_party/nearby/sharing/internal/impl/linux:__pkg__"],
deps = [
":comm",
@@ -127,6 +127,8 @@ cc_library(
"bluetooth_classic_device.cc",
"bluetooth_classic_medium.cc",
"bluetooth_classic_socket.cc",
"bluetooth_classic_server_socket.cc",
"bluetooth_devices.cc",
"bluetooth_pairing.cc",
"bluez.cc",
"dbus.cc",
@@ -185,6 +187,7 @@ cc_library(
"@nlohmann_json//:json",
"@libsystemd//:lib",
"@sdbus_cpp//:lib",
"@libcurl//:lib"
],
)
internal/platform/implementation/linux/timer.cc
+75 -59
View File
@@ -12,8 +12,15 @@
// See the License for the specific language governing permissions and
// limitations under the License.
#include <bits/types/struct_itimerspec.h>
#include <cerrno>
#include <cstring>
#include <ctime>
#include <signal.h>
#include <time.h>
#include "internal/platform/implementation/linux/submittable_executor.h"
#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"
@@ -21,100 +28,109 @@
namespace nearby {
namespace linux {
Timer::~Timer() { Stop(); }
static void timer_callback(union sigval val) {
absl::AnyInvocable<void()> *callback =
reinterpret_cast<absl::AnyInvocable<void()> *>(val.sival_ptr);
if (*callback != nullptr)
(*callback)();
}
Timer::~Timer() {
absl::MutexLock l(&mutex_);
if (timerid_.has_value())
if (timer_delete(*timerid_) < 0) {
NEARBY_LOGS(ERROR) << __func__ << ": Error deleting POSIX timer: "
<< std::strerror(errno);
}
}
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.";
if (delay < 0 || interval < 0) {
NEARBY_LOGS(ERROR) << __func__
<< ": Delay and interval cannot be negative.";
return false;
}
if (timer_queue_handle_) {
absl::MutexLock l(&mutex_);
if (timerid_.has_value()) {
NEARBY_LOGS(ERROR) << __func__
<< "Timer has already been created and armed.";
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);
struct sigevent ev;
ev.sigev_value.sival_ptr = &callback_;
ev.sigev_notify_function = timer_callback;
timer_t timerid;
absl::StatusOr<uint16_t> createStatus = timer_queue_handle_->CreateTimerQueueTimer(TimerRoutine,
&callback_, std::chrono::milliseconds(delay), std::chrono::milliseconds(interval), TimerQueue::WT_EXECUTEDEFAULT);
struct itimerspec spec;
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();
spec.it_value.tv_nsec = delay * 1000000;
spec.it_value.tv_sec = 0;
spec.it_interval.tv_nsec = interval * 1000000;
spec.it_interval.tv_sec = 0;
if (timer_create(CLOCK_MONOTONIC, &ev, &timerid) < 0) {
NEARBY_LOGS(ERROR) << __func__ << ": Error creating POSIX timer: "
<< std::strerror(errno);
return false;
}
handle_ = createStatus.value();
if (timer_settime(&timerid, 0, &spec, nullptr) < 0) {
NEARBY_LOGS(ERROR) << __func__ << ": Error arming POSIX timer: "
<< std::strerror(errno);
if (!timer_delete(&timerid)) {
NEARBY_LOGS(ERROR) << __func__ << ": error deleting POSIX timer: "
<< std::strerror(errno);
}
return false;
}
timerid_ = timerid;
return true;
}
bool Timer::Stop() {
absl::MutexLock lock(&mutex_);
if (!timer_queue_handle_) {
absl::MutexLock l(&mutex_);
if (!timerid_.has_value()) {
NEARBY_LOGS(WARNING) << __func__ << ": no timer created";
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();
if (!timer_delete(&*timerid_)) {
NEARBY_LOGS(ERROR) << __func__ << ": error deleting POSIX timer: "
<< std::strerror(errno);
return false;
}
timer_queue_handle_ = nullptr;
timerid_.reset();
return true;
}
bool Timer::FireNow() {
absl::MutexLock lock(&mutex_);
if (!timer_queue_handle_ || !callback_) {
if (!timerid_.has_value()) {
NEARBY_LOGS(ERROR) << __func__ << ": No timer has been created";
return false;
}
if (callback_ == nullptr) {
NEARBY_LOGS(ERROR) << __func__ << ": No callback has been set";
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_(); });
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
} // namespace linux
} // namespace nearby
internal/platform/implementation/linux/timer.h
+8 -12
View File
@@ -16,19 +16,21 @@
#define PLATFORM_IMPL_LINUX_TIMER_H_
#include <memory>
#include <optional>
#include <time.h>
#include <signal.h>
#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() : timerid_(nullptr) {} ;
~Timer() override;
bool Create(int delay, int interval,
@@ -37,17 +39,11 @@ class Timer : public api::Timer {
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_);
private:
absl::Mutex mutex_;
std::optional<timer_t> timerid_ 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;
std::unique_ptr<SubmittableExecutor> task_executor_;
};
} // namespace linux
internal/platform/implementation/linux/timer_queue.cc
-157
View File
@@ -1,157 +0,0 @@
// 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
-124
View File
@@ -1,124 +0,0 @@
// 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_