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

Convert windows TaskQueue implementation to pimpl convention.

Browse Source 
BUG=webrtc:8160,webrtc:8166

Review-Url: https://codereview.webrtc.org/3009133002
Cr-Commit-Position: refs/heads/master@{#19709}
This commit is contained in:
nisse
2017-09-06 04:38:22 -07:00
committed by Commit Bot
parent 5301e3c3db
commit 341c8e40b7
2 changed files with 146 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

195
webrtc/rtc_base/task_queue_win.cc
View File

@ -18,7 +18,11 @@
#include "webrtc/rtc_base/arraysize.h"
#include "webrtc/rtc_base/checks.h"
#include "webrtc/rtc_base/event.h"
#include "webrtc/rtc_base/logging.h"
#include "webrtc/rtc_base/platform_thread.h"
#include "webrtc/rtc_base/refcount.h"
#include "webrtc/rtc_base/refcountedobject.h"
#include "webrtc/rtc_base/safe_conversions.h"
#include "webrtc/rtc_base/timeutils.h"
@ -154,44 +158,97 @@ class MultimediaTimer {
} // namespace
class TaskQueue::ThreadState {
class TaskQueue::Impl : public RefCountInterface {
public:
explicit ThreadState(HANDLE in_queue) : in_queue_(in_queue) {}
~ThreadState() {}
Impl(const char* queue_name, TaskQueue* queue, Priority priority);
~Impl() override;
void RunThreadMain();
static TaskQueue::Impl* Current();
static TaskQueue* CurrentQueue();
// Used for DCHECKing the current queue.
bool IsCurrent() const;
template <class Closure,
typename std::enable_if<
std::is_copy_constructible<Closure>::value>::type* = nullptr>
void PostTask(const Closure& closure) {
PostTask(std::unique_ptr<QueuedTask>(new ClosureTask<Closure>(closure)));
}
void PostTask(std::unique_ptr<QueuedTask> task);
void PostTaskAndReply(std::unique_ptr<QueuedTask> task,
std::unique_ptr<QueuedTask> reply,
TaskQueue::Impl* reply_queue);
void PostDelayedTask(std::unique_ptr<QueuedTask> task, uint32_t milliseconds);
void RunPendingTasks();
private:
bool ProcessQueuedMessages();
void RunDueTasks();
void ScheduleNextTimer();
void CancelTimers();
static void ThreadMain(void* context);
// Since priority_queue<> by defult orders items in terms of
// largest->smallest, using std::less<>, and we want smallest->largest,
// we would like to use std::greater<> here. Alas it's only available in
// C++14 and later, so we roll our own compare template that that relies on
// operator<().
template <typename T>
struct greater {
bool operator()(const T& l, const T& r) { return l > r; }
class WorkerThread : public PlatformThread {
public:
WorkerThread(ThreadRunFunction func,
void* obj,
const char* thread_name,
ThreadPriority priority)
: PlatformThread(func, obj, thread_name, priority) {}
bool QueueAPC(PAPCFUNC apc_function, ULONG_PTR data) {
return PlatformThread::QueueAPC(apc_function, data);
}
};
MultimediaTimer timer_;
std::priority_queue<DelayedTaskInfo,
std::vector<DelayedTaskInfo>,
greater<DelayedTaskInfo>>
timer_tasks_;
UINT_PTR timer_id_ = 0;
class ThreadState {
public:
explicit ThreadState(HANDLE in_queue) : in_queue_(in_queue) {}
~ThreadState() {}
void RunThreadMain();
private:
bool ProcessQueuedMessages();
void RunDueTasks();
void ScheduleNextTimer();
void CancelTimers();
// Since priority_queue<> by defult orders items in terms of
// largest->smallest, using std::less<>, and we want smallest->largest,
// we would like to use std::greater<> here. Alas it's only available in
// C++14 and later, so we roll our own compare template that that relies on
// operator<().
template <typename T>
struct greater {
bool operator()(const T& l, const T& r) { return l > r; }
};
MultimediaTimer timer_;
std::priority_queue<DelayedTaskInfo,
std::vector<DelayedTaskInfo>,
greater<DelayedTaskInfo>>
timer_tasks_;
UINT_PTR timer_id_ = 0;
HANDLE in_queue_;
};
TaskQueue* const queue_;
WorkerThread thread_;
rtc::CriticalSection pending_lock_;
std::queue<std::unique_ptr<QueuedTask>> pending_ GUARDED_BY(pending_lock_);
HANDLE in_queue_;
};
TaskQueue::TaskQueue(const char* queue_name, Priority priority /*= NORMAL*/)
: thread_(&TaskQueue::ThreadMain,
TaskQueue::Impl::Impl(const char* queue_name,
TaskQueue* queue,
Priority priority)
: queue_(queue),
thread_(&TaskQueue::Impl::ThreadMain,
this,
queue_name,
TaskQueuePriorityToThreadPriority(priority)),
in_queue_(::CreateEvent(nullptr, true, false, nullptr)) {
in_queue_(::CreateEvent(nullptr, true, false, nullptr)) {
RTC_DCHECK(queue_name);
RTC_DCHECK(in_queue_);
thread_.Start();
@ -202,7 +259,7 @@ TaskQueue::TaskQueue(const char* queue_name, Priority priority /*= NORMAL*/)
event.Wait(Event::kForever);
}
TaskQueue::~TaskQueue() {
TaskQueue::Impl::~Impl() {
RTC_DCHECK(!IsCurrent());
while (!::PostThreadMessage(thread_.GetThreadRef(), WM_QUIT, 0, 0)) {
RTC_CHECK_EQ(ERROR_NOT_ENOUGH_QUOTA, ::GetLastError());
@ -213,22 +270,28 @@ TaskQueue::~TaskQueue() {
}
// static
TaskQueue* TaskQueue::Current() {
return static_cast<TaskQueue*>(::TlsGetValue(GetQueuePtrTls()));
TaskQueue::Impl* TaskQueue::Impl::Current() {
return static_cast<TaskQueue::Impl*>(::TlsGetValue(GetQueuePtrTls()));
}
bool TaskQueue::IsCurrent() const {
// static
TaskQueue* TaskQueue::Impl::CurrentQueue() {
TaskQueue::Impl* current = Current();
return current ? current->queue_ : nullptr;
}
bool TaskQueue::Impl::IsCurrent() const {
return IsThreadRefEqual(thread_.GetThreadRef(), CurrentThreadRef());
}
void TaskQueue::PostTask(std::unique_ptr<QueuedTask> task) {
void TaskQueue::Impl::PostTask(std::unique_ptr<QueuedTask> task) {
rtc::CritScope lock(&pending_lock_);
pending_.push(std::move(task));
::SetEvent(in_queue_);
}
void TaskQueue::PostDelayedTask(std::unique_ptr<QueuedTask> task,
uint32_t milliseconds) {
void TaskQueue::Impl::PostDelayedTask(std::unique_ptr<QueuedTask> task,
uint32_t milliseconds) {
if (!milliseconds) {
PostTask(std::move(task));
return;
@ -245,9 +308,9 @@ void TaskQueue::PostDelayedTask(std::unique_ptr<QueuedTask> task,
}
}
void TaskQueue::PostTaskAndReply(std::unique_ptr<QueuedTask> task,
std::unique_ptr<QueuedTask> reply,
TaskQueue* reply_queue) {
void TaskQueue::Impl::PostTaskAndReply(std::unique_ptr<QueuedTask> task,
std::unique_ptr<QueuedTask> reply,
TaskQueue::Impl* reply_queue) {
QueuedTask* task_ptr = task.release();
QueuedTask* reply_task_ptr = reply.release();
DWORD reply_thread_id = reply_queue->thread_.GetThreadRef();
@ -263,12 +326,7 @@ void TaskQueue::PostTaskAndReply(std::unique_ptr<QueuedTask> task,
});
}
void TaskQueue::PostTaskAndReply(std::unique_ptr<QueuedTask> task,
std::unique_ptr<QueuedTask> reply) {
return PostTaskAndReply(std::move(task), std::move(reply), Current());
}
void TaskQueue::RunPendingTasks() {
void TaskQueue::Impl::RunPendingTasks() {
while (true) {
std::unique_ptr<QueuedTask> task;
{
@ -285,12 +343,12 @@ void TaskQueue::RunPendingTasks() {
}
// static
void TaskQueue::ThreadMain(void* context) {
ThreadState state(static_cast<TaskQueue*>(context)->in_queue_);
void TaskQueue::Impl::ThreadMain(void* context) {
ThreadState state(static_cast<TaskQueue::Impl*>(context)->in_queue_);
state.RunThreadMain();
}
void TaskQueue::ThreadState::RunThreadMain() {
void TaskQueue::Impl::ThreadState::RunThreadMain() {
HANDLE handles[2] = { *timer_.event_for_wait(), in_queue_ };
while (true) {
// Make sure we do an alertable wait as that's required to allow APCs to run
@ -315,12 +373,12 @@ void TaskQueue::ThreadState::RunThreadMain() {
if (result == (WAIT_OBJECT_0 + 1)) {
::ResetEvent(in_queue_);
TaskQueue::Current()->RunPendingTasks();
TaskQueue::Impl::Current()->RunPendingTasks();
}
}
}
bool TaskQueue::ThreadState::ProcessQueuedMessages() {
bool TaskQueue::Impl::ThreadState::ProcessQueuedMessages() {
MSG msg = {};
// To protect against overly busy message queues, we limit the time
// we process tasks to a few milliseconds. If we don't do that, there's
@ -374,7 +432,7 @@ bool TaskQueue::ThreadState::ProcessQueuedMessages() {
return msg.message != WM_QUIT;
}
void TaskQueue::ThreadState::RunDueTasks() {
void TaskQueue::Impl::ThreadState::RunDueTasks() {
RTC_DCHECK(!timer_tasks_.empty());
auto now = GetTick();
do {
@ -386,7 +444,7 @@ void TaskQueue::ThreadState::RunDueTasks() {
} while (!timer_tasks_.empty());
}
void TaskQueue::ThreadState::ScheduleNextTimer() {
void TaskQueue::Impl::ThreadState::ScheduleNextTimer() {
RTC_DCHECK_EQ(timer_id_, 0);
if (timer_tasks_.empty())
return;
@ -398,7 +456,7 @@ void TaskQueue::ThreadState::ScheduleNextTimer() {
timer_id_ = ::SetTimer(nullptr, 0, milliseconds, nullptr);
}
void TaskQueue::ThreadState::CancelTimers() {
void TaskQueue::Impl::ThreadState::CancelTimers() {
timer_.Cancel();
if (timer_id_) {
::KillTimer(nullptr, timer_id_);
@ -406,4 +464,43 @@ void TaskQueue::ThreadState::CancelTimers() {
}
}
// Boilerplate for the PIMPL pattern.
TaskQueue::TaskQueue(const char* queue_name, Priority priority)
: impl_(new RefCountedObject<TaskQueue::Impl>(queue_name, this, priority)) {
}
TaskQueue::~TaskQueue() {}
// static
TaskQueue* TaskQueue::Current() {
return TaskQueue::Impl::CurrentQueue();
}
// Used for DCHECKing the current queue.
bool TaskQueue::IsCurrent() const {
return impl_->IsCurrent();
}
void TaskQueue::PostTask(std::unique_ptr<QueuedTask> task) {
return TaskQueue::impl_->PostTask(std::move(task));
}
void TaskQueue::PostTaskAndReply(std::unique_ptr<QueuedTask> task,
std::unique_ptr<QueuedTask> reply,
TaskQueue* reply_queue) {
return TaskQueue::impl_->PostTaskAndReply(std::move(task), std::move(reply),
reply_queue->impl_.get());
}
void TaskQueue::PostTaskAndReply(std::unique_ptr<QueuedTask> task,
std::unique_ptr<QueuedTask> reply) {
return TaskQueue::impl_->PostTaskAndReply(std::move(task), std::move(reply),
impl_.get());
}
void TaskQueue::PostDelayedTask(std::unique_ptr<QueuedTask> task,
uint32_t milliseconds) {
return TaskQueue::impl_->PostDelayedTask(std::move(task), milliseconds);
}
} // namespace rtc