| /* |
| * Copyright 2018 The WebRTC Project Authors. All rights reserved. |
| * |
| * Use of this source code is governed by a BSD-style license |
| * that can be found in the LICENSE file in the root of the source |
| * tree. An additional intellectual property rights grant can be found |
| * in the file PATENTS. All contributing project authors may |
| * be found in the AUTHORS file in the root of the source tree. |
| */ |
| |
| #include "rtc_base/task_queue_stdlib.h" |
| |
| #include <string.h> |
| |
| #include <algorithm> |
| #include <map> |
| #include <memory> |
| #include <queue> |
| #include <utility> |
| |
| #include "absl/strings/string_view.h" |
| #include "api/task_queue/queued_task.h" |
| #include "api/task_queue/task_queue_base.h" |
| #include "rtc_base/checks.h" |
| #include "rtc_base/event.h" |
| #include "rtc_base/logging.h" |
| #include "rtc_base/platform_thread.h" |
| #include "rtc_base/synchronization/mutex.h" |
| #include "rtc_base/thread_annotations.h" |
| #include "rtc_base/time_utils.h" |
| |
| namespace webrtc { |
| namespace { |
| |
| rtc::ThreadPriority TaskQueuePriorityToThreadPriority( |
| TaskQueueFactory::Priority priority) { |
| switch (priority) { |
| case TaskQueueFactory::Priority::HIGH: |
| return rtc::ThreadPriority::kRealtime; |
| case TaskQueueFactory::Priority::LOW: |
| return rtc::ThreadPriority::kLow; |
| case TaskQueueFactory::Priority::NORMAL: |
| return rtc::ThreadPriority::kNormal; |
| } |
| } |
| |
| class TaskQueueStdlib final : public TaskQueueBase { |
| public: |
| TaskQueueStdlib(absl::string_view queue_name, rtc::ThreadPriority priority); |
| ~TaskQueueStdlib() override = default; |
| |
| void Delete() override; |
| void PostTask(std::unique_ptr<QueuedTask> task) override; |
| void PostDelayedTask(std::unique_ptr<QueuedTask> task, |
| uint32_t milliseconds) override; |
| |
| private: |
| using OrderId = uint64_t; |
| |
| struct DelayedEntryTimeout { |
| int64_t next_fire_at_ms_{}; |
| OrderId order_{}; |
| |
| bool operator<(const DelayedEntryTimeout& o) const { |
| return std::tie(next_fire_at_ms_, order_) < |
| std::tie(o.next_fire_at_ms_, o.order_); |
| } |
| }; |
| |
| struct NextTask { |
| bool final_task_{false}; |
| std::unique_ptr<QueuedTask> run_task_; |
| int64_t sleep_time_ms_{}; |
| }; |
| |
| NextTask GetNextTask(); |
| |
| void ProcessTasks(); |
| |
| void NotifyWake(); |
| |
| // Indicates if the thread has started. |
| rtc::Event started_; |
| |
| // Indicates if the thread has stopped. |
| rtc::Event stopped_; |
| |
| // Signaled whenever a new task is pending. |
| rtc::Event flag_notify_; |
| |
| // Contains the active worker thread assigned to processing |
| // tasks (including delayed tasks). |
| rtc::PlatformThread thread_; |
| |
| Mutex pending_lock_; |
| |
| // Indicates if the worker thread needs to shutdown now. |
| bool thread_should_quit_ RTC_GUARDED_BY(pending_lock_){false}; |
| |
| // Holds the next order to use for the next task to be |
| // put into one of the pending queues. |
| OrderId thread_posting_order_ RTC_GUARDED_BY(pending_lock_){}; |
| |
| // The list of all pending tasks that need to be processed in the |
| // FIFO queue ordering on the worker thread. |
| std::queue<std::pair<OrderId, std::unique_ptr<QueuedTask>>> pending_queue_ |
| RTC_GUARDED_BY(pending_lock_); |
| |
| // The list of all pending tasks that need to be processed at a future |
| // time based upon a delay. On the off change the delayed task should |
| // happen at exactly the same time interval as another task then the |
| // task is processed based on FIFO ordering. std::priority_queue was |
| // considered but rejected due to its inability to extract the |
| // std::unique_ptr out of the queue without the presence of a hack. |
| std::map<DelayedEntryTimeout, std::unique_ptr<QueuedTask>> delayed_queue_ |
| RTC_GUARDED_BY(pending_lock_); |
| }; |
| |
| TaskQueueStdlib::TaskQueueStdlib(absl::string_view queue_name, |
| rtc::ThreadPriority priority) |
| : started_(/*manual_reset=*/false, /*initially_signaled=*/false), |
| stopped_(/*manual_reset=*/false, /*initially_signaled=*/false), |
| flag_notify_(/*manual_reset=*/false, /*initially_signaled=*/false), |
| thread_(rtc::PlatformThread::SpawnJoinable( |
| [this] { |
| CurrentTaskQueueSetter set_current(this); |
| ProcessTasks(); |
| }, |
| queue_name, |
| rtc::ThreadAttributes().SetPriority(priority))) { |
| started_.Wait(rtc::Event::kForever); |
| } |
| |
| void TaskQueueStdlib::Delete() { |
| RTC_DCHECK(!IsCurrent()); |
| |
| { |
| MutexLock lock(&pending_lock_); |
| thread_should_quit_ = true; |
| } |
| |
| NotifyWake(); |
| |
| stopped_.Wait(rtc::Event::kForever); |
| thread_.Finalize(); |
| delete this; |
| } |
| |
| void TaskQueueStdlib::PostTask(std::unique_ptr<QueuedTask> task) { |
| { |
| MutexLock lock(&pending_lock_); |
| OrderId order = thread_posting_order_++; |
| |
| pending_queue_.push(std::pair<OrderId, std::unique_ptr<QueuedTask>>( |
| order, std::move(task))); |
| } |
| |
| NotifyWake(); |
| } |
| |
| void TaskQueueStdlib::PostDelayedTask(std::unique_ptr<QueuedTask> task, |
| uint32_t milliseconds) { |
| auto fire_at = rtc::TimeMillis() + milliseconds; |
| |
| DelayedEntryTimeout delay; |
| delay.next_fire_at_ms_ = fire_at; |
| |
| { |
| MutexLock lock(&pending_lock_); |
| delay.order_ = ++thread_posting_order_; |
| delayed_queue_[delay] = std::move(task); |
| } |
| |
| NotifyWake(); |
| } |
| |
| TaskQueueStdlib::NextTask TaskQueueStdlib::GetNextTask() { |
| NextTask result{}; |
| |
| auto tick = rtc::TimeMillis(); |
| |
| MutexLock lock(&pending_lock_); |
| |
| if (thread_should_quit_) { |
| result.final_task_ = true; |
| return result; |
| } |
| |
| if (delayed_queue_.size() > 0) { |
| auto delayed_entry = delayed_queue_.begin(); |
| const auto& delay_info = delayed_entry->first; |
| auto& delay_run = delayed_entry->second; |
| if (tick >= delay_info.next_fire_at_ms_) { |
| if (pending_queue_.size() > 0) { |
| auto& entry = pending_queue_.front(); |
| auto& entry_order = entry.first; |
| auto& entry_run = entry.second; |
| if (entry_order < delay_info.order_) { |
| result.run_task_ = std::move(entry_run); |
| pending_queue_.pop(); |
| return result; |
| } |
| } |
| |
| result.run_task_ = std::move(delay_run); |
| delayed_queue_.erase(delayed_entry); |
| return result; |
| } |
| |
| result.sleep_time_ms_ = delay_info.next_fire_at_ms_ - tick; |
| } |
| |
| if (pending_queue_.size() > 0) { |
| auto& entry = pending_queue_.front(); |
| result.run_task_ = std::move(entry.second); |
| pending_queue_.pop(); |
| } |
| |
| return result; |
| } |
| |
| void TaskQueueStdlib::ProcessTasks() { |
| started_.Set(); |
| |
| while (true) { |
| auto task = GetNextTask(); |
| |
| if (task.final_task_) |
| break; |
| |
| if (task.run_task_) { |
| // process entry immediately then try again |
| QueuedTask* release_ptr = task.run_task_.release(); |
| if (release_ptr->Run()) |
| delete release_ptr; |
| |
| // attempt to sleep again |
| continue; |
| } |
| |
| if (0 == task.sleep_time_ms_) |
| flag_notify_.Wait(rtc::Event::kForever); |
| else |
| flag_notify_.Wait(task.sleep_time_ms_); |
| } |
| |
| stopped_.Set(); |
| } |
| |
| void TaskQueueStdlib::NotifyWake() { |
| // The queue holds pending tasks to complete. Either tasks are to be |
| // executed immediately or tasks are to be run at some future delayed time. |
| // For immediate tasks the task queue's thread is busy running the task and |
| // the thread will not be waiting on the flag_notify_ event. If no immediate |
| // tasks are available but a delayed task is pending then the thread will be |
| // waiting on flag_notify_ with a delayed time-out of the nearest timed task |
| // to run. If no immediate or pending tasks are available, the thread will |
| // wait on flag_notify_ until signaled that a task has been added (or the |
| // thread to be told to shutdown). |
| |
| // In all cases, when a new immediate task, delayed task, or request to |
| // shutdown the thread is added the flag_notify_ is signaled after. If the |
| // thread was waiting then the thread will wake up immediately and re-assess |
| // what task needs to be run next (i.e. run a task now, wait for the nearest |
| // timed delayed task, or shutdown the thread). If the thread was not waiting |
| // then the thread will remained signaled to wake up the next time any |
| // attempt to wait on the flag_notify_ event occurs. |
| |
| // Any immediate or delayed pending task (or request to shutdown the thread) |
| // must always be added to the queue prior to signaling flag_notify_ to wake |
| // up the possibly sleeping thread. This prevents a race condition where the |
| // thread is notified to wake up but the task queue's thread finds nothing to |
| // do so it waits once again to be signaled where such a signal may never |
| // happen. |
| flag_notify_.Set(); |
| } |
| |
| class TaskQueueStdlibFactory final : public TaskQueueFactory { |
| public: |
| std::unique_ptr<TaskQueueBase, TaskQueueDeleter> CreateTaskQueue( |
| absl::string_view name, |
| Priority priority) const override { |
| return std::unique_ptr<TaskQueueBase, TaskQueueDeleter>( |
| new TaskQueueStdlib(name, TaskQueuePriorityToThreadPriority(priority))); |
| } |
| }; |
| |
| } // namespace |
| |
| std::unique_ptr<TaskQueueFactory> CreateTaskQueueStdlibFactory() { |
| return std::make_unique<TaskQueueStdlibFactory>(); |
| } |
| |
| } // namespace webrtc |