diff --git a/modules/footstone/include/footstone/cv_driver.h b/modules/footstone/include/footstone/cv_driver.h index e79621c42..de6a16982 100644 --- a/modules/footstone/include/footstone/cv_driver.h +++ b/modules/footstone/include/footstone/cv_driver.h @@ -35,13 +35,12 @@ class CVDriver: public Driver { virtual ~CVDriver() = default; virtual void Notify() override; - virtual void WaitFor(const TimeDelta& delta) override; + virtual void WaitFor(const TimeDelta& delta, std::unique_lock& lock) override; virtual void Start() override; virtual void Terminate() override; private: std::condition_variable cv_; - std::mutex mutex_; }; } diff --git a/modules/footstone/include/footstone/driver.h b/modules/footstone/include/footstone/driver.h index 81417dbe2..c18869e67 100644 --- a/modules/footstone/include/footstone/driver.h +++ b/modules/footstone/include/footstone/driver.h @@ -23,7 +23,7 @@ #pragma once #include - +#include #include "footstone/time_delta.h" namespace footstone { @@ -35,7 +35,8 @@ class Driver { virtual ~Driver() = default; virtual void Notify() = 0; - virtual void WaitFor(const TimeDelta& delta) = 0; + std::mutex& Mutex() { return mutex_; } + virtual void WaitFor(const TimeDelta& delta, std::unique_lock& lock) = 0; virtual void Start() = 0; virtual void Terminate() = 0; @@ -60,6 +61,8 @@ class Driver { * */ bool is_exit_immediately_; + + std::mutex mutex_; }; } diff --git a/modules/footstone/include/footstone/platform/adr/looper_driver.h b/modules/footstone/include/footstone/platform/adr/looper_driver.h index 213dd9c30..e6f4e5053 100644 --- a/modules/footstone/include/footstone/platform/adr/looper_driver.h +++ b/modules/footstone/include/footstone/platform/adr/looper_driver.h @@ -36,7 +36,7 @@ class LooperDriver: public Driver { virtual ~LooperDriver(); virtual void Notify() override; - virtual void WaitFor(const TimeDelta& delta) override; + virtual void WaitFor(const TimeDelta& delta, std::unique_lock& lock) override; virtual void Start() override; virtual void Terminate() override; diff --git a/modules/footstone/include/footstone/platform/ios/looper_driver.h b/modules/footstone/include/footstone/platform/ios/looper_driver.h index 1ec849a04..1e07e2c90 100644 --- a/modules/footstone/include/footstone/platform/ios/looper_driver.h +++ b/modules/footstone/include/footstone/platform/ios/looper_driver.h @@ -35,10 +35,10 @@ class LooperDriver: public Driver { virtual ~LooperDriver(); virtual void Notify() override; - virtual void WaitFor(const TimeDelta& delta) override; + virtual void WaitFor(const TimeDelta& delta, std::unique_lock& lock) override; virtual void Start() override; virtual void Terminate() override; - + void OnTimerFire(CFRunLoopTimerRef timer); private: diff --git a/modules/footstone/include/footstone/task_runner.h b/modules/footstone/include/footstone/task_runner.h index 356d87b36..b83a124ec 100644 --- a/modules/footstone/include/footstone/task_runner.h +++ b/modules/footstone/include/footstone/task_runner.h @@ -122,6 +122,8 @@ class TaskRunner { std::unique_ptr PopIdleTask(); std::unique_ptr GetTopDelayTask(); std::unique_ptr GetNext(); + bool HasTask(); + bool HasMoreUrgentTask(TimeDelta min_wait_time, TimePoint now); std::queue> task_queue_; std::mutex queue_mutex_; diff --git a/modules/footstone/include/footstone/worker.h b/modules/footstone/include/footstone/worker.h index 31134f8b3..a4df43c9d 100644 --- a/modules/footstone/include/footstone/worker.h +++ b/modules/footstone/include/footstone/worker.h @@ -102,6 +102,9 @@ class Worker { void BalanceNoLock(); void SortNoLock(); + bool HasTask(); + bool HasMoreUrgentTask(TimeDelta min_wait_time, TimePoint now); + int32_t WorkerKeyCreate(uint32_t task_runner_id, const std::function& destruct); bool WorkerKeyDelete(uint32_t task_runner_id, int32_t key); bool WorkerSetSpecific(uint32_t task_runner_id, int32_t key, void *p); diff --git a/modules/footstone/src/cv_driver.cc b/modules/footstone/src/cv_driver.cc index 7356c419a..ff180f710 100644 --- a/modules/footstone/src/cv_driver.cc +++ b/modules/footstone/src/cv_driver.cc @@ -24,12 +24,11 @@ namespace footstone { inline namespace runner { void CVDriver::Notify() { + std::unique_lock lock(mutex_); cv_.notify_one(); } -void CVDriver::WaitFor(const TimeDelta& delta) { - std::unique_lock lock(mutex_); - +void CVDriver::WaitFor(const TimeDelta& delta, std::unique_lock& lock) { if (delta != TimeDelta::Max() && delta >= TimeDelta::Zero()) { cv_.wait_for(lock, std::chrono::nanoseconds(delta.ToNanoseconds())); } else { diff --git a/modules/footstone/src/platform/adr/looper_driver.cc b/modules/footstone/src/platform/adr/looper_driver.cc index 22ea02c9d..96a3b3711 100644 --- a/modules/footstone/src/platform/adr/looper_driver.cc +++ b/modules/footstone/src/platform/adr/looper_driver.cc @@ -57,7 +57,7 @@ void LooperDriver::Notify() { timerfd_settime(fd_, TFD_TIMER_ABSTIME, &spec, nullptr); } -void LooperDriver::WaitFor(const TimeDelta& delta) { +void LooperDriver::WaitFor(const TimeDelta& delta, std::unique_lock& lock) { auto nano_secs = delta.ToNanoseconds(); if (nano_secs < 1) { nano_secs = 1; diff --git a/modules/footstone/src/platform/ios/looper_driver.cc b/modules/footstone/src/platform/ios/looper_driver.cc index b817da01e..eac5e0d0b 100644 --- a/modules/footstone/src/platform/ios/looper_driver.cc +++ b/modules/footstone/src/platform/ios/looper_driver.cc @@ -56,7 +56,7 @@ void LooperDriver::Notify() { CFAbsoluteTimeGetCurrent()); } -void LooperDriver::WaitFor(const TimeDelta& delta) { +void LooperDriver::WaitFor(const TimeDelta& delta, std::unique_lock& lock) { CFRunLoopTimerSetNextFireDate( delayed_wake_timer_, CFAbsoluteTimeGetCurrent() + delta.ToSecondsF()); diff --git a/modules/footstone/src/task_runner.cc b/modules/footstone/src/task_runner.cc index b8a60590a..108ca7a63 100644 --- a/modules/footstone/src/task_runner.cc +++ b/modules/footstone/src/task_runner.cc @@ -266,5 +266,53 @@ void TaskRunner::RunnerDestroySpecifics() { return worker->WorkerDestroySpecific(task_runner_id); } +bool TaskRunner::HasTask() { + { + std::lock_guard lock(queue_mutex_); + if (!task_queue_.empty()) { + return true; + } + } + { + std::lock_guard lock(delay_mutex_); + if (!delayed_task_queue_.empty()) { + return true; + } + } + { + std::lock_guard lock(idle_mutex_); + if (!idle_task_queue_.empty()) { + return true; + } + } + return false; +} + +bool TaskRunner::HasMoreUrgentTask(TimeDelta min_wait_time, TimePoint now) { + { + std::lock_guard lock(queue_mutex_); + if (!task_queue_.empty()) { + return true; + } + } + { + std::lock_guard lock(idle_mutex_); + if (!idle_task_queue_.empty()) { + return true; + } + } + { + std::lock_guard lock(delay_mutex_); + if (!delayed_task_queue_.empty()) { + const DelayedEntry& delayed_task = delayed_task_queue_.top(); + if(delayed_task.first - now < min_wait_time) { + return true; + } + } + } + + return false; +} + } // namespace runner } // namespace footstone diff --git a/modules/footstone/src/worker.cc b/modules/footstone/src/worker.cc index b6b8e5fd9..d82b036db 100644 --- a/modules/footstone/src/worker.cc +++ b/modules/footstone/src/worker.cc @@ -402,10 +402,20 @@ std::unique_ptr Worker::GetNextTask() { })); return wrapper_idle_task; } + std::unique_lock lock(driver_->Mutex()); if (driver_->IsTerminated()) { return nullptr; } - driver_->WaitFor(min_wait_time_); + if (min_wait_time_ == TimeDelta::Max()) { + if (HasTask()) { + return nullptr; + } + } else { + if (HasMoreUrgentTask(min_wait_time_, now)) { + return nullptr; + } + } + driver_->WaitFor(min_wait_time_, lock); return nullptr; } @@ -532,5 +542,63 @@ void Worker::UpdateSpecific(uint32_t task_runner_id, specific_map_[task_runner_id] = array; // insert or update } +bool Worker::HasTask() { + bool has_found = false; + for (auto &group : running_group_list_) { + for (auto &runner : group) { + if (runner->HasTask()) { + has_found = true; + break; + } + } + if (has_found) { + break; + } + } + if (!has_found) { + for (auto &group : pending_group_list_) { + for (auto &runner : group) { + if (runner->HasTask()) { + has_found = true; + break; + } + } + if (has_found) { + break; + } + } + } + return has_found; +} + +bool Worker::HasMoreUrgentTask(TimeDelta min_wait_time, TimePoint now) { + bool has_found = false; + for (auto &group : running_group_list_) { + for (auto &runner : group) { + if (runner->HasMoreUrgentTask(min_wait_time, now)) { + has_found = true; + break; + } + } + if (has_found) { + break; + } + } + if (!has_found) { + for (auto &group : pending_group_list_) { + for (auto &runner : group) { + if (runner->HasMoreUrgentTask(min_wait_time, now)) { + has_found = true; + break; + } + } + if (has_found) { + break; + } + } + } + return has_found; +} + } // namespace runner } // namespace footstone