Abstract over async handles (#536)

Introduce an `AsyncHandle` type to the host api, and refactor the event loop to use them instead.

I reworked the interface to the underlying `fastly_async_io_select` host call, so that it's now a bit more clear when the timeout has expired. I've also added comments to `process_pending_async_tasks` to better explain the interactions between the two action queues.

Author: elliottt

runtime/js-compute-runtime/builtins/request-response.cpp

@@ -1063,14 +1063,16 @@ HttpReq Request::request_handle(JSObject *obj) {
       JS::GetReservedSlot(obj, static_cast<uint32_t>(Request::Slots::Request)).toInt32());
 }
 
-fastly_pending_request_handle_t Request::pending_handle(JSObject *obj) {
+HttpPendingReq Request::pending_handle(JSObject *obj) {
+  HttpPendingReq res;
+
   JS::Value handle_val =
       JS::GetReservedSlot(obj, static_cast<uint32_t>(Request::Slots::PendingRequest));
   if (handle_val.isInt32()) {
-    return handle_val.toInt32();
+    res = HttpPendingReq(handle_val.toInt32());
   }
 
-  return INVALID_HANDLE;
+  return res;
 }
 
 bool Request::is_downstream(JSObject *obj) {

runtime/js-compute-runtime/builtins/request-response.h

@@ -141,7 +141,7 @@ class Request final : public BuiltinImpl<Request> {
   static bool apply_cache_override(JSContext *cx, JS::HandleObject self);
 
   static HttpReq request_handle(JSObject *obj);
-  static fastly_pending_request_handle_t pending_handle(JSObject *obj);
+  static HttpPendingReq pending_handle(JSObject *obj);
   static bool is_downstream(JSObject *obj);
   static JSString *backend(JSObject *obj);
   static const JSFunctionSpec static_methods[];

runtime/js-compute-runtime/fastly-world/fastly_world_adapter.cpp

@@ -685,7 +685,10 @@ bool fastly_async_io_select(fastly_list_async_handle_t *hs, uint32_t timeout_ms,
     }
     return false;
   }
-  ret->is_some = true;
+
+  // The result is only valid if the timeout didn't expire.
+  ret->is_some = ret->val != UINT32_MAX;
+
   return true;
 }
 bool fastly_async_io_is_ready(fastly_async_handle_t handle, bool *ret, fastly_error_t *err) {

runtime/js-compute-runtime/fastly.wit

@@ -462,8 +462,8 @@ default world fastly-world {
     ///
     /// The timeout is specified in milliseconds, or 0 if no timeout is desired.
     ///
-    /// Returns the _index_ (not handle!) of the first object that is ready, or u32::MAX if the
-    /// timeout expires before any objects are ready for I/O.
+    /// Returns the _index_ (not handle!) of the first object that is ready, or
+    /// none if the timeout expires before any objects are ready for I/O.
     async-io-select: func(hs: list<async-handle>, timeout-ms: u32) -> result<option<u32>, error>
 
     /// Returns 1 if the given async item is "ready" for its associated I/O action, 0 otherwise.

runtime/js-compute-runtime/host_interface/host_api.cpp

@@ -15,6 +15,41 @@ fastly_world_string_t string_view_to_world_string(std::string_view str) {
 
 } // namespace
 
+Result<bool> AsyncHandle::is_ready() const {
+  Result<bool> res;
+
+  fastly_error_t err;
+  bool ret;
+  if (!fastly_async_io_is_ready(this->handle, &ret, &err)) {
+    res.emplace_err(err);
+  } else {
+    res.emplace(ret);
+  }
+
+  return res;
+}
+
+Result<std::optional<uint32_t>> AsyncHandle::select(const std::vector<AsyncHandle> &handles,
+                                                    uint32_t timeout_ms) {
+  Result<std::optional<uint32_t>> res;
+
+  static_assert(sizeof(AsyncHandle) == sizeof(fastly_async_handle_t));
+  fastly_list_async_handle_t hs{
+      .ptr = reinterpret_cast<fastly_async_handle_t *>(const_cast<AsyncHandle *>(handles.data())),
+      .len = handles.size()};
+  fastly_option_u32_t ret;
+  fastly_error_t err;
+  if (!fastly_async_io_select(&hs, timeout_ms, &ret, &err)) {
+    res.emplace_err(err);
+  } else if (ret.is_some) {
+    res.emplace(ret.val);
+  } else {
+    res.emplace(std::nullopt);
+  }
+
+  return res;
+}
+
 Result<HttpBody> HttpBody::make() {
   Result<HttpBody> res;
 
@@ -101,6 +136,8 @@ Result<Void> HttpBody::close() {
   return res;
 }
 
+AsyncHandle HttpBody::async_handle() const { return AsyncHandle{this->handle}; }
+
 namespace {
 
 template <auto header_names_get>
@@ -187,6 +224,38 @@ Result<Void> generic_header_remove(auto handle, std::string_view name) {
 
 } // namespace
 
+Result<std::optional<Response>> HttpPendingReq::poll() {
+  Result<std::optional<Response>> res;
+
+  fastly_error_t err;
+  fastly_option_response_t ret;
+  if (!fastly_http_req_pending_req_poll(this->handle, &ret, &err)) {
+    res.emplace_err(err);
+  } else if (ret.is_some) {
+    res.emplace(ret.val);
+  } else {
+    res.emplace(std::nullopt);
+  }
+
+  return res;
+}
+
+Result<Response> HttpPendingReq::wait() {
+  Result<Response> res;
+
+  fastly_error_t err;
+  fastly_response_t ret;
+  if (!fastly_http_req_pending_req_wait(this->handle, &ret, &err)) {
+    res.emplace_err(err);
+  } else {
+    res.emplace(ret);
+  }
+
+  return res;
+}
+
+AsyncHandle HttpPendingReq::async_handle() const { return AsyncHandle{this->handle}; }
+
 Result<HttpReq> HttpReq::make() {
   Result<HttpReq> res;
 
@@ -201,6 +270,51 @@ Result<HttpReq> HttpReq::make() {
   return res;
 }
 
+Result<Response> HttpReq::send(HttpBody body, std::string_view backend) {
+  Result<Response> res;
+
+  fastly_error_t err;
+  fastly_response_t ret;
+  fastly_world_string_t backend_str = string_view_to_world_string(backend);
+  if (!fastly_http_req_send(this->handle, body.handle, &backend_str, &ret, &err)) {
+    res.emplace_err(err);
+  } else {
+    res.emplace(ret);
+  }
+
+  return res;
+}
+
+Result<HttpPendingReq> HttpReq::send_async(HttpBody body, std::string_view backend) {
+  Result<HttpPendingReq> res;
+
+  fastly_error_t err;
+  fastly_pending_request_handle_t ret;
+  fastly_world_string_t backend_str = string_view_to_world_string(backend);
+  if (!fastly_http_req_send_async(this->handle, body.handle, &backend_str, &ret, &err)) {
+    res.emplace_err(err);
+  } else {
+    res.emplace(ret);
+  }
+
+  return res;
+}
+
+Result<HttpPendingReq> HttpReq::send_async_streaming(HttpBody body, std::string_view backend) {
+  Result<HttpPendingReq> res;
+
+  fastly_error_t err;
+  fastly_pending_request_handle_t ret;
+  fastly_world_string_t backend_str = string_view_to_world_string(backend);
+  if (!fastly_http_req_send_async_streaming(this->handle, body.handle, &backend_str, &ret, &err)) {
+    res.emplace_err(err);
+  } else {
+    res.emplace(ret);
+  }
+
+  return res;
+}
+
 Result<Void> HttpReq::set_method(std::string_view method) {
   Result<Void> res;
 

runtime/js-compute-runtime/host_interface/host_api.h

@@ -90,6 +90,33 @@ struct HostString final {
   operator std::string_view() { return std::string_view(this->ptr.get(), this->len); }
 };
 
+/// Common methods for async handles.
+class AsyncHandle {
+public:
+  static constexpr fastly_async_handle_t invalid = UINT32_MAX - 1;
+
+  fastly_async_handle_t handle;
+
+  AsyncHandle() = default;
+  explicit AsyncHandle(fastly_async_handle_t handle) : handle{handle} {}
+
+  /// Check to see if this handle is ready.
+  Result<bool> is_ready() const;
+
+  /// Return the index in handles of the `AsyncHandle` that's ready. If the select call finishes
+  /// successfully and returns `std::nullopt`, the timeout has expired.
+  ///
+  /// If the timeout is `0`, two behaviors are possible
+  ///   * if handles is empty, an error will be returned immediately
+  ///   * otherwise, block until a handle is ready and return its index
+  ///
+  /// If the timeout is non-zero, two behaviors are possible
+  ///   * no handle becomes ready within timeout, and the successful `std::nullopt` is returned
+  ///   * a handle becomes ready within the timeout, and its index is returned.
+  static Result<std::optional<uint32_t>> select(const std::vector<AsyncHandle> &handles,
+                                                uint32_t timeout_ms);
+};
+
 /// A convenience wrapper for the host calls involving http bodies.
 class HttpBody final {
 public:
@@ -101,6 +128,7 @@ class HttpBody final {
 
   HttpBody() = default;
   explicit HttpBody(fastly_body_handle_t handle) : handle{handle} {}
+  explicit HttpBody(AsyncHandle async) : handle{async.handle} {}
 
   /// Returns true when this body handle is valid.
   bool valid() const { return this->handle != invalid; }
@@ -125,6 +153,30 @@ class HttpBody final {
 
   /// Close this handle, and reset internal state to invalid.
   Result<Void> close();
+
+  AsyncHandle async_handle() const;
+};
+
+struct Response;
+
+class HttpPendingReq final {
+public:
+  static constexpr fastly_pending_request_handle_t invalid = UINT32_MAX - 1;
+
+  fastly_pending_request_handle_t handle = invalid;
+
+  HttpPendingReq() = default;
+  explicit HttpPendingReq(fastly_pending_request_handle_t handle) : handle{handle} {}
+  explicit HttpPendingReq(AsyncHandle async) : handle{async.handle} {}
+
+  /// Poll for the response to this request.
+  Result<std::optional<Response>> poll();
+
+  /// Block until the response is ready.
+  Result<Response> wait();
+
+  /// Fetch the AsyncHandle for this pending request.
+  AsyncHandle async_handle() const;
 };
 
 class HttpBase {
@@ -155,6 +207,15 @@ class HttpReq final : public HttpBase {
 
   static Result<HttpReq> make();
 
+  /// Send this request synchronously, and wait for the response.
+  Result<Response> send(HttpBody body, std::string_view backend);
+
+  /// Send this request asynchronously.
+  Result<HttpPendingReq> send_async(HttpBody body, std::string_view backend);
+
+  /// Send this request asynchronously, and allow sending additional data through the body.
+  Result<HttpPendingReq> send_async_streaming(HttpBody body, std::string_view backend);
+
   /// Get the http version used for this request.
 
   /// Set the request method.
@@ -196,4 +257,13 @@ class HttpResp final : public HttpBase {
   Result<Void> remove_header(std::string_view name) override;
 };
 
+/// The pair of a response and its body.
+struct Response {
+  HttpResp resp;
+  HttpBody body;
+
+  Response() = default;
+  explicit Response(fastly_response_t resp) : resp{HttpResp{resp.f0}}, body{HttpBody{resp.f1}} {}
+};
+
 #endif