Add internal locks for I/O primitives

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

@@ -12,6 +12,7 @@
 #include <wasi/file_utils.h>
 #include <common/errors.h>
 #include <string.h>
+#include "lock.h"
 #endif
 
 ssize_t read(int fildes, void *buf, size_t nbyte) {
@@ -33,6 +34,7 @@ ssize_t read(int fildes, void *buf, size_t nbyte) {
     wasi_read_t read;
     if (entry.vtable->get_read_stream(entry.data, &read) < 0)
       return -1;
+    defer STRONG_UNLOCK(*read.state->lock);
     return __wasilibc_read(&read, buf, nbyte);
   }
   if (entry.vtable->recvfrom)

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

@@ -12,6 +12,7 @@
 #include <wasi/file_utils.h>
 #include <common/errors.h>
 #include <time.h>
+#include "lock.h"
 #endif
 
 ssize_t write(int fildes, const void *buf, size_t nbyte) {
@@ -34,6 +35,7 @@ ssize_t write(int fildes, const void *buf, size_t nbyte) {
     wasi_write_t write;
     if (entry.vtable->get_write_stream(entry.data, &write) < 0)
       return -1;
+    defer STRONG_UNLOCK(*write.state->lock);
     return __wasilibc_write(&write, buf, nbyte);
   }
   if (entry.vtable->sendto)

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

@@ -13,6 +13,8 @@
 
 #ifdef __wasip3__
 
+#include "lock.h"
+
 /// 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.
@@ -26,6 +28,9 @@
 /// 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)
+/// A thread is actively blocked in an I/O operation and no other operation
+/// can be performed.
+#define WASIP3_IO_BLOCKED (1 << 5)
 
 /// Helper structure to package up state related to a wasip3 `stream<u8>`.
 ///
@@ -41,16 +46,43 @@ typedef struct wasip3_io_state_t {
   size_t buf_start;
   /// End of `buf` that has data in-flight or ready.
   size_t buf_end;
+
+#ifdef _REENTRANT
+  /// Pointer to the containing object's lock, which protects all the state
+  /// above. This is acquired and held through all I/O operations, but dropped
+  /// while an operation that may indefinitely block is in flight.
+  volatile int *lock;
+  /// An internal lock used to hold a queue of threads waiting to do blocking
+  /// I/O while some other thread is already doing blocking I/O.
+  DECLARE_STRONG_LOCK(blocking_lock);
+  /// Number of threads that are waiting on `blocking_lock` to take their turn
+  /// doing blocking I/O. This count notably disallows a concurrent
+  /// shutdown/reset/etc of this stream, for example `shutdown` fails while
+  /// something is blocking and so does `seek`.
+  size_t blocking_waiters;
+#endif
 } 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) {
+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;
 }
 
+/// Initializes `state` with the `stream` provided, using `lock` — the
+/// containing object's lock — to protect the state's fields.
+#ifdef _REENTRANT
+#define wasip3_io_state_init(state, stream, lock_ptr)                          \
+  do {                                                                         \
+    __wasip3_io_state_init(state, stream);                                     \
+    (state)->lock = &(lock_ptr);                                               \
+  } while (0)
+#else
+#define wasip3_io_state_init(state, stream, lock)                              \
+  __wasip3_io_state_init(state, stream)
+#endif
+
 /// 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;
@@ -60,6 +92,7 @@ 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) {
+  STRONG_ASSERT_EMPTY(state->blocking_lock);
   if (state->flags & WASIP3_IO_INPROGRESS)
     filesystem_stream_u8_cancel_read(state->stream);
   if (state->buf)
@@ -73,6 +106,7 @@ static inline void wasip3_read_state_close(wasip3_io_state_t *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) {
+  STRONG_ASSERT_EMPTY(state->blocking_lock);
   if (state->flags & WASIP3_IO_INPROGRESS)
     filesystem_stream_u8_cancel_write(state->stream);
   if (state->buf)
@@ -81,6 +115,17 @@ static inline void wasip3_write_state_close(wasip3_io_state_t *state) {
     filesystem_stream_u8_drop_writable(state->stream);
   memset(state, 0, sizeof(*state));
 }
+
+/// Tests whether there is active or pending blocking I/O on this stream which
+/// notably should prevent it from being closed.
+static inline bool wasip3_io_state_blocked(wasip3_io_state_t *state) {
+#ifdef _REENTRANT
+  return (state->flags & WASIP3_IO_BLOCKED) || state->blocking_waiters > 0;
+#else
+  (void)state;
+  return false;
+#endif
+}
 #endif
 
 // Metadata for WASI reads which is used to delegate to `__wasilibc_read(...)`
@@ -149,6 +194,11 @@ typedef struct descriptor_vtable_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.
+  ///
+  /// When threading is enabled for WASIp3+ this additionally acquires the
+  /// internal lock for the object and the caller must unlock the
+  /// `read->state->lock` object on success. On failure no locks are held when
+  /// this returns.
   int (*get_read_stream)(void *, wasi_read_t *);
   /// Same as `get_read_stream`, but for output streams.
   int (*get_write_stream)(void *, wasi_write_t *);

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

@@ -57,6 +57,11 @@ __wasilibc_poll_add_output_stream(poll_state_t *state,
 
 #ifdef __wasip3__
 
+/// Callback invoked after `__wasilibc_poll_add` when an event is ready.
+///
+/// This additionally serves as a destructor of sorts where if no event arrives
+/// before the `poll` is torn down then this is invoked with `state` and
+/// `event` as NULL.
 typedef void (*poll_ready_t)(void *data, poll_state_t *state,
                              wasip3_event_t *event);
 

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

@@ -5,6 +5,7 @@
 
 #ifndef __wasip1__
 
+#include "lock.h"
 #include <sys/socket.h>
 #include <wasi/sockets_utils.h>
 
@@ -33,6 +34,10 @@ typedef struct {
   sockets_method_tcp_socket_connect_args_t args;
   sockets_result_void_error_code_t result;
   wasip3_subtask_t subtask;
+  /// Whether `subtask` is currently registered in some `poll`'s waitable-set.
+  /// Subtasks can be in at most one waitable-set at a time, so while this is
+  /// set no other `poll` can register this socket.
+  bool polling;
 #endif
 } tcp_socket_state_connecting_t;
 
@@ -43,11 +48,16 @@ typedef struct {
 #define TCP_LISTENING_ACCEPTING (1 << 1)
 /// The `accept_result` field is valid and ready to be processed.
 #define TCP_LISTENING_ACCEPT_READY (1 << 2)
+/// Whether or not a thread is blocked in `accept` on this socket meaning that
+/// the stream is not suitable for any other operations.
+#define TCP_LISTENING_BLOCKING (1 << 3)
 
 typedef struct {
 #ifdef __wasip2__
   int dummy;
 #else
+  /// Queue for concurrent blocking calls to accept to wait on.
+  DECLARE_STRONG_LOCK(blocking_lock);
   // The `stream<tcp-socket>` that this is reading to receive accepted sockets.
   sockets_stream_own_tcp_socket_t stream;
   /// In-flight result of the read of `stream`.
@@ -104,6 +114,7 @@ typedef struct {
 
 typedef struct {
   descriptor_refcnt_t refcnt;
+  DECLARE_STRONG_LOCK(lock);
   sockets_own_tcp_socket_t socket;
   tcp_socket_state_t state;
 #ifdef __wasip2__

libc-bottom-half/sources/descriptor_table.c

@@ -15,9 +15,6 @@
 
 #define MINSIZE 8
 
-// FIXME: Remove after #825 lands
-#define STRONG_ASSERT_HELD(lock) ((void)0)
-
 typedef struct {
   bool occupied;
   descriptor_table_entry_t entry;

libc-bottom-half/sources/file.c

@@ -1,16 +1,18 @@
+#include "libc/sys/stat/stat_impl.h"
+#include "lock.h"
 #include <assert.h>
 #include <common/errors.h>
 #include <errno.h>
 #include <fcntl.h>
+#include <stddefer.h>
 #include <wasi/descriptor_table.h>
 #include <wasi/file.h>
 #include <wasi/wasip2.h>
 #include <wasi/wasip3_block.h>
 
-#include "libc/sys/stat/stat_impl.h"
-
 typedef struct {
   descriptor_refcnt_t refcnt;
+  DECLARE_STRONG_LOCK(lock);
   filesystem_own_descriptor_t file_handle;
   // Current position in stream, relative to the beginning of the
   // *file_handle*, measured in bytes
@@ -36,6 +38,7 @@ typedef struct {
 
 static void file_close_streams(void *data) {
   file_t *file = (file_t *)data;
+
 #ifdef __wasip2__
   if (file->read_pollable.__handle != 0) {
     poll_pollable_drop_own(file->read_pollable);
@@ -69,6 +72,7 @@ static void file_close_streams(void *data) {
 
 static void file_free(void *data) {
   file_t *file = (file_t *)data;
+  STRONG_ASSERT_EMPTY(file->lock);
   file_close_streams(data);
   filesystem_descriptor_drop_own(file->file_handle);
   free(file);
@@ -77,6 +81,7 @@ static void file_free(void *data) {
 #ifndef __wasip2__
 static int file_read_eof(void *data) {
   file_t *file = (file_t *)data;
+  STRONG_ASSERT_HELD(file->lock);
 
   if (file->read_result != 0) {
     filesystem_result_void_error_code_t result;
@@ -96,6 +101,10 @@ static int file_read_eof(void *data) {
 
 static int file_get_read_stream(void *data, wasi_read_t *read) {
   file_t *file = (file_t *)data;
+  STRONG_LOCK(file->lock);
+  // .. intentionally don't unlock `file->lock` as this function lets the
+  // caller do that (see `descriptor_table.h` for details of this callback).
+
 #ifdef __wasip2__
   if (file->read_stream.__handle == 0) {
     filesystem_error_code_t error_code;
@@ -115,7 +124,7 @@ static int file_get_read_stream(void *data, wasi_read_t *read) {
     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, &result);
-    wasip3_io_state_init(&file->read, result.f0);
+    wasip3_io_state_init(&file->read, result.f0, file->lock);
     file->read_result = result.f1;
   }
   read->state = &file->read;
@@ -131,6 +140,7 @@ static int file_get_read_stream(void *data, wasi_read_t *read) {
 #ifndef __wasip2__
 static int file_write_eof(void *data) {
   file_t *file = (file_t *)data;
+  STRONG_ASSERT_HELD(file->lock);
 
   if (file->write_result) {
     filesystem_result_void_error_code_t result;
@@ -152,6 +162,10 @@ static int file_write_eof(void *data) {
 
 static int file_get_write_stream(void *data, wasi_write_t *write) {
   file_t *file = (file_t *)data;
+  STRONG_LOCK(file->lock);
+  // .. intentionally don't unlock `file->lock` as this function lets the
+  // caller do that (see `descriptor_table.h` for details of this callback).
+
 #ifdef __wasip2__
   if (file->write_stream.__handle == 0) {
     filesystem_error_code_t error_code;
@@ -186,7 +200,7 @@ static int file_get_write_stream(void *data, wasi_write_t *write) {
           filesystem_borrow_descriptor(file->file_handle), write_read,
           file->offset);
     }
-    wasip3_io_state_init(&file->write, writer);
+    wasip3_io_state_init(&file->write, writer, file->lock);
   }
   write->state = &file->write;
   write->eof = file_write_eof;
@@ -230,6 +244,7 @@ static int file_fstat(void *data, struct stat *buf) {
 }
 
 static int file_seek_end(file_t *file) {
+  STRONG_ASSERT_HELD(file->lock);
   filesystem_descriptor_stat_t stat;
   filesystem_error_code_t error;
   bool ok = filesystem_method_descriptor_stat(
@@ -245,6 +260,22 @@ static int file_seek_end(file_t *file) {
 static off_t file_seek(void *data, off_t offset, int whence) {
   file_t *file = (file_t *)data;
 
+  STRONG_LOCK(file->lock);
+  defer STRONG_UNLOCK(file->lock);
+
+#ifndef __wasip2__
+  // If another thread is blocked in an I/O operation on this file then
+  // disallow this concurrent operation: `file_close_streams` below can't
+  // cancel a blocking operation in another thread, and supporting the seek
+  // would additionally require synchronizing updates to `offset` which
+  // currently isn't done.
+  if (wasip3_io_state_blocked(&file->read) ||
+      wasip3_io_state_blocked(&file->write)) {
+    errno = EOPNOTSUPP;
+    return -1;
+  }
+#endif
+
   // If this file is in append mode, reset our knowledge of the current cursor
   // to the current end of the file.
   if ((file->oflag & O_APPEND) && file_seek_end(file) < 0)
@@ -283,6 +314,9 @@ static off_t file_seek(void *data, off_t offset, int whence) {
 
 static int file_set_blocking(void *data, bool blocking) {
   file_t *file = (file_t *)data;
+  STRONG_LOCK(file->lock);
+  defer STRONG_UNLOCK(file->lock);
+
   if (blocking)
     file->oflag &= ~O_NONBLOCK;
   else
@@ -292,6 +326,8 @@ static int file_set_blocking(void *data, bool blocking) {
 
 static int file_fcntl_getfl(void *data) {
   file_t *file = (file_t *)data;
+  STRONG_LOCK(file->lock);
+  defer STRONG_UNLOCK(file->lock);
 
   // Get the flags of the descriptor
   filesystem_descriptor_flags_t flags;
@@ -321,6 +357,9 @@ static int file_fcntl_getfl(void *data) {
 
 static int file_fcntl_setfl(void *data, int flags) {
   file_t *file = (file_t *)data;
+  STRONG_LOCK(file->lock);
+  defer STRONG_UNLOCK(file->lock);
+
   flags &= O_NONBLOCK | O_APPEND;
   file->oflag = (file->oflag & ~(O_NONBLOCK | O_APPEND)) | flags;
   return 0;