[LibOS] Fix *_event functions

They all lacked error checking and `wait_event` was completely broken:
it was reading from non-blocking pipe and treating EAGAIN as
successfully waited-for event.

LibOS/shim/include/shim_internal.h

@@ -305,13 +305,19 @@ static inline void enable_preempt(shim_tcb_t* tcb) {
     __enable_preempt(tcb);
 }
 
-static inline void create_event(AEVENTTYPE* e) {
-    if (!e->event)
-        e->event = DkStreamOpen(URI_PREFIX_PIPE, PAL_ACCESS_RDWR, 0, 0, PAL_OPTION_NONBLOCK);
-}
-
-static inline bool event_created(AEVENTTYPE* e) {
-    return e->event != NULL;
+/*
+ * These events have counting semaphore semantics:
+ * - `set_event(e, n)` increases value of the semaphore by `n`,
+ * - `wait_event(e)` decreases value by 1 (blocking if it's 0),
+ * - `clear_event(e)` decreases value to 0, without blocking - this operation is not atomic.
+ * Note that using `clear_event` probably requires external locking to avoid races.
+ */
+static inline int create_event(AEVENTTYPE* e) {
+    e->event = DkStreamOpen(URI_PREFIX_PIPE, PAL_ACCESS_RDWR, 0, 0, 0);
+    if (!e->event) {
+        return -PAL_ERRNO();
+    }
+    return 0;
 }
 
 static inline PAL_HANDLE event_handle(AEVENTTYPE* e) {
@@ -325,31 +331,90 @@ static inline void destroy_event(AEVENTTYPE* e) {
     }
 }
 
-static inline void set_event(AEVENTTYPE* e, int n) {
-    if (e->event) {
-        char bytes[n];
-        DkStreamWrite(e->event, 0, n, bytes, NULL);
+static inline int set_event(AEVENTTYPE* e, size_t n) {
+    /* TODO: this should be changed into an assert, once we make sure it does not happen (old
+     * version handled it). */
+    if (!e->event) {
+        return -EINVAL;
     }
+
+    char bytes[n];
+    memset(bytes, '\0', n);
+    while (n > 0) {
+        PAL_NUM ret = DkStreamWrite(e->event, 0, n, bytes, NULL);
+        if (ret == PAL_STREAM_ERROR) {
+            int err = PAL_ERRNO();
+            if (err == EINTR || err == EAGAIN || err == EWOULDBLOCK) {
+                continue;
+            }
+            return -err;
+        }
+        n -= ret;
+    }
+
+    return 0;
 }
 
-static inline void wait_event(AEVENTTYPE* e) {
-    if (e->event) {
+static inline int wait_event(AEVENTTYPE* e) {
+    /* TODO: this should be changed into an assert, once we make sure it does not happen (old
+     * version handled it). */
+    if (!e->event) {
+        return -EINVAL;
+    }
+
+    int err = 0;
+    do {
         char byte;
-        int n = 0;
-        do {
-            n = DkStreamRead(e->event, 0, 1, &byte, NULL, 0);
-        } while (!n);
-    }
+        PAL_NUM ret = DkStreamRead(e->event, 0, 1, &byte, NULL, 0);
+        err = ret == PAL_STREAM_ERROR ? PAL_ERRNO() : 0;
+    } while (err == EINTR || err == EAGAIN || err == EWOULDBLOCK);
+
+    return -err;
 }
 
-static inline void clear_event(AEVENTTYPE* e) {
-    if (e->event) {
-        char bytes[100];
-        int n;
-        do {
-            n = DkStreamRead(e->event, 0, 100, bytes, NULL, 0);
-        } while (n == 100);
+static inline int clear_event(AEVENTTYPE* e) {
+    /* TODO: this should be changed into an assert, once we make sure it does not happen (old
+     * version handled it). */
+    if (!e->event) {
+        return -EINVAL;
     }
+
+    while (1) {
+        PAL_HANDLE handle = e->event;
+        PAL_FLG ievent = PAL_WAIT_READ;
+        PAL_FLG revent = 0;
+
+        shim_get_tcb()->pal_errno = PAL_ERROR_SUCCESS;
+        PAL_BOL ret = DkStreamsWaitEvents(1, &handle, &ievent, &revent, /*timeout=*/0);
+        if (!ret) {
+            int err = PAL_ERRNO();
+            if (err == EINTR) {
+                continue;
+            } else if (!err || err == EAGAIN || err == EWOULDBLOCK) {
+                break;
+            }
+            return -err;
+        }
+
+        /* Even if `revent` has `PAL_WAIT_ERROR` marked, let `DkSitreamRead()` report the error
+         * below. */
+        assert(revent);
+
+        char bytes[100];
+        PAL_NUM n = DkStreamRead(e->event, 0, sizeof(bytes), bytes, NULL, 0);
+        if (n == PAL_STREAM_ERROR) {
+            int err = PAL_ERRNO();
+            if (err == EINTR) {
+                continue;
+            } else if (err == EAGAIN || err == EWOULDBLOCK) {
+                /* This should not happen, since we polled above  ... */
+                break;
+            }
+            return -err;
+        }
+    }
+
+    return 0;
 }
 
 /* reference counter APIs */

LibOS/shim/src/bookkeep/shim_handle.c

@@ -803,8 +803,12 @@ BEGIN_RS_FUNC(handle) {
                 return -EINVAL;
             }
             break;
-        case TYPE_EPOLL:
-            create_event(&hdl->info.epoll.event);
+        case TYPE_EPOLL: ;
+            int ret = create_event(&hdl->info.epoll.event);
+            if (ret < 0) {
+                return ret;
+            }
+
             struct shim_epoll_item* epoll_item;
             size_t count = 0;
             LISTP_FOR_EACH_ENTRY(epoll_item, &hdl->info.epoll.fds, list) {

LibOS/shim/src/ipc/shim_ipc_helper.c

@@ -176,13 +176,17 @@ int init_ipc_helper(void) {
     if (!create_lock(&ipc_helper_lock)) {
         return -ENOMEM;
     }
-    create_event(&install_new_event);
+
+    int ret = create_event(&install_new_event);
+    if (ret < 0) {
+        return ret;
+    }
 
     /* some IPC ports were already added before this point, so spawn IPC helper thread (and enable
      * locking mechanisms if not done already since we are going in multi-threaded mode) */
     enable_locking();
     lock(&ipc_helper_lock);
-    int ret = create_ipc_helper();
+    ret = create_ipc_helper();
     unlock(&ipc_helper_lock);
 
     return ret;

LibOS/shim/src/shim_async.c

@@ -126,7 +126,10 @@ int init_async(void) {
     if (!create_lock(&async_helper_lock)) {
         return -ENOMEM;
     }
-    create_event(&install_new_event);
+    int ret = create_event(&install_new_event);
+    if (ret < 0) {
+        return ret;
+    }
 
     /* enable locking mechanisms since we are going in multi-threaded mode */
     enable_locking();

LibOS/shim/src/shim_init.c

@@ -55,7 +55,7 @@ noreturn void __abort(void) {
 }
 
 static int pal_errno_to_unix_errno[PAL_ERROR_NATIVE_COUNT + 1] = {
-    [0]                         = 0,
+    [PAL_ERROR_SUCCESS]         = 0,
     [PAL_ERROR_NOTIMPLEMENTED]  = ENOSYS,
     [PAL_ERROR_NOTDEFINED]      = ENOSYS,
     [PAL_ERROR_NOTSUPPORT]      = EACCES,

LibOS/shim/src/sys/shim_epoll.c

@@ -48,9 +48,14 @@ int shim_do_epoll_create1(int flags) {
     set_handle_fs(hdl, &epoll_builtin_fs);
     epoll->fds_count = 0;
     __atomic_store_n(&epoll->waiter_cnt, 0, __ATOMIC_RELAXED);
-    create_event(&epoll->event);
     INIT_LISTP(&epoll->fds);
 
+    int ret = create_event(&epoll->event);
+    if (ret < 0) {
+        put_handle(hdl);
+        return ret;
+    }
+
     int vfd = set_new_fd_handle(hdl, (flags & EPOLL_CLOEXEC) ? FD_CLOEXEC : 0, NULL);
     put_handle(hdl);
     return vfd;
@@ -72,6 +77,7 @@ static void notify_epoll_waiters(struct shim_epoll_handle* epoll) {
      * (`shim_do_epoll_wait`), what if one threads consumes multiple events? */
     size_t waiters = __atomic_load_n(&epoll->waiter_cnt, __ATOMIC_RELAXED);
     if (waiters) {
+        /* TODO: this needs error checking. */
         set_event(&epoll->event, waiters);
     }
 }
@@ -369,7 +375,11 @@ int shim_do_epoll_wait(int epfd, struct __kernel_epoll_event* events, int maxeve
         if (event_handle_update) {
             /* retry if epoll was updated concurrently (similar to Linux semantics) */
             unlock(&epoll_hdl->lock);
-            wait_event(&epoll->event);
+            int ret = wait_event(&epoll->event);
+            if (ret < 0) {
+                put_handle(epoll_hdl);
+                return ret;
+            }
             lock(&epoll_hdl->lock);
         } else {
             /* no need to retry, exit the while loop */