diff --git a/include/ucore/iomux_waker.h b/include/ucore/iomux_waker.h new file mode 100644 index 0000000..e5e6c8d --- /dev/null +++ b/include/ucore/iomux_waker.h @@ -0,0 +1,73 @@ +#ifndef UC_IOMUX_WAKE_H_ +#define UC_IOMUX_WAKE_H_ + +#include +#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 + diff --git a/src/iomux_waker.c b/src/iomux_waker.c new file mode 100644 index 0000000..172b1fa --- /dev/null +++ b/src/iomux_waker.c @@ -0,0 +1,121 @@ +#include +#include +#include +#include +#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; +} +