wasip3: Implement read/write timeouts (WebAssembly#762)

This is needed for socket options with TCP/UDP to implement timed-out
reads/writes. This plumbs the relevant timeout to the waitable-set.wait
invocation and handles the timeout outwards from there. This is a bit
tricky since timeouts/cancellation can all race with actual completion
of the operation, which is handled throughout and along the way, too.

Author: alexcrichton

cmake/bindings.cmake

@@ -131,7 +131,6 @@ add_custom_target(
       "--async=-wasi:filesystem/types@${wasip3-version}#[method]descriptor.set-times-at"
       "--async=-wasi:filesystem/types@${wasip3-version}#[method]descriptor.set-times"
       "--async=-wasi:clocks/monotonic-clock@${wasip3-version}#wait-until"
-      "--async=-wasi:clocks/monotonic-clock@${wasip3-version}#wait-for"
       "--async=-wasi:sockets/ip-name-lookup@${wasip3-version}#resolve-addresses"
       ${CMAKE_SOURCE_DIR}/wasi/p3/wit
   COMMAND cmake -E copy wasip3.h ${bottom_half}/headers/public/wasi/__generated_wasip3.h

expected/wasm32-wasip3/defined-symbols.txt

@@ -302,7 +302,7 @@ __wasilibc_fd_renumber
 __wasilibc_find_abspath
 __wasilibc_find_relpath
 __wasilibc_find_relpath_alloc
-__wasilibc_future_block_on
+__wasilibc_future_block_on_timeout
 __wasilibc_get_environ
 __wasilibc_get_service_entry_by_name
 __wasilibc_get_service_entry_by_port
@@ -340,7 +340,7 @@ __wasilibc_rmdirat
 __wasilibc_sockaddr_to_wasi
 __wasilibc_sockaddr_validate
 __wasilibc_stat
-__wasilibc_stream_block_on
+__wasilibc_stream_block_on_timeout
 __wasilibc_subtask_await_nonblocking
 __wasilibc_subtask_block_on_and_drop
 __wasilibc_tell

libc-bottom-half/cloudlibc/src/libc/time/clock_nanosleep.c

@@ -8,6 +8,7 @@
 #include <wasi/api.h>
 #include <errno.h>
 #include <time.h>
+#include <wasi/wasip3_block.h>
 
 #ifdef __wasip1__
 static_assert(TIMER_ABSTIME == __WASI_SUBCLOCKFLAGS_SUBSCRIPTION_CLOCK_ABSTIME,
@@ -61,7 +62,9 @@ int clock_nanosleep(clockid_t clock_id, int flags, const struct timespec *rqtp,
   if (flags & TIMER_ABSTIME) {
     monotonic_clock_wait_until(duration);
   } else {
-    monotonic_clock_wait_for(duration);
+    wasip3_subtask_status_t status = monotonic_clock_wait_for(duration);
+    if (WASIP3_SUBTASK_STATE(status) != WASIP3_SUBTASK_RETURNED)
+      __wasilibc_subtask_block_on_and_drop(WASIP3_SUBTASK_HANDLE(status));
   }
 #endif
   return 0;

libc-bottom-half/headers/private/wasi/descriptor_table.h

@@ -4,10 +4,58 @@
 #include <wasi/api.h>
 
 #ifndef __wasip1__
+#include <assert.h>
 #include <netinet/in.h>
 #include <sys/stat.h>
 #include <wasi/poll.h>
 
+#ifdef __wasip3__
+
+/// Helper structure to package up state related to a wasip3 `stream<u8>`.
+///
+/// This is used by various helpers to coordinate reading/writing/etc on a
+/// stream. This simultaneously represents both readers and writers.
+typedef struct wasip3_io_state_t {
+  uint32_t stream;
+  bool done;
+} wasip3_io_state_t;
+
+/// Initializes `state` with the `stream` provided.
+static inline void wasip3_io_state_init(wasip3_io_state_t *state,
+                                        uint32_t stream) {
+  assert(stream != 0);
+  state->stream = stream;
+  state->done = false;
+}
+
+/// Tests whether `state` has been initialized with a stream yet.
+static inline bool wasip3_io_state_present(wasip3_io_state_t *state) {
+  return state->stream != 0;
+}
+
+/// Closes out the streams/etc internal to `state`.
+///
+/// Internally the stream must be a reader-half of a `stream<u8>`.
+static inline void wasip3_read_state_close(wasip3_io_state_t *state) {
+  if (state->stream != 0) {
+    filesystem_stream_u8_drop_readable(state->stream);
+    state->stream = 0;
+  }
+  state->done = false;
+}
+
+/// Closes out the streams/etc internal to `state`.
+///
+/// Internally the stream must be a writer-half of a `stream<u8>`.
+static inline void wasip3_write_state_close(wasip3_io_state_t *state) {
+  if (state->stream != 0) {
+    filesystem_stream_u8_drop_writable(state->stream);
+    state->stream = 0;
+  }
+  state->done = false;
+}
+#endif
+
 // Metadata for WASI reads which is used to delegate to `__wasilibc_read(...)`
 // to perform the actual read of a stream.
 typedef struct wasi_read_t {
@@ -25,8 +73,7 @@ typedef struct wasi_read_t {
   // initialized as-necessary.
   poll_own_pollable_t *pollable;
 #else
-  // The `stream<u8>` that's being read.
-  filesystem_stream_u8_t stream;
+  wasip3_io_state_t *state;
 
   // A callback/ptr pair to invoke when EOF is reached to set errno and return
   // an error code.
@@ -45,22 +92,14 @@ typedef struct wasi_write_t {
   streams_borrow_output_stream_t output;
   poll_own_pollable_t *pollable;
 #else
-  // The actual stream that is being written to.
-  filesystem_stream_u8_writer_t output;
-  // An indicator if `output` has been closed/dropped.
-  bool *done;
+  wasip3_io_state_t *state;
   // A callback/ptr pair to invoke when EOF is reached to set errno and return
   // an error code.
   int (*eof)(void *);
   void *eof_data;
 #endif
 } wasi_write_t;
 
-#ifdef __wasip3__
-// create an alias to distinguish the handle type in the API
-typedef uint32_t waitable_t;
-#endif
-
 /**
  * Operations that are required of all descriptors registered as file
  * descriptors.

libc-bottom-half/headers/private/wasi/tcp.h

@@ -55,17 +55,16 @@ typedef struct {
 #else
   // The bytes that are being received on this socket in addition to the future
   // of the result to read when the socket hits EOF to see if there's an error.
-  sockets_stream_u8_t receive;
+  wasip3_io_state_t receive;
   sockets_future_result_void_error_code_t receive_result;
 
   // The stream to write to in order to send out bytes, along with the `send`
   // subtask that was created connected to the other half of `send`. The
   // result of the subtask at `send_result` will get filled in when
   // `send_subtask` is finished.
-  sockets_stream_u8_writer_t send;
+  wasip3_io_state_t send;
   wasip3_subtask_t send_subtask;
   sockets_result_void_error_code_t send_result;
-  bool send_done;
 #endif
 } tcp_socket_state_connected_t;
 

libc-bottom-half/headers/private/wasi/wasip3_block.h

@@ -7,6 +7,8 @@
 #include <stdint.h>
 #include <wasi/descriptor_table.h> // for waitable_t
 
+typedef wasip3_waitable_status_t (*wasip3_cancel_t)(uint32_t);
+
 // Waits for a subtask to return
 void __wasilibc_subtask_block_on_and_drop(wasip3_subtask_t subtask);
 
@@ -37,8 +39,22 @@ int __wasilibc_subtask_await_nonblocking(wasip3_subtask_status_t status);
 /// This function will block if `status` indicates that the future is blocked,
 /// and this function won't return until the future has become ready again. This
 /// is suitable, for example, for waiting for future reads to complete.
-void __wasilibc_future_block_on(wasip3_subtask_status_t status,
-                                uint32_t future);
+///
+/// If `timeout` is nonzero then this function will block for at most that
+/// amount of time. The `cancel` callback must be set when `timeout` is set and
+/// that will be used to cancel the in-flight operation on this future.
+int __wasilibc_future_block_on_timeout(wasip3_waitable_status_t status,
+                                       uint32_t future,
+                                       monotonic_clock_duration_t timeout,
+                                       wasip3_cancel_t cancel);
+
+// Convenience wrapper around `*_timeout` above which doesn't take a timeout.
+static inline void __wasilibc_future_block_on(wasip3_waitable_status_t status,
+                                              uint32_t future) {
+  ssize_t ret = __wasilibc_future_block_on_timeout(status, future, 0, NULL);
+  (void)ret;
+  assert(ret >= 0);
+}
 
 /// Waits for `stream` to be resolved after a previous operation yielded
 /// `status` as a result.
@@ -49,8 +65,24 @@ void __wasilibc_future_block_on(wasip3_subtask_status_t status,
 ///
 /// The `closed` variable is set based on the result of the operation to
 /// communicate what was received from the component model.
-size_t __wasilibc_stream_block_on(wasip3_subtask_status_t status,
-                                  uint32_t stream, bool *closed);
-#endif
+///
+/// If `timeout` is nonzero then this function will block for at most that
+/// amount of time. The `cancel` callback must be set when `timeout` is set and
+/// that will be used to cancel the in-flight operation on this stream.
+ssize_t __wasilibc_stream_block_on_timeout(wasip3_waitable_status_t status,
+                                           uint32_t stream, bool *closed,
+                                           monotonic_clock_duration_t timeout,
+                                           wasip3_cancel_t cancel);
+
+// Convenience wrapper around `*_timeout` above which doesn't take a timeout.
+static inline size_t __wasilibc_stream_block_on(wasip3_waitable_status_t status,
+                                                uint32_t stream, bool *closed) {
+  ssize_t ret =
+      __wasilibc_stream_block_on_timeout(status, stream, closed, 0, NULL);
+  assert(ret >= 0);
+  return ret;
+}
+
+#endif // __wasip3__
 
 #endif // WASI_WASIP3_BLOCK_H

libc-bottom-half/headers/public/wasi/__generated_wasip3.h

@@ -26,11 +26,11 @@ typedef struct {
   streams_own_pollable_t read_pollable;
   streams_own_pollable_t write_pollable;
 #else
-  filesystem_tuple2_stream_u8_future_result_void_error_code_t read;
-  filesystem_stream_u8_writer_t write;
+  wasip3_io_state_t read;
+  filesystem_future_result_void_error_code_t read_result;
+  wasip3_io_state_t write;
   wasip3_subtask_t write_subtask;
   filesystem_result_void_error_code_t write_pending_result;
-  bool write_done;
 #endif
 } file_t;
 
@@ -54,17 +54,11 @@ static void file_close_streams(void *data) {
     file->write_stream.__handle = 0;
   }
 #else
-  if (file->read.f0 != 0) {
-    filesystem_stream_u8_drop_readable(file->read.f0);
-    file->read.f0 = 0;
-  }
-  if (file->read.f1 != 0) {
-    filesystem_future_result_void_error_code_drop_readable(file->read.f1);
-    file->read.f1 = 0;
-  }
-  if (file->write != 0) {
-    filesystem_stream_u8_drop_writable(file->write);
-    file->write = 0;
+  wasip3_read_state_close(&file->read);
+  wasip3_write_state_close(&file->write);
+  if (file->read_result != 0) {
+    filesystem_future_result_void_error_code_drop_readable(file->read_result);
+    file->read_result = 0;
   }
   if (file->write_subtask != 0) {
     // TODO: this should use `wasip3_subtask_cancel` but right now that's buggy
@@ -76,7 +70,6 @@ static void file_close_streams(void *data) {
     __wasilibc_subtask_block_on_and_drop(file->write_subtask);
     file->write_subtask = 0;
   }
-  file->write_done = false;
 #endif
 }
 
@@ -91,13 +84,13 @@ static void file_free(void *data) {
 static int file_read_eof(void *data) {
   file_t *file = (file_t *)data;
 
-  if (file->read.f1 != 0) {
+  if (file->read_result != 0) {
     filesystem_result_void_error_code_t result;
-    __wasilibc_future_block_on(
-        filesystem_future_result_void_error_code_read(file->read.f1, &result),
-        file->read.f1);
-    filesystem_future_result_void_error_code_drop_readable(file->read.f1);
-    file->read.f1 = 0;
+    __wasilibc_future_block_on(filesystem_future_result_void_error_code_read(
+                                   file->read_result, &result),
+                               file->read_result);
+    filesystem_future_result_void_error_code_drop_readable(file->read_result);
+    file->read_result = 0;
     if (result.is_err) {
       translate_error(result.val.err);
       return -1;
@@ -123,12 +116,15 @@ static int file_get_read_stream(void *data, wasi_read_t *read) {
   read->input = streams_borrow_input_stream(file->read_stream);
   read->pollable = &file->read_pollable;
 #else
-  if (file->read.f0 == 0) {
+  if (!wasip3_io_state_present(&file->read)) {
+    assert(!file->read_result);
+    filesystem_tuple2_stream_u8_future_result_void_error_code_t result;
     filesystem_method_descriptor_read_via_stream(
-        filesystem_borrow_descriptor(file->file_handle), file->offset,
-        &file->read);
+        filesystem_borrow_descriptor(file->file_handle), file->offset, &result);
+    wasip3_io_state_init(&file->read, result.f0);
+    file->read_result = result.f1;
   }
-  read->stream = file->read.f0;
+  read->state = &file->read;
   read->eof = file_read_eof;
   read->eof_data = data;
 #endif
@@ -178,9 +174,10 @@ static int file_get_write_stream(void *data, wasi_write_t *write) {
   write->output = streams_borrow_output_stream(file->write_stream);
   write->pollable = &file->write_pollable;
 #else
-  if (file->write == 0) {
+  if (!wasip3_io_state_present(&file->write)) {
     assert(file->write_subtask == 0);
-    filesystem_stream_u8_t write_read = filesystem_stream_u8_new(&file->write);
+    filesystem_stream_u8_writer_t writer;
+    filesystem_stream_u8_t write_read = filesystem_stream_u8_new(&writer);
     wasip3_subtask_status_t status;
     if (file->oflag & O_APPEND) {
       status = filesystem_method_descriptor_append_via_stream(
@@ -193,11 +190,11 @@ static int file_get_write_stream(void *data, wasi_write_t *write) {
     }
     if (WASIP3_SUBTASK_STATE(status) != WASIP3_SUBTASK_RETURNED)
       file->write_subtask = WASIP3_SUBTASK_HANDLE(status);
+    wasip3_io_state_init(&file->write, writer);
   }
-  write->output = file->write;
+  write->state = &file->write;
   write->eof = file_write_eof;
   write->eof_data = data;
-  write->done = &file->write_done;
 #endif
   write->offset = &file->offset;
   write->timeout = 0;

libc-bottom-half/sources/file.c

@@ -1,8 +1,3 @@
-// TODO(wasip3) these are load-bearing assertions functionality-wise, but once
-// the assertions below are removed and the implementation is filled in then
-// this should be removed.
-#undef NDEBUG
-
 #include <assert.h>
 #include <common/errors.h>
 #include <errno.h>
@@ -164,15 +159,15 @@ ssize_t __wasilibc_write(wasi_write_t *write, const void *buffer,
     }
   }
 #elif defined(__wasip3__)
-  assert(write->blocking);     // TODO(wasip3)
-  assert(write->timeout == 0); // TODO(wasip3)
+  assert(write->blocking); // TODO(wasip3)
 
   // Perform writes until either a non-zero-length write completes or the stream
   // is closed.
-  while (!*write->done) {
-    size_t amount = __wasilibc_stream_block_on(
-        filesystem_stream_u8_write(write->output, buffer, length),
-        write->output, write->done);
+  while (!write->state->done) {
+    ssize_t amount = __wasilibc_stream_block_on_timeout(
+        filesystem_stream_u8_write(write->state->stream, buffer, length),
+        write->state->stream, &write->state->done, write->timeout,
+        filesystem_stream_u8_cancel_write);
     if (amount > 0 || length == 0) {
       if (write->offset)
         *write->offset += amount;
@@ -250,16 +245,15 @@ ssize_t __wasilibc_read(wasi_read_t *read, void *buffer, size_t length) {
     }
   }
 #elif defined(__wasip3__)
-  assert(read->blocking);     // TODO(wasip3)
-  assert(read->timeout == 0); // TODO(wasip3)
+  assert(read->blocking); // TODO(wasip3)
 
   // Attempt reads until a nonzero-length read is encountered or the stream is
   // closed.
-  bool closed = false;
-  while (!closed) {
-    size_t amount = __wasilibc_stream_block_on(
-        filesystem_stream_u8_read(read->stream, buffer, length), read->stream,
-        &closed);
+  while (!read->state->done) {
+    size_t amount = __wasilibc_stream_block_on_timeout(
+        filesystem_stream_u8_read(read->state->stream, buffer, length),
+        read->state->stream, &read->state->done, read->timeout,
+        filesystem_stream_u8_cancel_read);
     if (amount > 0 || length == 0) {
       if (read->offset)
         *read->offset += amount;