blob: bd20d074ef0e41a8f2a0d006cd006c9d2603111e [file] [log] [blame]
// Copyright 2018 The Chromium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#include "base/fuchsia/async_dispatcher.h"
#include <lib/async/default.h>
#include <lib/async/task.h>
#include <lib/async/wait.h>
#include <lib/zx/handle.h>
#include <lib/zx/time.h>
#include <zircon/syscalls.h>
#include "base/fuchsia/fuchsia_logging.h"
#include "starboard/types.h"
namespace base {
namespace {
template <typename T>
uintptr_t key_from_ptr(T* ptr) {
return reinterpret_cast<uintptr_t>(ptr);
};
} // namespace
class AsyncDispatcher::ExceptionState : public LinkNode<ExceptionState> {
public:
explicit ExceptionState(AsyncDispatcher* async_dispatcher) {
async_dispatcher->exception_list_.Append(this);
}
~ExceptionState() { RemoveFromList(); }
async_exception_t* exception() {
// ExceptionState objects are allocated in-place in the |state| field of an
// enclosing async_exceptionwait_t, so async_exception_t address can be
// calculated by subtracting state offset in async_exception_t from |this|.
static_assert(std::is_standard_layout<async_exception_t>(),
"async_wait_t is expected to have standard layout.");
return reinterpret_cast<async_exception_t*>(
reinterpret_cast<uint8_t*>(this) - offsetof(async_exception_t, state));
}
private:
DISALLOW_COPY_AND_ASSIGN(ExceptionState);
};
class AsyncDispatcher::WaitState : public LinkNode<WaitState> {
public:
explicit WaitState(AsyncDispatcher* async_dispatcher) {
async_dispatcher->wait_list_.Append(this);
}
~WaitState() { RemoveFromList(); }
async_wait_t* wait() {
// WaitState objects are allocated in-place in the |state| field of an
// enclosing async_wait_t, so async_wait_t address can be calculated by
// subtracting state offset in async_wait_t from |this|.
static_assert(std::is_standard_layout<async_wait_t>(),
"async_wait_t is expected to have standard layout.");
return reinterpret_cast<async_wait_t*>(reinterpret_cast<uint8_t*>(this) -
offsetof(async_wait_t, state));
}
private:
DISALLOW_COPY_AND_ASSIGN(WaitState);
};
class AsyncDispatcher::TaskState : public LinkNode<TaskState> {
public:
explicit TaskState(LinkNode<TaskState>* previous_task) {
InsertAfter(previous_task);
}
~TaskState() { RemoveFromList(); }
async_task_t* task() {
// TaskState objects are allocated in-place in the |state| field of an
// enclosing async_task_t, so async_task_t address can be calculated by
// subtracting state offset in async_task_t from |this|.
static_assert(std::is_standard_layout<async_task_t>(),
"async_task_t is expected to have standard layout.");
return reinterpret_cast<async_task_t*>(reinterpret_cast<uint8_t*>(this) -
offsetof(async_task_t, state));
}
private:
DISALLOW_COPY_AND_ASSIGN(TaskState);
};
AsyncDispatcher::AsyncDispatcher() : ops_storage_({}) {
zx_status_t status = zx::port::create(0u, &port_);
ZX_DCHECK(status == ZX_OK, status);
status = zx::timer::create(0u, ZX_CLOCK_MONOTONIC, &timer_);
ZX_DCHECK(status == ZX_OK, status);
status = timer_.wait_async(port_, key_from_ptr(&timer_), ZX_TIMER_SIGNALED,
ZX_WAIT_ASYNC_REPEATING);
ZX_DCHECK(status == ZX_OK, status);
status = zx::event::create(0, &stop_event_);
ZX_DCHECK(status == ZX_OK, status);
status = stop_event_.wait_async(port_, key_from_ptr(&stop_event_),
ZX_EVENT_SIGNALED, ZX_WAIT_ASYNC_REPEATING);
ZX_DCHECK(status == ZX_OK, status);
ops_storage_.version = ASYNC_OPS_V2;
ops_storage_.v1.now = NowOp;
ops_storage_.v1.begin_wait = BeginWaitOp;
ops_storage_.v1.cancel_wait = CancelWaitOp;
ops_storage_.v1.post_task = PostTaskOp;
ops_storage_.v1.cancel_task = CancelTaskOp;
ops_storage_.v1.queue_packet = QueuePacketOp;
ops_storage_.v1.set_guest_bell_trap = SetGuestBellTrapOp;
ops_storage_.v2.bind_exception_port = BindExceptionPortOp;
ops_storage_.v2.unbind_exception_port = UnbindExceptionPortOp;
ops = &ops_storage_;
DCHECK(!async_get_default_dispatcher());
async_set_default_dispatcher(this);
}
AsyncDispatcher::~AsyncDispatcher() {
DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
DCHECK_EQ(async_get_default_dispatcher(), this);
// Some waits and tasks may be canceled while the dispatcher is being
// destroyed, so pop-from-head until none remain.
while (!exception_list_.empty()) {
ExceptionState* state = exception_list_.head()->value();
async_exception_t* exception = state->exception();
state->~ExceptionState();
exception->handler(this, exception, ZX_ERR_CANCELED, nullptr);
}
while (!wait_list_.empty()) {
WaitState* state = wait_list_.head()->value();
async_wait_t* wait = state->wait();
state->~WaitState();
wait->handler(this, wait, ZX_ERR_CANCELED, nullptr);
}
while (!task_list_.empty()) {
TaskState* state = task_list_.head()->value();
async_task_t* task = state->task();
state->~TaskState();
task->handler(this, task, ZX_ERR_CANCELED);
}
async_set_default_dispatcher(nullptr);
}
zx_status_t AsyncDispatcher::DispatchOrWaitUntil(zx_time_t deadline) {
DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
zx_port_packet_t packet = {};
zx_status_t status = port_.wait(zx::time(deadline), &packet);
if (status != ZX_OK)
return status;
if (ZX_PKT_IS_SIGNAL_ONE(packet.type) || ZX_PKT_IS_SIGNAL_REP(packet.type)) {
if (packet.key == key_from_ptr(&timer_)) {
// |timer_| has expired.
DCHECK(packet.signal.observed & ZX_TIMER_SIGNALED);
DispatchTasks();
return ZX_OK;
} else if (packet.key == key_from_ptr(&stop_event_)) {
// Stop() was called.
DCHECK(packet.signal.observed & ZX_EVENT_SIGNALED);
status = zx_object_signal(stop_event_.get(), ZX_EVENT_SIGNALED, 0);
ZX_DCHECK(status == ZX_OK, status);
return ZX_ERR_CANCELED;
} else {
DCHECK_EQ(packet.type, ZX_PKT_TYPE_SIGNAL_ONE);
auto* wait = reinterpret_cast<async_wait_t*>(packet.key);
// Clean the state before invoking the handler: it may destroy |*wait|.
auto* state = reinterpret_cast<WaitState*>(&wait->state);
state->~WaitState();
wait->handler(this, wait, packet.status, &packet.signal);
return ZX_OK;
}
} else if (ZX_PKT_IS_EXCEPTION(packet.type)) {
auto* exception = reinterpret_cast<async_exception_t*>(packet.key);
// |exception| may have been deleted by the time |handler| returns.
exception->handler(this, exception, packet.status, &packet);
return ZX_OK;
}
NOTREACHED();
return ZX_ERR_INTERNAL;
}
void AsyncDispatcher::Stop() {
// Can be called on any thread.
zx_status_t status =
zx_object_signal(stop_event_.get(), 0, ZX_EVENT_SIGNALED);
ZX_DCHECK(status == ZX_OK, status);
}
zx_time_t AsyncDispatcher::NowOp(async_dispatcher_t* async) {
DCHECK(async);
return zx_clock_get(ZX_CLOCK_MONOTONIC);
}
zx_status_t AsyncDispatcher::BeginWaitOp(async_dispatcher_t* async,
async_wait_t* wait) {
return static_cast<AsyncDispatcher*>(async)->BeginWait(wait);
}
zx_status_t AsyncDispatcher::CancelWaitOp(async_dispatcher_t* async,
async_wait_t* wait) {
return static_cast<AsyncDispatcher*>(async)->CancelWait(wait);
}
zx_status_t AsyncDispatcher::PostTaskOp(async_dispatcher_t* async,
async_task_t* task) {
return static_cast<AsyncDispatcher*>(async)->PostTask(task);
}
zx_status_t AsyncDispatcher::CancelTaskOp(async_dispatcher_t* async,
async_task_t* task) {
return static_cast<AsyncDispatcher*>(async)->CancelTask(task);
}
zx_status_t AsyncDispatcher::QueuePacketOp(async_dispatcher_t* async,
async_receiver_t* receiver,
const zx_packet_user_t* data) {
return ZX_ERR_NOT_SUPPORTED;
}
zx_status_t AsyncDispatcher::SetGuestBellTrapOp(async_dispatcher_t* async,
async_guest_bell_trap_t* trap,
zx_handle_t guest,
zx_vaddr_t addr,
size_t length) {
return ZX_ERR_NOT_SUPPORTED;
}
zx_status_t AsyncDispatcher::BindExceptionPortOp(async_dispatcher_t* async,
async_exception_t* exception) {
return static_cast<AsyncDispatcher*>(async)->BindExceptionPort(exception);
}
zx_status_t AsyncDispatcher::UnbindExceptionPortOp(
async_dispatcher_t* async,
async_exception_t* exception) {
return static_cast<AsyncDispatcher*>(async)->UnbindExceptionPort(exception);
}
zx_status_t AsyncDispatcher::BeginWait(async_wait_t* wait) {
DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
static_assert(sizeof(AsyncDispatcher::WaitState) <= sizeof(async_state_t),
"WaitState is too big");
WaitState* state = new (&wait->state) WaitState(this);
zx_status_t status =
zx::unowned_handle(wait->object)
->wait_async(port_, reinterpret_cast<uintptr_t>(wait), wait->trigger,
ZX_WAIT_ASYNC_ONCE);
if (status != ZX_OK)
state->~WaitState();
return status;
}
zx_status_t AsyncDispatcher::CancelWait(async_wait_t* wait) {
DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
if (!wait->state.reserved[0])
return ZX_ERR_NOT_FOUND;
zx_status_t status = port_.cancel(*zx::unowned_handle(wait->object),
reinterpret_cast<uintptr_t>(wait));
if (status == ZX_OK) {
auto* state = reinterpret_cast<WaitState*>(&(wait->state));
state->~WaitState();
}
return status;
}
zx_status_t AsyncDispatcher::PostTask(async_task_t* task) {
// Can be called on any thread.
AutoLock lock(lock_);
// Find correct position for the new task in |task_list_| to keep the list
// sorted by deadline. This implementation has O(N) complexity, but it's
// acceptable - async task are not expected to be used frequently.
// TODO(sergeyu): Consider using a more efficient data structure if tasks
// performance becomes important.
LinkNode<TaskState>* node;
for (node = task_list_.head(); node != task_list_.end();
node = node->previous()) {
if (task->deadline >= node->value()->task()->deadline)
break;
}
static_assert(sizeof(AsyncDispatcher::TaskState) <= sizeof(async_state_t),
"TaskState is too big");
// Will insert new task after |node|.
new (&task->state) TaskState(node);
if (reinterpret_cast<TaskState*>(&task->state) == task_list_.head()) {
// Task inserted at head. Earliest deadline changed.
RestartTimerLocked();
}
return ZX_OK;
}
zx_status_t AsyncDispatcher::CancelTask(async_task_t* task) {
DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
AutoLock lock(lock_);
if (!task->state.reserved[0])
return ZX_ERR_NOT_FOUND;
auto* state = reinterpret_cast<TaskState*>(&task->state);
state->~TaskState();
return ZX_OK;
}
zx_status_t AsyncDispatcher::BindExceptionPort(async_exception_t* exception) {
DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
static_assert(
sizeof(AsyncDispatcher::ExceptionState) <= sizeof(async_state_t),
"ExceptionState is too big");
ExceptionState* state = new (&exception->state) ExceptionState(this);
zx_status_t status = zx_task_bind_exception_port(
exception->task, port_.get(), reinterpret_cast<uintptr_t>(exception),
exception->options);
if (status != ZX_OK)
state->~ExceptionState();
return status;
}
zx_status_t AsyncDispatcher::UnbindExceptionPort(async_exception_t* exception) {
DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
if (!exception->state.reserved[0])
return ZX_ERR_NOT_FOUND;
zx_status_t status = zx_task_bind_exception_port(
exception->task, ZX_HANDLE_INVALID,
reinterpret_cast<uintptr_t>(exception), exception->options);
if (status == ZX_OK) {
auto* state = reinterpret_cast<ExceptionState*>(&exception->state);
state->~ExceptionState();
}
return status;
}
void AsyncDispatcher::DispatchTasks() {
// Snapshot now value to set implicit bound for the tasks that will run before
// DispatchTasks() returns. This also helps to avoid calling zx_clock_get()
// more than necessary.
zx_time_t now = zx_clock_get(ZX_CLOCK_MONOTONIC);
while (true) {
async_task_t* task;
{
AutoLock lock(lock_);
if (task_list_.empty())
break;
TaskState* task_state = task_list_.head()->value();
task = task_state->task();
if (task->deadline > now) {
RestartTimerLocked();
break;
}
task_state->~TaskState();
// ~TaskState() is expected to reset the state to 0. The destructor
// removes the task from the |task_list_| and LinkNode::RemoveFromList()
// sets both its fields to nullptr, which is equivalent to resetting the
// state to 0.
DCHECK_EQ(task->state.reserved[0], 0u);
}
// The handler is responsible for freeing the |task| or it may reuse it.
task->handler(this, task, ZX_OK);
}
}
void AsyncDispatcher::RestartTimerLocked() {
lock_.AssertAcquired();
if (task_list_.empty())
return;
zx_time_t deadline = task_list_.head()->value()->task()->deadline;
zx_status_t status = timer_.set(zx::time(deadline), zx::duration());
ZX_DCHECK(status == ZX_OK, status);
}
} // namespace base