wasip2: Deduplicate stream code in read/write/TCP (WebAssembly#735)

This commit extends the support in WebAssembly#734 by deduplicating the separate
paths for reading a stream shared between `read` and `recvfrom`, for
example. The `get_{read,write}_stream` methods now return more metadata
and the TCP `recvfrom` implementation delegates to an internal
`__wasilibc_read` to perform the actual read. This opens up the door to
supporting nonblocking reads/writes on other streams in the future, for
example.

Author: alexcrichton

expected/wasm32-wasip2/defined-symbols.txt

@@ -334,7 +334,7 @@ __wasilibc_poll_ready
 __wasilibc_populate_preopens
 __wasilibc_pthread_self
 __wasilibc_random
-__wasilibc_read_stream
+__wasilibc_read
 __wasilibc_rename_newat
 __wasilibc_rename_oldat
 __wasilibc_reset_preopens
@@ -348,7 +348,7 @@ __wasilibc_unspecified_addr
 __wasilibc_utimens
 __wasilibc_wasi_family_to_libc
 __wasilibc_wasi_to_sockaddr
-__wasilibc_write_stream
+__wasilibc_write
 __wasm_call_dtors
 __wcscoll_l
 __wcsftime_l

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

@@ -217,11 +217,10 @@ static int poll_impl(struct pollfd *fds, size_t nfds, int timeout) {
     if (pollfd->events & POLLRDNORM) {
       if (entry->vtable->get_read_stream) {
         streams_borrow_input_stream_t input;
-        poll_own_pollable_t *pollable;
-        if (entry->vtable->get_read_stream(entry->data, &input, NULL,
-                                           &pollable) < 0)
+        wasip2_read_t read;
+        if (entry->vtable->get_read_stream(entry->data, &read) < 0)
           return -1;
-        if (__wasilibc_poll_add_input_stream(&state, input, pollable) < 0)
+        if (__wasilibc_poll_add_input_stream(&state, read.input, read.pollable) < 0)
           return -1;
       } else {
         errno = EOPNOTSUPP;
@@ -231,12 +230,10 @@ static int poll_impl(struct pollfd *fds, size_t nfds, int timeout) {
 
     if (pollfd->events & POLLWRNORM) {
       if (entry->vtable->get_write_stream) {
-        streams_borrow_output_stream_t output;
-        poll_own_pollable_t *pollable;
-        if (entry->vtable->get_write_stream(entry->data, &output, NULL,
-                                            &pollable) < 0)
+        wasip2_write_t write;
+        if (entry->vtable->get_write_stream(entry->data, &write) < 0)
           return -1;
-        if (__wasilibc_poll_add_output_stream(&state, output, pollable) < 0)
+        if (__wasilibc_poll_add_output_stream(&state, write.output, write.pollable) < 0)
           return -1;
       } else {
         errno = EOPNOTSUPP;

libc-bottom-half/cloudlibc/src/libc/unistd/read.c

@@ -28,41 +28,19 @@ ssize_t read(int fildes, void *buf, size_t nbyte) {
   }
   return bytes_read;
 #elif defined(__wasip2__)
-  // First, check to see if this is a socket, in which case we defer to `recvfrom`:
   descriptor_table_entry_t *entry = descriptor_table_get_ref(fildes);
   if (!entry)
     return -1;
-  if (entry->vtable->recvfrom != NULL)
+  if (entry->vtable->get_read_stream) {
+    wasip2_read_t read;
+    if (entry->vtable->get_read_stream(entry->data, &read) < 0)
+      return -1;
+    return __wasilibc_read(&read, buf, nbyte);
+  }
+  if (entry->vtable->recvfrom)
     return entry->vtable->recvfrom(entry->data, buf, nbyte, 0, NULL, NULL);
-
-  bool ok = false;
-
-  // Translate the file descriptor to an internal handle
-  streams_borrow_input_stream_t input_stream;
-  off_t *off;
-  if (__wasilibc_read_stream(fildes, &input_stream, &off, NULL) < 0)
-    return -1;
-
-  // Set up a WASI list of bytes to receive the results
-  wasip2_list_u8_t contents;
-
-  // Read the bytes
-  streams_stream_error_t stream_error;
-  ok = streams_method_input_stream_blocking_read(input_stream,
-                                                 nbyte,
-                                                 &contents,
-                                                 &stream_error);
-  if (!ok)
-    return wasip2_handle_read_error(stream_error);
-
-  // Copy the bytes allocated in the canonical ABI to `buf`
-  memcpy(buf, contents.ptr, contents.len);
-  wasip2_list_u8_free(&contents);
-
-  // Update the offset
-  if (off)
-    *off += contents.len;
-  return contents.len;
+  errno = EOPNOTSUPP;
+  return -1;
 #elif defined(__wasip3__)
   filesystem_tuple2_stream_u8_future_result_void_error_code_t *stream;
   off_t *off;

libc-bottom-half/cloudlibc/src/libc/unistd/write.c

@@ -29,58 +29,19 @@ ssize_t write(int fildes, const void *buf, size_t nbyte) {
   }
   return bytes_written;
 #elif defined(__wasip2__)
-  // First, check to see if this is a socket, in which case we defer to `sendto`:
   descriptor_table_entry_t *entry = descriptor_table_get_ref(fildes);
   if (!entry)
     return -1;
-  if (entry->vtable->sendto != NULL)
-    return entry->vtable->sendto(entry->data, buf, nbyte, 0, NULL, 0);
-
-  streams_borrow_output_stream_t output_stream;
-  poll_borrow_pollable_t pollable;
-  bool ok = false;
-  filesystem_error_code_t error_code;
-
-  // Translate the file descriptor to an internal handle
-  off_t *off;
-  if (__wasilibc_write_stream(fildes, &output_stream, &off, &pollable) < 0)
-    return -1;
-
-  // Check readiness for writing
-  uint64_t num_bytes_permitted = 0;
-  streams_stream_error_t stream_error;
-  while (num_bytes_permitted == 0) {
-    ok = streams_method_output_stream_check_write(output_stream,
-                                                  &num_bytes_permitted,
-                                                  &stream_error);
-    if (!ok)
-      return wasip2_handle_write_error(stream_error);
-
-    if (num_bytes_permitted == 0)
-      poll_method_pollable_block(pollable);
+  if (entry->vtable->get_write_stream) {
+    wasip2_write_t write;
+    if (entry->vtable->get_write_stream(entry->data, &write) < 0)
+      return -1;
+    return __wasilibc_write(&write, buf, nbyte);
   }
-
-  // Convert the buffer to a WASI list of bytes
-  wasip2_list_u8_t contents;
-  contents.len = num_bytes_permitted < nbyte ? num_bytes_permitted : nbyte;
-  contents.ptr = (uint8_t*) buf;
-
-  // Write the bytes to the stream
-  ok = streams_method_output_stream_write(output_stream,
-                                          &contents,
-                                          &stream_error);
-  if (!ok)
-    return wasip2_handle_write_error(stream_error);
-
-  ok = streams_method_output_stream_blocking_flush(output_stream,
-                                                   &stream_error);
-  if (!ok)
-    return wasip2_handle_write_error(stream_error);
-
-
-  if (off)
-    *off += contents.len;
-  return contents.len;
+  if (entry->vtable->sendto)
+    return entry->vtable->sendto(entry->data, buf, nbyte, 0, NULL, 0);
+  errno = EOPNOTSUPP;
+  return -1;
 #elif defined(__wasip3__)
   wasip3_write_t *write_end;
   off_t *off;

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

@@ -8,6 +8,34 @@
 #include <sys/stat.h>
 #include <netinet/in.h>
 
+#ifdef __wasip2__
+// Metadata for WASI reads which is used to delegate to `__wasilibc_read(...)`
+// to perform the actual read of a stream.
+typedef struct wasip2_read_t {
+  // The `wasi:io/streams.input-stream` that this is reading from.
+  streams_borrow_input_stream_t input;
+  // An optional pointer to the internal offset of this stream, updated on
+  // successful reads.
+  off_t *offset;
+  // A required pointer to an owned pollable for `input`. This is lazily
+  // initialized as-necessary.
+  poll_own_pollable_t *pollable;
+  // Whether or not this read will use blocking I/O.
+  bool blocking;
+  // The timeout, in nanoseconds, for this operation.
+  monotonic_clock_duration_t timeout;
+} wasip2_read_t;
+
+// Same as `wasip2_read_t`, but for writes.
+typedef struct wasip2_write_t {
+  streams_borrow_output_stream_t output;
+  off_t *offset;
+  poll_own_pollable_t *pollable;
+  bool blocking;
+  monotonic_clock_duration_t timeout;
+} wasip2_write_t;
+#endif
+
 #ifdef __wasip3__
 // create an alias to distinguish the handle type in the API
 typedef uint32_t waitable_t;
@@ -44,14 +72,12 @@ typedef struct descriptor_vtable_t {
   // Generic I/O
 
 #ifdef __wasip2__
-  /// Looks up a `wasi:io/streams.input-stream` object and stores it in
-  /// the first argument. If provide also stores a pointer to the internal
-  /// `off_t` offset and `pollable` for this object. The returned pointers
-  /// point within the descriptor itself.
-  int (*get_read_stream)(void*, streams_borrow_input_stream_t*, off_t**, poll_own_pollable_t**);
-
+  /// Looks up metadata to perform a read operation for this stream. This is used
+  /// to implement the `read` syscall, for example, and is also used with `poll`
+  /// when waiting for readability.
+  int (*get_read_stream)(void*, wasip2_read_t*);
   /// Same as `get_read_stream`, but for output streams.
-  int (*get_write_stream)(void*, streams_borrow_output_stream_t*, off_t**, poll_own_pollable_t**);
+  int (*get_write_stream)(void*, wasip2_write_t*);
 #endif
 #ifdef __wasip3__
   int (*get_read_stream3)(void*, filesystem_tuple2_stream_u8_future_result_void_error_code_t **out, off_t** off);

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

@@ -65,18 +65,16 @@ static int fd_to_file_handle(int fd, filesystem_borrow_descriptor_t* result) {
 #endif
 
 #ifdef __wasip2__
-// Gets an `output-stream` borrow from the `fd` provided.
-int __wasilibc_write_stream(int fd,
-                            streams_borrow_output_stream_t *out,
-                            off_t **off,
-                            poll_borrow_pollable_t *pollable);
-
-// Gets an `input-stream` borrow from the `fd` provided.
-int __wasilibc_read_stream(int fd,
-                           streams_borrow_input_stream_t *out,
-                           off_t **off,
-                           poll_borrow_pollable_t *pollable);
+// Reads from `read` into `buf`/`len`
+//
+// This perform the read configured by `read`, e.g. whether it's blocking or
+// not, and places the result in the specified buffer. Used to implement
+// `read` and `recvfrom`, for example.
+ssize_t __wasilibc_read(wasip2_read_t *read, void *buf, size_t len);
+// Same as `__wasilibc_read`, but for writes.
+ssize_t __wasilibc_write(wasip2_write_t *write, const void *buf, size_t len);
 #endif
+
 #ifdef __wasip3__
 int __wasilibc_write_stream3(int fildes, wasip3_write_t **write_end, off_t **off);
 int __wasilibc_read_stream3(int fildes, filesystem_tuple2_stream_u8_future_result_void_error_code_t **stream, off_t **off);