support non-blocking wasip2 sockets in read and write (#734)

For descriptors which have `sendto` and `recvfrom` in their vtables, use
those for writing and reading respectively since they know how to handle
non-blocking descriptors.

In the future, we could generalize this to support non-blocking
descriptors which aren't sockets, also.

Author: dicej

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

@@ -28,6 +28,13 @@ 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)
+    return entry->vtable->recvfrom(entry->data, buf, nbyte, 0, NULL, NULL);
+
   bool ok = false;
 
   // Translate the file descriptor to an internal handle

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

@@ -29,11 +29,17 @@ 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;
-  descriptor_table_entry_t* entry = 0;
 
   // Translate the file descriptor to an internal handle
   off_t *off;

test/src/poll-connect.c

@@ -2,7 +2,10 @@
 #include <errno.h>
 #include <netinet/in.h>
 #include <poll.h>
+#include <stdlib.h>
+#include <sys/ioctl.h>
 #include <sys/socket.h>
+#include <unistd.h>
 
 #define TEST(c)                                                                \
   do {                                                                         \
@@ -73,6 +76,9 @@ int main() {
   // finishes accepting.
   int loop_count = 0;
   while ((server_client == -1 || fds[0].events) && loop_count < 10) {
+    if (t_status)
+      exit(t_status);
+
     ++loop_count;
 
     int fd_count = 1;
@@ -101,23 +107,97 @@ int main() {
   }
   ASSERT(loop_count < 10);
 
-  // Send some data from the server to the client.
-  uint8_t data[5] = {1, 2, 3, 4, 5};
-  TEST(send(server_client, data, sizeof(data), 0) == sizeof(data));
+  int one = 1;
+  TEST(ioctl(server_client, FIONBIO, &one) != -1);
+
+  // Test that we can `recv` what we `send`
+  {
+    // Send some data from the server to the client.
+    uint8_t data[5] = {1, 2, 3, 4, 5};
+    TEST(send(server_client, data, sizeof(data), 0) == sizeof(data));
+
+    fds[0].events = POLLRDNORM;
+    errno = 0;
+    int count = poll(fds, 1, 100);
+    TEST2(count == 1);
+    // Now the server has sent something, so the client socket should be
+    // readable.
+    ASSERT((fds[0].revents & POLLRDNORM) != 0);
+
+    uint8_t received[sizeof(data)];
+    TEST(recv(client, received, sizeof(data), 0) == sizeof(data));
+
+    // Assert that what was received matches what was sent.
+    for (int i = 0; i < sizeof(data); ++i) {
+      ASSERT(received[i] == data[i]);
+    }
+  }
 
-  fds[0].events = POLLRDNORM;
-  errno = 0;
-  int count = poll(fds, 1, 100);
-  TEST2(count == 1);
-  // Now the server has sent something, so the client socket should be readable.
-  ASSERT((fds[0].revents & POLLRDNORM) != 0);
+  // Test that we can `read` what we `write` and that neither of them block.
+  {
+    size_t data_len = 16 * 1024;
+    errno = 0;
+    uint8_t *received = malloc(data_len);
+    TEST2(received != NULL);
+
+    // Nothing should be readable, yet.
+    TEST(read(client, received, data_len) == -1 &&
+         (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINPROGRESS));
+
+    errno = 0;
+    uint8_t *data = malloc(data_len);
+    TEST2(data != NULL);
+
+    for (size_t i = 0; i < data_len; ++i) {
+      data[i] = 42;
+    }
 
-  uint8_t received[sizeof(data)];
-  TEST(recv(client, received, sizeof(data), 0) == sizeof(data));
+    // Write until we hit backpressure
+    ssize_t write_total = 0;
+    while (1) {
+      if (t_status)
+        exit(t_status);
 
-  // Assert that what was received matches what was sent.
-  for (int i = 0; i < sizeof(data); ++i) {
-    ASSERT(received[i] == data[i]);
+      errno = 0;
+      ssize_t count = write(server_client, data, data_len);
+      if (count == -1) {
+        TEST2(errno == EAGAIN || errno == EWOULDBLOCK || errno == EINPROGRESS);
+        break;
+      } else {
+        write_total += count;
+      }
+    }
+
+    fds[0].events = POLLRDNORM;
+
+    // Read what we wrote
+    ssize_t read_total = 0;
+    while (1) {
+      if (t_status)
+        exit(t_status);
+
+      {
+        errno = 0;
+        int count = poll(fds, 1, 100);
+        TEST2(count == 1);
+        ASSERT((fds[0].revents & POLLRDNORM) != 0);
+      }
+
+      errno = 0;
+      ssize_t count = read(client, received, data_len);
+      TEST2(count != -1 || errno == EAGAIN || errno == EWOULDBLOCK ||
+            errno == EINPROGRESS);
+
+      if (count > 0) {
+        read_total += count;
+        for (ssize_t i = 0; i < count; ++i) {
+          ASSERT(received[i] == 42);
+        }
+        if (read_total == data_len) {
+          break;
+        }
+      }
+    }
   }
 
   return t_status;