[Fuchsia] Add AsyncDispatcher.

Added AsyncDispatcher class which implements async dispatcher interface
for Fuchsia. It will be used in MessagePumpFuchsia. This will allow to
run FIDL on Chromium threads.

The implementation is based largely on the default async loop. That code
has been translated to C++ with all features that we don't need in
chromium removed.

Bug: 831384
Change-Id: I76459d364e1a4864dbe6f76f582b9fb87c79e027
Reviewed-on: https://chromium-review.googlesource.com/1003073
Commit-Queue: Sergey Ulanov <sergeyu@chromium.org>
Reviewed-by: Daniel Cheng <dcheng@chromium.org>
Reviewed-by: Wez <wez@chromium.org>
Cr-Commit-Position: refs/heads/master@{#550416}

base/BUILD.gn

@@ -1318,6 +1318,8 @@ jumbo_component("base") {
       "base_paths_fuchsia.h",
       "debug/stack_trace_fuchsia.cc",
       "files/file_path_watcher_fuchsia.cc",
+      "fuchsia/async_dispatcher.cc",
+      "fuchsia/async_dispatcher.h",
       "fuchsia/default_job.cc",
       "fuchsia/default_job.h",
       "fuchsia/fuchsia_logging.cc",
@@ -1345,6 +1347,9 @@ jumbo_component("base") {
     ]
 
     libs = [ "launchpad" ]
+
+    public_deps += [ "//third_party/fuchsia-sdk:async" ]
+    deps += [ "//third_party/fuchsia-sdk:async" ]
   }
 
   # NaCl.
@@ -2511,6 +2516,9 @@ test("base_unittests") {
       "files/file_locking_unittest.cc",
       "posix/unix_domain_socket_unittest.cc",
     ]
+
+    sources += [ "fuchsia/async_dispatcher_unittest.cc" ]
+    deps += [ "//third_party/fuchsia-sdk:async" ]
   }
 
   if (is_android) {

base/fuchsia/async_dispatcher.h

+// 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.
+
+#ifndef BASE_FUCHSIA_ASYNC_DISPATCHER_H_
+#define BASE_FUCHSIA_ASYNC_DISPATCHER_H_
+
+#include <lib/async/default.h>
+#include <lib/async/dispatcher.h>
+
+#include "base/containers/linked_list.h"
+#include "base/fuchsia/scoped_zx_handle.h"
+#include "base/macros.h"
+#include "base/synchronization/lock.h"
+#include "base/threading/thread_checker.h"
+
+namespace base {
+
+// Implementation of dispatcher for Fuchsia's async library. It's necessary to
+// run Fuchsia's library on chromium threads.
+class AsyncDispatcher : public async_t {
+ public:
+  AsyncDispatcher();
+  ~AsyncDispatcher();
+
+  // Returns after running one or more tasks or waits until |deadline|.
+  // Returns |ZX_OK| if some tasks were executed, |ZX_ERR_TIMED_OUT| - the
+  // deadline expired, |ZX_ERR_CANCELED| - Stop() was called.
+  zx_status_t DispatchOrWaitUntil(zx_time_t deadline);
+
+  // If Run() is being executed then it will return as soon as possible (e.g.
+  // finishing running the current task), otherwise the following Run() call
+  // will quit immediately instead of waiting until deadline expires.
+  void Stop();
+
+ private:
+  class WaitState;
+  class TaskState;
+
+  static zx_time_t NowOp(async_t* async);
+  static zx_status_t BeginWaitOp(async_t* async, async_wait_t* wait);
+  static zx_status_t CancelWaitOp(async_t* async, async_wait_t* wait);
+  static zx_status_t PostTaskOp(async_t* async, async_task_t* task);
+  static zx_status_t CancelTaskOp(async_t* async, async_task_t* task);
+  static zx_status_t QueuePacketOp(async_t* async,
+                                   async_receiver_t* receiver,
+                                   const zx_packet_user_t* data);
+  static zx_status_t SetGuestBellTrapOp(async_t* async,
+                                        async_guest_bell_trap_t* trap,
+                                        zx_handle_t guest,
+                                        zx_vaddr_t addr,
+                                        size_t length);
+
+  // async_ops_t implementation. Called by corresponding *Op() methods above.
+  zx_status_t BeginWait(async_wait_t* wait);
+  zx_status_t CancelWait(async_wait_t* wait);
+  zx_status_t PostTask(async_task_t* task);
+  zx_status_t CancelTask(async_task_t* task);
+
+  // Runs tasks in |task_list_| that have deadline in the past.
+  void DispatchTasks();
+
+  // Must be called while |lock_| is held.
+  void RestartTimerLocked();
+
+  THREAD_CHECKER(thread_checker_);
+
+  ScopedZxHandle port_;
+  ScopedZxHandle timer_;
+  ScopedZxHandle stop_event_;
+
+  LinkedList<WaitState> wait_list_;
+
+  // |lock_| must be held when accessing |task_list_|.
+  base::Lock lock_;
+
+  LinkedList<TaskState> task_list_;
+
+  DISALLOW_COPY_AND_ASSIGN(AsyncDispatcher);
+};
+
+}  // namespace base
+
+#endif  // BASE_FUCHSIA_ASYNC_DISPATCHER_H_
\ No newline at end of file

base/fuchsia/async_dispatcher_unittest.cc

+// 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 "base/callback.h"
+#include "base/fuchsia/scoped_zx_handle.h"
+#include "base/test/test_timeouts.h"
+#include "base/time/time.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace base {
+
+namespace {
+
+struct TestTask : public async_task_t {
+  explicit TestTask() {
+    state = ASYNC_STATE_INIT;
+    handler = &TaskProc;
+    deadline = 0;
+  }
+
+  static void TaskProc(async_t* async, async_task_t* task, zx_status_t status);
+
+  int num_calls = 0;
+  int repeats = 1;
+  OnceClosure on_call;
+  zx_status_t last_status = ZX_OK;
+};
+
+// static
+void TestTask::TaskProc(async_t* async,
+                        async_task_t* task,
+                        zx_status_t status) {
+  EXPECT_EQ(async, async_get_default());
+  EXPECT_TRUE(status == ZX_OK || status == ZX_ERR_CANCELED)
+      << "status: " << status;
+
+  auto* test_task = static_cast<TestTask*>(task);
+  test_task->num_calls++;
+  test_task->last_status = status;
+
+  if (!test_task->on_call.is_null())
+    std::move(test_task->on_call).Run();
+
+  if (test_task->num_calls < test_task->repeats)
+    async_post_task(async, task);
+};
+
+struct TestWait : public async_wait_t {
+  TestWait(zx_handle_t handle,
+           zx_signals_t signals) {
+    state = ASYNC_STATE_INIT;
+    handler = &HandleProc;
+    object = handle;
+    trigger = signals;
+  }
+
+  static void HandleProc(async_t* async,
+                         async_wait_t* wait,
+                         zx_status_t status,
+                         const zx_packet_signal_t* signal);
+  int num_calls = 0;
+  OnceClosure on_call;
+  zx_status_t last_status = ZX_OK;
+};
+
+// static
+void TestWait::HandleProc(async_t* async,
+                          async_wait_t* wait,
+                          zx_status_t status,
+                          const zx_packet_signal_t* signal) {
+  EXPECT_EQ(async, async_get_default());
+  EXPECT_TRUE(status == ZX_OK || status == ZX_ERR_CANCELED)
+      << "status: " << status;
+
+  auto* test_wait = static_cast<TestWait*>(wait);
+
+  test_wait->num_calls++;
+  test_wait->last_status = status;
+
+  if (!test_wait->on_call.is_null())
+    std::move(test_wait->on_call).Run();
+}
+
+}  // namespace
+
+class AsyncDispatcherTest : public testing::Test {
+ public:
+  AsyncDispatcherTest() {
+    dispatcher_ = std::make_unique<AsyncDispatcher>();
+
+    async_ = async_get_default();
+    EXPECT_TRUE(async_);
+
+    EXPECT_EQ(zx_socket_create(ZX_SOCKET_DATAGRAM, socket1_.receive(),
+                               socket2_.receive()),
+              ZX_OK);
+  }
+
+  ~AsyncDispatcherTest() override = default;
+
+  void RunUntilIdle() {
+    while (true) {
+      zx_status_t status = dispatcher_->DispatchOrWaitUntil(0);
+      if (status != ZX_OK) {
+        EXPECT_EQ(status, ZX_ERR_TIMED_OUT);
+        break;
+      }
+    }
+  }
+
+ protected:
+  std::unique_ptr<AsyncDispatcher> dispatcher_;
+
+  async_t* async_ = nullptr;
+
+  base::ScopedZxHandle socket1_;
+  base::ScopedZxHandle socket2_;
+};
+
+TEST_F(AsyncDispatcherTest, PostTask) {
+  TestTask task;
+  ASSERT_EQ(async_post_task(async_, &task), ZX_OK);
+  dispatcher_->DispatchOrWaitUntil(0);
+  EXPECT_EQ(task.num_calls, 1);
+  EXPECT_EQ(task.last_status, ZX_OK);
+}
+
+TEST_F(AsyncDispatcherTest, TaskRepeat) {
+  TestTask task;
+  task.repeats = 2;
+  ASSERT_EQ(async_post_task(async_, &task), ZX_OK);
+  RunUntilIdle();
+  EXPECT_EQ(task.num_calls, 2);
+  EXPECT_EQ(task.last_status, ZX_OK);
+}
+
+TEST_F(AsyncDispatcherTest, DelayedTask) {
+  TestTask task;
+  constexpr auto kDelay = TimeDelta::FromMilliseconds(5);
+  TimeTicks started = TimeTicks::Now();
+  task.deadline = zx_deadline_after(kDelay.InNanoseconds());
+  ASSERT_EQ(async_post_task(async_, &task), ZX_OK);
+  zx_status_t status = dispatcher_->DispatchOrWaitUntil(zx_deadline_after(
+      (kDelay + TestTimeouts::tiny_timeout()).InNanoseconds()));
+  EXPECT_EQ(status, ZX_OK);
+  EXPECT_GE(TimeTicks::Now() - started, kDelay);
+}
+
+TEST_F(AsyncDispatcherTest, CancelTask) {
+  TestTask task;
+  ASSERT_EQ(async_post_task(async_, &task), ZX_OK);
+  ASSERT_EQ(async_cancel_task(async_, &task), ZX_OK);
+  RunUntilIdle();
+  EXPECT_EQ(task.num_calls, 0);
+}
+
+TEST_F(AsyncDispatcherTest, TaskObserveShutdown) {
+  TestTask task;
+  ASSERT_EQ(async_post_task(async_, &task), ZX_OK);
+  dispatcher_.reset();
+
+  EXPECT_EQ(task.num_calls, 1);
+  EXPECT_EQ(task.last_status, ZX_ERR_CANCELED);
+}
+
+TEST_F(AsyncDispatcherTest, Wait) {
+  TestWait wait(socket1_.get(), ZX_SOCKET_READABLE);
+  EXPECT_EQ(async_begin_wait(async_, &wait), ZX_OK);
+
+  // Handler shouldn't be called because the event wasn't signaled.
+  RunUntilIdle();
+  EXPECT_EQ(wait.num_calls, 0);
+
+  char byte = 0;
+  EXPECT_EQ(zx_socket_write(socket2_.get(), /*options=*/0, &byte, sizeof(byte),
+                            /*actual=*/nullptr),
+            ZX_OK);
+
+  zx_status_t status = dispatcher_->DispatchOrWaitUntil(
+      zx_deadline_after(TestTimeouts::tiny_timeout().InNanoseconds()));
+  EXPECT_EQ(status, ZX_OK);
+
+  EXPECT_EQ(wait.num_calls, 1);
+  EXPECT_EQ(wait.last_status, ZX_OK);
+}
+
+TEST_F(AsyncDispatcherTest, CancelWait) {
+  TestWait wait(socket1_.get(), ZX_SOCKET_READABLE);
+  EXPECT_EQ(async_begin_wait(async_, &wait), ZX_OK);
+
+  char byte = 0;
+  EXPECT_EQ(zx_socket_write(socket2_.get(), /*options=*/0, &byte, sizeof(byte),
+                            /*actual=*/nullptr),
+            ZX_OK);
+
+  EXPECT_EQ(async_cancel_wait(async_, &wait), ZX_OK);
+
+  RunUntilIdle();
+  EXPECT_EQ(wait.num_calls, 0);
+}
+
+TEST_F(AsyncDispatcherTest, WaitShutdown) {
+  TestWait wait(socket1_.get(), ZX_SOCKET_READABLE);
+  EXPECT_EQ(async_begin_wait(async_, &wait), ZX_OK);
+  RunUntilIdle();
+  dispatcher_.reset();
+
+  EXPECT_EQ(wait.num_calls, 1);
+  EXPECT_EQ(wait.last_status, ZX_ERR_CANCELED);
+}
+
+}  // namespace base

third_party/fuchsia-sdk/BUILD.gn

+# 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.
+
+assert(is_fuchsia)
+
+config("sdk_lib_dirs_config") {
+  lib_dirs = [ "sdk/arch/${target_cpu}/lib" ]
+}
+
+config("async_pkg_config") {
+  visibility = [ ":async" ]
+
+  configs = [ ":sdk_lib_dirs_config" ]
+
+  include_dirs = [
+    "sdk/pkg/async/include",
+    "sdk/pkg/async-default/include",
+  ]
+
+  libs = [ "async.default" ]
+}
+
+static_library("async") {
+  public_configs = [ ":async_pkg_config" ]
+  sources = [
+    "sdk/pkg/async/ops.c",
+  ]
+}