wasip3: Fix fallback code when buffering writes (#788)

This commit fixes mistakes from #782 where the fallback code for
buffering writes wasn't handled properly. Specifically when a completion
event came in through `poll` it didn't fully update the internal state
of the stream, such as kicking off further writes or cleaning up the
internal buffer. This is fixed by refactoring the logic already present
during a normal `write` to get executed in this event.

Testing this is unfortunately not trivial. There aren't really all that
many streams we have access to natively in Wasmtime, and TCP is "well
behaved" and doesn't execute any of these fallbacks anyway. The closest
way to test this is filesystem-based streams, but Wasmtime's default
behavior is to lie and just do blocking operations on these streams
regardless. Through changes in Wasmtime, and flags to Wasmtime, however,
it's possible to Wasmtime in a mode where filesystem streams are "ill
behaved" where zero-length reads don't signal readiness, triggering
these fallback paths. With those changes the added test here, and minor
additionally support for nonblocking reads/writes, exercise these paths
and triggers the previous bug.

Author: alexcrichton

libc-bottom-half/sources/file.c

@@ -123,7 +123,7 @@ static int file_get_read_stream(void *data, wasi_read_t *read) {
 #endif
   read->offset = &file->offset;
   read->timeout = 0;
-  read->blocking = true;
+  read->blocking = (file->oflag & O_NONBLOCK) == 0;
   return 0;
 }
 
@@ -193,7 +193,7 @@ static int file_get_write_stream(void *data, wasi_write_t *write) {
 #endif
   write->offset = &file->offset;
   write->timeout = 0;
-  write->blocking = true;
+  write->blocking = (file->oflag & O_NONBLOCK) == 0;
   return 0;
 }
 

libc-bottom-half/sources/file_utils.c

@@ -121,6 +121,45 @@ static size_t wasip3_io_update_event(wasip3_io_state_t *state,
   return wasip3_io_update_code(state, event->code);
 }
 
+/// When `event` has happened due to a completion of a pending write, this
+/// function will advance `state` forward.
+///
+/// This attempts to perform any follow-up writes as necessary if the pending
+/// write ended up coming in short. Additionally this will clear out the
+/// internal buffered data once it reaches
+/// completion.
+///
+/// Returns `true` if this stream is ready for more writes, or `false` if
+/// there's still a pending write in-flight.
+static bool wasip3_advance_pending_write(wasip3_io_state_t *state,
+                                         wasip3_event_t *event) {
+  // Update the I/O internal state given the result of the write.
+  state->buf_start += wasip3_io_update_event(state, event);
+
+  // While there's remaining writes to perform, kick those off here. Once a
+  // write blocks we bail out of this loop as there's I/O in-progress.
+  while (!(state->flags & WASIP3_IO_DONE) &&
+         state->buf_start != state->buf_end) {
+    wasip3_waitable_status_t status =
+        filesystem_stream_u8_write(state->stream, state->buf + state->buf_start,
+                                   state->buf_end - state->buf_start);
+    state->flags |= WASIP3_IO_INPROGRESS;
+    if (status == WASIP3_WAITABLE_STATUS_BLOCKED)
+      return false;
+    state->buf_start += wasip3_io_update_code(state, status);
+  }
+
+  // Everything should be done now at this point, meaning that the stream is
+  // closed or we've written the entire buffer. Clean up internal state and
+  // return to indicate there's no more pending I/O.
+  assert((state->flags & WASIP3_IO_DONE) || state->buf_start == state->buf_end);
+  free(state->buf);
+  state->buf = NULL;
+  state->buf_start = 0;
+  state->buf_end = 0;
+  return true;
+}
+
 /// Attempts to resolve any pending write that may be in-progress on `write`.
 ///
 /// This may notably end up issuing more writes to finish a buffered write that
@@ -154,40 +193,19 @@ static int wasip3_write_resolve_pending(wasi_write_t *write) {
       }
     }
 
-    // Update the I/O internal state given the result of the write.
-    state->buf_start += wasip3_io_update_event(state, &event);
-
-    // While there's remaining writes to perform, kick those off here. If a
-    // write is blocked then nonblocking mode returns as such and blocking
-    // mode breaks out to turn this outer `while (1)` loop again to block
-    // on the result. If the write finishes immediately then state is updated
-    // again and then further continues.
-    while (!(state->flags & WASIP3_IO_DONE) &&
-           state->buf_start != state->buf_end) {
-      wasip3_waitable_status_t status = filesystem_stream_u8_write(
-          state->stream, state->buf + state->buf_start,
-          state->buf_end - state->buf_start);
-      state->flags |= WASIP3_IO_INPROGRESS;
-      if (status == WASIP3_WAITABLE_STATUS_BLOCKED) {
-        if (write->blocking) {
-          break;
-        } else {
-          errno = EWOULDBLOCK;
-          return -1;
-        }
-      }
-      state->buf_start += wasip3_io_update_code(state, status);
-    }
-
-    // We either broke out of the `while` loop normally, or we hit the `break`
-    // in the loop which wants to continue to the top of this outer loop. Test
-    // which it is here and act accordingly.
-    if ((state->flags & WASIP3_IO_DONE) || state->buf_start == state->buf_end) {
-      free(state->buf);
-      state->buf = NULL;
-      state->buf_start = 0;
-      state->buf_end = 0;
+    // Update the internal status of this stream with the `event` we have now
+    // learned. If the stream is complete at this point then go ahead and
+    // return.
+    if (wasip3_advance_pending_write(state, &event))
       return 0;
+
+    // If the write isn't blocking then a pending I/O op is kicked off from
+    // above and there's no point in turning the loop and re-polling. Bail out
+    // here with EWOULDBLOCK.
+    if (!write->blocking) {
+      assert(state->flags & WASIP3_IO_INPROGRESS);
+      errno = EWOULDBLOCK;
+      return -1;
     }
   }
 
@@ -684,8 +702,15 @@ static void wasip3_poll_read_ready(void *data, poll_state_t *state,
 static void wasip3_poll_write_ready(void *data, poll_state_t *state,
                                     wasip3_event_t *event) {
   wasip3_io_state_t *iostate = (wasip3_io_state_t *)data;
-  iostate->buf_start += wasip3_io_update_event(iostate, event);
-  __wasilibc_poll_ready(state, POLLWRNORM);
+
+  // Update our state with this event, and if the pending write is fully
+  // complete then this is now ready for writing again. Otherwise there's still
+  // a pending write so our job is complete trying to advance things a bit.
+  if (wasip3_advance_pending_write(iostate, event)) {
+    __wasilibc_poll_ready(state, POLLWRNORM);
+  } else {
+    assert(iostate->flags & WASIP3_IO_INPROGRESS);
+  }
 }
 
 static int wasip3_stream_poll(wasip3_io_state_t *iostate, poll_state_t *state,

test/CMakeLists.txt

@@ -133,7 +133,7 @@ endfunction()
 function(register_test test_name executable_name)
   set(options FS NETWORK SETJMP)
   set(oneValueArgs CLIENT PASS_REGULAR_EXPRESSION)
-  set(multiValueArgs ARGV ENV)
+  set(multiValueArgs ARGV ENV ENGINE_ARG)
   cmake_parse_arguments(PARSE_ARGV 1 arg "${options}" "${oneValueArgs}" "${multiValueArgs}")
 
   set(wasmtime_args)
@@ -148,6 +148,9 @@ function(register_test test_name executable_name)
   foreach(env IN LISTS arg_ENV)
     list(APPEND wasmtime_args --env ${env})
   endforeach()
+  foreach(arg IN LISTS arg_ENGINE_ARG)
+    list(APPEND wasmtime_args ${arg})
+  endforeach()
   if (TARGET_TRIPLE MATCHES "-threads")
     list(APPEND wasmtime_args --wasi threads --wasm shared-memory)
   endif()
@@ -327,6 +330,11 @@ add_wasilibc_test(fsync.c FS)
 add_wasilibc_test(ftruncate.c FS)
 add_wasilibc_test(fts.c FS)
 add_wasilibc_test(fwscanf.c FS)
+# Note that `-Wtimeout` here is passed to force Wasmtime to avoid blocking the
+# main thread when doing file I/O. This is used to exercise this test's
+# behavior where file streams use fallback code for nonblocking reads/writes.
+add_wasilibc_test(fs-nonblocking.c FS ENGINE_ARG -Wtimeout=10s)
+set_tests_properties(fs-nonblocking.wasm PROPERTIES LABELS v8fail)
 add_wasilibc_test(getentropy.c)
 add_wasilibc_test(hello.c PASS_REGULAR_EXPRESSION "Hello, World!")
 add_wasilibc_test(ioctl.c FS)

test/src/fs-nonblocking.c

@@ -0,0 +1,56 @@
+#include "test.h"
+#include <dirent.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <poll.h>
+#include <stdbool.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/stat.h>
+#include <unistd.h>
+
+#define TEST(c)                                                                \
+  do {                                                                         \
+    errno = 0;                                                                 \
+    if (!(c))                                                                  \
+      t_error("%s failed (errno = %d)\n", #c, errno);                          \
+  } while (0)
+
+int main() {
+  int fd;
+  TEST((fd = open("howdy.txt", O_RDWR | O_NONBLOCK | O_CREAT, 0755)) > 0);
+
+  int rc, size = 0;
+
+  struct pollfd pollfd;
+  int max = 32 * 1024;
+  pollfd.fd = fd;
+
+  pollfd.events = POLLWRNORM;
+  for (int i = 0; size < max && i < 100; i++) {
+    while (size < max && (rc = write(fd, "hello", 5)) > 0)
+      size += rc;
+    TEST(poll(&pollfd, 1, -1) != -1);
+  }
+
+  TEST(size > 0);
+
+  struct stat stat;
+  TEST(fstat(fd, &stat) == 0);
+  TEST(stat.st_size == size);
+
+  TEST(lseek(fd, 0, SEEK_SET) != -1);
+
+  char buf[5];
+  pollfd.events = POLLRDNORM;
+  int remaining = size;
+  while (remaining) {
+    while ((rc = read(fd, buf, sizeof(buf))) > 0)
+      remaining -= rc;
+    TEST(poll(&pollfd, 1, -1) != -1);
+  }
+
+  TEST(close(fd) != -1);
+  return t_status;
+}