zqz/platform-external-webrtc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Revamp the ProcessThreadImpl implementation.

Browse Source 
* Add a new WakeUp method that gives a module a chance to be called back right away on the worker thread.
* Wrote unit tests for the class.
* Significantly reduce the amount of locking.
  - ProcessThreadImpl itself does a lot less locking.
  - Reimplemented the way we keep track of when to make calls to Process.
    This reduces the amount of calls to TimeUntilNextProcess and since most implementations of that function grab a lock, this means less locking.
* Renamed ProcessThread::CreateProcessThread to ProcessThread::Create.
* Added thread checks for Start/Stop.  Threading model of other functions is now documented.
* We now log an error if an implementation of TimeUntilNextProcess returns a negative value (some implementations do, but the method should only return a positive nr of ms).
* Removed the DestroyProcessThread method and instead force callers to use scoped_ptr<> to maintain object lifetime.

BUG=2822
R=henrika@webrtc.org

Review URL: https://webrtc-codereview.appspot.com/35999004

Cr-Commit-Position: refs/heads/master@{#8261}
git-svn-id: http://webrtc.googlecode.com/svn/trunk@8261 4adac7df-926f-26a2-2b94-8c16560cd09d
This commit is contained in:
tommi@webrtc.org
2015-02-06 09:44:12 +00:00
parent 75025434bf
commit 0c3e12b7bf
15 changed files with 463 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files

250
webrtc/modules/utility/source/process_thread_impl.cc
View File

@ -8,163 +8,161 @@
* be found in the AUTHORS file in the root of the source tree.
*/
#include "webrtc/modules/interface/module.h"
#include "webrtc/modules/utility/source/process_thread_impl.h"
#include "webrtc/base/checks.h"
#include "webrtc/modules/interface/module.h"
#include "webrtc/system_wrappers/interface/logging.h"
#include "webrtc/system_wrappers/interface/tick_util.h"
namespace webrtc {
ProcessThread::~ProcessThread()
{
namespace {
int64_t GetNextCallbackTime(Module* module, int64_t time_now) {
int64_t interval = module->TimeUntilNextProcess();
// Currently some implementations erroneously return error codes from
// TimeUntilNextProcess(). So, as is, we correct that and log an error.
if (interval < 0) {
LOG(LS_ERROR) << "TimeUntilNextProcess returned an invalid value "
<< interval;
interval = 0;
}
return time_now + interval;
}
}
ProcessThread* ProcessThread::CreateProcessThread()
{
return new ProcessThreadImpl();
}
ProcessThread::~ProcessThread() {}
void ProcessThread::DestroyProcessThread(ProcessThread* module)
{
delete module;
// static
rtc::scoped_ptr<ProcessThread> ProcessThread::Create() {
return rtc::scoped_ptr<ProcessThread>(new ProcessThreadImpl()).Pass();
}
ProcessThreadImpl::ProcessThreadImpl()
: _timeEvent(*EventWrapper::Create()),
_critSectModules(CriticalSectionWrapper::CreateCriticalSection()),
_thread(NULL)
{
: wake_up_(EventWrapper::Create()), stop_(false) {
}
ProcessThreadImpl::~ProcessThreadImpl()
{
delete _critSectModules;
delete &_timeEvent;
ProcessThreadImpl::~ProcessThreadImpl() {
DCHECK(thread_checker_.CalledOnValidThread());
DCHECK(!thread_.get());
DCHECK(!stop_);
}
int32_t ProcessThreadImpl::Start()
{
CriticalSectionScoped lock(_critSectModules);
if(_thread)
{
return -1;
}
_thread = ThreadWrapper::CreateThread(Run, this, kNormalPriority,
"ProcessThread");
unsigned int id;
int32_t retVal = _thread->Start(id);
if(retVal >= 0)
{
return 0;
}
delete _thread;
_thread = NULL;
int32_t ProcessThreadImpl::Start() {
DCHECK(thread_checker_.CalledOnValidThread());
if (thread_.get())
return -1;
DCHECK(!stop_);
thread_.reset(ThreadWrapper::CreateThread(
&ProcessThreadImpl::Run, this, kNormalPriority, "ProcessThread"));
unsigned int id;
if (!thread_->Start(id)) {
thread_.reset();
return -1;
}
return 0;
}
int32_t ProcessThreadImpl::Stop()
{
_critSectModules->Enter();
if(_thread)
{
ThreadWrapper* thread = _thread;
_thread = NULL;
_timeEvent.Set();
_critSectModules->Leave();
if(thread->Stop())
{
delete thread;
} else {
return -1;
}
} else {
_critSectModules->Leave();
}
int32_t ProcessThreadImpl::Stop() {
DCHECK(thread_checker_.CalledOnValidThread());
if(!thread_.get())
return 0;
{
rtc::CritScope lock(&lock_);
stop_ = true;
}
wake_up_->Set();
thread_->Stop();
thread_.reset();
stop_ = false;
return 0;
}
int32_t ProcessThreadImpl::RegisterModule(Module* module)
{
CriticalSectionScoped lock(_critSectModules);
void ProcessThreadImpl::WakeUp(Module* module) {
// Allowed to be called on any thread.
{
rtc::CritScope lock(&lock_);
ModuleCallback cb(module);
const auto& found = std::find(modules_.begin(), modules_.end(), cb);
DCHECK(found != modules_.end()) << "programmer error?";
(*found).next_callback = 0;
}
wake_up_->Set();
}
int32_t ProcessThreadImpl::RegisterModule(Module* module) {
// Allowed to be called on any thread.
{
rtc::CritScope lock(&lock_);
// Only allow module to be registered once.
for (ModuleList::iterator iter = _modules.begin();
iter != _modules.end(); ++iter) {
if(module == *iter)
{
return -1;
}
}
ModuleCallback cb(module);
if (std::find(modules_.begin(), modules_.end(), cb) != modules_.end())
return -1;
modules_.push_front(cb);
}
_modules.push_front(module);
// Wake the thread calling ProcessThreadImpl::Process() to update the
// waiting time. The waiting time for the just registered module may be
// shorter than all other registered modules.
wake_up_->Set();
// Wake the thread calling ProcessThreadImpl::Process() to update the
// waiting time. The waiting time for the just registered module may be
// shorter than all other registered modules.
_timeEvent.Set();
return 0;
return 0;
}
int32_t ProcessThreadImpl::DeRegisterModule(const Module* module)
{
CriticalSectionScoped lock(_critSectModules);
for (ModuleList::iterator iter = _modules.begin();
iter != _modules.end(); ++iter) {
if(module == *iter)
{
_modules.erase(iter);
return 0;
}
}
return -1;
int32_t ProcessThreadImpl::DeRegisterModule(const Module* module) {
// Allowed to be called on any thread.
rtc::CritScope lock(&lock_);
modules_.remove_if([&module](const ModuleCallback& m) {
return m.module == module;
});
return 0;
}
bool ProcessThreadImpl::Run(void* obj)
{
return static_cast<ProcessThreadImpl*>(obj)->Process();
// static
bool ProcessThreadImpl::Run(void* obj) {
return static_cast<ProcessThreadImpl*>(obj)->Process();
}
bool ProcessThreadImpl::Process()
{
// Wait for the module that should be called next, but don't block thread
// longer than 100 ms.
int64_t minTimeToNext = 100;
{
CriticalSectionScoped lock(_critSectModules);
for (ModuleList::iterator iter = _modules.begin();
iter != _modules.end(); ++iter) {
int64_t timeToNext = (*iter)->TimeUntilNextProcess();
if(minTimeToNext > timeToNext)
{
minTimeToNext = timeToNext;
}
}
}
bool ProcessThreadImpl::Process() {
int64_t now = TickTime::MillisecondTimestamp();
int64_t next_checkpoint = now + (1000 * 60);
{
rtc::CritScope lock(&lock_);
if (stop_)
return false;
for (auto& m : modules_) {
// TODO(tommi): Would be good to measure the time TimeUntilNextProcess
// takes and dcheck if it takes too long (e.g. >=10ms). Ideally this
// operation should not require taking a lock, so querying all modules
// should run in a matter of nanoseconds.
if (m.next_callback == 0)
m.next_callback = GetNextCallbackTime(m.module, now);
if(minTimeToNext > 0)
{
if(kEventError ==
_timeEvent.Wait(static_cast<unsigned long>(minTimeToNext)))
{
return true;
}
CriticalSectionScoped lock(_critSectModules);
if(!_thread)
{
return false;
}
if (m.next_callback <= now) {
m.module->Process();
// Use a new 'now' reference to calculate when the next callback
// should occur. We'll continue to use 'now' above for the baseline
// of calculating how long we should wait, to reduce variance.
auto new_now = TickTime::MillisecondTimestamp();
m.next_callback = GetNextCallbackTime(m.module, new_now);
}
if (m.next_callback < next_checkpoint)
next_checkpoint = m.next_callback;
}
{
CriticalSectionScoped lock(_critSectModules);
for (ModuleList::iterator iter = _modules.begin();
iter != _modules.end(); ++iter) {
int64_t timeToNext = (*iter)->TimeUntilNextProcess();
if(timeToNext < 1)
{
(*iter)->Process();
}
}
}
return true;
}
auto time_to_wait = next_checkpoint - TickTime::MillisecondTimestamp();
if (time_to_wait > 0)
wake_up_->Wait(static_cast<unsigned long>(time_to_wait));
return true;
}
} // namespace webrtc