Internal change

PiperOrigin-RevId: 370573757
This commit is contained in:
hai007
2021-04-26 17:07:53 -07:00
committed by Copybara-Service
parent 07eab88962
commit e1b1c67141
5 changed files with 167 additions and 84 deletions
+2
View File
@@ -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 = [
+59
View File
@@ -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
+5 -28
View File
@@ -15,11 +15,9 @@
#ifndef PLATFORM_PUBLIC_MONITORED_RUNNABLE_H_
#define PLATFORM_PUBLIC_MONITORED_RUNNABLE_H_
#include <utility>
#include <string>
#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();
@@ -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
+9 -56
View File
@@ -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<const std::string, absl::Time> pending_jobs_