From e1b1c671416858c85bbac732e1ae2482c659f834 Mon Sep 17 00:00:00 2001 From: hai007 Date: Mon, 26 Apr 2021 17:07:27 -0700 Subject: [PATCH] Internal change PiperOrigin-RevId: 370573757 --- cpp/platform/public/BUILD | 2 + cpp/platform/public/monitored_runnable.cc | 59 +++++++++++++ cpp/platform/public/monitored_runnable.h | 33 ++------ cpp/platform/public/pending_job_registry.cc | 92 +++++++++++++++++++++ cpp/platform/public/pending_job_registry.h | 65 ++------------- 5 files changed, 167 insertions(+), 84 deletions(-) create mode 100644 cpp/platform/public/monitored_runnable.cc create mode 100644 cpp/platform/public/pending_job_registry.cc diff --git a/cpp/platform/public/BUILD b/cpp/platform/public/BUILD index 58dde048..008dae9e 100644 --- a/cpp/platform/public/BUILD +++ b/cpp/platform/public/BUILD @@ -17,6 +17,8 @@ load("//tools/build_defs/cc:cc_fake_binary.bzl", "cc_fake_binary") cc_library( name = "types", srcs = [ + "monitored_runnable.cc", + "pending_job_registry.cc", "pipe.cc", ], hdrs = [ diff --git a/cpp/platform/public/monitored_runnable.cc b/cpp/platform/public/monitored_runnable.cc new file mode 100644 index 00000000..7478f70e --- /dev/null +++ b/cpp/platform/public/monitored_runnable.cc @@ -0,0 +1,59 @@ +// Copyright 2020 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/public/monitored_runnable.h" + +#include "platform/public/logging.h" +#include "platform/public/pending_job_registry.h" + +namespace location { +namespace nearby { + +namespace { +absl::Duration kMinReportedStartDelay = absl::Seconds(5); +absl::Duration kMinReportedTaskDuration = absl::Seconds(10); +} // namespace + +MonitoredRunnable::MonitoredRunnable(Runnable&& runnable) + : runnable_{runnable} {} + +MonitoredRunnable::MonitoredRunnable(const std::string& name, + Runnable&& runnable) + : name_{name}, runnable_{runnable} { + PendingJobRegistry::GetInstance().AddPendingJob(name_, post_time_); +} + +MonitoredRunnable::~MonitoredRunnable() = default; + +void MonitoredRunnable::operator()() const { + auto start_time = SystemClock::ElapsedRealtime(); + auto start_delay = start_time - post_time_; + if (start_delay >= kMinReportedStartDelay) { + NEARBY_LOGS(INFO) << "Task: \"" << name_ << "\" started after " + << absl::ToInt64Seconds(start_delay) << " seconds"; + } + PendingJobRegistry::GetInstance().RemovePendingJob(name_, post_time_); + PendingJobRegistry::GetInstance().AddRunningJob(name_, post_time_); + runnable_(); + auto task_duration = SystemClock::ElapsedRealtime() - start_time; + if (task_duration >= kMinReportedTaskDuration) { + NEARBY_LOGS(INFO) << "Task: \"" << name_ << "\" finished after " + << absl::ToInt64Seconds(task_duration) << " seconds"; + } + PendingJobRegistry::GetInstance().RemoveRunningJob(name_, post_time_); + PendingJobRegistry::GetInstance().ListJobs(); +} + +} // namespace nearby +} // namespace location diff --git a/cpp/platform/public/monitored_runnable.h b/cpp/platform/public/monitored_runnable.h index 63e3475d..b7630b62 100644 --- a/cpp/platform/public/monitored_runnable.h +++ b/cpp/platform/public/monitored_runnable.h @@ -15,11 +15,9 @@ #ifndef PLATFORM_PUBLIC_MONITORED_RUNNABLE_H_ #define PLATFORM_PUBLIC_MONITORED_RUNNABLE_H_ -#include +#include #include "platform/base/runnable.h" -#include "platform/public/logging.h" -#include "platform/public/pending_job_registry.h" #include "platform/public/system_clock.h" #include "absl/time/time.h" @@ -32,34 +30,13 @@ namespace nearby { // to run for longer periods of time (minutes). class MonitoredRunnable { public: - explicit MonitoredRunnable(Runnable&& runnable) : runnable_{runnable} {} - MonitoredRunnable(const std::string& name, Runnable&& runnable) - : name_{name}, runnable_{runnable} { - PendingJobRegistry::GetInstance().AddPendingJob(name_, post_time_); - } + explicit MonitoredRunnable(Runnable&& runnable); + MonitoredRunnable(const std::string& name, Runnable&& runnable); + ~MonitoredRunnable(); - void operator()() const { - auto start_time = SystemClock::ElapsedRealtime(); - auto start_delay = start_time - post_time_; - if (start_delay >= kMinReportedStartDelay) { - NEARBY_LOGS(INFO) << "Task: \"" << name_ << "\" started after " - << absl::ToInt64Seconds(start_delay) << " seconds"; - } - PendingJobRegistry::GetInstance().RemovePendingJob(name_, post_time_); - PendingJobRegistry::GetInstance().AddRunningJob(name_, post_time_); - runnable_(); - auto task_duration = SystemClock::ElapsedRealtime() - start_time; - if (task_duration >= kMinReportedTaskDuration) { - NEARBY_LOGS(INFO) << "Task: \"" << name_ << "\" finished after " - << absl::ToInt64Seconds(task_duration) << " seconds"; - } - PendingJobRegistry::GetInstance().RemoveRunningJob(name_, post_time_); - PendingJobRegistry::GetInstance().ListJobs(); - } + void operator()() const; private: - static constexpr absl::Duration kMinReportedStartDelay = absl::Seconds(5); - static constexpr absl::Duration kMinReportedTaskDuration = absl::Seconds(10); const std::string name_; Runnable runnable_; absl::Time post_time_ = SystemClock::ElapsedRealtime(); diff --git a/cpp/platform/public/pending_job_registry.cc b/cpp/platform/public/pending_job_registry.cc new file mode 100644 index 00000000..f4edd1a1 --- /dev/null +++ b/cpp/platform/public/pending_job_registry.cc @@ -0,0 +1,92 @@ +// Copyright 2020 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/public/pending_job_registry.h" + +#include "platform/public/logging.h" +#include "platform/public/mutex_lock.h" +#include "platform/public/system_clock.h" + +namespace location { +namespace nearby { + +namespace { +absl::Duration kMinReportInterval = absl::Seconds(60); +absl::Duration kReportPendingJobsOlderThan = absl::Seconds(40); +absl::Duration kReportRunningJobsOlderThan = absl::Seconds(60); +} // namespace + +PendingJobRegistry& PendingJobRegistry::GetInstance() { + static PendingJobRegistry* instance = new PendingJobRegistry(); + return *instance; +} + +PendingJobRegistry::PendingJobRegistry() = default; + +PendingJobRegistry::~PendingJobRegistry() = default; + +void PendingJobRegistry::AddPendingJob(const std::string& name, + absl::Time post_time) { + MutexLock lock(&mutex_); + pending_jobs_.emplace(CreateKey(name, post_time), post_time); +} + +void PendingJobRegistry::RemovePendingJob(const std::string& name, + absl::Time post_time) { + MutexLock lock(&mutex_); + pending_jobs_.erase(CreateKey(name, post_time)); +} + +void PendingJobRegistry::AddRunningJob(const std::string& name, + absl::Time post_time) { + MutexLock lock(&mutex_); + running_jobs_.emplace(CreateKey(name, post_time), + SystemClock::ElapsedRealtime()); +} + +void PendingJobRegistry::RemoveRunningJob(const std::string& name, + absl::Time post_time) { + MutexLock lock(&mutex_); + running_jobs_.erase(CreateKey(name, post_time)); +} + +void PendingJobRegistry::ListJobs() { + auto current_time = SystemClock::ElapsedRealtime(); + if (current_time - list_jobs_time_ < kMinReportInterval) + return; + MutexLock lock(&mutex_); + for (auto& job : pending_jobs_) { + auto age = current_time - job.second; + if (age >= kReportPendingJobsOlderThan) { + NEARBY_LOGS(INFO) << "Task \"" << job.first << "\" is waiting for " + << absl::ToInt64Seconds(age) << " s"; + } + } + for (auto& job : running_jobs_) { + auto age = current_time - job.second; + if (age >= kReportRunningJobsOlderThan) { + NEARBY_LOGS(INFO) << "Task \"" << job.first << "\" is running for " + << absl::ToInt64Seconds(age) << " s"; + } + } + list_jobs_time_ = current_time; +} + +std::string PendingJobRegistry::CreateKey(const std::string& name, + absl::Time post_time) { + return name + "." + std::to_string(absl::ToUnixNanos(post_time)); +} + +} // namespace nearby +} // namespace location diff --git a/cpp/platform/public/pending_job_registry.h b/cpp/platform/public/pending_job_registry.h index 1cb79cb9..c90f3d3c 100644 --- a/cpp/platform/public/pending_job_registry.h +++ b/cpp/platform/public/pending_job_registry.h @@ -15,10 +15,7 @@ #ifndef PLATFORM_PUBLIC_PENDING_JOB_REGISTRY_H_ #define PLATFORM_PUBLIC_PENDING_JOB_REGISTRY_H_ -#include "platform/public/logging.h" #include "platform/public/mutex.h" -#include "platform/public/mutex_lock.h" -#include "platform/public/system_clock.h" #include "absl/base/thread_annotations.h" #include "absl/time/time.h" @@ -29,64 +26,20 @@ namespace nearby { // tasks that are either waiting too long for their turn or they never finish class PendingJobRegistry { public: - static PendingJobRegistry& GetInstance() { - static PendingJobRegistry* instance = new PendingJobRegistry(); - return *instance; - } + static PendingJobRegistry& GetInstance(); - void AddPendingJob(const std::string& name, absl::Time post_time) { - MutexLock lock(&mutex_); - pending_jobs_.emplace(CreateKey(name, post_time), post_time); - } + ~PendingJobRegistry(); - void RemovePendingJob(const std::string& name, absl::Time post_time) { - MutexLock lock(&mutex_); - pending_jobs_.erase(CreateKey(name, post_time)); - } - - void AddRunningJob(const std::string& name, absl::Time post_time) { - MutexLock lock(&mutex_); - running_jobs_.emplace(CreateKey(name, post_time), - SystemClock::ElapsedRealtime()); - } - - void RemoveRunningJob(const std::string& name, absl::Time post_time) { - MutexLock lock(&mutex_); - running_jobs_.erase(CreateKey(name, post_time)); - } - - void ListJobs() { - auto current_time = SystemClock::ElapsedRealtime(); - if (current_time - list_jobs_time_ < kMinReportInterval) return; - MutexLock lock(&mutex_); - for (auto& job : pending_jobs_) { - auto age = current_time - job.second; - if (age >= kReportPendingJobsOlderThan) { - NEARBY_LOGS(INFO) << "Task \"" << job.first << "\" is waiting for " - << absl::ToInt64Seconds(age) << " s"; - } - } - for (auto& job : running_jobs_) { - auto age = current_time - job.second; - if (age >= kReportRunningJobsOlderThan) { - NEARBY_LOGS(INFO) << "Task \"" << job.first << "\" is running for " - << absl::ToInt64Seconds(age) << " s"; - } - } - list_jobs_time_ = current_time; - } + void AddPendingJob(const std::string& name, absl::Time post_time); + void RemovePendingJob(const std::string& name, absl::Time post_time); + void AddRunningJob(const std::string& name, absl::Time post_time); + void RemoveRunningJob(const std::string& name, absl::Time post_time); + void ListJobs(); private: - PendingJobRegistry() = default; - static constexpr absl::Duration kMinReportInterval = absl::Seconds(60); - static constexpr absl::Duration kReportPendingJobsOlderThan = - absl::Seconds(40); - static constexpr absl::Duration kReportRunningJobsOlderThan = - absl::Seconds(60); + PendingJobRegistry(); - std::string CreateKey(const std::string& name, absl::Time post_time) { - return name + "." + std::to_string(absl::ToUnixNanos(post_time)); - } + std::string CreateKey(const std::string& name, absl::Time post_time); Mutex mutex_; absl::flat_hash_map pending_jobs_