diff options
Diffstat (limited to '')
| -rw-r--r-- | src/nxt_pollset_engine.c | 647 |
1 files changed, 647 insertions, 0 deletions
diff --git a/src/nxt_pollset_engine.c b/src/nxt_pollset_engine.c new file mode 100644 index 00000000..571ad794 --- /dev/null +++ b/src/nxt_pollset_engine.c @@ -0,0 +1,647 @@ + +/* + * Copyright (C) Igor Sysoev + * Copyright (C) NGINX, Inc. + */ + +#include <nxt_main.h> + + +/* + * pollset has been introduced in AIX 5L 5.3. + * + * pollset_create() returns a pollset_t descriptor which is not + * a file descriptor, so it cannot be added to another pollset. + * The first pollset_create() call returns 0. + */ + + +#define NXT_POLLSET_ADD 0 +#define NXT_POLLSET_UPDATE 1 +#define NXT_POLLSET_CHANGE 2 +#define NXT_POLLSET_DELETE 3 + + +static nxt_int_t nxt_pollset_create(nxt_event_engine_t *engine, + nxt_uint_t mchanges, nxt_uint_t mevents); +static void nxt_pollset_free(nxt_event_engine_t *engine); +static void nxt_pollset_enable(nxt_event_engine_t *engine, nxt_fd_event_t *ev); +static void nxt_pollset_disable(nxt_event_engine_t *engine, nxt_fd_event_t *ev); +static nxt_bool_t nxt_pollset_close(nxt_event_engine_t *engine, + nxt_fd_event_t *ev); +static void nxt_pollset_enable_read(nxt_event_engine_t *engine, + nxt_fd_event_t *ev); +static void nxt_pollset_enable_write(nxt_event_engine_t *engine, + nxt_fd_event_t *ev); +static void nxt_pollset_disable_read(nxt_event_engine_t *engine, + nxt_fd_event_t *ev); +static void nxt_pollset_disable_write(nxt_event_engine_t *engine, + nxt_fd_event_t *ev); +static void nxt_pollset_block_read(nxt_event_engine_t *engine, + nxt_fd_event_t *ev); +static void nxt_pollset_block_write(nxt_event_engine_t *engine, + nxt_fd_event_t *ev); +static void nxt_pollset_oneshot_read(nxt_event_engine_t *engine, + nxt_fd_event_t *ev); +static void nxt_pollset_oneshot_write(nxt_event_engine_t *engine, + nxt_fd_event_t *ev); +static void nxt_pollset_change(nxt_event_engine_t *engine, nxt_fd_event_t *ev, + nxt_uint_t op, nxt_uint_t events); +static nxt_int_t nxt_pollset_commit_changes(nxt_event_engine_t *engine); +static void nxt_pollset_change_error(nxt_event_engine_t *engine, + nxt_fd_event_t *ev); +static void nxt_pollset_remove(nxt_event_engine_t *engine, nxt_fd_t fd); +static nxt_int_t nxt_pollset_write(nxt_event_engine_t *engine, + struct poll_ctl *ctl, int n); +static void nxt_pollset_poll(nxt_event_engine_t *engine, nxt_msec_t timeout); + + +const nxt_event_interface_t nxt_pollset_engine = { + "pollset", + nxt_pollset_create, + nxt_pollset_free, + nxt_pollset_enable, + nxt_pollset_disable, + nxt_pollset_disable, + nxt_pollset_close, + nxt_pollset_enable_read, + nxt_pollset_enable_write, + nxt_pollset_disable_read, + nxt_pollset_disable_write, + nxt_pollset_block_read, + nxt_pollset_block_write, + nxt_pollset_oneshot_read, + nxt_pollset_oneshot_write, + nxt_pollset_enable_read, + NULL, + NULL, + NULL, + NULL, + nxt_pollset_poll, + + &nxt_unix_event_conn_io, + + NXT_NO_FILE_EVENTS, + NXT_NO_SIGNAL_EVENTS, +}; + + +static nxt_int_t +nxt_pollset_create(nxt_event_engine_t *engine, nxt_uint_t mchanges, + nxt_uint_t mevents) +{ + void *changes; + + engine->u.pollset.ps = -1; + engine->u.pollset.mchanges = mchanges; + engine->u.pollset.mevents = mevents; + + changes = nxt_malloc(sizeof(nxt_pollset_change_t) * mchanges); + if (changes == NULL) { + goto fail; + } + + engine->u.pollset.changes = changes; + + /* + * NXT_POLLSET_CHANGE requires two struct poll_ctl's + * for PS_DELETE and subsequent PS_ADD. + */ + changes = nxt_malloc(2 * sizeof(struct poll_ctl) * mchanges); + if (changes == NULL) { + goto fail; + } + + engine->u.pollset.write_changes = changes; + + engine->u.pollset.events = nxt_malloc(sizeof(struct pollfd) * mevents); + if (engine->u.pollset.events == NULL) { + goto fail; + } + + engine->u.pollset.ps = pollset_create(-1); + + if (engine->u.pollset.ps == -1) { + nxt_log(&engine->task, NXT_LOG_CRIT, "pollset_create() failed %E", + nxt_errno); + goto fail; + } + + nxt_debug(&engine->task, "pollset_create(): %d", engine->u.pollset.ps); + + return NXT_OK; + +fail: + + nxt_pollset_free(engine); + + return NXT_ERROR; +} + + +static void +nxt_pollset_free(nxt_event_engine_t *engine) +{ + pollset_t ps; + + ps = engine->u.pollset.ps; + + nxt_debug(&engine->task, "pollset %d free", ps); + + if (ps != -1 && pollset_destroy(ps) != 0) { + nxt_log(&engine->task, NXT_LOG_CRIT, "pollset_destroy(%d) failed %E", + ps, nxt_errno); + } + + nxt_free(engine->u.pollset.events); + nxt_free(engine->u.pollset.write_changes); + nxt_free(engine->u.pollset.changes); + nxt_fd_event_hash_destroy(&engine->u.pollset.fd_hash); + + nxt_memzero(&engine->u.pollset, sizeof(nxt_pollset_engine_t)); +} + + +static void +nxt_pollset_enable(nxt_event_engine_t *engine, nxt_fd_event_t *ev) +{ + ev->read = NXT_EVENT_ACTIVE; + ev->write = NXT_EVENT_ACTIVE; + + nxt_pollset_change(engine, ev, NXT_POLLSET_ADD, POLLIN | POLLOUT); +} + + +static void +nxt_pollset_disable(nxt_event_engine_t *engine, nxt_fd_event_t *ev) +{ + if (ev->read != NXT_EVENT_INACTIVE || ev->write != NXT_EVENT_INACTIVE) { + + ev->read = NXT_EVENT_INACTIVE; + ev->write = NXT_EVENT_INACTIVE; + + nxt_pollset_change(engine, ev, NXT_POLLSET_DELETE, 0); + } +} + + +/* + * A closed descriptor must be deleted from a pollset, otherwise next + * pollset_poll() will return POLLNVAL on it. However, pollset_ctl() + * allows to delete the already closed file descriptor from the pollset + * using PS_DELETE, so the removal can be batched, pollset_ctl(2): + * + * After a file descriptor is added to a pollset, the file descriptor will + * not be removed until a pollset_ctl call with the cmd of PS_DELETE is + * executed. The file descriptor remains in the pollset even if the file + * descriptor is closed. A pollset_poll operation on a pollset containing + * a closed file descriptor returns a POLLNVAL event for that file + * descriptor. If the file descriptor is later allocated to a new object, + * the new object will be polled on future pollset_poll calls. + */ + +static nxt_bool_t +nxt_pollset_close(nxt_event_engine_t *engine, nxt_fd_event_t *ev) +{ + nxt_pollset_disable(engine, ev); + + return ev->changing; +} + + +static void +nxt_pollset_enable_read(nxt_event_engine_t *engine, nxt_fd_event_t *ev) +{ + nxt_uint_t op, events; + + if (ev->read != NXT_EVENT_BLOCKED) { + + events = POLLIN; + + if (ev->write == NXT_EVENT_INACTIVE) { + op = NXT_POLLSET_ADD; + + } else if (ev->write == NXT_EVENT_BLOCKED) { + ev->write = NXT_EVENT_INACTIVE; + op = NXT_POLLSET_CHANGE; + + } else { + op = NXT_POLLSET_UPDATE; + events = POLLIN | POLLOUT; + } + + nxt_pollset_change(engine, ev, op, events); + } + + ev->read = NXT_EVENT_ACTIVE; +} + + +static void +nxt_pollset_enable_write(nxt_event_engine_t *engine, nxt_fd_event_t *ev) +{ + nxt_uint_t op, events; + + if (ev->write != NXT_EVENT_BLOCKED) { + + events = POLLOUT; + + if (ev->read == NXT_EVENT_INACTIVE) { + op = NXT_POLLSET_ADD; + + } else if (ev->read == NXT_EVENT_BLOCKED) { + ev->read = NXT_EVENT_INACTIVE; + op = NXT_POLLSET_CHANGE; + + } else { + op = NXT_POLLSET_UPDATE; + events = POLLIN | POLLOUT; + } + + nxt_pollset_change(engine, ev, op, events); + } + + ev->write = NXT_EVENT_ACTIVE; +} + + +static void +nxt_pollset_disable_read(nxt_event_engine_t *engine, nxt_fd_event_t *ev) +{ + nxt_uint_t op, events; + + ev->read = NXT_EVENT_INACTIVE; + + if (ev->write <= NXT_EVENT_BLOCKED) { + ev->write = NXT_EVENT_INACTIVE; + op = NXT_POLLSET_DELETE; + events = POLLREMOVE; + + } else { + op = NXT_POLLSET_CHANGE; + events = POLLOUT; + } + + nxt_pollset_change(engine, ev, op, events); +} + + +static void +nxt_pollset_disable_write(nxt_event_engine_t *engine, nxt_fd_event_t *ev) +{ + nxt_uint_t op, events; + + ev->write = NXT_EVENT_INACTIVE; + + if (ev->read <= NXT_EVENT_BLOCKED) { + ev->read = NXT_EVENT_INACTIVE; + op = NXT_POLLSET_DELETE; + events = POLLREMOVE; + + } else { + op = NXT_POLLSET_CHANGE; + events = POLLIN; + } + + nxt_pollset_change(engine, ev, op, events); +} + + +static void +nxt_pollset_block_read(nxt_event_engine_t *engine, nxt_fd_event_t *ev) +{ + if (ev->read != NXT_EVENT_INACTIVE) { + ev->read = NXT_EVENT_BLOCKED; + } +} + + +static void +nxt_pollset_block_write(nxt_event_engine_t *engine, nxt_fd_event_t *ev) +{ + if (ev->write != NXT_EVENT_INACTIVE) { + ev->write = NXT_EVENT_BLOCKED; + } +} + + +static void +nxt_pollset_oneshot_read(nxt_event_engine_t *engine, nxt_fd_event_t *ev) +{ + nxt_pollset_enable_read(engine, ev); + + ev->read = NXT_EVENT_ONESHOT; +} + + +static void +nxt_pollset_oneshot_write(nxt_event_engine_t *engine, nxt_fd_event_t *ev) +{ + nxt_pollset_enable_write(engine, ev); + + ev->write = NXT_EVENT_ONESHOT; +} + + +/* + * PS_ADD adds only a new file descriptor to a pollset. + * PS_DELETE removes a file descriptor from a pollset. + * + * PS_MOD can add a new file descriptor or modify events for a file + * descriptor which is already in a pollset. However, modified events + * are always ORed, so to delete an event for a file descriptor, + * the file descriptor must be removed using PS_DELETE and then + * added again without the event. + */ + +static void +nxt_pollset_change(nxt_event_engine_t *engine, nxt_fd_event_t *ev, + nxt_uint_t op, nxt_uint_t events) +{ + nxt_pollset_change_t *change; + + nxt_debug(ev->task, "pollset %d change fd:%d op:%ui ev:%04Xi", + engine->u.pollset.ps, ev->fd, op, events); + + if (engine->u.pollset.nchanges >= engine->u.pollset.mchanges) { + (void) nxt_pollset_commit_changes(engine); + } + + ev->changing = 1; + + change = &engine->u.pollset.changes[engine->u.pollset.nchanges++]; + change->op = op; + change->cmd = (op == NXT_POLLSET_DELETE) ? PS_DELETE : PS_MOD; + change->events = events; + change->event = ev; +} + + +static nxt_int_t +nxt_pollset_commit_changes(nxt_event_engine_t *engine) +{ + size_t n; + nxt_int_t ret, retval; + nxt_fd_event_t *ev; + struct poll_ctl *ctl, *write_changes; + nxt_pollset_change_t *change, *end; + + nxt_debug(&engine->task, "pollset %d changes:%ui", + engine->u.pollset.ps, engine->u.pollset.nchanges); + + retval = NXT_OK; + n = 0; + write_changes = engine->u.pollset.write_changes; + change = engine->u.pollset.changes; + end = change + engine->u.pollset.nchanges; + + do { + ev = change->event; + ev->changing = 0; + + nxt_debug(&engine->task, "pollset fd:%d op:%d ev:%04Xd", + ev->fd, change->op, change->events); + + if (change->op == NXT_POLLSET_CHANGE) { + ctl = &write_changes[n++]; + ctl->cmd = PS_DELETE; + ctl->events = 0; + ctl->fd = ev->fd; + } + + ctl = &write_changes[n++]; + ctl->cmd = change->cmd; + ctl->events = change->events; + ctl->fd = ev->fd; + + change++; + + } while (change < end); + + change = engine->u.pollset.changes; + end = change + engine->u.pollset.nchanges; + + ret = nxt_pollset_write(engine, write_changes, n); + + if (nxt_slow_path(ret != NXT_OK)) { + + do { + nxt_pollset_change_error(engine, change->event); + change++; + } while (change < end); + + engine->u.pollset.nchanges = 0; + + return NXT_ERROR; + } + + do { + ev = change->event; + + if (change->op == NXT_POLLSET_ADD) { + ret = nxt_fd_event_hash_add(&engine->u.pollset.fd_hash, ev->fd, ev); + + if (nxt_slow_path(ret != NXT_OK)) { + nxt_pollset_change_error(engine, ev); + retval = NXT_ERROR; + } + + } else if (change->op == NXT_POLLSET_DELETE) { + nxt_fd_event_hash_delete(&engine->task, &engine->u.pollset.fd_hash, + ev->fd, 0); + } + + /* Nothing to do for NXT_POLLSET_UPDATE and NXT_POLLSET_CHANGE. */ + + change++; + + } while (change < end); + + engine->u.pollset.nchanges = 0; + + return retval; +} + + +static void +nxt_pollset_change_error(nxt_event_engine_t *engine, nxt_fd_event_t *ev) +{ + ev->read = NXT_EVENT_INACTIVE; + ev->write = NXT_EVENT_INACTIVE; + + nxt_work_queue_add(&engine->fast_work_queue, ev->error_handler, + ev->task, ev, ev->data); + + nxt_fd_event_hash_delete(&engine->task, &engine->u.pollset.fd_hash, + ev->fd, 1); + + nxt_pollset_remove(engine, ev->fd); +} + + +static void +nxt_pollset_remove(nxt_event_engine_t *engine, nxt_fd_t fd) +{ + int n; + struct pollfd pfd; + struct poll_ctl ctl; + + pfd.fd = fd; + pfd.events = 0; + pfd.revents = 0; + + n = pollset_query(engine->u.pollset.ps, &pfd); + + nxt_debug(&engine->task, "pollset_query(%d, %d): %d", + engine->u.pollset.ps, fd, n); + + if (n == 0) { + /* The file descriptor is not in the pollset. */ + return; + } + + if (n == -1) { + nxt_log(&engine->task, NXT_LOG_CRIT, "pollset_query(%d, %d) failed %E", + engine->u.pollset.ps, fd, nxt_errno); + /* Fall through. */ + } + + /* n == 1: The file descriptor is in the pollset. */ + + nxt_debug(&engine->task, "pollset %d remove fd:%d", + engine->u.pollset.ps, fd); + + ctl.cmd = PS_DELETE; + ctl.events = 0; + ctl.fd = fd; + + nxt_pollset_write(engine, &ctl, 1); +} + + +static nxt_int_t +nxt_pollset_write(nxt_event_engine_t *engine, struct poll_ctl *ctl, int n) +{ + pollset_t ps; + + ps = engine->u.pollset.ps; + + nxt_debug(&engine->task, "pollset_ctl(%d) changes:%d", ps, n); + + nxt_set_errno(0); + + n = pollset_ctl(ps, ctl, n); + + if (nxt_fast_path(n == 0)) { + return NXT_OK; + } + + nxt_log(&engine->task, NXT_LOG_CRIT, "pollset_ctl(%d) failed: %d %E", + ps, n, nxt_errno); + + return NXT_ERROR; +} + + +static void +nxt_pollset_poll(nxt_event_engine_t *engine, nxt_msec_t timeout) +{ + int nevents; + nxt_fd_t fd; + nxt_int_t i; + nxt_err_t err; + nxt_uint_t events, level; + struct pollfd *pfd; + nxt_fd_event_t *ev; + + if (engine->u.pollset.nchanges != 0) { + if (nxt_pollset_commit_changes(engine) != NXT_OK) { + /* Error handlers have been enqueued on failure. */ + timeout = 0; + } + } + + nxt_debug(&engine->task, "pollset_poll(%d) timeout:%M", + engine->u.pollset.ps, timeout); + + nevents = pollset_poll(engine->u.pollset.ps, engine->u.pollset.events, + engine->u.pollset.mevents, timeout); + + err = (nevents == -1) ? nxt_errno : 0; + + nxt_thread_time_update(engine->task.thread); + + nxt_debug(&engine->task, "pollset_poll(%d): %d", + engine->u.pollset.ps, nevents); + + if (nevents == -1) { + level = (err == NXT_EINTR) ? NXT_LOG_INFO : NXT_LOG_CRIT; + + nxt_log(&engine->task, level, "pollset_poll(%d) failed %E", + engine->u.pollset.ps, err); + + return; + } + + for (i = 0; i < nevents; i++) { + + pfd = &engine->u.pollset.events[i]; + fd = pfd->fd; + events = pfd->revents; + + ev = nxt_fd_event_hash_get(&engine->task, &engine->u.pollset.fd_hash, + fd); + + if (nxt_slow_path(ev == NULL)) { + nxt_log(&engine->task, NXT_LOG_CRIT, + "pollset_poll(%d) returned invalid " + "fd:%d ev:%04Xd rev:%04uXi", + engine->u.pollset.ps, fd, pfd->events, events); + + nxt_pollset_remove(engine, fd); + continue; + } + + nxt_debug(ev->task, "pollset: fd:%d ev:%04uXi", fd, events); + + if (nxt_slow_path(events & (POLLERR | POLLHUP | POLLNVAL)) != 0) { + nxt_log(ev->task, NXT_LOG_CRIT, + "pollset_poll(%d) error fd:%d ev:%04Xd rev:%04uXi", + engine->u.pollset.ps, fd, pfd->events, events); + + nxt_work_queue_add(&engine->fast_work_queue, ev->error_handler, + ev->task, ev, ev->data); + continue; + } + + if (events & POLLIN) { + ev->read_ready = 1; + + if (ev->read != NXT_EVENT_BLOCKED) { + nxt_work_queue_add(ev->read_work_queue, ev->read_handler, + ev->task, ev, ev->data); + } + + if (ev->read == NXT_EVENT_BLOCKED + || ev->read == NXT_EVENT_ONESHOT) + { + nxt_pollset_disable_read(engine, ev); + } + } + + if (events & POLLOUT) { + ev->write_ready = 1; + + if (ev->write != NXT_EVENT_BLOCKED) { + nxt_work_queue_add(ev->write_work_queue, ev->write_handler, + ev->task, ev, ev->data); + } + + if (ev->write == NXT_EVENT_BLOCKED + || ev->write == NXT_EVENT_ONESHOT) + { + nxt_pollset_disable_write(engine, ev); + } + } + } +} |