Add IOMuxWaker

include/ucore/iomux_waker.h

@@ -0,0 +1,73 @@
+#ifndef UC_IOMUX_WAKE_H_
+#define UC_IOMUX_WAKE_H_
+
+#include <pthread.h>
+#include "iomux.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+struct IOMuxWaker;
+typedef void (*uc_waker_cb)(struct IOMuxWaker *wk);
+
+/** These are primitives to wake up an IOMux.
+ * An IOMux event loop can be woken up from e.g. an unix signal handler
+ * or another thread. The IOMuxWaker itself is not thread safe, it's members
+ * must not be accessed from anything other than the loop it's connected to.
+ * Only uc_iomux_waker_signal() can be called from another thread/signal handler
+ * than the IOMux it's connected to.
+ * 
+ * Note that several signals to wake up a loop can get merged, the callback
+ * handler can be called only once for several signals.
+ */
+struct IOMuxWaker {
+    //private fields
+    int fds[2];
+    struct IOMuxFD read_fd;
+    uc_waker_cb wake_cb;
+
+    //public fields
+    struct IOMux *mux;
+    void *cookie;
+};
+
+/** Initialize an IOMuxWaker.
+ * Must be called from the thread running the IOMux.
+ *
+ * @mux the IOMux to be woken up.
+ * @wk the waker to initialize
+ * @cb callback to invoke in the mux when signalled
+ * @return 0 on success, errno value if initializing fails
+ */
+int uc_iomux_waker_init(struct IOMux *mux, struct IOMuxWaker *wk, uc_waker_cb cb);
+
+/** Destroy an IOMuxWaker.
+ *  Must previously been successfully initialized.
+ *  Must be called from the thread running the IOMux.
+ *  Care must be taken so the IOMuxWaker is destroyed
+ *  without race conditions with a thread that wants to _signal the
+ *  waker.
+ * 
+ * @param wk
+ * @return 0 on success, errno on failure (Things are seriously
+ *         wrong if this fails)
+ */
+int uc_iomux_waker_destroy(struct IOMuxWaker *wk);
+
+/** Signal an IOMux to wake up and run its callback handler.
+ * Is async signal safe.
+ * Several signals in a short time may result in the callback
+ * being run only once.
+ * 
+ * @param wk 
+ * @return 0 on success, errno on failure (Things are seriously
+ *         wrong if this fails)
+ */
+int uc_iomux_waker_signal(struct IOMuxWaker *wk);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
+

src/iomux_waker.c

@@ -0,0 +1,121 @@
+#include <errno.h>
+#include <string.h>
+#include <unistd.h>
+#include <assert.h>
+#include "ucore/iomux_waker.h"
+
+static void uc_waker_pipe_cb(struct IOMux *mux, struct IOMuxFD *fd, unsigned int event)
+{
+    char buf[259];
+    struct IOMuxWaker *wk;
+    int rc;
+
+    assert(event & MUX_EV_READ);
+    assert((event & MUX_EV_WRITE) == 0);
+    wk = fd->cookie_ptr;
+
+    //we want to to drain the pipe and process all
+    //events
+    do {
+        rc = read(fd->fd, buf, sizeof buf);
+
+        if (rc < (int)sizeof buf) { //avoid unneeded read()s
+            break;
+        }
+
+    } while (rc > 0); // if we get EINTR, we rather handle it next round
+
+    if (rc == -1 && errno != EWOULDBLOCK && errno != EINTR) {
+        assert(0);
+    }
+
+    wk->wake_cb(wk);
+}
+
+static int uc_install_wake_handler(struct IOMuxWaker *wk)
+{
+    struct IOMuxFD *fd = &wk->read_fd;
+
+    memset(fd, 0, sizeof *fd);
+
+    fd->fd = wk->fds[0];
+    fd->callback = uc_waker_pipe_cb;
+    fd->cookie_ptr = wk;
+    fd->what = MUX_EV_READ;
+
+    return iomux_register_fd(wk->mux, fd);
+}
+
+int uc_iomux_waker_init(struct IOMux *mux, struct IOMuxWaker *wk, uc_waker_cb cb)
+{
+    int rc;
+
+    assert(wk != NULL);
+    assert(cb != NULL);
+
+    memset(wk, 0 , sizeof *wk);
+
+    rc = pipe(wk->fds);
+    if (rc != 0) {
+        return errno;
+    }
+    
+    wk->mux = mux;
+    wk->wake_cb = cb;
+
+    if (uc_install_wake_handler(wk) == 0) {
+        return 0;
+    } else {
+        int saved_errno;
+
+        saved_errno = errno;
+        close(wk->fds[0]);
+        close(wk->fds[1]);
+
+        return saved_errno;
+    }
+}
+
+int uc_iomux_waker_destroy(struct IOMuxWaker *wk)
+{
+    int rc;
+
+    assert(wk != NULL);
+
+    close(wk->fds[0]);
+    close(wk->fds[1]);
+
+    rc = iomux_unregister_fd(wk->mux, &wk->read_fd);
+    assert(rc == 0);
+
+    wk->mux = NULL;
+    wk->wake_cb = NULL;
+    wk->fds[0] = wk->fds[1] = -1;
+
+    assert(rc == 0);
+
+    return rc;
+}
+
+int uc_iomux_waker_signal(struct IOMuxWaker *wk)
+{
+    int rc;
+    static const char x = 'X';
+
+    assert(wk != NULL);
+
+    do {
+        rc = write(wk->fds[1], &x, 1);
+    } while (rc == -1 && errno == EINTR);
+
+    //if we would have blocked, the pipe is already
+    //signalled and it should wake up anyhow
+    if (rc == -1 && errno != EWOULDBLOCK) {
+        return errno;
+    } else if (rc == 0) { //Can't happen
+        assert(0);
+    }
+
+    return 0;
+}
+