wasip3: Finish nonblocking I/O implementation for TCP

expected/wasm32-wasip3/defined-symbols.txt

@@ -336,6 +336,7 @@ __wasilibc_populate_preopens
 __wasilibc_pthread_self
 __wasilibc_random
 __wasilibc_read
+__wasilibc_read_poll
 __wasilibc_rename_newat
 __wasilibc_rename_oldat
 __wasilibc_reset_preopens
@@ -350,9 +351,11 @@ __wasilibc_tell
 __wasilibc_unlinkat
 __wasilibc_unspecified_addr
 __wasilibc_utimens
+__wasilibc_waitable_block_on
 __wasilibc_wasi_family_to_libc
 __wasilibc_wasi_to_sockaddr
 __wasilibc_write
+__wasilibc_write_poll
 __wasm_call_dtors
 __wcscoll_l
 __wcsftime_l

libc-bottom-half/cloudlibc/src/libc/poll/poll.c

@@ -372,20 +372,30 @@ static int poll_impl(struct pollfd *fds, size_t nfds, int timeout) {
       goto out;
     }
 
-
     if (events & POLLRDNORM) {
-      // TODO(wasip3): use `get_read_stream` and use that I/O to do something.
-      // Requires a lot more integration with nonblocking I/O to get that
-      // working.
-      errno = EOPNOTSUPP;
-      goto out;
+      if (entry->vtable->get_read_stream) {
+        wasi_read_t read;
+        if (entry->vtable->get_read_stream(entry->data, &read) < 0)
+          goto out;
+        if (__wasilibc_read_poll(read.state, &state) < 0)
+          goto out;
+      } else {
+        errno = EOPNOTSUPP;
+        goto out;
+      }
     }
 
     if (events & POLLWRNORM) {
-      // TODO(wasip3): use `get_write_stream` to implement this (see
-      // `POLLRDNORM` above).
-      errno = EOPNOTSUPP;
-      goto out;
+      if (entry->vtable->get_write_stream) {
+        wasi_write_t write;
+        if (entry->vtable->get_write_stream(entry->data, &write) < 0)
+          goto out;
+        if (__wasilibc_write_poll(write.state, &state) < 0)
+          goto out;
+      } else {
+        errno = EOPNOTSUPP;
+        goto out;
+      }
     }
   }
 
@@ -451,10 +461,14 @@ static int poll_impl(struct pollfd *fds, size_t nfds, int timeout) {
         if (p->waitable != event.waitable)
           continue;
         state.pollfd = p->pollfd;
-        // Clear the `waitable` since this event has fired and it doesn't need
-        // to be join'd to 0 below. Then invoke the custom callback originally
-        // added for this which will handle any necessary completion logic and
-        // updating `state.pollfd` with various events.
+        // Remove this waitable from the `waitable-set` as the `ready`
+        // operation might end up deleting the handle. Set the list here to 0
+        // so it's not removed down below.
+        //
+        // Then invoke the custom callback originally added for this
+        // which will handle any necessary completion logic and updating
+        // `state.pollfd` with various events.
+        wasip3_waitable_join(p->waitable, 0);
         p->waitable = 0;
         p->ready(p->ready_data, &state, &event);
       }

libc-bottom-half/cloudlibc/src/libc/sys/select/pselect.c

@@ -3,9 +3,7 @@
 // SPDX-License-Identifier: BSD-2-Clause
 
 #include <common/time.h>
-
 #include <sys/select.h>
-
 #include <wasi/api.h>
 #include <errno.h>
 #include <poll.h>

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

@@ -6,26 +6,49 @@
 #ifndef __wasip1__
 #include <assert.h>
 #include <netinet/in.h>
+#include <stdlib.h>
+#include <string.h>
 #include <sys/stat.h>
 #include <wasi/poll.h>
 
 #ifdef __wasip3__
 
+/// The stream is complete and no further operations are allowed on it.
+#define WASIP3_IO_DONE (1 << 0)
+/// An I/O operation, be it a read or write, is in flight.
+#define WASIP3_IO_INPROGRESS (1 << 1)
+/// The in-flight I/O operation is a zero-length operation. This means that if
+/// it's finished we expect that the next operation finishes immediately.
+#define WASIP3_IO_ZERO_INPROGRESS (1 << 2)
+/// A zero-sized in-flight I/O operation just finished so the next I/O op
+/// should finish immediately. If it doesn't it'll turn on the next flag.
+#define WASIP3_IO_SHOULD_BE_READY (1 << 3)
+/// This stream isn't compatible with zero-length reads/writes signaling
+/// readiness, so libc must buffer data internally for reads/writes.
+#define WASIP3_IO_MUST_BUFFER (1 << 4)
+
 /// 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;
+  /// Bitset of `WASIP3_IO_*` flags.
+  uint32_t flags;
+  /// Malloc'd buffer used for reads/writes. NULL if not present.
+  uint8_t *buf;
+  /// Start of `buf` that has data in-flight or ready.
+  size_t buf_start;
+  /// End of `buf` that has data in-flight or ready.
+  size_t buf_end;
 } 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);
+  memset(state, 0, sizeof(*state));
   state->stream = stream;
-  state->done = false;
 }
 
 /// Tests whether `state` has been initialized with a stream yet.
@@ -37,22 +60,26 @@ static inline bool wasip3_io_state_present(wasip3_io_state_t *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) {
+  if (state->flags & WASIP3_IO_INPROGRESS)
+    filesystem_stream_u8_cancel_read(state->stream);
+  if (state->buf)
+    free(state->buf);
+  if (state->stream != 0)
     filesystem_stream_u8_drop_readable(state->stream);
-    state->stream = 0;
-  }
-  state->done = false;
+  memset(state, 0, sizeof(*state));
 }
 
 /// 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) {
+  if (state->flags & WASIP3_IO_INPROGRESS)
+    filesystem_stream_u8_cancel_write(state->stream);
+  if (state->buf)
+    free(state->buf);
+  if (state->stream != 0)
     filesystem_stream_u8_drop_writable(state->stream);
-    state->stream = 0;
-  }
-  state->done = false;
+  memset(state, 0, sizeof(*state));
 }
 #endif
 
@@ -195,6 +222,7 @@ typedef struct descriptor_vtable_t {
   /// and `get_write_stream`, if present, to handle `POLL{RD,WR}NORM` events.
   int (*poll_register)(void *, poll_state_t *state, short events);
 
+#ifdef __wasip2__
   /// Invoked when `poll` has already run and detected that this object was
   /// ready. The `events` provided are the same as those provided to
   /// `__wasilibc_poll_add` originally. This function should call
@@ -204,6 +232,7 @@ typedef struct descriptor_vtable_t {
   /// If this function is not provided then `events` will automatically
   /// be placed into the `revents` field of `pollfd`.
   int (*poll_finish)(void *, poll_state_t *state, short events);
+#endif // __wasip2__
 } descriptor_vtable_t;
 
 /// A "fat pointer" which is placed inside of the descriptor table.

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

@@ -9,6 +9,7 @@
 #include <errno.h>
 #include <fcntl.h>
 #include <wasi/descriptor_table.h>
+#include <wasi/poll.h>
 #include <wasi/wasip2.h>
 
 #ifdef __wasip2__
@@ -119,6 +120,18 @@ dir_entry_type_to_d_type(filesystem_descriptor_type_t *ty) {
   }
 }
 
+#ifndef __wasip2__
+/// Starts the process of `poll` for `iostate` assuming it's backed by a
+/// readable stream.
+///
+/// This will handle all the internal logic for zero-length reads/writes/etc
+/// and add to `state` as necessary.
+int __wasilibc_read_poll(wasip3_io_state_t *iostate, poll_state_t *state);
+
+/// Write dual of `__wasilibc_read_poll`.
+int __wasilibc_write_poll(wasip3_io_state_t *iostate, poll_state_t *state);
+#endif // !__wasip2__
+
 #endif // __wasip2__ || __wasip3__
 
 #endif // __wasi_file_utils_h

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

@@ -35,13 +35,25 @@ typedef struct {
   wasip3_subtask_t subtask;
 #endif
 } tcp_socket_state_connecting_t;
+
+/// The `stream` in `tcp_socket_state_listening_t` is finished and should not
+/// be read again.
+#define TCP_LISTENING_DONE (1 << 0)
+/// There is an active read on `stream` which has yet to complete.
+#define TCP_LISTENING_ACCEPTING (1 << 1)
+/// The `accept_result` field is valid and ready to be processed.
+#define TCP_LISTENING_ACCEPT_READY (1 << 2)
+
 typedef struct {
 #ifdef __wasip2__
   int dummy;
 #else
   // The `stream<tcp-socket>` that this is reading to receive accepted sockets.
   sockets_stream_own_tcp_socket_t stream;
-  bool done;
+  /// In-flight result of the read of `stream`.
+  sockets_own_tcp_socket_t accept_result;
+  /// Bitset of `TCP_LISTENING_*`.
+  uint32_t flags;
 #endif
 } tcp_socket_state_listening_t;
 

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

@@ -9,6 +9,14 @@
 
 typedef wasip3_waitable_status_t (*wasip3_cancel_t)(uint32_t);
 
+/// Wait at most `timeout` duration for `waitable` to become ready.
+///
+/// If `timeout` is 0 this will wait indefinitely. Returns `true` if `waitable`
+/// has become ready. Returns `false` if `timeout` elapsed. Upon returning
+/// `true` the `event` field is filled in.
+bool __wasilibc_waitable_block_on(uint32_t waitable, wasip3_event_t *event,
+                                  monotonic_clock_duration_t timeout);
+
 // Waits for a subtask to return
 void __wasilibc_subtask_block_on_and_drop(wasip3_subtask_t subtask);