Separates simulated TaskQueue and simulated ProcessThread.
The overlap in functionality is quite limited and separating the functionality makes it a bit easier to follow each. This prepares for adding a SimulatedThread class in a follow up CL. Bug: webrtc:11255 Change-Id: I83c754bd570113dfb582098bb4d39e27bb4f4d87 Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/165688 Reviewed-by: Jonas Olsson <jonasolsson@webrtc.org> Commit-Queue: Sebastian Jansson <srte@webrtc.org> Cr-Commit-Position: refs/heads/master@{#30229}
This commit is contained in:

committed by
Commit Bot

parent
ce0ea49001
commit
53cd9e2645
@ -16,6 +16,10 @@ if (rtc_include_tests) {
|
||||
"external_time_controller.h",
|
||||
"real_time_controller.cc",
|
||||
"real_time_controller.h",
|
||||
"simulated_process_thread.cc",
|
||||
"simulated_process_thread.h",
|
||||
"simulated_task_queue.cc",
|
||||
"simulated_task_queue.h",
|
||||
"simulated_time_controller.cc",
|
||||
"simulated_time_controller.h",
|
||||
]
|
||||
|
181
test/time_controller/simulated_process_thread.cc
Normal file
181
test/time_controller/simulated_process_thread.cc
Normal file
@ -0,0 +1,181 @@
|
||||
/*
|
||||
* Copyright (c) 2020 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 "test/time_controller/simulated_process_thread.h"
|
||||
|
||||
#include <algorithm>
|
||||
#include <utility>
|
||||
|
||||
namespace webrtc {
|
||||
namespace {
|
||||
// Helper function to remove from a std container by value.
|
||||
template <class C>
|
||||
bool RemoveByValue(C* vec, typename C::value_type val) {
|
||||
auto it = std::find(vec->begin(), vec->end(), val);
|
||||
if (it == vec->end())
|
||||
return false;
|
||||
vec->erase(it);
|
||||
return true;
|
||||
}
|
||||
} // namespace
|
||||
SimulatedProcessThread::SimulatedProcessThread(
|
||||
sim_time_impl::SimulatedTimeControllerImpl* handler,
|
||||
absl::string_view name)
|
||||
: handler_(handler), name_(new char[name.size()]) {
|
||||
std::copy_n(name.begin(), name.size(), name_);
|
||||
}
|
||||
|
||||
SimulatedProcessThread::~SimulatedProcessThread() {
|
||||
handler_->Unregister(this);
|
||||
delete[] name_;
|
||||
}
|
||||
|
||||
void SimulatedProcessThread::RunReady(Timestamp at_time) {
|
||||
TokenTaskQueue::CurrentTaskQueueSetter set_current(this);
|
||||
rtc::CritScope lock(&lock_);
|
||||
for (auto it = delayed_modules_.begin();
|
||||
it != delayed_modules_.end() && it->first <= at_time;
|
||||
it = delayed_modules_.erase(it)) {
|
||||
for (auto module : it->second) {
|
||||
ready_modules_.push_back(module);
|
||||
}
|
||||
}
|
||||
if (!ready_modules_.empty()) {
|
||||
for (auto* module : ready_modules_) {
|
||||
module->Process();
|
||||
delayed_modules_[GetNextTime(module, at_time)].push_back(module);
|
||||
}
|
||||
next_run_time_ = delayed_modules_.begin()->first;
|
||||
} else {
|
||||
next_run_time_ = Timestamp::PlusInfinity();
|
||||
}
|
||||
ready_modules_.clear();
|
||||
|
||||
while (!queue_.empty()) {
|
||||
std::unique_ptr<QueuedTask> task = std::move(queue_.front());
|
||||
queue_.pop_front();
|
||||
lock_.Leave();
|
||||
bool should_delete = task->Run();
|
||||
RTC_CHECK(should_delete);
|
||||
lock_.Enter();
|
||||
}
|
||||
}
|
||||
void SimulatedProcessThread::Start() {
|
||||
std::vector<Module*> starting;
|
||||
{
|
||||
rtc::CritScope lock(&lock_);
|
||||
if (process_thread_running_)
|
||||
return;
|
||||
process_thread_running_ = true;
|
||||
starting.swap(stopped_modules_);
|
||||
}
|
||||
for (auto& module : starting)
|
||||
module->ProcessThreadAttached(this);
|
||||
|
||||
Timestamp at_time = handler_->CurrentTime();
|
||||
rtc::CritScope lock(&lock_);
|
||||
for (auto& module : starting)
|
||||
delayed_modules_[GetNextTime(module, at_time)].push_back(module);
|
||||
|
||||
if (!ready_modules_.empty() || !queue_.empty()) {
|
||||
next_run_time_ = Timestamp::MinusInfinity();
|
||||
} else if (!delayed_modules_.empty()) {
|
||||
next_run_time_ = delayed_modules_.begin()->first;
|
||||
} else {
|
||||
next_run_time_ = Timestamp::PlusInfinity();
|
||||
}
|
||||
}
|
||||
|
||||
void SimulatedProcessThread::Stop() {
|
||||
std::vector<Module*> stopping;
|
||||
{
|
||||
rtc::CritScope lock(&lock_);
|
||||
process_thread_running_ = false;
|
||||
|
||||
for (auto* ready : ready_modules_)
|
||||
stopped_modules_.push_back(ready);
|
||||
ready_modules_.clear();
|
||||
|
||||
for (auto& delayed : delayed_modules_) {
|
||||
for (auto mod : delayed.second)
|
||||
stopped_modules_.push_back(mod);
|
||||
}
|
||||
delayed_modules_.clear();
|
||||
|
||||
stopping = stopped_modules_;
|
||||
}
|
||||
for (auto& module : stopping)
|
||||
module->ProcessThreadAttached(nullptr);
|
||||
}
|
||||
|
||||
void SimulatedProcessThread::WakeUp(Module* module) {
|
||||
rtc::CritScope lock(&lock_);
|
||||
// If we already are planning to run this module as soon as possible, we don't
|
||||
// need to do anything.
|
||||
for (auto mod : ready_modules_)
|
||||
if (mod == module)
|
||||
return;
|
||||
|
||||
for (auto it = delayed_modules_.begin(); it != delayed_modules_.end(); ++it) {
|
||||
if (RemoveByValue(&it->second, module))
|
||||
break;
|
||||
}
|
||||
Timestamp next_time = GetNextTime(module, handler_->CurrentTime());
|
||||
delayed_modules_[next_time].push_back(module);
|
||||
next_run_time_ = std::min(next_run_time_, next_time);
|
||||
}
|
||||
|
||||
void SimulatedProcessThread::RegisterModule(Module* module,
|
||||
const rtc::Location& from) {
|
||||
module->ProcessThreadAttached(this);
|
||||
rtc::CritScope lock(&lock_);
|
||||
if (!process_thread_running_) {
|
||||
stopped_modules_.push_back(module);
|
||||
} else {
|
||||
Timestamp next_time = GetNextTime(module, handler_->CurrentTime());
|
||||
delayed_modules_[next_time].push_back(module);
|
||||
next_run_time_ = std::min(next_run_time_, next_time);
|
||||
}
|
||||
}
|
||||
|
||||
void SimulatedProcessThread::DeRegisterModule(Module* module) {
|
||||
bool modules_running;
|
||||
{
|
||||
rtc::CritScope lock(&lock_);
|
||||
if (!process_thread_running_) {
|
||||
RemoveByValue(&stopped_modules_, module);
|
||||
} else {
|
||||
bool removed = RemoveByValue(&ready_modules_, module);
|
||||
if (!removed) {
|
||||
for (auto& pair : delayed_modules_) {
|
||||
if (RemoveByValue(&pair.second, module))
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
modules_running = process_thread_running_;
|
||||
}
|
||||
if (modules_running)
|
||||
module->ProcessThreadAttached(nullptr);
|
||||
}
|
||||
|
||||
void SimulatedProcessThread::PostTask(std::unique_ptr<QueuedTask> task) {
|
||||
rtc::CritScope lock(&lock_);
|
||||
queue_.emplace_back(std::move(task));
|
||||
next_run_time_ = Timestamp::MinusInfinity();
|
||||
}
|
||||
|
||||
Timestamp SimulatedProcessThread::GetNextTime(Module* module,
|
||||
Timestamp at_time) {
|
||||
CurrentTaskQueueSetter set_current(this);
|
||||
return at_time + TimeDelta::ms(module->TimeUntilNextProcess());
|
||||
}
|
||||
|
||||
} // namespace webrtc
|
66
test/time_controller/simulated_process_thread.h
Normal file
66
test/time_controller/simulated_process_thread.h
Normal file
@ -0,0 +1,66 @@
|
||||
/*
|
||||
* Copyright (c) 2020 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.
|
||||
*/
|
||||
#ifndef TEST_TIME_CONTROLLER_SIMULATED_PROCESS_THREAD_H_
|
||||
#define TEST_TIME_CONTROLLER_SIMULATED_PROCESS_THREAD_H_
|
||||
|
||||
#include <deque>
|
||||
#include <list>
|
||||
#include <map>
|
||||
#include <memory>
|
||||
#include <vector>
|
||||
|
||||
#include "test/time_controller/simulated_time_controller.h"
|
||||
|
||||
namespace webrtc {
|
||||
|
||||
class SimulatedProcessThread : public TokenTaskQueue,
|
||||
public ProcessThread,
|
||||
public sim_time_impl::SimulatedSequenceRunner {
|
||||
public:
|
||||
SimulatedProcessThread(sim_time_impl::SimulatedTimeControllerImpl* handler,
|
||||
absl::string_view name);
|
||||
virtual ~SimulatedProcessThread();
|
||||
void RunReady(Timestamp at_time) override;
|
||||
|
||||
Timestamp GetNextRunTime() const override {
|
||||
rtc::CritScope lock(&lock_);
|
||||
return next_run_time_;
|
||||
}
|
||||
|
||||
TaskQueueBase* GetAsTaskQueue() override { return this; }
|
||||
|
||||
// ProcessThread interface
|
||||
void Start() override;
|
||||
void Stop() override;
|
||||
void WakeUp(Module* module) override;
|
||||
void RegisterModule(Module* module, const rtc::Location& from) override;
|
||||
void DeRegisterModule(Module* module) override;
|
||||
void PostTask(std::unique_ptr<QueuedTask> task) override;
|
||||
|
||||
private:
|
||||
Timestamp GetNextTime(Module* module, Timestamp at_time);
|
||||
|
||||
sim_time_impl::SimulatedTimeControllerImpl* const handler_;
|
||||
// Using char* to be debugger friendly.
|
||||
char* name_;
|
||||
rtc::CriticalSection lock_;
|
||||
Timestamp next_run_time_ RTC_GUARDED_BY(lock_) = Timestamp::PlusInfinity();
|
||||
|
||||
std::deque<std::unique_ptr<QueuedTask>> queue_;
|
||||
|
||||
bool process_thread_running_ RTC_GUARDED_BY(lock_) = false;
|
||||
std::vector<Module*> stopped_modules_ RTC_GUARDED_BY(lock_);
|
||||
std::vector<Module*> ready_modules_ RTC_GUARDED_BY(lock_);
|
||||
std::map<Timestamp, std::list<Module*>> delayed_modules_
|
||||
RTC_GUARDED_BY(lock_);
|
||||
};
|
||||
} // namespace webrtc
|
||||
|
||||
#endif // TEST_TIME_CONTROLLER_SIMULATED_PROCESS_THREAD_H_
|
81
test/time_controller/simulated_task_queue.cc
Normal file
81
test/time_controller/simulated_task_queue.cc
Normal file
@ -0,0 +1,81 @@
|
||||
/*
|
||||
* Copyright (c) 2020 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 "test/time_controller/simulated_task_queue.h"
|
||||
|
||||
#include <algorithm>
|
||||
#include <utility>
|
||||
|
||||
namespace webrtc {
|
||||
|
||||
SimulatedTaskQueue::SimulatedTaskQueue(
|
||||
sim_time_impl::SimulatedTimeControllerImpl* handler,
|
||||
absl::string_view name)
|
||||
: handler_(handler), name_(new char[name.size()]) {
|
||||
std::copy_n(name.begin(), name.size(), name_);
|
||||
}
|
||||
|
||||
SimulatedTaskQueue::~SimulatedTaskQueue() {
|
||||
handler_->Unregister(this);
|
||||
delete[] name_;
|
||||
}
|
||||
|
||||
void SimulatedTaskQueue::Delete() {
|
||||
{
|
||||
rtc::CritScope lock(&lock_);
|
||||
ready_tasks_.clear();
|
||||
delayed_tasks_.clear();
|
||||
}
|
||||
delete this;
|
||||
}
|
||||
|
||||
void SimulatedTaskQueue::RunReady(Timestamp at_time) {
|
||||
rtc::CritScope lock(&lock_);
|
||||
for (auto it = delayed_tasks_.begin();
|
||||
it != delayed_tasks_.end() && it->first <= at_time;
|
||||
it = delayed_tasks_.erase(it)) {
|
||||
for (auto& task : it->second) {
|
||||
ready_tasks_.emplace_back(std::move(task));
|
||||
}
|
||||
}
|
||||
CurrentTaskQueueSetter set_current(this);
|
||||
while (!ready_tasks_.empty()) {
|
||||
std::unique_ptr<QueuedTask> ready = std::move(ready_tasks_.front());
|
||||
ready_tasks_.pop_front();
|
||||
lock_.Leave();
|
||||
bool delete_task = ready->Run();
|
||||
if (delete_task) {
|
||||
ready.reset();
|
||||
} else {
|
||||
ready.release();
|
||||
}
|
||||
lock_.Enter();
|
||||
}
|
||||
if (!delayed_tasks_.empty()) {
|
||||
next_run_time_ = delayed_tasks_.begin()->first;
|
||||
} else {
|
||||
next_run_time_ = Timestamp::PlusInfinity();
|
||||
}
|
||||
}
|
||||
|
||||
void SimulatedTaskQueue::PostTask(std::unique_ptr<QueuedTask> task) {
|
||||
rtc::CritScope lock(&lock_);
|
||||
ready_tasks_.emplace_back(std::move(task));
|
||||
next_run_time_ = Timestamp::MinusInfinity();
|
||||
}
|
||||
|
||||
void SimulatedTaskQueue::PostDelayedTask(std::unique_ptr<QueuedTask> task,
|
||||
uint32_t milliseconds) {
|
||||
rtc::CritScope lock(&lock_);
|
||||
Timestamp target_time = handler_->CurrentTime() + TimeDelta::ms(milliseconds);
|
||||
delayed_tasks_[target_time].push_back(std::move(task));
|
||||
next_run_time_ = std::min(next_run_time_, target_time);
|
||||
}
|
||||
|
||||
} // namespace webrtc
|
60
test/time_controller/simulated_task_queue.h
Normal file
60
test/time_controller/simulated_task_queue.h
Normal file
@ -0,0 +1,60 @@
|
||||
/*
|
||||
* Copyright (c) 2020 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.
|
||||
*/
|
||||
#ifndef TEST_TIME_CONTROLLER_SIMULATED_TASK_QUEUE_H_
|
||||
#define TEST_TIME_CONTROLLER_SIMULATED_TASK_QUEUE_H_
|
||||
|
||||
#include <deque>
|
||||
#include <map>
|
||||
#include <memory>
|
||||
#include <vector>
|
||||
|
||||
#include "test/time_controller/simulated_time_controller.h"
|
||||
|
||||
namespace webrtc {
|
||||
|
||||
class SimulatedTaskQueue : public TaskQueueBase,
|
||||
public sim_time_impl::SimulatedSequenceRunner {
|
||||
public:
|
||||
SimulatedTaskQueue(sim_time_impl::SimulatedTimeControllerImpl* handler,
|
||||
absl::string_view name);
|
||||
|
||||
~SimulatedTaskQueue();
|
||||
|
||||
void RunReady(Timestamp at_time) override;
|
||||
|
||||
Timestamp GetNextRunTime() const override {
|
||||
rtc::CritScope lock(&lock_);
|
||||
return next_run_time_;
|
||||
}
|
||||
TaskQueueBase* GetAsTaskQueue() override { return this; }
|
||||
|
||||
// TaskQueueBase interface
|
||||
void Delete() override;
|
||||
void PostTask(std::unique_ptr<QueuedTask> task) override;
|
||||
void PostDelayedTask(std::unique_ptr<QueuedTask> task,
|
||||
uint32_t milliseconds) override;
|
||||
|
||||
private:
|
||||
sim_time_impl::SimulatedTimeControllerImpl* const handler_;
|
||||
// Using char* to be debugger friendly.
|
||||
char* name_;
|
||||
|
||||
rtc::CriticalSection lock_;
|
||||
|
||||
std::deque<std::unique_ptr<QueuedTask>> ready_tasks_ RTC_GUARDED_BY(lock_);
|
||||
std::map<Timestamp, std::vector<std::unique_ptr<QueuedTask>>> delayed_tasks_
|
||||
RTC_GUARDED_BY(lock_);
|
||||
|
||||
Timestamp next_run_time_ RTC_GUARDED_BY(lock_) = Timestamp::PlusInfinity();
|
||||
};
|
||||
|
||||
} // namespace webrtc
|
||||
|
||||
#endif // TEST_TIME_CONTROLLER_SIMULATED_TASK_QUEUE_H_
|
@ -12,278 +12,29 @@
|
||||
#include <algorithm>
|
||||
#include <deque>
|
||||
#include <list>
|
||||
#include <map>
|
||||
#include <memory>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <vector>
|
||||
|
||||
#include "absl/strings/string_view.h"
|
||||
#include "test/time_controller/simulated_process_thread.h"
|
||||
#include "test/time_controller/simulated_task_queue.h"
|
||||
|
||||
namespace webrtc {
|
||||
namespace {
|
||||
// Helper function to remove from a std container by value.
|
||||
template <class C>
|
||||
bool RemoveByValue(C& vec, typename C::value_type val) {
|
||||
auto it = std::find(vec.begin(), vec.end(), val);
|
||||
if (it == vec.end())
|
||||
bool RemoveByValue(C* vec, typename C::value_type val) {
|
||||
auto it = std::find(vec->begin(), vec->end(), val);
|
||||
if (it == vec->end())
|
||||
return false;
|
||||
vec.erase(it);
|
||||
vec->erase(it);
|
||||
return true;
|
||||
}
|
||||
} // namespace
|
||||
|
||||
namespace sim_time_impl {
|
||||
class SimulatedSequenceRunner : public ProcessThread, public TaskQueueBase {
|
||||
public:
|
||||
SimulatedSequenceRunner(SimulatedTimeControllerImpl* handler,
|
||||
absl::string_view queue_name)
|
||||
: handler_(handler), name_(queue_name) {}
|
||||
~SimulatedSequenceRunner() override { handler_->Unregister(this); }
|
||||
|
||||
// Provides next run time.
|
||||
Timestamp GetNextRunTime() const;
|
||||
|
||||
// Iterates through delayed tasks and modules and moves them to the ready set
|
||||
// if they are supposed to execute by |at time|.
|
||||
void UpdateReady(Timestamp at_time);
|
||||
// Runs all ready tasks and modules and updates next run time.
|
||||
void Run(Timestamp at_time);
|
||||
|
||||
// TaskQueueBase interface
|
||||
void Delete() override;
|
||||
// Note: PostTask is also in ProcessThread interface.
|
||||
void PostTask(std::unique_ptr<QueuedTask> task) override;
|
||||
void PostDelayedTask(std::unique_ptr<QueuedTask> task,
|
||||
uint32_t milliseconds) override;
|
||||
|
||||
// ProcessThread interface
|
||||
void Start() override;
|
||||
void Stop() override;
|
||||
void WakeUp(Module* module) override;
|
||||
void RegisterModule(Module* module, const rtc::Location& from) override;
|
||||
void DeRegisterModule(Module* module) override;
|
||||
// Promoted to public for use in SimulatedTimeControllerImpl::YieldExecution.
|
||||
using CurrentTaskQueueSetter = TaskQueueBase::CurrentTaskQueueSetter;
|
||||
|
||||
private:
|
||||
Timestamp GetCurrentTime() const { return handler_->CurrentTime(); }
|
||||
void RunReadyTasks(Timestamp at_time) RTC_LOCKS_EXCLUDED(lock_);
|
||||
void RunReadyModules(Timestamp at_time) RTC_EXCLUSIVE_LOCKS_REQUIRED(lock_);
|
||||
void UpdateNextRunTime() RTC_EXCLUSIVE_LOCKS_REQUIRED(lock_);
|
||||
Timestamp GetNextTime(Module* module, Timestamp at_time);
|
||||
|
||||
SimulatedTimeControllerImpl* const handler_;
|
||||
const std::string name_;
|
||||
|
||||
rtc::CriticalSection lock_;
|
||||
|
||||
std::deque<std::unique_ptr<QueuedTask>> ready_tasks_ RTC_GUARDED_BY(lock_);
|
||||
std::map<Timestamp, std::vector<std::unique_ptr<QueuedTask>>> delayed_tasks_
|
||||
RTC_GUARDED_BY(lock_);
|
||||
|
||||
bool process_thread_running_ RTC_GUARDED_BY(lock_) = false;
|
||||
std::vector<Module*> stopped_modules_ RTC_GUARDED_BY(lock_);
|
||||
std::vector<Module*> ready_modules_ RTC_GUARDED_BY(lock_);
|
||||
std::map<Timestamp, std::list<Module*>> delayed_modules_
|
||||
RTC_GUARDED_BY(lock_);
|
||||
|
||||
Timestamp next_run_time_ RTC_GUARDED_BY(lock_) = Timestamp::PlusInfinity();
|
||||
};
|
||||
|
||||
Timestamp SimulatedSequenceRunner::GetNextRunTime() const {
|
||||
rtc::CritScope lock(&lock_);
|
||||
return next_run_time_;
|
||||
}
|
||||
|
||||
void SimulatedSequenceRunner::UpdateReady(Timestamp at_time) {
|
||||
rtc::CritScope lock(&lock_);
|
||||
for (auto it = delayed_tasks_.begin();
|
||||
it != delayed_tasks_.end() && it->first <= at_time;
|
||||
it = delayed_tasks_.erase(it)) {
|
||||
for (auto& task : it->second) {
|
||||
ready_tasks_.emplace_back(std::move(task));
|
||||
}
|
||||
}
|
||||
for (auto it = delayed_modules_.begin();
|
||||
it != delayed_modules_.end() && it->first <= at_time;
|
||||
it = delayed_modules_.erase(it)) {
|
||||
for (auto module : it->second) {
|
||||
ready_modules_.push_back(module);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void SimulatedSequenceRunner::Run(Timestamp at_time) {
|
||||
RunReadyTasks(at_time);
|
||||
rtc::CritScope lock(&lock_);
|
||||
RunReadyModules(at_time);
|
||||
UpdateNextRunTime();
|
||||
}
|
||||
|
||||
void SimulatedSequenceRunner::Delete() {
|
||||
{
|
||||
rtc::CritScope lock(&lock_);
|
||||
ready_tasks_.clear();
|
||||
delayed_tasks_.clear();
|
||||
}
|
||||
delete this;
|
||||
}
|
||||
|
||||
void SimulatedSequenceRunner::RunReadyTasks(Timestamp at_time) {
|
||||
std::deque<std::unique_ptr<QueuedTask>> ready_tasks;
|
||||
{
|
||||
rtc::CritScope lock(&lock_);
|
||||
ready_tasks.swap(ready_tasks_);
|
||||
}
|
||||
if (!ready_tasks.empty()) {
|
||||
CurrentTaskQueueSetter set_current(this);
|
||||
for (auto& ready : ready_tasks) {
|
||||
bool delete_task = ready->Run();
|
||||
if (delete_task) {
|
||||
ready.reset();
|
||||
} else {
|
||||
ready.release();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void SimulatedSequenceRunner::RunReadyModules(Timestamp at_time) {
|
||||
if (!ready_modules_.empty()) {
|
||||
CurrentTaskQueueSetter set_current(this);
|
||||
for (auto* module : ready_modules_) {
|
||||
module->Process();
|
||||
delayed_modules_[GetNextTime(module, at_time)].push_back(module);
|
||||
}
|
||||
}
|
||||
ready_modules_.clear();
|
||||
}
|
||||
|
||||
void SimulatedSequenceRunner::UpdateNextRunTime() {
|
||||
if (!ready_tasks_.empty() || !ready_modules_.empty()) {
|
||||
next_run_time_ = Timestamp::MinusInfinity();
|
||||
} else {
|
||||
next_run_time_ = Timestamp::PlusInfinity();
|
||||
if (!delayed_tasks_.empty())
|
||||
next_run_time_ = std::min(next_run_time_, delayed_tasks_.begin()->first);
|
||||
if (!delayed_modules_.empty())
|
||||
next_run_time_ =
|
||||
std::min(next_run_time_, delayed_modules_.begin()->first);
|
||||
}
|
||||
}
|
||||
|
||||
void SimulatedSequenceRunner::PostTask(std::unique_ptr<QueuedTask> task) {
|
||||
rtc::CritScope lock(&lock_);
|
||||
ready_tasks_.emplace_back(std::move(task));
|
||||
next_run_time_ = Timestamp::MinusInfinity();
|
||||
}
|
||||
|
||||
void SimulatedSequenceRunner::PostDelayedTask(std::unique_ptr<QueuedTask> task,
|
||||
uint32_t milliseconds) {
|
||||
rtc::CritScope lock(&lock_);
|
||||
Timestamp target_time = GetCurrentTime() + TimeDelta::ms(milliseconds);
|
||||
delayed_tasks_[target_time].push_back(std::move(task));
|
||||
next_run_time_ = std::min(next_run_time_, target_time);
|
||||
}
|
||||
|
||||
void SimulatedSequenceRunner::Start() {
|
||||
std::vector<Module*> starting;
|
||||
{
|
||||
rtc::CritScope lock(&lock_);
|
||||
if (process_thread_running_)
|
||||
return;
|
||||
process_thread_running_ = true;
|
||||
starting.swap(stopped_modules_);
|
||||
}
|
||||
for (auto& module : starting)
|
||||
module->ProcessThreadAttached(this);
|
||||
|
||||
Timestamp at_time = GetCurrentTime();
|
||||
rtc::CritScope lock(&lock_);
|
||||
for (auto& module : starting)
|
||||
delayed_modules_[GetNextTime(module, at_time)].push_back(module);
|
||||
UpdateNextRunTime();
|
||||
}
|
||||
|
||||
void SimulatedSequenceRunner::Stop() {
|
||||
std::vector<Module*> stopping;
|
||||
{
|
||||
rtc::CritScope lock(&lock_);
|
||||
process_thread_running_ = false;
|
||||
|
||||
for (auto* ready : ready_modules_)
|
||||
stopped_modules_.push_back(ready);
|
||||
ready_modules_.clear();
|
||||
|
||||
for (auto& delayed : delayed_modules_) {
|
||||
for (auto mod : delayed.second)
|
||||
stopped_modules_.push_back(mod);
|
||||
}
|
||||
delayed_modules_.clear();
|
||||
|
||||
stopping = stopped_modules_;
|
||||
}
|
||||
for (auto& module : stopping)
|
||||
module->ProcessThreadAttached(nullptr);
|
||||
}
|
||||
|
||||
void SimulatedSequenceRunner::WakeUp(Module* module) {
|
||||
rtc::CritScope lock(&lock_);
|
||||
// If we already are planning to run this module as soon as possible, we don't
|
||||
// need to do anything.
|
||||
for (auto mod : ready_modules_)
|
||||
if (mod == module)
|
||||
return;
|
||||
|
||||
for (auto it = delayed_modules_.begin(); it != delayed_modules_.end(); ++it) {
|
||||
if (RemoveByValue(it->second, module))
|
||||
break;
|
||||
}
|
||||
Timestamp next_time = GetNextTime(module, GetCurrentTime());
|
||||
delayed_modules_[next_time].push_back(module);
|
||||
next_run_time_ = std::min(next_run_time_, next_time);
|
||||
}
|
||||
|
||||
void SimulatedSequenceRunner::RegisterModule(Module* module,
|
||||
const rtc::Location& from) {
|
||||
module->ProcessThreadAttached(this);
|
||||
rtc::CritScope lock(&lock_);
|
||||
if (!process_thread_running_) {
|
||||
stopped_modules_.push_back(module);
|
||||
} else {
|
||||
Timestamp next_time = GetNextTime(module, GetCurrentTime());
|
||||
delayed_modules_[next_time].push_back(module);
|
||||
next_run_time_ = std::min(next_run_time_, next_time);
|
||||
}
|
||||
}
|
||||
|
||||
void SimulatedSequenceRunner::DeRegisterModule(Module* module) {
|
||||
bool modules_running;
|
||||
{
|
||||
rtc::CritScope lock(&lock_);
|
||||
if (!process_thread_running_) {
|
||||
RemoveByValue(stopped_modules_, module);
|
||||
} else {
|
||||
bool removed = RemoveByValue(ready_modules_, module);
|
||||
if (!removed) {
|
||||
for (auto& pair : delayed_modules_) {
|
||||
if (RemoveByValue(pair.second, module))
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
modules_running = process_thread_running_;
|
||||
}
|
||||
if (modules_running)
|
||||
module->ProcessThreadAttached(nullptr);
|
||||
}
|
||||
|
||||
Timestamp SimulatedSequenceRunner::GetNextTime(Module* module,
|
||||
Timestamp at_time) {
|
||||
CurrentTaskQueueSetter set_current(this);
|
||||
return at_time + TimeDelta::ms(module->TimeUntilNextProcess());
|
||||
}
|
||||
|
||||
SimulatedTimeControllerImpl::SimulatedTimeControllerImpl(Timestamp start_time)
|
||||
: thread_id_(rtc::CurrentThreadId()), current_time_(start_time) {}
|
||||
@ -296,8 +47,8 @@ SimulatedTimeControllerImpl::CreateTaskQueue(
|
||||
TaskQueueFactory::Priority priority) const {
|
||||
// TODO(srte): Remove the const cast when the interface is made mutable.
|
||||
auto mutable_this = const_cast<SimulatedTimeControllerImpl*>(this);
|
||||
auto task_queue = std::unique_ptr<SimulatedSequenceRunner, TaskQueueDeleter>(
|
||||
new SimulatedSequenceRunner(mutable_this, name));
|
||||
auto task_queue = std::unique_ptr<SimulatedTaskQueue, TaskQueueDeleter>(
|
||||
new SimulatedTaskQueue(mutable_this, name));
|
||||
rtc::CritScope lock(&mutable_this->lock_);
|
||||
mutable_this->runners_.push_back(task_queue.get());
|
||||
return task_queue;
|
||||
@ -307,7 +58,7 @@ std::unique_ptr<ProcessThread> SimulatedTimeControllerImpl::CreateProcessThread(
|
||||
const char* thread_name) {
|
||||
rtc::CritScope lock(&lock_);
|
||||
auto process_thread =
|
||||
std::make_unique<SimulatedSequenceRunner>(this, thread_name);
|
||||
std::make_unique<SimulatedProcessThread>(this, thread_name);
|
||||
runners_.push_back(process_thread.get());
|
||||
return process_thread;
|
||||
}
|
||||
@ -319,8 +70,7 @@ void SimulatedTimeControllerImpl::YieldExecution() {
|
||||
// the thread local task queue reference. This ensures that thread checkers
|
||||
// won't think we are executing on the yielding task queue. It also ensure
|
||||
// that TaskQueueBase::Current() won't return the yielding task queue.
|
||||
SimulatedSequenceRunner::CurrentTaskQueueSetter reset_queue(nullptr);
|
||||
RTC_DCHECK_RUN_ON(&thread_checker_);
|
||||
TokenTaskQueue::CurrentTaskQueueSetter reset_queue(nullptr);
|
||||
// When we yield, we don't want to risk executing further tasks on the
|
||||
// currently executing task queue. If there's a ready task that also yields,
|
||||
// it's added to this set as well and only tasks on the remaining task
|
||||
@ -333,7 +83,6 @@ void SimulatedTimeControllerImpl::YieldExecution() {
|
||||
}
|
||||
|
||||
void SimulatedTimeControllerImpl::RunReadyRunners() {
|
||||
RTC_DCHECK_RUN_ON(&thread_checker_);
|
||||
rtc::CritScope lock(&lock_);
|
||||
RTC_DCHECK_EQ(rtc::CurrentThreadId(), thread_id_);
|
||||
Timestamp current_time = CurrentTime();
|
||||
@ -345,21 +94,20 @@ void SimulatedTimeControllerImpl::RunReadyRunners() {
|
||||
// runners.
|
||||
while (true) {
|
||||
for (auto* runner : runners_) {
|
||||
if (yielded_.find(runner) == yielded_.end() &&
|
||||
if (yielded_.find(runner->GetAsTaskQueue()) == yielded_.end() &&
|
||||
runner->GetNextRunTime() <= current_time) {
|
||||
ready_runners_.push_back(runner);
|
||||
}
|
||||
}
|
||||
if (ready_runners_.empty())
|
||||
return;
|
||||
break;
|
||||
while (!ready_runners_.empty()) {
|
||||
auto* runner = ready_runners_.front();
|
||||
ready_runners_.pop_front();
|
||||
runner->UpdateReady(current_time);
|
||||
// Note that the Run function might indirectly cause a call to
|
||||
// Note that the RunReady function might indirectly cause a call to
|
||||
// Unregister() which will recursively grab |lock_| again to remove items
|
||||
// from |ready_runners_|.
|
||||
runner->Run(current_time);
|
||||
runner->RunReady(current_time);
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -390,11 +138,10 @@ void SimulatedTimeControllerImpl::AdvanceTime(Timestamp target_time) {
|
||||
|
||||
void SimulatedTimeControllerImpl::Unregister(SimulatedSequenceRunner* runner) {
|
||||
rtc::CritScope lock(&lock_);
|
||||
bool removed = RemoveByValue(runners_, runner);
|
||||
bool removed = RemoveByValue(&runners_, runner);
|
||||
RTC_CHECK(removed);
|
||||
RemoveByValue(ready_runners_, runner);
|
||||
RemoveByValue(&ready_runners_, runner);
|
||||
}
|
||||
|
||||
} // namespace sim_time_impl
|
||||
|
||||
GlobalSimulatedTimeController::GlobalSimulatedTimeController(
|
||||
|
@ -16,6 +16,7 @@
|
||||
#include <utility>
|
||||
#include <vector>
|
||||
|
||||
#include "absl/strings/string_view.h"
|
||||
#include "api/test/time_controller.h"
|
||||
#include "api/units/timestamp.h"
|
||||
#include "modules/include/module.h"
|
||||
@ -27,9 +28,21 @@
|
||||
#include "rtc_base/thread_checker.h"
|
||||
|
||||
namespace webrtc {
|
||||
|
||||
namespace sim_time_impl {
|
||||
class SimulatedSequenceRunner;
|
||||
class SimulatedSequenceRunner {
|
||||
public:
|
||||
virtual ~SimulatedSequenceRunner() = default;
|
||||
// Provides next run time.
|
||||
virtual Timestamp GetNextRunTime() const = 0;
|
||||
// Runs all ready tasks and modules and updates next run time.
|
||||
virtual void RunReady(Timestamp at_time) = 0;
|
||||
|
||||
// All implementations also implements TaskQueueBase in some form, but if we'd
|
||||
// inherit from it in this interface we'd run into issues with double
|
||||
// inheritance. Therefore we simply allow the implementations to provide a
|
||||
// casted pointer to themself.
|
||||
virtual TaskQueueBase* GetAsTaskQueue() = 0;
|
||||
};
|
||||
|
||||
class SimulatedTimeControllerImpl : public TaskQueueFactory,
|
||||
public rtc::YieldInterface {
|
||||
@ -47,6 +60,7 @@ class SimulatedTimeControllerImpl : public TaskQueueFactory,
|
||||
void YieldExecution() override;
|
||||
// Create process thread with the name |thread_name|.
|
||||
std::unique_ptr<ProcessThread> CreateProcessThread(const char* thread_name);
|
||||
|
||||
// Runs all runners in |runners_| that has tasks or modules ready for
|
||||
// execution.
|
||||
void RunReadyRunners();
|
||||
@ -61,7 +75,6 @@ class SimulatedTimeControllerImpl : public TaskQueueFactory,
|
||||
|
||||
private:
|
||||
const rtc::PlatformThreadId thread_id_;
|
||||
rtc::ThreadChecker thread_checker_;
|
||||
rtc::CriticalSection time_lock_;
|
||||
Timestamp current_time_ RTC_GUARDED_BY(time_lock_);
|
||||
rtc::CriticalSection lock_;
|
||||
@ -71,11 +84,27 @@ class SimulatedTimeControllerImpl : public TaskQueueFactory,
|
||||
// runners can removed from here by Unregister().
|
||||
std::list<SimulatedSequenceRunner*> ready_runners_ RTC_GUARDED_BY(lock_);
|
||||
|
||||
// Task queues on which YieldExecution has been called.
|
||||
std::unordered_set<TaskQueueBase*> yielded_ RTC_GUARDED_BY(thread_checker_);
|
||||
// Runners on which YieldExecution has been called.
|
||||
std::unordered_set<TaskQueueBase*> yielded_;
|
||||
};
|
||||
} // namespace sim_time_impl
|
||||
|
||||
// Used to satisfy sequence checkers for non task queue sequences.
|
||||
class TokenTaskQueue : public TaskQueueBase {
|
||||
public:
|
||||
// Promoted to public
|
||||
using CurrentTaskQueueSetter = TaskQueueBase::CurrentTaskQueueSetter;
|
||||
|
||||
void Delete() override { RTC_NOTREACHED(); }
|
||||
void PostTask(std::unique_ptr<QueuedTask> /*task*/) override {
|
||||
RTC_NOTREACHED();
|
||||
}
|
||||
void PostDelayedTask(std::unique_ptr<QueuedTask> /*task*/,
|
||||
uint32_t /*milliseconds*/) override {
|
||||
RTC_NOTREACHED();
|
||||
}
|
||||
};
|
||||
|
||||
// TimeController implementation using completely simulated time. Task queues
|
||||
// and process threads created by this controller will run delayed activities
|
||||
// when AdvanceTime() is called. Overrides the global clock backing
|
||||
@ -90,6 +119,7 @@ class GlobalSimulatedTimeController : public TimeController {
|
||||
TaskQueueFactory* GetTaskQueueFactory() override;
|
||||
std::unique_ptr<ProcessThread> CreateProcessThread(
|
||||
const char* thread_name) override;
|
||||
|
||||
void AdvanceTime(TimeDelta duration) override;
|
||||
|
||||
private:
|
||||
|
Reference in New Issue
Block a user