Initial version.

1 files changed, 646 insertions, 0 deletions

diff --git a/src/nxt_eventport.c b/src/nxt_eventport.c
new file mode 100644
index 00000000..02984573
--- /dev/null
+++ b/src/nxt_eventport.c
@@ -0,0 +1,646 @@
+
+/*
+ * Copyright (C) Igor Sysoev
+ * Copyright (C) NGINX, Inc.
+ */
+
+#include <nxt_main.h>
+
+
+/*
+ * The event ports have been introduced in Solaris 10.
+ * The PORT_SOURCE_MQ and PORT_SOURCE_FILE sources have
+ * been added in OpenSolaris.
+ */
+
+
+static nxt_event_set_t *nxt_eventport_create(nxt_event_signals_t *signals,
+    nxt_uint_t mchanges, nxt_uint_t mevents);
+static void nxt_eventport_free(nxt_event_set_t *event_set);
+static void nxt_eventport_enable(nxt_event_set_t *event_set,
+    nxt_event_fd_t *ev);
+static void nxt_eventport_disable(nxt_event_set_t *event_set,
+    nxt_event_fd_t *ev);
+static void nxt_eventport_close(nxt_event_set_t *event_set, nxt_event_fd_t *ev);
+static void nxt_eventport_drop_changes(nxt_event_set_t *event_set,
+    nxt_event_fd_t *ev);
+static void nxt_eventport_enable_read(nxt_event_set_t *event_set,
+    nxt_event_fd_t *ev);
+static void nxt_eventport_enable_write(nxt_event_set_t *event_set,
+    nxt_event_fd_t *ev);
+static void nxt_eventport_enable_event(nxt_event_set_t *event_set,
+    nxt_event_fd_t *ev, nxt_uint_t events);
+static void nxt_eventport_disable_read(nxt_event_set_t *event_set,
+    nxt_event_fd_t *ev);
+static void nxt_eventport_disable_write(nxt_event_set_t *event_set,
+    nxt_event_fd_t *ev);
+static void nxt_eventport_disable_event(nxt_event_set_t *event_set,
+    nxt_event_fd_t *ev);
+static nxt_int_t nxt_eventport_commit_changes(nxt_thread_t *thr,
+    nxt_eventport_event_set_t *es);
+static void nxt_eventport_error_handler(nxt_thread_t *thr, void *obj,
+    void *data);
+static void nxt_eventport_block_read(nxt_event_set_t *event_set,
+    nxt_event_fd_t *ev);
+static void nxt_eventport_block_write(nxt_event_set_t *event_set,
+    nxt_event_fd_t *ev);
+static void nxt_eventport_oneshot_read(nxt_event_set_t *event_set,
+    nxt_event_fd_t *ev);
+static void nxt_eventport_oneshot_write(nxt_event_set_t *event_set,
+    nxt_event_fd_t *ev);
+static void nxt_eventport_enable_accept(nxt_event_set_t *event_set,
+    nxt_event_fd_t *ev);
+static nxt_int_t nxt_eventport_enable_post(nxt_event_set_t *event_set,
+    nxt_work_handler_t handler);
+static void nxt_eventport_signal(nxt_event_set_t *event_set, nxt_uint_t signo);
+static void nxt_eventport_poll(nxt_thread_t *thr, nxt_event_set_t *event_set,
+    nxt_msec_t timeout);
+
+
+const nxt_event_set_ops_t  nxt_eventport_event_set = {
+    "eventport",
+    nxt_eventport_create,
+    nxt_eventport_free,
+    nxt_eventport_enable,
+    nxt_eventport_disable,
+    nxt_eventport_disable,
+    nxt_eventport_close,
+    nxt_eventport_enable_read,
+    nxt_eventport_enable_write,
+    nxt_eventport_disable_read,
+    nxt_eventport_disable_write,
+    nxt_eventport_block_read,
+    nxt_eventport_block_write,
+    nxt_eventport_oneshot_read,
+    nxt_eventport_oneshot_write,
+    nxt_eventport_enable_accept,
+    NULL,
+    NULL,
+    nxt_eventport_enable_post,
+    nxt_eventport_signal,
+    nxt_eventport_poll,
+
+    &nxt_unix_event_conn_io,
+
+    NXT_NO_FILE_EVENTS,
+    NXT_NO_SIGNAL_EVENTS,
+};
+
+
+static nxt_event_set_t *
+nxt_eventport_create(nxt_event_signals_t *signals, nxt_uint_t mchanges,
+    nxt_uint_t mevents)
+{
+    nxt_event_set_t            *event_set;
+    nxt_eventport_event_set_t  *es;
+
+    event_set = nxt_zalloc(sizeof(nxt_eventport_event_set_t));
+    if (event_set == NULL) {
+        return NULL;
+    }
+
+    es = &event_set->eventport;
+
+    es->port = -1;
+    es->mchanges = mchanges;
+    es->mevents = mevents;
+
+    es->changes = nxt_malloc(sizeof(nxt_eventport_change_t) * mchanges);
+    if (es->changes == NULL) {
+        goto fail;
+    }
+
+    es->events = nxt_malloc(sizeof(port_event_t) * mevents);
+    if (es->events == NULL) {
+        goto fail;
+    }
+
+    es->port = port_create();
+    if (es->port == -1) {
+        nxt_main_log_emerg("port_create() failed %E", nxt_errno);
+        goto fail;
+    }
+
+    nxt_main_log_debug("port_create(): %d", es->port);
+
+    if (signals != NULL) {
+        es->signal_handler = signals->handler;
+    }
+
+    return event_set;
+
+fail:
+
+    nxt_eventport_free(event_set);
+
+    return NULL;
+}
+
+
+static void
+nxt_eventport_free(nxt_event_set_t *event_set)
+{
+    nxt_eventport_event_set_t  *es;
+
+    es = &event_set->eventport;
+
+    nxt_main_log_debug("eventport %d free", es->port);
+
+    if (es->port != -1) {
+        if (close(es->port) != 0) {
+            nxt_main_log_emerg("eventport close(%d) failed %E",
+                               es->port, nxt_errno);
+        }
+    }
+
+    nxt_free(es->events);
+    nxt_free(es);
+}
+
+
+static void
+nxt_eventport_enable(nxt_event_set_t *event_set, nxt_event_fd_t *ev)
+{
+    ev->read = NXT_EVENT_DEFAULT;
+    ev->write = NXT_EVENT_DEFAULT;
+
+    nxt_eventport_enable_event(event_set, ev, POLLIN | POLLOUT);
+}
+
+
+static void
+nxt_eventport_disable(nxt_event_set_t *event_set, nxt_event_fd_t *ev)
+{
+    if (ev->read != NXT_EVENT_INACTIVE || ev->write != NXT_EVENT_INACTIVE) {
+
+        ev->read = NXT_EVENT_INACTIVE;
+        ev->write = NXT_EVENT_INACTIVE;
+
+        nxt_eventport_disable_event(event_set, ev);
+    }
+}
+
+
+static void
+nxt_eventport_close(nxt_event_set_t *event_set, nxt_event_fd_t *ev)
+{
+    ev->read = NXT_EVENT_INACTIVE;
+    ev->write = NXT_EVENT_INACTIVE;
+
+    nxt_eventport_drop_changes(event_set, ev);
+}
+
+
+static void
+nxt_eventport_drop_changes(nxt_event_set_t *event_set, nxt_event_fd_t *ev)
+{
+    nxt_eventport_change_t     *dst, *src, *end;
+    nxt_eventport_event_set_t  *es;
+
+    es = &event_set->eventport;
+
+    dst = es->changes;
+    end = dst + es->nchanges;
+
+    for (src = dst; src < end; src++) {
+
+        if (src->event == ev) {
+            continue;
+        }
+
+        if (dst != src) {
+            *dst = *src;
+        }
+
+        dst++;
+    }
+
+    es->nchanges -= end - dst;
+}
+
+
+static void
+nxt_eventport_enable_read(nxt_event_set_t *event_set, nxt_event_fd_t *ev)
+{
+    nxt_uint_t  events;
+
+    if (ev->read != NXT_EVENT_BLOCKED) {
+        events = (ev->write == NXT_EVENT_INACTIVE) ? POLLIN:
+                                                     (POLLIN | POLLOUT);
+        nxt_eventport_enable_event(event_set, ev, events);
+    }
+
+    ev->read = NXT_EVENT_DEFAULT;
+}
+
+
+static void
+nxt_eventport_enable_write(nxt_event_set_t *event_set, nxt_event_fd_t *ev)
+{
+    nxt_uint_t  events;
+
+    if (ev->write != NXT_EVENT_BLOCKED) {
+        events = (ev->read == NXT_EVENT_INACTIVE) ? POLLOUT:
+                                                    (POLLIN | POLLOUT);
+        nxt_eventport_enable_event(event_set, ev, events);
+    }
+
+    ev->write = NXT_EVENT_DEFAULT;
+}
+
+
+/*
+ * eventport changes are batched to improve instruction and data
+ * cache locality of several port_associate() and port_dissociate()
+ * calls followed by port_getn() call.
+ */
+
+static void
+nxt_eventport_enable_event(nxt_event_set_t *event_set, nxt_event_fd_t *ev,
+    nxt_uint_t events)
+{
+    nxt_eventport_change_t     *ch;
+    nxt_eventport_event_set_t  *es;
+
+    es = &event_set->eventport;
+
+    nxt_log_debug(ev->log, "port %d set event: fd:%d ev:%04XD u:%p",
+                  es->port, ev->fd, events, ev);
+
+    if (es->nchanges >= es->mchanges) {
+        (void) nxt_eventport_commit_changes(nxt_thread(), es);
+    }
+
+    ch = &es->changes[es->nchanges++];
+    ch->fd = ev->fd;
+    ch->events = events;
+    ch->event = ev;
+}
+
+
+static void
+nxt_eventport_disable_read(nxt_event_set_t *event_set, nxt_event_fd_t *ev)
+{
+    ev->read = NXT_EVENT_INACTIVE;
+
+    if (ev->write == NXT_EVENT_INACTIVE) {
+        nxt_eventport_disable_event(event_set, ev);
+
+    } else {
+        nxt_eventport_enable_event(event_set, ev, POLLOUT);
+    }
+}
+
+
+static void
+nxt_eventport_disable_write(nxt_event_set_t *event_set, nxt_event_fd_t *ev)
+{
+    ev->write = NXT_EVENT_INACTIVE;
+
+    if (ev->read == NXT_EVENT_INACTIVE) {
+        nxt_eventport_disable_event(event_set, ev);
+
+    } else {
+        nxt_eventport_enable_event(event_set, ev, POLLIN);
+    }
+}
+
+
+static void
+nxt_eventport_disable_event(nxt_event_set_t *event_set, nxt_event_fd_t *ev)
+{
+    nxt_eventport_change_t     *ch;
+    nxt_eventport_event_set_t  *es;
+
+    es = &event_set->eventport;
+
+    nxt_log_debug(ev->log, "port %d disable event : fd:%d", es->port, ev->fd);
+
+    if (es->nchanges >= es->mchanges) {
+        (void) nxt_eventport_commit_changes(nxt_thread(), es);
+    }
+
+    ch = &es->changes[es->nchanges++];
+    ch->fd = ev->fd;
+    ch->events = 0;
+    ch->event = ev;
+}
+
+
+static nxt_int_t
+nxt_eventport_commit_changes(nxt_thread_t *thr, nxt_eventport_event_set_t *es)
+{
+    int                     ret;
+    nxt_int_t               retval;
+    nxt_event_fd_t          *ev;
+    nxt_eventport_change_t  *ch, *end;
+
+    nxt_log_debug(thr->log, "eventport %d changes:%ui", es->port, es->nchanges);
+
+    retval = NXT_OK;
+    ch = es->changes;
+    end = ch + es->nchanges;
+
+    do {
+        ev = ch->event;
+
+        if (ch->events != 0) {
+            nxt_log_debug(ev->log, "port_associate(%d): fd:%d ev:%04XD u:%p",
+                          es->port, ch->fd, ch->events, ev);
+
+            ret = port_associate(es->port, PORT_SOURCE_FD, ch->fd,
+                                 ch->events, ev);
+            if (ret == 0) {
+                goto next;
+            }
+
+            nxt_log_alert(ev->log,
+                          "port_associate(%d, %d, %d, %04XD) failed %E",
+                          es->port, PORT_SOURCE_FD, ch->fd, ch->events,
+                          nxt_errno);
+
+        } else {
+            nxt_log_debug(ev->log, "port_dissociate(%d): fd:%d",
+                          es->port, ch->fd);
+
+            if (port_dissociate(es->port, PORT_SOURCE_FD, ch->fd) == 0) {
+                goto next;
+            }
+
+            nxt_log_alert(ev->log, "port_dissociate(%d, %d, %d) failed %E",
+                          es->port, PORT_SOURCE_FD, ch->fd, nxt_errno);
+        }
+
+        nxt_thread_work_queue_add(thr, &thr->work_queue.main,
+                                  nxt_eventport_error_handler,
+                                  ev, ev->data, ev->log);
+
+        retval = NXT_ERROR;
+
+    next:
+
+        ch++;
+
+    } while (ch < end);
+
+    es->nchanges = 0;
+
+    return retval;
+}
+
+
+static void
+nxt_eventport_error_handler(nxt_thread_t *thr, void *obj, void *data)
+{
+    nxt_event_fd_t  *ev;
+
+    ev = obj;
+
+    ev->read = NXT_EVENT_INACTIVE;
+    ev->write = NXT_EVENT_INACTIVE;
+
+    ev->error_handler(thr, ev, data);
+}
+
+
+static void
+nxt_eventport_block_read(nxt_event_set_t *event_set, nxt_event_fd_t *ev)
+{
+    if (ev->read != NXT_EVENT_INACTIVE) {
+        ev->read = NXT_EVENT_BLOCKED;
+    }
+}
+
+
+static void
+nxt_eventport_block_write(nxt_event_set_t *event_set, nxt_event_fd_t *ev)
+{
+    if (ev->write != NXT_EVENT_INACTIVE) {
+        ev->write = NXT_EVENT_BLOCKED;
+    }
+}
+
+
+static void
+nxt_eventport_oneshot_read(nxt_event_set_t *event_set, nxt_event_fd_t *ev)
+{
+    if (ev->read == NXT_EVENT_INACTIVE) {
+        ev->read = NXT_EVENT_DEFAULT;
+
+        nxt_eventport_enable_event(event_set, ev, POLLIN);
+    }
+}
+
+
+static void
+nxt_eventport_oneshot_write(nxt_event_set_t *event_set, nxt_event_fd_t *ev)
+{
+    if (ev->write == NXT_EVENT_INACTIVE) {
+        ev->write = NXT_EVENT_DEFAULT;
+
+        nxt_eventport_enable_event(event_set, ev, POLLOUT);
+    }
+}
+
+
+static void
+nxt_eventport_enable_accept(nxt_event_set_t *event_set, nxt_event_fd_t *ev)
+{
+    ev->read = NXT_EVENT_LEVEL;
+
+    nxt_eventport_enable_event(event_set, ev, POLLIN);
+}
+
+
+static nxt_int_t
+nxt_eventport_enable_post(nxt_event_set_t *event_set,
+    nxt_work_handler_t handler)
+{
+    event_set->eventport.post_handler = handler;
+
+    return NXT_OK;
+}
+
+
+static void
+nxt_eventport_signal(nxt_event_set_t *event_set, nxt_uint_t signo)
+{
+    nxt_eventport_event_set_t  *es;
+
+    es = &event_set->eventport;
+
+    nxt_thread_log_debug("port_send(%d, %ui)", es->port, signo);
+
+    if (port_send(es->port, signo, NULL) != 0) {
+        nxt_thread_log_alert("port_send(%d) failed %E", es->port, nxt_errno);
+    }
+}
+
+
+static void
+nxt_eventport_poll(nxt_thread_t *thr, nxt_event_set_t *event_set,
+    nxt_msec_t timeout)
+{
+    int                        n, events, signo;
+    uint_t                     nevents;
+    nxt_err_t                  err;
+    nxt_uint_t                 i, level;
+    timespec_t                 ts, *tp;
+    port_event_t               *event;
+    nxt_event_fd_t             *ev;
+    nxt_work_handler_t         handler;
+    nxt_eventport_event_set_t  *es;
+
+    es = &event_set->eventport;
+
+    if (es->nchanges != 0) {
+        if (nxt_eventport_commit_changes(thr, es) != NXT_OK) {
+            /* Error handlers have been enqueued on failure. */
+            timeout = 0;
+        }
+    }
+
+    if (timeout == NXT_INFINITE_MSEC) {
+        tp = NULL;
+
+    } else {
+        ts.tv_sec = timeout / 1000;
+        ts.tv_nsec = (timeout % 1000) * 1000000;
+        tp = &ts;
+    }
+
+    nxt_log_debug(thr->log, "port_getn(%d) timeout: %M", es->port, timeout);
+
+    /*
+     * A trap for possible error when Solaris does not update nevents
+     * if ETIME or EINTR is returned.  This issue will be logged as
+     * "unexpected port_getn() event".
+     *
+     * The details are in OpenSolaris mailing list thread "port_getn()
+     * and timeouts - is this a bug or an undocumented feature?"
+     */
+    event = &es->events[0];
+    event->portev_events = -1; /* invalid port events */
+    event->portev_source = -1; /* invalid port source */
+    event->portev_object = -1;
+    event->portev_user = (void *) -1;
+
+    nevents = 1;
+    n = port_getn(es->port, es->events, es->mevents, &nevents, tp);
+
+    /*
+     * 32-bit port_getn() on Solaris 10 x86 returns large negative
+     * values instead of 0 when returning immediately.
+     */
+    err = (n < 0) ? nxt_errno : 0;
+
+    nxt_thread_time_update(thr);
+
+    if (n == -1) {
+        if (err == NXT_ETIME || err == NXT_EINTR) {
+            if (nevents != 0) {
+                nxt_log_alert(thr->log, "port_getn(%d) failed %E, events:%ud",
+                              es->port, err, nevents);
+            }
+        }
+
+        if (err != NXT_ETIME) {
+            level = (err == NXT_EINTR) ? NXT_LOG_INFO : NXT_LOG_ALERT;
+            nxt_log_error(level, thr->log, "port_getn(%d) failed %E",
+                          es->port, err);
+
+            if (err != NXT_EINTR) {
+                return;
+            }
+        }
+    }
+
+    nxt_log_debug(thr->log, "port_getn(%d) events: %d", es->port, nevents);
+
+    for (i = 0; i < nevents; i++) {
+        event = &es->events[i];
+
+        switch (event->portev_source) {
+
+        case PORT_SOURCE_FD:
+            ev = event->portev_user;
+            events = event->portev_events;
+
+            nxt_log_debug(ev->log, "eventport: fd:%d ev:%04Xd u:%p rd:%d wr:%d",
+                          event->portev_object, events, ev,
+                          ev->read, ev->write);
+
+            if (nxt_slow_path(events & (POLLERR | POLLHUP | POLLNVAL)) != 0) {
+                nxt_log_alert(ev->log,
+                              "port_getn(%d) error fd:%d events:%04Xud",
+                              es->port, ev->fd, events);
+
+                nxt_thread_work_queue_add(thr, &thr->work_queue.main,
+                                          nxt_eventport_error_handler,
+                                          ev, ev->data, ev->log);
+                continue;
+            }
+
+            if (events & POLLIN) {
+                ev->read_ready = 1;
+
+                if (ev->read != NXT_EVENT_BLOCKED) {
+                    nxt_thread_work_queue_add(thr, ev->read_work_queue,
+                                              ev->read_handler,
+                                              ev, ev->data, ev->log);
+
+                }
+
+                if (ev->read != NXT_EVENT_LEVEL) {
+                    ev->read = NXT_EVENT_INACTIVE;
+                }
+            }
+
+            if (events & POLLOUT) {
+                ev->write_ready = 1;
+
+                if (ev->write != NXT_EVENT_BLOCKED) {
+                    nxt_thread_work_queue_add(thr, ev->write_work_queue,
+                                              ev->write_handler,
+                                              ev, ev->data, ev->log);
+                }
+
+                ev->write = NXT_EVENT_INACTIVE;
+            }
+
+            /*
+             * Reactivate counterpart direction, because the
+             * eventport is oneshot notification facility.
+             */
+            events = (ev->read == NXT_EVENT_INACTIVE) ? 0 : POLLIN;
+            events |= (ev->write == NXT_EVENT_INACTIVE) ? 0 : POLLOUT;
+
+            if (events != 0) {
+                nxt_eventport_enable_event(event_set, ev, events);
+            }
+
+            break;
+
+        case PORT_SOURCE_USER:
+            nxt_log_debug(thr->log, "eventport: user ev:%d u:%p",
+                          event->portev_events, event->portev_user);
+
+            signo = event->portev_events;
+
+            handler = (signo == 0) ? es->post_handler : es->signal_handler;
+
+            nxt_thread_work_queue_add(thr, &thr->work_queue.main, handler,
+                                      (void *) (uintptr_t) signo, NULL,
+                                      thr->log);
+
+            break;
+
+        default:
+            nxt_log_alert(thr->log, "unexpected port_getn(%d) event: "
+                          "ev:%d src:%d obj:%p u:%p",
+                          es->port, event->portev_events,
+                          event->portev_source, event->portev_object,
+                          event->portev_user);
+        }
+    }
+}