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

Add support for multimedia timers to TaskQueue on Windows.

Browse Source 
Multimedia timers are higher precision than WM_TIMER, but they're also
a limited resource and more costly. So this implementation is a best
effort implementation that falls back on WM_TIMER when multimedia
timers aren't available.

A possible future change could be to make high precision timers in a
TaskQueue, optional. The reason for doing so would be for TaskQueues
that don't need high precision timers, won't eat up timers from TQ
instances that really need it.

BUG=webrtc:7151

Review-Url: https://codereview.webrtc.org/2691973002
Cr-Commit-Position: refs/heads/master@{#16661}
This commit is contained in:
tommi
2017-02-17 02:47:11 -08:00
committed by Commit bot
parent fc5d22c86f
commit f9d9154808
3 changed files with 202 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
webrtc/base/task_queue.h
View File

@ -178,6 +178,11 @@ class LOCKABLE TaskQueue {
void PostTaskAndReply(std::unique_ptr<QueuedTask> task,
std::unique_ptr<QueuedTask> reply);
// Schedules a task to execute a specified number of milliseconds from when
// the call is made. The precision should be considered as "best effort"
// and in some cases, such as on Windows when all high precision timers have
// been used up, can be off by as much as 15 millseconds (although 8 would be
// more likely). This can be mitigated by limiting the use of delayed tasks.
void PostDelayedTask(std::unique_ptr<QueuedTask> task, uint32_t milliseconds);
template <class Closure>
@ -185,6 +190,7 @@ class LOCKABLE TaskQueue {
PostTask(std::unique_ptr<QueuedTask>(new ClosureTask<Closure>(closure)));
}
// See documentation above for performance expectations.
template <class Closure>
void PostDelayedTask(const Closure& closure, uint32_t milliseconds) {
PostDelayedTask(
@ -254,10 +260,12 @@ class LOCKABLE TaskQueue {
dispatch_queue_t queue_;
QueueContext* const context_;
#elif defined(WEBRTC_WIN)
class MultimediaTimer;
typedef std::unordered_map<UINT_PTR, std::unique_ptr<QueuedTask>>
DelayedTasks;
static bool ThreadMain(void* context);
static bool ProcessQueuedMessages(DelayedTasks* delayed_tasks);
static bool ProcessQueuedMessages(DelayedTasks* delayed_tasks,
std::vector<MultimediaTimer>* timers);
class WorkerThread : public PlatformThread {
public:

4
webrtc/base/task_queue_unittest.cc
View File

@ -112,14 +112,14 @@ TEST(TaskQueueTest, PostMultipleDelayed) {
TaskQueue queue(kQueueName);
std::vector<std::unique_ptr<Event>> events;
for (int i = 0; i < 10; ++i) {
for (int i = 0; i < 100; ++i) {
events.push_back(std::unique_ptr<Event>(new Event(false, false)));
queue.PostDelayedTask(
Bind(&CheckCurrent, kQueueName, events.back().get(), &queue), 10);
}
for (const auto& e : events)
EXPECT_TRUE(e->Wait(100));
EXPECT_TRUE(e->Wait(1000));
}
TEST(TaskQueueTest, PostDelayedAfterDestruct) {

213
webrtc/base/task_queue_win.cc
View File

@ -10,8 +10,12 @@
#include "webrtc/base/task_queue.h"
#include <mmsystem.h>
#include <string.h>
#include <algorithm>
#include "webrtc/base/arraysize.h"
#include "webrtc/base/checks.h"
#include "webrtc/base/logging.h"
@ -29,7 +33,7 @@ BOOL CALLBACK InitializeTls(PINIT_ONCE init_once, void* param, void** context) {
DWORD GetQueuePtrTls() {
static INIT_ONCE init_once = INIT_ONCE_STATIC_INIT;
InitOnceExecuteOnce(&init_once, InitializeTls, nullptr, nullptr);
::InitOnceExecuteOnce(&init_once, InitializeTls, nullptr, nullptr);
return g_queue_ptr_tls;
}
@ -40,13 +44,107 @@ struct ThreadStartupData {
void CALLBACK InitializeQueueThread(ULONG_PTR param) {
MSG msg;
PeekMessage(&msg, NULL, WM_USER, WM_USER, PM_NOREMOVE);
::PeekMessage(&msg, nullptr, WM_USER, WM_USER, PM_NOREMOVE);
ThreadStartupData* data = reinterpret_cast<ThreadStartupData*>(param);
TlsSetValue(GetQueuePtrTls(), data->thread_context);
::TlsSetValue(GetQueuePtrTls(), data->thread_context);
data->started->Set();
}
} // namespace
class TaskQueue::MultimediaTimer {
public:
// kMaxTimers defines the limit of how many MultimediaTimer instances should
// be created.
// Background: The maximum number of supported handles for Wait functions, is
// MAXIMUM_WAIT_OBJECTS - 1 (63).
// There are some ways to work around the limitation but as it turns out, the
// limit of concurrently active multimedia timers per process, is much lower,
// or 16. So there isn't much value in going to the lenghts required to
// overcome the Wait limitations.
// kMaxTimers is larger than 16 though since it is possible that 'complete' or
// signaled timers that haven't been handled, are counted as part of
// kMaxTimers and thus a multimedia timer can actually be queued even though
// as far as we're concerned, there are more than 16 that are pending.
static const int kMaxTimers = MAXIMUM_WAIT_OBJECTS - 1;
// Controls how many MultimediaTimer instances a queue can hold before
// attempting to garbage collect (GC) timers that aren't in use.
static const int kInstanceThresholdGC = 8;
MultimediaTimer() : event_(::CreateEvent(nullptr, false, false, nullptr)) {}
MultimediaTimer(MultimediaTimer&& timer)
: event_(timer.event_),
timer_id_(timer.timer_id_),
task_(std::move(timer.task_)) {
RTC_DCHECK(event_);
timer.event_ = nullptr;
timer.timer_id_ = 0;
}
~MultimediaTimer() { Close(); }
// Implementing this operator is required because of the way
// some stl algorithms work, such as std::rotate().
MultimediaTimer& operator=(MultimediaTimer&& timer) {
if (this != &timer) {
Close();
event_ = timer.event_;
timer.event_ = nullptr;
task_ = std::move(timer.task_);
timer_id_ = timer.timer_id_;
timer.timer_id_ = 0;
}
return *this;
}
bool StartOneShotTimer(std::unique_ptr<QueuedTask> task, UINT delay_ms) {
RTC_DCHECK_EQ(0, timer_id_);
RTC_DCHECK(event_ != nullptr);
RTC_DCHECK(!task_.get());
RTC_DCHECK(task.get());
task_ = std::move(task);
timer_id_ =
::timeSetEvent(delay_ms, 0, reinterpret_cast<LPTIMECALLBACK>(event_), 0,
TIME_ONESHOT | TIME_CALLBACK_EVENT_SET);
return timer_id_ != 0;
}
std::unique_ptr<QueuedTask> Cancel() {
if (timer_id_) {
::timeKillEvent(timer_id_);
timer_id_ = 0;
}
return std::move(task_);
}
void OnEventSignaled() {
RTC_DCHECK_NE(0, timer_id_);
timer_id_ = 0;
task_->Run() ? task_.reset() : static_cast<void>(task_.release());
}
HANDLE event() const { return event_; }
bool is_active() const { return timer_id_ != 0; }
private:
void Close() {
Cancel();
if (event_) {
::CloseHandle(event_);
event_ = nullptr;
}
}
HANDLE event_ = nullptr;
MMRESULT timer_id_ = 0;
std::unique_ptr<QueuedTask> task_;
RTC_DISALLOW_COPY_AND_ASSIGN(MultimediaTimer);
};
TaskQueue::TaskQueue(const char* queue_name)
: thread_(&TaskQueue::ThreadMain, this, queue_name) {
RTC_DCHECK(queue_name);
@ -60,7 +158,7 @@ TaskQueue::TaskQueue(const char* queue_name)
TaskQueue::~TaskQueue() {
RTC_DCHECK(!IsCurrent());
while (!PostThreadMessage(thread_.GetThreadRef(), WM_QUIT, 0, 0)) {
while (!::PostThreadMessage(thread_.GetThreadRef(), WM_QUIT, 0, 0)) {
RTC_CHECK_EQ(ERROR_NOT_ENOUGH_QUOTA, ::GetLastError());
Sleep(1);
}
@ -69,7 +167,7 @@ TaskQueue::~TaskQueue() {
// static
TaskQueue* TaskQueue::Current() {
return static_cast<TaskQueue*>(TlsGetValue(GetQueuePtrTls()));
return static_cast<TaskQueue*>(::TlsGetValue(GetQueuePtrTls()));
}
// static
@ -83,8 +181,8 @@ bool TaskQueue::IsCurrent() const {
}
void TaskQueue::PostTask(std::unique_ptr<QueuedTask> task) {
if (PostThreadMessage(thread_.GetThreadRef(), WM_RUN_TASK, 0,
reinterpret_cast<LPARAM>(task.get()))) {
if (::PostThreadMessage(thread_.GetThreadRef(), WM_RUN_TASK, 0,
reinterpret_cast<LPARAM>(task.get()))) {
task.release();
}
}
@ -100,8 +198,8 @@ void TaskQueue::PostDelayedTask(std::unique_ptr<QueuedTask> task,
#else
wparam = milliseconds;
#endif
if (PostThreadMessage(thread_.GetThreadRef(), WM_QUEUE_DELAYED_TASK, wparam,
reinterpret_cast<LPARAM>(task.get()))) {
if (::PostThreadMessage(thread_.GetThreadRef(), WM_QUEUE_DELAYED_TASK, wparam,
reinterpret_cast<LPARAM>(task.get()))) {
task.release();
}
}
@ -117,8 +215,8 @@ void TaskQueue::PostTaskAndReply(std::unique_ptr<QueuedTask> task,
delete task_ptr;
// If the thread's message queue is full, we can't queue the task and will
// have to drop it (i.e. delete).
if (!PostThreadMessage(reply_thread_id, WM_RUN_TASK, 0,
reinterpret_cast<LPARAM>(reply_task_ptr))) {
if (!::PostThreadMessage(reply_thread_id, WM_RUN_TASK, 0,
reinterpret_cast<LPARAM>(reply_task_ptr))) {
delete reply_task_ptr;
}
});
@ -131,25 +229,69 @@ void TaskQueue::PostTaskAndReply(std::unique_ptr<QueuedTask> task,
// static
bool TaskQueue::ThreadMain(void* context) {
HANDLE timer_handles[MultimediaTimer::kMaxTimers];
// Active multimedia timers.
std::vector<MultimediaTimer> mm_timers;
// Tasks that have been queued by using SetTimer/WM_TIMER.
DelayedTasks delayed_tasks;
while (true) {
DWORD result = ::MsgWaitForMultipleObjectsEx(0, nullptr, INFINITE,
RTC_DCHECK(mm_timers.size() <= arraysize(timer_handles));
DWORD count = 0;
for (const auto& t : mm_timers) {
if (!t.is_active())
break;
timer_handles[count++] = t.event();
}
// Make sure we do an alertable wait as that's required to allow APCs to run
// (e.g. required for InitializeQueueThread and stopping the thread in
// PlatformThread).
DWORD result = ::MsgWaitForMultipleObjectsEx(count, timer_handles, INFINITE,
QS_ALLEVENTS, MWMO_ALERTABLE);
RTC_CHECK_NE(WAIT_FAILED, result);
if (result == WAIT_OBJECT_0) {
if (!ProcessQueuedMessages(&delayed_tasks))
// If we're not waiting for any timers, then count will be equal to
// WAIT_OBJECT_0. If we're waiting for timers, then |count| represents
// "One more than the number of timers", which means that there's a
// message in the queue that needs to be handled.
// If |result| is less than |count|, then its value will be the index of the
// timer that has been signaled.
if (result == (WAIT_OBJECT_0 + count)) {
if (!ProcessQueuedMessages(&delayed_tasks, &mm_timers))
break;
} else if (result < (WAIT_OBJECT_0 + count)) {
mm_timers[result].OnEventSignaled();
RTC_DCHECK(!mm_timers[result].is_active());
// Reuse timer events by moving inactive timers to the back of the vector.
// When new delayed tasks are queued, they'll get reused.
if (mm_timers.size() > 1) {
auto it = mm_timers.begin() + result;
std::rotate(it, it + 1, mm_timers.end());
}
// Collect some garbage.
if (mm_timers.size() > MultimediaTimer::kInstanceThresholdGC) {
const auto inactive = std::find_if(
mm_timers.begin(), mm_timers.end(),
[](const MultimediaTimer& t) { return !t.is_active(); });
if (inactive != mm_timers.end()) {
// Since inactive timers are always moved to the back, we can
// safely delete all timers following the first inactive one.
mm_timers.erase(inactive, mm_timers.end());
}
}
} else {
RTC_DCHECK_EQ(WAIT_IO_COMPLETION, result);
}
}
return false;
}
// static
bool TaskQueue::ProcessQueuedMessages(DelayedTasks* delayed_tasks) {
bool TaskQueue::ProcessQueuedMessages(DelayedTasks* delayed_tasks,
std::vector<MultimediaTimer>* timers) {
MSG msg = {};
while (PeekMessage(&msg, nullptr, 0, 0, PM_REMOVE) &&
while (::PeekMessage(&msg, nullptr, 0, 0, PM_REMOVE) &&
msg.message != WM_QUIT) {
if (!msg.hwnd) {
switch (msg.message) {
@ -160,7 +302,8 @@ bool TaskQueue::ProcessQueuedMessages(DelayedTasks* delayed_tasks) {
break;
}
case WM_QUEUE_DELAYED_TASK: {
QueuedTask* task = reinterpret_cast<QueuedTask*>(msg.lParam);
std::unique_ptr<QueuedTask> task(
reinterpret_cast<QueuedTask*>(msg.lParam));
uint32_t milliseconds = msg.wParam & 0xFFFFFFFF;
#if defined(_WIN64)
// Subtract the time it took to queue the timer.
@ -169,12 +312,38 @@ bool TaskQueue::ProcessQueuedMessages(DelayedTasks* delayed_tasks) {
milliseconds =
post_time > milliseconds ? 0 : milliseconds - post_time;
#endif
UINT_PTR timer_id = SetTimer(nullptr, 0, milliseconds, nullptr);
delayed_tasks->insert(std::make_pair(timer_id, task));
bool timer_queued = false;
if (timers->size() < MultimediaTimer::kMaxTimers) {
MultimediaTimer* timer = nullptr;
auto available = std::find_if(
timers->begin(), timers->end(),
[](const MultimediaTimer& t) { return !t.is_active(); });
if (available != timers->end()) {
timer = &(*available);
} else {
timers->emplace_back();
timer = &timers->back();
}
timer_queued =
timer->StartOneShotTimer(std::move(task), milliseconds);
if (!timer_queued) {
// No more multimedia timers can be queued.
// Detach the task and fall back on SetTimer.
task = timer->Cancel();
}
}
// When we fail to use multimedia timers, we fall back on the more
// coarse SetTimer/WM_TIMER approach.
if (!timer_queued) {
UINT_PTR timer_id = ::SetTimer(nullptr, 0, milliseconds, nullptr);
delayed_tasks->insert(std::make_pair(timer_id, task.release()));
}
break;
}
case WM_TIMER: {
KillTimer(nullptr, msg.wParam);
::KillTimer(nullptr, msg.wParam);
auto found = delayed_tasks->find(msg.wParam);
RTC_DCHECK(found != delayed_tasks->end());
if (!found->second->Run())
@ -187,8 +356,8 @@ bool TaskQueue::ProcessQueuedMessages(DelayedTasks* delayed_tasks) {
break;
}
} else {
TranslateMessage(&msg);
DispatchMessage(&msg);
::TranslateMessage(&msg);
::DispatchMessage(&msg);
}
}
return msg.message != WM_QUIT;