Update bundled libverto to 0.2.5

7 files changed, 363 insertions, 63 deletions

diff --git a/src/util/k5ev/verto-k5ev.c b/src/util/k5ev/verto-k5ev.c
index bd3003b04f..35c94d66ed 100644
--- a/src/util/k5ev/verto-k5ev.c
+++ b/src/util/k5ev/verto-k5ev.c
@@ -94,12 +94,40 @@ k5ev_ctx_reinitialize(verto_mod_ctx *ctx)
 static void
 libev_callback(EV_P_ ev_watcher *w, int revents)
 {
-    if (verto_get_type(w->data) == VERTO_EV_TYPE_CHILD)
+    verto_ev_flag state = VERTO_EV_FLAG_NONE;
+
+    if (verto_get_type(w->data)== VERTO_EV_TYPE_CHILD)
         verto_set_proc_status(w->data, ((ev_child*) w)->rstatus);
 
+    if (revents & EV_READ)
+        state |= VERTO_EV_FLAG_IO_READ;
+    if (revents & EV_WRITE)
+        state |= VERTO_EV_FLAG_IO_WRITE;
+    if (revents & EV_ERROR)
+        state |= VERTO_EV_FLAG_IO_ERROR;
+
+    verto_set_fd_state(w->data, state);
     verto_fire(w->data);
 }
 
+static void
+k5ev_ctx_set_flags(verto_mod_ctx *ctx, const verto_ev *ev,
+		   verto_mod_ev *evpriv)
+{
+    if (verto_get_type(ev) == VERTO_EV_TYPE_IO) {
+        int events = EV_NONE;
+
+        if (verto_get_flags(ev) & VERTO_EV_FLAG_IO_READ)
+            events |= EV_READ;
+        if (verto_get_flags(ev) & VERTO_EV_FLAG_IO_WRITE)
+            events |= EV_WRITE;
+
+        ev_io_stop(ctx, (ev_io*) evpriv);
+        ev_io_set(((ev_io*) evpriv), verto_get_fd(ev), events);
+        ev_io_start(ctx, (ev_io*) evpriv);
+    }
+}
+
 #define setuptype(type, ...) \
     w.type = malloc(sizeof(ev_ ## type)); \
     if (w.type) { \
@@ -120,17 +148,12 @@ k5ev_ctx_add(verto_mod_ctx *ctx, const verto_ev *ev, verto_ev_flag *flags)
        ev_child *child;
     } w;
     ev_tstamp interval;
-    int events = EV_NONE;
 
     w.watcher = NULL;
     *flags |= VERTO_EV_FLAG_PERSIST;
     switch (verto_get_type(ev)) {
         case VERTO_EV_TYPE_IO:
-            if (verto_get_flags(ev) & VERTO_EV_FLAG_IO_READ)
-                events |= EV_READ;
-            if (verto_get_flags(ev) & VERTO_EV_FLAG_IO_WRITE)
-                events |= EV_WRITE;
-            setuptype(io, libev_callback, verto_get_fd(ev), events);
+            setuptype(io, libev_callback, verto_get_fd(ev), EV_NONE);
         case VERTO_EV_TYPE_TIMEOUT:
             interval = ((ev_tstamp) verto_get_interval(ev)) / 1000.0;
             setuptype(timer, libev_callback, interval, interval);
@@ -145,8 +168,10 @@ k5ev_ctx_add(verto_mod_ctx *ctx, const verto_ev *ev, verto_ev_flag *flags)
             break; /* Not supported */
     }
 
-    if (w.watcher)
+    if (w.watcher) {
         w.watcher->data = (void*) ev;
+        k5ev_ctx_set_flags(ctx, ev, w.watcher);
+    }
     return w.watcher;
 }
 
diff --git a/src/util/k5ev/verto-libev.c b/src/util/k5ev/verto-libev.c
index 4e6e816b9f..9c7c32449c 100644
--- a/src/util/k5ev/verto-libev.c
+++ b/src/util/k5ev/verto-libev.c
@@ -78,12 +78,40 @@ libev_ctx_reinitialize(verto_mod_ctx *ctx)
 static void
 libev_callback(EV_P_ ev_watcher *w, int revents)
 {
-    if (verto_get_type(w->data) == VERTO_EV_TYPE_CHILD)
+    verto_ev_flag state = VERTO_EV_FLAG_NONE;
+
+    if (verto_get_type(w->data)== VERTO_EV_TYPE_CHILD)
         verto_set_proc_status(w->data, ((ev_child*) w)->rstatus);
 
+    if (revents & EV_READ)
+        state |= VERTO_EV_FLAG_IO_READ;
+    if (revents & EV_WRITE)
+        state |= VERTO_EV_FLAG_IO_WRITE;
+    if (revents & EV_ERROR)
+        state |= VERTO_EV_FLAG_IO_ERROR;
+
+    verto_set_fd_state(w->data, state);
     verto_fire(w->data);
 }
 
+static void
+libev_ctx_set_flags(verto_mod_ctx *ctx, const verto_ev *ev,
+                    verto_mod_ev *evpriv)
+{
+    if (verto_get_type(ev) == VERTO_EV_TYPE_IO) {
+        int events = EV_NONE;
+
+        if (verto_get_flags(ev) & VERTO_EV_FLAG_IO_READ)
+            events |= EV_READ;
+        if (verto_get_flags(ev) & VERTO_EV_FLAG_IO_WRITE)
+            events |= EV_WRITE;
+
+        ev_io_stop(ctx, (ev_io*) evpriv);
+        ev_io_set(((ev_io*) evpriv), verto_get_fd(ev), events);
+        ev_io_start(ctx, (ev_io*) evpriv);
+    }
+}
+
 #define setuptype(type, ...) \
     w.type = malloc(sizeof(ev_ ## type)); \
     if (w.type) { \
@@ -104,17 +132,12 @@ libev_ctx_add(verto_mod_ctx *ctx, const verto_ev *ev, verto_ev_flag *flags)
        ev_child *child;
     } w;
     ev_tstamp interval;
-    int events = EV_NONE;
 
     w.watcher = NULL;
     *flags |= VERTO_EV_FLAG_PERSIST;
     switch (verto_get_type(ev)) {
         case VERTO_EV_TYPE_IO:
-            if (verto_get_flags(ev) & VERTO_EV_FLAG_IO_READ)
-                events |= EV_READ;
-            if (verto_get_flags(ev) & VERTO_EV_FLAG_IO_WRITE)
-                events |= EV_WRITE;
-            setuptype(io, libev_callback, verto_get_fd(ev), events);
+            setuptype(io, libev_callback, verto_get_fd(ev), EV_NONE);
         case VERTO_EV_TYPE_TIMEOUT:
             interval = ((ev_tstamp) verto_get_interval(ev)) / 1000.0;
             setuptype(timer, libev_callback, interval, interval);
@@ -129,8 +152,10 @@ libev_ctx_add(verto_mod_ctx *ctx, const verto_ev *ev, verto_ev_flag *flags)
             break; /* Not supported */
     }
 
-    if (w.watcher)
+    if (w.watcher) {
         w.watcher->data = (void*) ev;
+        libev_ctx_set_flags(ctx, ev, w.watcher);
+    }
     return w.watcher;
 }
 
diff --git a/src/util/verto/libverto.exports b/src/util/verto/libverto.exports
index 1e487744a3..098156fa2f 100644
--- a/src/util/verto/libverto.exports
+++ b/src/util/verto/libverto.exports
@@ -9,7 +9,9 @@ verto_default
 verto_del
 verto_fire
 verto_free
+verto_get_ctx
 verto_get_fd
+verto_get_fd_state
 verto_get_flags
 verto_get_interval
 verto_get_private
@@ -22,6 +24,8 @@ verto_new
 verto_reinitialize
 verto_run
 verto_run_once
+verto_set_allocator
 verto_set_default
+verto_set_fd_state
 verto_set_private
 verto_set_proc_status
diff --git a/src/util/verto/module.c b/src/util/verto/module.c
index 8b81646596..d5977cbbc9 100644
--- a/src/util/verto/module.c
+++ b/src/util/verto/module.c
@@ -40,6 +40,51 @@ dllerror(void) {
     LocalFree(msg);
     return amsg;
 }
+#elif defined(aix)
+#include "sys/ldr.h"
+
+struct Dl_info {
+  const char* dli_fname;
+};
+
+static int
+dladdr(void* s, Dl_info* i)
+{
+    static const size_t bufSize = 4096;
+    G__FastAllocString buf(bufSize);
+    char* pldi = buf;
+    int r;
+
+    r = loadquery(L_GETINFO, pldi, bufSize);
+    if (r == -1) {
+        i->dli_fname = NULL;
+        return 0;
+    }
+
+    for (ld_info* ldi = (ld_info*) buf;
+         ldi->ldinfo_next;
+         ldi += ldi->ldinfo_next) {
+        char* textBegin = (char*) ldi->ldinfo_textorg;
+        if (textBegin < s) {
+            char* textEnd = textBegin + ldi->ldinfo_textsize;
+            if (textEnd > s) {
+                i->dli_fname = ldi->ldinfo_filename;
+                return 1;
+            }
+        }
+    }
+
+    // First is main(), skip.
+    ld_info* ldi = (ld_info*) pldi;
+    while (ldi->ldinfo_next) {
+        pldi += ldi->ldinfo_next;
+        ldi = (ld_info*) pldi;
+
+    }
+
+    i->dli_fname = NULL;
+    return 0;
+}
 #else
 #define _GNU_SOURCE
 #include <stdlib.h>
@@ -80,14 +125,14 @@ module_get_filename_for_symbol(void *addr, char **filename)
 
     if (!GetModuleFileNameA(mod, tmp, MAX_PATH))
         return 0;
-#else  /* WIN32 */
+#else
     const char *tmp;
     Dl_info dlinfo;
 
     if (!dladdr(addr, &dlinfo))
         return 0;
     tmp = dlinfo.dli_fname;
-#endif /* WIN32 */
+#endif
 
     if (filename) {
         *filename = strdup(tmp);
diff --git a/src/util/verto/verto-module.h b/src/util/verto/verto-module.h
index b0e7232b1e..4ef3b8af60 100644
--- a/src/util/verto/verto-module.h
+++ b/src/util/verto/verto-module.h
@@ -35,7 +35,7 @@ typedef void verto_mod_ctx;
 typedef void verto_mod_ev;
 #endif
 
-#define VERTO_MODULE_VERSION 2
+#define VERTO_MODULE_VERSION 3
 #define VERTO_MODULE_TABLE(name) verto_module_table_ ## name
 #define VERTO_MODULE(name, symb, types) \
     static verto_ctx_funcs name ## _funcs = { \
@@ -46,6 +46,7 @@ typedef void verto_mod_ev;
         name ## _ctx_run_once, \
         name ## _ctx_break, \
         name ## _ctx_reinitialize, \
+        name ## _ctx_set_flags, \
         name ## _ctx_add, \
         name ## _ctx_del \
     }; \
@@ -75,6 +76,9 @@ typedef struct {
     /* Required */ void (*ctx_run_once)(verto_mod_ctx *ctx);
     /* Optional */ void (*ctx_break)(verto_mod_ctx *ctx);
     /* Optional */ void (*ctx_reinitialize)(verto_mod_ctx *ctx);
+    /* Optional */ void (*ctx_set_flags)(verto_mod_ctx *ctx,
+                                         const verto_ev *ev,
+                                         verto_mod_ev *modev);
     /* Required */ verto_mod_ev *(*ctx_add)(verto_mod_ctx *ctx,
                                             const verto_ev *ev,
                                             verto_ev_flag *flags);
@@ -166,4 +170,19 @@ verto_fire(verto_ev *ev);
 void
 verto_set_proc_status(verto_ev *ev, verto_proc_status status);
 
+/**
+ * Sets the state of the fd which caused this event to fire.
+ *
+ * This function does nothing if the verto_ev is not a io type.
+ *
+ * Only the flags VERTO_EV_FLAG_IO_(READ|WRITE|ERROR) are supported. All other
+ * flags are unset.
+ *
+ * @see verto_add_io()
+ * @param ev The verto_ev to set the state in.
+ * @param state The fd state.
+ */
+void
+verto_set_fd_state(verto_ev *ev, verto_ev_flag state);
+
 #endif /* VERTO_MODULE_H_ */
diff --git a/src/util/verto/verto.c b/src/util/verto/verto.c
index af4172687d..44ea4373fc 100644
--- a/src/util/verto/verto.c
+++ b/src/util/verto/verto.c
@@ -22,7 +22,7 @@
  * SOFTWARE.
  */
 
-#define _GNU_SOURCE /* For dladdr(), asprintf() */
+#define _GNU_SOURCE /* For asprintf() */
 
 #include <stdio.h>
 #include <stdlib.h>
@@ -45,6 +45,9 @@
 #define  _str(s) # s
 #define __str(s) _str(s)
 
+/* Remove flags we can emulate */
+#define make_actual(flags) ((flags) & ~(VERTO_EV_FLAG_PERSIST|VERTO_EV_FLAG_IO_CLOSE_FD))
+
 struct verto_ctx {
     size_t ref;
     verto_mod_ctx *ctx;
@@ -59,6 +62,11 @@ typedef struct {
     verto_proc_status status;
 } verto_child;
 
+typedef struct {
+    int fd;
+    verto_ev_flag state;
+} verto_io;
+
 struct verto_ev {
     verto_ev *next;
     verto_ctx *ctx;
@@ -72,7 +80,7 @@ struct verto_ev {
     size_t depth;
     int deleted;
     union {
-        int fd;
+        verto_io io;
         int signal;
         time_t interval;
         verto_child child;
@@ -88,7 +96,27 @@ struct module_record {
     verto_ctx *defctx;
 };
 
+
+#ifdef BUILTIN_MODULE
+#define _MODTABLE(n) verto_module_table_ ## n
+#define MODTABLE(n) _MODTABLE(n)
+/*
+ * This symbol can be used when embedding verto.c in a library along with a
+ * built-in private module, to preload the module instead of dynamically
+ * linking it in later.  Define to verto_module_table_<modulename>.
+ */
+extern verto_module MODTABLE(BUILTIN_MODULE);
+static module_record builtin_record = {
+    NULL, &MODTABLE(BUILTIN_MODULE), NULL, "", NULL
+};
+static module_record *loaded_modules = &builtin_record;
+#else
 static module_record *loaded_modules;
+#endif
+
+static void *(*resize_cb)(void *mem, size_t size);
+static int resize_cb_hierarchical;
+
 #ifdef HAVE_PTHREAD
 static pthread_mutex_t loaded_modules_mutex = PTHREAD_MUTEX_INITIALIZER;
 #define mutex_lock(x) pthread_mutex_lock(x)
@@ -98,6 +126,16 @@ static pthread_mutex_t loaded_modules_mutex = PTHREAD_MUTEX_INITIALIZER;
 #define mutex_unlock(x)
 #endif
 
+#define vfree(mem) vresize(mem, 0)
+static void *
+vresize(void *mem, size_t size)
+{
+    if (!resize_cb)
+        resize_cb = &realloc;
+    return (*resize_cb)(mem, size);
+}
+
+#ifndef BUILTIN_MODULE
 static int
 int_vasprintf(char **strp, const char *fmt, va_list ap) {
     va_list apc;
@@ -125,19 +163,6 @@ int_asprintf(char **strp, const char *fmt, ...) {
 }
 
 static char *
-int_get_table_name(const char *suffix)
-{
-    char *tmp;
-
-    tmp = malloc(strlen(suffix) + strlen(__str(VERTO_MODULE_TABLE())) + 1);
-    if (tmp) {
-        strcpy(tmp, __str(VERTO_MODULE_TABLE()));
-        strcat(tmp, suffix);
-    }
-    return tmp;
-}
-
-static char *
 int_get_table_name_from_filename(const char *filename)
 {
     char *bn = NULL, *tmp = NULL;
@@ -160,7 +185,8 @@ int_get_table_name_from_filename(const char *filename)
     if (tmp) {
         if (strchr(tmp+1, '.')) {
             *strchr(tmp+1, '.') = '\0';
-            tmp = int_get_table_name(tmp + 1);
+            if (int_asprintf(&tmp, "%s%s", __str(VERTO_MODULE_TABLE()), tmp + 1) < 0)
+                tmp = NULL;
         } else
             tmp = NULL;
     }
@@ -225,13 +251,13 @@ do_load_file(const char *filename, int reqsym, verto_ev_type reqtypes,
     mutex_unlock(&loaded_modules_mutex);
 
     /* Create our module record */
-    tmp = *record = malloc(sizeof(module_record));
+    tmp = *record = vresize(NULL, sizeof(module_record));
     if (!tmp)
         return 0;
     memset(tmp, 0, sizeof(module_record));
     tmp->filename = strdup(filename);
     if (!tmp->filename) {
-        free(tmp);
+        vfree(tmp);
         return 0;
     }
 
@@ -239,7 +265,7 @@ do_load_file(const char *filename, int reqsym, verto_ev_type reqtypes,
     tblname = int_get_table_name_from_filename(filename);
     if (!tblname) {
         free(tblname);
-        free(tmp);
+        vfree(tmp);
         return 0;
     }
 
@@ -252,7 +278,7 @@ do_load_file(const char *filename, int reqsym, verto_ev_type reqtypes,
         free(error);
         module_close(tmp->dll);
         free(tblname);
-        free(tmp);
+        vfree(tmp);
         return 0;
     }
 
@@ -311,14 +337,17 @@ do_load_dir(const char *dirname, const char *prefix, const char *suffix,
     closedir(dir);
     return *record != NULL;
 }
+#endif
 
 static int
 load_module(const char *impl, verto_ev_type reqtypes, module_record **record)
 {
     int success = 0;
+#ifndef BUILTIN_MODULE
     char *prefix = NULL;
     char *suffix = NULL;
     char *tmp = NULL;
+#endif
 
     /* Check the cache */
     mutex_lock(&loaded_modules_mutex);
@@ -341,6 +370,7 @@ load_module(const char *impl, verto_ev_type reqtypes, module_record **record)
     }
     mutex_unlock(&loaded_modules_mutex);
 
+#ifndef BUILTIN_MODULE
     if (!module_get_filename_for_symbol(verto_convert_module, &prefix))
         return 0;
 
@@ -394,14 +424,14 @@ load_module(const char *impl, verto_ev_type reqtypes, module_record **record)
                 success = do_load_dir(dname, prefix, suffix, 1, reqtypes,
                                       record);
                 if (!success) {
-#ifdef DEFAULT_LIBRARY
+#ifdef DEFAULT_MODULE
                     /* Attempt to find the default module */
-                    success = load_module(DEFAULT_LIBRARY, reqtypes, record);
+                    success = load_module(DEFAULT_MODULE, reqtypes, record);
                     if (!success)
-#endif /* DEFAULT_LIBRARY */
-                    /* Attempt to load any plugin (we're desperate) */
-                    success = do_load_dir(dname, prefix, suffix, 0,
-                                          reqtypes, record);
+#endif /* DEFAULT_MODULE */
+                        /* Attempt to load any plugin (we're desperate) */
+                        success = do_load_dir(dname, prefix, suffix, 0,
+                                              reqtypes, record);
                 }
             }
 
@@ -411,6 +441,7 @@ load_module(const char *impl, verto_ev_type reqtypes, module_record **record)
 
     free(suffix);
     free(prefix);
+#endif /* BUILTIN_MODULE */
     return success;
 }
 
@@ -423,7 +454,7 @@ make_ev(verto_ctx *ctx, verto_callback *callback,
     if (!ctx || !callback)
         return NULL;
 
-    ev = malloc(sizeof(verto_ev));
+    ev = vresize(NULL, sizeof(verto_ev));
     if (ev) {
         memset(ev, 0, sizeof(verto_ev));
         ev->ctx        = ctx;
@@ -502,6 +533,17 @@ verto_set_default(const char *impl, verto_ev_type reqtypes)
     return load_module(impl, reqtypes, &mr);
 }
 
+int
+verto_set_allocator(void *(*resize)(void *mem, size_t size),
+                    int hierarchical)
+{
+    if (resize_cb || !resize)
+        return 0;
+    resize_cb = resize;
+    resize_cb_hierarchical = hierarchical;
+    return 1;
+}
+
 void
 verto_free(verto_ctx *ctx)
 {
@@ -520,7 +562,7 @@ verto_free(verto_ctx *ctx)
     if (!ctx->deflt || !ctx->module->funcs->ctx_default)
         ctx->module->funcs->ctx_free(ctx->ctx);
 
-    free(ctx);
+    vfree(ctx);
 }
 
 void
@@ -583,7 +625,7 @@ verto_reinitialize(verto_ctx *ctx)
 
     /* Recreate events that were marked forkable */
     for (tmp = ctx->events; tmp; tmp = tmp->next) {
-        tmp->actual = tmp->flags;
+        tmp->actual = make_actual(tmp->flags);
         tmp->ev = ctx->module->funcs->ctx_add(ctx->ctx, tmp, &tmp->actual);
         if (!tmp->ev)
             error = 0;
@@ -596,10 +638,10 @@ verto_reinitialize(verto_ctx *ctx)
     ev = make_ev(ctx, callback, type, flags); \
     if (ev) { \
         set; \
-        ev->actual = ev->flags; \
+        ev->actual = make_actual(ev->flags); \
         ev->ev = ctx->module->funcs->ctx_add(ctx->ctx, ev, &ev->actual); \
         if (!ev->ev) { \
-            free(ev); \
+            vfree(ev); \
             return NULL; \
         } \
         push_ev(ctx, ev); \
@@ -614,7 +656,7 @@ verto_add_io(verto_ctx *ctx, verto_ev_flag flags,
     if (fd < 0 || !(flags & (VERTO_EV_FLAG_IO_READ | VERTO_EV_FLAG_IO_WRITE)))
         return NULL;
 
-    doadd(ev, ev->option.fd = fd, VERTO_EV_TYPE_IO);
+    doadd(ev, ev->option.io.fd = fd, VERTO_EV_TYPE_IO);
     return ev;
 }
 
@@ -704,14 +746,43 @@ verto_get_flags(const verto_ev *ev)
     return ev->flags;
 }
 
+void
+verto_set_flags(verto_ev *ev, verto_ev_flag flags)
+{
+    if (!ev)
+        return;
+
+    ev->flags  &= ~_VERTO_EV_FLAG_MUTABLE_MASK;
+    ev->flags  |= flags & _VERTO_EV_FLAG_MUTABLE_MASK;
+
+    /* If setting flags isn't supported, just rebuild the event */
+    if (!ev->ctx->module->funcs->ctx_set_flags) {
+        ev->ctx->module->funcs->ctx_del(ev->ctx->ctx, ev, ev->ev);
+        ev->actual = make_actual(ev->flags);
+        ev->ev = ev->ctx->module->funcs->ctx_add(ev->ctx->ctx, ev, &ev->actual);
+        assert(ev->ev); /* Here is the main reason why modules should */
+        return;         /* implement set_flags(): we cannot fail gracefully. */
+    }
+
+    ev->actual &= ~_VERTO_EV_FLAG_MUTABLE_MASK;
+    ev->actual |= flags & _VERTO_EV_FLAG_MUTABLE_MASK;
+    ev->ctx->module->funcs->ctx_set_flags(ev->ctx->ctx, ev, ev->ev);
+}
+
 int
 verto_get_fd(const verto_ev *ev)
 {
     if (ev && (ev->type == VERTO_EV_TYPE_IO))
-        return ev->option.fd;
+        return ev->option.io.fd;
     return -1;
 }
 
+verto_ev_flag
+verto_get_fd_state(const verto_ev *ev)
+{
+    return ev->option.io.state;
+}
+
 time_t
 verto_get_interval(const verto_ev *ev)
 {
@@ -741,6 +812,12 @@ verto_get_proc_status(const verto_ev *ev)
     return ev->option.child.status;
 }
 
+verto_ctx *
+verto_get_ctx(const verto_ev *ev)
+{
+    return ev->ctx;
+}
+
 void
 verto_del(verto_ev *ev)
 {
@@ -760,7 +837,13 @@ verto_del(verto_ev *ev)
         ev->onfree(ev->ctx, ev);
     ev->ctx->module->funcs->ctx_del(ev->ctx->ctx, ev, ev->ev);
     remove_ev(&(ev->ctx->events), ev);
-    free(ev);
+
+    if ((ev->type == VERTO_EV_TYPE_IO) &&
+        (ev->flags & VERTO_EV_FLAG_IO_CLOSE_FD) &&
+        !(ev->actual & VERTO_EV_FLAG_IO_CLOSE_FD))
+        close(ev->option.io.fd);
+
+    vfree(ev);
 }
 
 verto_ev_type
@@ -806,7 +889,7 @@ verto_convert_module(const verto_module *module, int deflt, verto_mod_ctx *mctx)
             goto error;
     }
 
-    ctx = malloc(sizeof(verto_ctx));
+    ctx = vresize(NULL, sizeof(verto_ctx));
     if (!ctx)
         goto error;
     memset(ctx, 0, sizeof(verto_ctx));
@@ -836,9 +919,9 @@ verto_convert_module(const verto_module *module, int deflt, verto_mod_ctx *mctx)
         }
         mutex_unlock(&loaded_modules_mutex);
 
-        *tmp = malloc(sizeof(module_record));
+        *tmp = vresize(NULL, sizeof(module_record));
         if (!*tmp) {
-            free(ctx);
+            vfree(ctx);
             goto error;
         }
 
@@ -867,12 +950,19 @@ verto_fire(verto_ev *ev)
     if (ev->depth == 0) {
         if (!(ev->flags & VERTO_EV_FLAG_PERSIST) || ev->deleted)
             verto_del(ev);
-        else if (!ev->actual & VERTO_EV_FLAG_PERSIST) {
-            ev->actual = ev->flags;
-            priv = ev->ctx->module->funcs->ctx_add(ev->ctx->ctx, ev, &ev->actual);
-            assert(priv); /* TODO: create an error callback */
-            ev->ctx->module->funcs->ctx_del(ev->ctx->ctx, ev, ev->ev);
-            ev->ev = priv;
+        else {
+            if (!(ev->actual & VERTO_EV_FLAG_PERSIST)) {
+                ev->actual = make_actual(ev->flags);
+                priv = ev->ctx->module->funcs->ctx_add(ev->ctx->ctx, ev, &ev->actual);
+                assert(priv); /* TODO: create an error callback */
+                ev->ctx->module->funcs->ctx_del(ev->ctx->ctx, ev, ev->ev);
+                ev->ev = priv;
+            }
+
+            if (ev->type == VERTO_EV_TYPE_IO)
+                ev->option.io.state = VERTO_EV_FLAG_NONE;
+            if (ev->type == VERTO_EV_TYPE_CHILD)
+                ev->option.child.status = 0;
         }
     }
 }
@@ -883,3 +973,19 @@ verto_set_proc_status(verto_ev *ev, verto_proc_status status)
     if (ev && ev->type == VERTO_EV_TYPE_CHILD)
         ev->option.child.status = status;
 }
+
+void
+verto_set_fd_state(verto_ev *ev, verto_ev_flag state)
+{
+    /* Filter out only the io flags */
+    state = state & (VERTO_EV_FLAG_IO_READ |
+                     VERTO_EV_FLAG_IO_WRITE |
+                     VERTO_EV_FLAG_IO_ERROR);
+
+    /* Don't report read/write if the socket is closed */
+    if (state & VERTO_EV_FLAG_IO_ERROR)
+        state = VERTO_EV_FLAG_IO_ERROR;
+
+    if (ev && ev->type == VERTO_EV_TYPE_IO)
+        ev->option.io.state = state;
+}
diff --git a/src/util/verto/verto.h b/src/util/verto/verto.h
index 23d9a20969..5540367717 100644
--- a/src/util/verto/verto.h
+++ b/src/util/verto/verto.h
@@ -39,6 +39,11 @@ typedef int verto_proc_status;
 
 #define VERTO_SIG_IGN ((verto_callback *) 1)
 
+#ifdef __cplusplus
+extern "C"
+{
+#endif /* __cplusplus */
+
 typedef struct verto_ctx verto_ctx;
 typedef struct verto_ev verto_ev;
 
@@ -59,8 +64,15 @@ typedef enum {
     VERTO_EV_FLAG_PRIORITY_HIGH = 1 << 3,
     VERTO_EV_FLAG_IO_READ = 1 << 4,
     VERTO_EV_FLAG_IO_WRITE = 1 << 5,
+    VERTO_EV_FLAG_IO_ERROR = 1 << 7,
+    VERTO_EV_FLAG_IO_CLOSE_FD = 1 << 8,
     VERTO_EV_FLAG_REINITIABLE = 1 << 6,
-    _VERTO_EV_FLAG_MAX = VERTO_EV_FLAG_REINITIABLE
+    _VERTO_EV_FLAG_MUTABLE_MASK = VERTO_EV_FLAG_PRIORITY_LOW
+                                  | VERTO_EV_FLAG_PRIORITY_MEDIUM
+                                  | VERTO_EV_FLAG_PRIORITY_HIGH
+                                  | VERTO_EV_FLAG_IO_READ
+                                  | VERTO_EV_FLAG_IO_WRITE,
+    _VERTO_EV_FLAG_MAX = VERTO_EV_FLAG_IO_CLOSE_FD
 } verto_ev_flag;
 
 typedef void (verto_callback)(verto_ctx *ctx, verto_ev *ev);
@@ -171,6 +183,25 @@ int
 verto_set_default(const char *impl, verto_ev_type reqtypes);
 
 /**
+ * Sets the allocator to use for verto_ctx and verto_ev objects.
+ *
+ * If you plan to set the allocator, you MUST call this function before any
+ * other verto_*() calls.
+ *
+ * @see verto_new()
+ * @see verto_default()
+ * @see verto_add_io()
+ * @see verto_add_timeout()
+ * @see verto_add_idle()
+ * @see verto_add_signal()
+ * @see verto_add_child()
+ * @param resize The allocator to use (behaves like realloc())
+ * @param hierarchical Zero if the allocator is not hierarchical
+ */
+int
+verto_set_allocator(void *(*resize)(void *mem, size_t size), int hierarchical);
+
+/**
  * Frees a verto_ctx.
  *
  * When called on a default verto_ctx, the reference will be freed but the
@@ -237,6 +268,8 @@ verto_reinitialize(verto_ctx *ctx);
  * VERTO_EV_FLAG_PERSIST is not provided, the event will be freed automatically
  * after its execution. In either case, you may call verto_del() at any time
  * to prevent the event from executing.
+ * If VERTO_EV_FLAG_IO_CLOSE_FD is provided the passed in fd is automatically
+ * closed when the event is freed with verto_del()
  *
  * NOTE: On Windows, the underlying select() only works with sockets. As such,
  * any attempt to add a non-socket io event on Windows will produce undefined
@@ -400,6 +433,7 @@ verto_get_type(const verto_ev *ev);
  * @see verto_add_idle()
  * @see verto_add_signal()
  * @see verto_add_child()
+ * @see verto_set_flags()
  * @param ev The verto_ev
  * @return The verto_ev type
  */
@@ -407,6 +441,24 @@ verto_ev_flag
 verto_get_flags(const verto_ev *ev);
 
 /**
+ * Sets the flags associated with the given verto_ev.
+ *
+ * See _VERTO_EV_FLAG_MUTABLE_MASK for the flags that can be changed
+ * with this function. All others will be ignored.
+ *
+ * @see verto_add_io()
+ * @see verto_add_timeout()
+ * @see verto_add_idle()
+ * @see verto_add_signal()
+ * @see verto_add_child()
+ * @see verto_get_flags()
+ * @param ev The verto_ev
+ * @param flags The flags for the event
+ */
+void
+verto_set_flags(verto_ev *ev, verto_ev_flag flags);
+
+/**
  * Gets the file descriptor associated with a read/write verto_ev.
  *
  * @see verto_add_io()
@@ -417,6 +469,16 @@ int
 verto_get_fd(const verto_ev *ev);
 
 /**
+ * Gets the file descriptor state from when the event fires.
+ *
+ * @see verto_add_io()
+ * @param ev The verto_ev to retrieve the fd state from.
+ * @return The fd state.
+ */
+verto_ev_flag
+verto_get_fd_state(const verto_ev *ev);
+
+/**
  * Gets the interval associated with a timeout verto_ev.
  *
  * @see verto_add_timeout()
@@ -457,6 +519,17 @@ verto_proc_status
 verto_get_proc_status(const verto_ev *ev);
 
 /**
+ * Gets the verto_ctx associated with a verto_ev.
+ *
+ * This is a borrowed reference, don't attempt to free it!
+ *
+ * @param ev The verto_ev to retrieve the verto_ctx from.
+ * @return The verto_ctx.
+ */
+verto_ctx *
+verto_get_ctx(const verto_ev *ev);
+
+/**
  * Removes an event from from the event context and frees it.
  *
  * The event and its contents cannot be used after this call.
@@ -480,4 +553,7 @@ verto_del(verto_ev *ev);
 verto_ev_type
 verto_get_supported_types(verto_ctx *ctx);
 
+#ifdef __cplusplus
+} /* extern "C" */
+#endif /* __cplusplus */
 #endif /* VERTO_H_ */