Convert mac/ios TaskQueue implementation to pimpl convention.

BUG=webrtc:8160, webrtc:8166

Review-Url: https://codereview.webrtc.org/3013433002
Cr-Commit-Position: refs/heads/master@{#19743}
This commit is contained in:
nisse 2017-09-08 05:12:51 -07:00 committed by Commit Bot
parent a64685325c
commit 66d07c54fe
2 changed files with 135 additions and 84 deletions

View File

@ -16,10 +16,6 @@
#include <queue>
#include <type_traits>
#if defined(WEBRTC_MAC)
#include <dispatch/dispatch.h>
#endif
#include "webrtc/rtc_base/constructormagic.h"
#include "webrtc/rtc_base/criticalsection.h"
#include "webrtc/rtc_base/scoped_ref_ptr.h"
@ -236,16 +232,8 @@ class RTC_LOCKABLE TaskQueue {
}
private:
#if defined(WEBRTC_MAC)
struct QueueContext;
struct TaskContext;
struct PostTaskAndReplyContext;
dispatch_queue_t queue_;
QueueContext* const context_;
#else
class Impl;
const scoped_refptr<Impl> impl_;
#endif
RTC_DISALLOW_COPY_AND_ASSIGN(TaskQueue);
};

View File

@ -16,8 +16,12 @@
#include <string.h>
#include <dispatch/dispatch.h>
#include "webrtc/rtc_base/checks.h"
#include "webrtc/rtc_base/logging.h"
#include "webrtc/rtc_base/refcount.h"
#include "webrtc/rtc_base/refcountedobject.h"
#include "webrtc/rtc_base/task_queue_posix.h"
namespace rtc {
@ -40,79 +44,103 @@ int TaskQueuePriorityToGCD(Priority priority) {
using internal::GetQueuePtrTls;
using internal::AutoSetCurrentQueuePtr;
struct TaskQueue::QueueContext {
explicit QueueContext(TaskQueue* q) : queue(q), is_active(true) {}
class TaskQueue::Impl : public RefCountInterface {
public:
Impl(const char* queue_name, TaskQueue* task_queue, Priority priority);
~Impl() override;
static void SetNotActive(void* context) {
QueueContext* qc = static_cast<QueueContext*>(context);
qc->is_active = false;
}
static TaskQueue* Current();
static void DeleteContext(void* context) {
QueueContext* qc = static_cast<QueueContext*>(context);
delete qc;
}
// Used for DCHECKing the current queue.
bool IsCurrent() const;
TaskQueue* const queue;
bool is_active;
};
void PostTask(std::unique_ptr<QueuedTask> task);
void PostTaskAndReply(std::unique_ptr<QueuedTask> task,
std::unique_ptr<QueuedTask> reply,
TaskQueue::Impl* reply_queue);
struct TaskQueue::TaskContext {
TaskContext(QueueContext* queue_ctx, std::unique_ptr<QueuedTask> task)
: queue_ctx(queue_ctx), task(std::move(task)) {}
virtual ~TaskContext() {}
void PostDelayedTask(std::unique_ptr<QueuedTask> task, uint32_t milliseconds);
static void RunTask(void* context) {
std::unique_ptr<TaskContext> tc(static_cast<TaskContext*>(context));
if (tc->queue_ctx->is_active) {
AutoSetCurrentQueuePtr set_current(tc->queue_ctx->queue);
if (!tc->task->Run())
tc->task.release();
private:
struct QueueContext {
explicit QueueContext(TaskQueue* q) : queue(q), is_active(true) {}
static void SetNotActive(void* context) {
QueueContext* qc = static_cast<QueueContext*>(context);
qc->is_active = false;
}
}
QueueContext* const queue_ctx;
std::unique_ptr<QueuedTask> task;
};
// Special case context for holding two tasks, a |first_task| + the task
// that's owned by the parent struct, TaskContext, that then becomes the
// second (i.e. 'reply') task.
struct TaskQueue::PostTaskAndReplyContext : public TaskQueue::TaskContext {
explicit PostTaskAndReplyContext(QueueContext* first_queue_ctx,
std::unique_ptr<QueuedTask> first_task,
QueueContext* second_queue_ctx,
std::unique_ptr<QueuedTask> second_task)
: TaskContext(second_queue_ctx, std::move(second_task)),
first_queue_ctx(first_queue_ctx),
first_task(std::move(first_task)),
reply_queue_(second_queue_ctx->queue->queue_) {
// Retain the reply queue for as long as this object lives.
// If we don't, we may have memory leaks and/or failures.
dispatch_retain(reply_queue_);
}
~PostTaskAndReplyContext() override { dispatch_release(reply_queue_); }
static void RunTask(void* context) {
auto* rc = static_cast<PostTaskAndReplyContext*>(context);
if (rc->first_queue_ctx->is_active) {
AutoSetCurrentQueuePtr set_current(rc->first_queue_ctx->queue);
if (!rc->first_task->Run())
rc->first_task.release();
static void DeleteContext(void* context) {
QueueContext* qc = static_cast<QueueContext*>(context);
delete qc;
}
// Post the reply task. This hands the work over to the parent struct.
// This task will eventually delete |this|.
dispatch_async_f(rc->reply_queue_, rc, &TaskContext::RunTask);
}
QueueContext* const first_queue_ctx;
std::unique_ptr<QueuedTask> first_task;
dispatch_queue_t reply_queue_;
TaskQueue* const queue;
bool is_active;
};
struct TaskContext {
TaskContext(QueueContext* queue_ctx, std::unique_ptr<QueuedTask> task)
: queue_ctx(queue_ctx), task(std::move(task)) {}
virtual ~TaskContext() {}
static void RunTask(void* context) {
std::unique_ptr<TaskContext> tc(static_cast<TaskContext*>(context));
if (tc->queue_ctx->is_active) {
AutoSetCurrentQueuePtr set_current(tc->queue_ctx->queue);
if (!tc->task->Run())
tc->task.release();
}
}
QueueContext* const queue_ctx;
std::unique_ptr<QueuedTask> task;
};
// Special case context for holding two tasks, a |first_task| + the task
// that's owned by the parent struct, TaskContext, that then becomes the
// second (i.e. 'reply') task.
struct PostTaskAndReplyContext : public TaskContext {
explicit PostTaskAndReplyContext(QueueContext* first_queue_ctx,
std::unique_ptr<QueuedTask> first_task,
QueueContext* second_queue_ctx,
std::unique_ptr<QueuedTask> second_task)
: TaskContext(second_queue_ctx, std::move(second_task)),
first_queue_ctx(first_queue_ctx),
first_task(std::move(first_task)),
reply_queue_(second_queue_ctx->queue->impl_->queue_) {
// Retain the reply queue for as long as this object lives.
// If we don't, we may have memory leaks and/or failures.
dispatch_retain(reply_queue_);
}
~PostTaskAndReplyContext() override { dispatch_release(reply_queue_); }
static void RunTask(void* context) {
auto* rc = static_cast<PostTaskAndReplyContext*>(context);
if (rc->first_queue_ctx->is_active) {
AutoSetCurrentQueuePtr set_current(rc->first_queue_ctx->queue);
if (!rc->first_task->Run())
rc->first_task.release();
}
// Post the reply task. This hands the work over to the parent struct.
// This task will eventually delete |this|.
dispatch_async_f(rc->reply_queue_, rc, &TaskContext::RunTask);
}
QueueContext* const first_queue_ctx;
std::unique_ptr<QueuedTask> first_task;
dispatch_queue_t reply_queue_;
};
dispatch_queue_t queue_;
QueueContext* const context_;
};
TaskQueue::TaskQueue(const char* queue_name, Priority priority /*= NORMAL*/)
TaskQueue::Impl::Impl(const char* queue_name,
TaskQueue* task_queue,
Priority priority)
: queue_(dispatch_queue_create(queue_name, DISPATCH_QUEUE_SERIAL)),
context_(new QueueContext(this)) {
context_(new QueueContext(task_queue)) {
RTC_DCHECK(queue_name);
RTC_CHECK(queue_);
dispatch_set_context(queue_, context_);
@ -125,7 +153,7 @@ TaskQueue::TaskQueue(const char* queue_name, Priority priority /*= NORMAL*/)
queue_, dispatch_get_global_queue(TaskQueuePriorityToGCD(priority), 0));
}
TaskQueue::~TaskQueue() {
TaskQueue::Impl::~Impl() {
RTC_DCHECK(!IsCurrent());
// Implementation/behavioral note:
// Dispatch queues are reference counted via calls to dispatch_retain and
@ -141,39 +169,74 @@ TaskQueue::~TaskQueue() {
}
// static
TaskQueue* TaskQueue::Current() {
TaskQueue* TaskQueue::Impl::Current() {
return static_cast<TaskQueue*>(pthread_getspecific(GetQueuePtrTls()));
}
bool TaskQueue::IsCurrent() const {
bool TaskQueue::Impl::IsCurrent() const {
RTC_DCHECK(queue_);
return this == Current();
const TaskQueue* current = Current();
return current && this == current->impl_.get();
}
void TaskQueue::PostTask(std::unique_ptr<QueuedTask> task) {
void TaskQueue::Impl::PostTask(std::unique_ptr<QueuedTask> task) {
auto* context = new TaskContext(context_, std::move(task));
dispatch_async_f(queue_, context, &TaskContext::RunTask);
}
void TaskQueue::PostDelayedTask(std::unique_ptr<QueuedTask> task,
uint32_t milliseconds) {
void TaskQueue::Impl::PostDelayedTask(std::unique_ptr<QueuedTask> task,
uint32_t milliseconds) {
auto* context = new TaskContext(context_, std::move(task));
dispatch_after_f(
dispatch_time(DISPATCH_TIME_NOW, milliseconds * NSEC_PER_MSEC), queue_,
context, &TaskContext::RunTask);
}
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) {
auto* context = new PostTaskAndReplyContext(
context_, std::move(task), reply_queue->context_, std::move(reply));
dispatch_async_f(queue_, context, &PostTaskAndReplyContext::RunTask);
}
// 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::Current();
}
// 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 PostTaskAndReply(std::move(task), std::move(reply), Current());
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