From 2e88b4aa7fd439b874fe9f91aa1561717b11fc22 Mon Sep 17 00:00:00 2001 From: Vibhav Pant Date: Tue, 29 Aug 2023 13:07:30 +0530 Subject: [PATCH] Minor refactor. --- .../platform/implementation/linux/executor.cc | 17 +++++++--------- .../platform/implementation/linux/executor.h | 4 +--- .../linux/scheduled_executor.cc | 20 ++++++++++--------- .../linux/submittable_executor.cc | 4 +--- .../linux/submittable_executor.h | 3 +-- 5 files changed, 21 insertions(+), 27 deletions(-) diff --git a/internal/platform/implementation/linux/executor.cc b/internal/platform/implementation/linux/executor.cc index 37dfcd9c..c222d1b2 100644 --- a/internal/platform/implementation/linux/executor.cc +++ b/internal/platform/implementation/linux/executor.cc @@ -16,21 +16,18 @@ #include +#include "internal/platform/implementation/linux/thread_pool.h" #include "internal/platform/logging.h" namespace nearby { namespace linux { - -Executor::Executor() : Executor(1) {} - -Executor::Executor(int32_t max_concurrency) - : max_concurrency_(max_concurrency) { - assert(max_concurrency_ >= 1); - thread_pool_ = linux::ThreadPool::Create(max_concurrency); +Executor::Executor(size_t max_concurrency) + : thread_pool_(std::make_unique(max_concurrency)) { + assert(max_concurrency >= 1); assert(thread_pool_ != nullptr); } -void Executor::Execute(Runnable&& runnable) { +void Executor::Execute(Runnable &&runnable) { if (shut_down_) { NEARBY_LOGS(VERBOSE) << "Warning: " << __func__ << ": Attempt to execute on a shut down pool."; @@ -51,5 +48,5 @@ void Executor::Shutdown() { thread_pool_ = nullptr; } -} // namespace linux -} // namespace nearby +} // namespace linux +} // namespace nearby diff --git a/internal/platform/implementation/linux/executor.h b/internal/platform/implementation/linux/executor.h index bd96ef06..7552945a 100644 --- a/internal/platform/implementation/linux/executor.h +++ b/internal/platform/implementation/linux/executor.h @@ -28,8 +28,7 @@ namespace linux { // Executor. class Executor : public api::Executor { public: - Executor(); - explicit Executor(int max_concurrency); + Executor(size_t max_concurrency = 1); // Before returning from destructor, executor must wait for all pending // jobs to finish. @@ -41,7 +40,6 @@ class Executor : public api::Executor { private: std::unique_ptr thread_pool_ = nullptr; std::atomic shut_down_ = false; - int32_t max_concurrency_; }; } // namespace linux diff --git a/internal/platform/implementation/linux/scheduled_executor.cc b/internal/platform/implementation/linux/scheduled_executor.cc index ff8b3a1d..e8a0be76 100644 --- a/internal/platform/implementation/linux/scheduled_executor.cc +++ b/internal/platform/implementation/linux/scheduled_executor.cc @@ -32,8 +32,8 @@ ScheduledExecutor::ScheduledExecutor() // We want Cancelable to live until both caller and executor are done with it. // Exclusive ownership model does not work for this case; // using std:shared_ptr<> instead of std::unique_ptr<>. -std::shared_ptr ScheduledExecutor::Schedule( - Runnable&& runnable, absl::Duration duration) { +std::shared_ptr +ScheduledExecutor::Schedule(Runnable &&runnable, absl::Duration duration) { if (shut_down_) { NEARBY_LOGS(ERROR) << __func__ << ": Attempt to Schedule on a shut down executor."; @@ -42,9 +42,11 @@ std::shared_ptr ScheduledExecutor::Schedule( } // Cleans completed tasks - std::remove_if( - scheduled_tasks_.begin(), scheduled_tasks_.end(), - [](std::shared_ptr& task) { return task->IsDone(); }); + scheduled_tasks_.erase( + std::remove_if( + scheduled_tasks_.begin(), scheduled_tasks_.end(), + [](std::shared_ptr &task) { return task->IsDone(); }), + scheduled_tasks_.end()); std::shared_ptr task = std::make_shared(std::move(runnable), duration); @@ -54,7 +56,7 @@ std::shared_ptr ScheduledExecutor::Schedule( return task; } -void ScheduledExecutor::Execute(Runnable&& runnable) { +void ScheduledExecutor::Execute(Runnable &&runnable) { if (shut_down_) { NEARBY_LOGS(ERROR) << __func__ << ": Attempt to Execute on a shut down executor."; @@ -67,7 +69,7 @@ void ScheduledExecutor::Execute(Runnable&& runnable) { void ScheduledExecutor::Shutdown() { if (!shut_down_) { shut_down_ = true; - for (auto& task : scheduled_tasks_) { + for (auto &task : scheduled_tasks_) { task->Cancel(); } @@ -78,5 +80,5 @@ void ScheduledExecutor::Shutdown() { NEARBY_LOGS(ERROR) << __func__ << ": Attempt to Shutdown on a shut down executor."; } -} // namespace linux -} // namespace nearby +} // namespace linux +} // namespace nearby diff --git a/internal/platform/implementation/linux/submittable_executor.cc b/internal/platform/implementation/linux/submittable_executor.cc index 3b721522..d9eb5ca2 100644 --- a/internal/platform/implementation/linux/submittable_executor.cc +++ b/internal/platform/implementation/linux/submittable_executor.cc @@ -20,9 +20,7 @@ namespace nearby { namespace linux { -SubmittableExecutor::SubmittableExecutor() : SubmittableExecutor(1) {} - -SubmittableExecutor::SubmittableExecutor(int32_t max_concurrancy) +SubmittableExecutor::SubmittableExecutor(size_t max_concurrancy) : executor_(std::make_unique(max_concurrancy)), shut_down_(false) {} diff --git a/internal/platform/implementation/linux/submittable_executor.h b/internal/platform/implementation/linux/submittable_executor.h index 63b1ef0b..133c144e 100644 --- a/internal/platform/implementation/linux/submittable_executor.h +++ b/internal/platform/implementation/linux/submittable_executor.h @@ -27,8 +27,7 @@ namespace linux { // Platform must override bool submit(absl::AnyInvocable) method. class SubmittableExecutor : public api::SubmittableExecutor { public: - SubmittableExecutor(); - SubmittableExecutor(int32_t maxConcurrancy); + SubmittableExecutor(size_t maxConcurrancy = 1); ~SubmittableExecutor() override = default; // Submit a callable (with no delay).