initial commit

This commit is contained in:
2022-10-31 21:37:30 +09:00
commit 4de58743c2
1251 changed files with 252671 additions and 0 deletions
+561
View File
@@ -0,0 +1,561 @@
/*
* Copyright (C) Igor Sysoev
* Copyright (C) Nginx, Inc.
*/
#include <ngx_config.h>
#include <ngx_core.h>
#include <ngx_event.h>
#if (NGX_TEST_BUILD_DEVPOLL)
/* Solaris declarations */
#ifndef POLLREMOVE
#define POLLREMOVE 0x0800
#endif
#define DP_POLL 0xD001
#define DP_ISPOLLED 0xD002
struct dvpoll {
struct pollfd *dp_fds;
int dp_nfds;
int dp_timeout;
};
#endif
typedef struct {
ngx_uint_t changes;
ngx_uint_t events;
} ngx_devpoll_conf_t;
static ngx_int_t ngx_devpoll_init(ngx_cycle_t *cycle, ngx_msec_t timer);
static void ngx_devpoll_done(ngx_cycle_t *cycle);
static ngx_int_t ngx_devpoll_add_event(ngx_event_t *ev, ngx_int_t event,
ngx_uint_t flags);
static ngx_int_t ngx_devpoll_del_event(ngx_event_t *ev, ngx_int_t event,
ngx_uint_t flags);
static ngx_int_t ngx_devpoll_set_event(ngx_event_t *ev, ngx_int_t event,
ngx_uint_t flags);
static ngx_int_t ngx_devpoll_process_events(ngx_cycle_t *cycle,
ngx_msec_t timer, ngx_uint_t flags);
static void *ngx_devpoll_create_conf(ngx_cycle_t *cycle);
static char *ngx_devpoll_init_conf(ngx_cycle_t *cycle, void *conf);
static int dp = -1;
static struct pollfd *change_list, *event_list;
static ngx_uint_t nchanges, max_changes, nevents;
static ngx_event_t **change_index;
static ngx_str_t devpoll_name = ngx_string("/dev/poll");
static ngx_command_t ngx_devpoll_commands[] = {
{ ngx_string("devpoll_changes"),
NGX_EVENT_CONF|NGX_CONF_TAKE1,
ngx_conf_set_num_slot,
0,
offsetof(ngx_devpoll_conf_t, changes),
NULL },
{ ngx_string("devpoll_events"),
NGX_EVENT_CONF|NGX_CONF_TAKE1,
ngx_conf_set_num_slot,
0,
offsetof(ngx_devpoll_conf_t, events),
NULL },
ngx_null_command
};
static ngx_event_module_t ngx_devpoll_module_ctx = {
&devpoll_name,
ngx_devpoll_create_conf, /* create configuration */
ngx_devpoll_init_conf, /* init configuration */
{
ngx_devpoll_add_event, /* add an event */
ngx_devpoll_del_event, /* delete an event */
ngx_devpoll_add_event, /* enable an event */
ngx_devpoll_del_event, /* disable an event */
NULL, /* add an connection */
NULL, /* delete an connection */
NULL, /* trigger a notify */
ngx_devpoll_process_events, /* process the events */
ngx_devpoll_init, /* init the events */
ngx_devpoll_done, /* done the events */
}
};
ngx_module_t ngx_devpoll_module = {
NGX_MODULE_V1,
&ngx_devpoll_module_ctx, /* module context */
ngx_devpoll_commands, /* module directives */
NGX_EVENT_MODULE, /* module type */
NULL, /* init master */
NULL, /* init module */
NULL, /* init process */
NULL, /* init thread */
NULL, /* exit thread */
NULL, /* exit process */
NULL, /* exit master */
NGX_MODULE_V1_PADDING
};
static ngx_int_t
ngx_devpoll_init(ngx_cycle_t *cycle, ngx_msec_t timer)
{
size_t n;
ngx_devpoll_conf_t *dpcf;
dpcf = ngx_event_get_conf(cycle->conf_ctx, ngx_devpoll_module);
if (dp == -1) {
dp = open("/dev/poll", O_RDWR);
if (dp == -1) {
ngx_log_error(NGX_LOG_EMERG, cycle->log, ngx_errno,
"open(/dev/poll) failed");
return NGX_ERROR;
}
}
if (max_changes < dpcf->changes) {
if (nchanges) {
n = nchanges * sizeof(struct pollfd);
if (write(dp, change_list, n) != (ssize_t) n) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"write(/dev/poll) failed");
return NGX_ERROR;
}
nchanges = 0;
}
if (change_list) {
ngx_free(change_list);
}
change_list = ngx_alloc(sizeof(struct pollfd) * dpcf->changes,
cycle->log);
if (change_list == NULL) {
return NGX_ERROR;
}
if (change_index) {
ngx_free(change_index);
}
change_index = ngx_alloc(sizeof(ngx_event_t *) * dpcf->changes,
cycle->log);
if (change_index == NULL) {
return NGX_ERROR;
}
}
max_changes = dpcf->changes;
if (nevents < dpcf->events) {
if (event_list) {
ngx_free(event_list);
}
event_list = ngx_alloc(sizeof(struct pollfd) * dpcf->events,
cycle->log);
if (event_list == NULL) {
return NGX_ERROR;
}
}
nevents = dpcf->events;
ngx_io = ngx_os_io;
ngx_event_actions = ngx_devpoll_module_ctx.actions;
ngx_event_flags = NGX_USE_LEVEL_EVENT|NGX_USE_FD_EVENT;
return NGX_OK;
}
static void
ngx_devpoll_done(ngx_cycle_t *cycle)
{
if (close(dp) == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"close(/dev/poll) failed");
}
dp = -1;
ngx_free(change_list);
ngx_free(event_list);
ngx_free(change_index);
change_list = NULL;
event_list = NULL;
change_index = NULL;
max_changes = 0;
nchanges = 0;
nevents = 0;
}
static ngx_int_t
ngx_devpoll_add_event(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags)
{
#if (NGX_DEBUG)
ngx_connection_t *c;
#endif
#if (NGX_READ_EVENT != POLLIN)
event = (event == NGX_READ_EVENT) ? POLLIN : POLLOUT;
#endif
#if (NGX_DEBUG)
c = ev->data;
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"devpoll add event: fd:%d ev:%04Xi", c->fd, event);
#endif
ev->active = 1;
return ngx_devpoll_set_event(ev, event, 0);
}
static ngx_int_t
ngx_devpoll_del_event(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags)
{
ngx_event_t *e;
ngx_connection_t *c;
c = ev->data;
#if (NGX_READ_EVENT != POLLIN)
event = (event == NGX_READ_EVENT) ? POLLIN : POLLOUT;
#endif
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"devpoll del event: fd:%d ev:%04Xi", c->fd, event);
if (ngx_devpoll_set_event(ev, POLLREMOVE, flags) == NGX_ERROR) {
return NGX_ERROR;
}
ev->active = 0;
if (flags & NGX_CLOSE_EVENT) {
e = (event == POLLIN) ? c->write : c->read;
if (e) {
e->active = 0;
}
return NGX_OK;
}
/* restore the pair event if it exists */
if (event == POLLIN) {
e = c->write;
event = POLLOUT;
} else {
e = c->read;
event = POLLIN;
}
if (e && e->active) {
return ngx_devpoll_set_event(e, event, 0);
}
return NGX_OK;
}
static ngx_int_t
ngx_devpoll_set_event(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags)
{
size_t n;
ngx_connection_t *c;
c = ev->data;
ngx_log_debug3(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"devpoll fd:%d ev:%04Xi fl:%04Xi", c->fd, event, flags);
if (nchanges >= max_changes) {
ngx_log_error(NGX_LOG_WARN, ev->log, 0,
"/dev/pool change list is filled up");
n = nchanges * sizeof(struct pollfd);
if (write(dp, change_list, n) != (ssize_t) n) {
ngx_log_error(NGX_LOG_ALERT, ev->log, ngx_errno,
"write(/dev/poll) failed");
return NGX_ERROR;
}
nchanges = 0;
}
change_list[nchanges].fd = c->fd;
change_list[nchanges].events = (short) event;
change_list[nchanges].revents = 0;
change_index[nchanges] = ev;
ev->index = nchanges;
nchanges++;
if (flags & NGX_CLOSE_EVENT) {
n = nchanges * sizeof(struct pollfd);
if (write(dp, change_list, n) != (ssize_t) n) {
ngx_log_error(NGX_LOG_ALERT, ev->log, ngx_errno,
"write(/dev/poll) failed");
return NGX_ERROR;
}
nchanges = 0;
}
return NGX_OK;
}
static ngx_int_t
ngx_devpoll_process_events(ngx_cycle_t *cycle, ngx_msec_t timer,
ngx_uint_t flags)
{
int events, revents, rc;
size_t n;
ngx_fd_t fd;
ngx_err_t err;
ngx_int_t i;
ngx_uint_t level, instance;
ngx_event_t *rev, *wev;
ngx_queue_t *queue;
ngx_connection_t *c;
struct pollfd pfd;
struct dvpoll dvp;
/* NGX_TIMER_INFINITE == INFTIM */
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"devpoll timer: %M", timer);
if (nchanges) {
n = nchanges * sizeof(struct pollfd);
if (write(dp, change_list, n) != (ssize_t) n) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"write(/dev/poll) failed");
return NGX_ERROR;
}
nchanges = 0;
}
dvp.dp_fds = event_list;
dvp.dp_nfds = (int) nevents;
dvp.dp_timeout = timer;
events = ioctl(dp, DP_POLL, &dvp);
err = (events == -1) ? ngx_errno : 0;
if (flags & NGX_UPDATE_TIME || ngx_event_timer_alarm) {
ngx_time_update();
}
if (err) {
if (err == NGX_EINTR) {
if (ngx_event_timer_alarm) {
ngx_event_timer_alarm = 0;
return NGX_OK;
}
level = NGX_LOG_INFO;
} else {
level = NGX_LOG_ALERT;
}
ngx_log_error(level, cycle->log, err, "ioctl(DP_POLL) failed");
return NGX_ERROR;
}
if (events == 0) {
if (timer != NGX_TIMER_INFINITE) {
return NGX_OK;
}
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
"ioctl(DP_POLL) returned no events without timeout");
return NGX_ERROR;
}
for (i = 0; i < events; i++) {
fd = event_list[i].fd;
revents = event_list[i].revents;
c = ngx_cycle->files[fd];
if (c == NULL || c->fd == -1) {
pfd.fd = fd;
pfd.events = 0;
pfd.revents = 0;
rc = ioctl(dp, DP_ISPOLLED, &pfd);
switch (rc) {
case -1:
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"ioctl(DP_ISPOLLED) failed for socket %d, event %04Xd",
fd, revents);
break;
case 0:
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
"phantom event %04Xd for closed and removed socket %d",
revents, fd);
break;
default:
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
"unexpected event %04Xd for closed and removed socket %d, "
"ioctl(DP_ISPOLLED) returned rc:%d, fd:%d, event %04Xd",
revents, fd, rc, pfd.fd, pfd.revents);
pfd.fd = fd;
pfd.events = POLLREMOVE;
pfd.revents = 0;
if (write(dp, &pfd, sizeof(struct pollfd))
!= (ssize_t) sizeof(struct pollfd))
{
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"write(/dev/poll) for %d failed", fd);
}
if (close(fd) == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"close(%d) failed", fd);
}
break;
}
continue;
}
ngx_log_debug3(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"devpoll: fd:%d, ev:%04Xd, rev:%04Xd",
fd, event_list[i].events, revents);
if (revents & (POLLERR|POLLHUP|POLLNVAL)) {
ngx_log_debug3(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"ioctl(DP_POLL) error fd:%d ev:%04Xd rev:%04Xd",
fd, event_list[i].events, revents);
}
if (revents & ~(POLLIN|POLLOUT|POLLERR|POLLHUP|POLLNVAL)) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
"strange ioctl(DP_POLL) events "
"fd:%d ev:%04Xd rev:%04Xd",
fd, event_list[i].events, revents);
}
if (revents & (POLLERR|POLLHUP|POLLNVAL)) {
/*
* if the error events were returned, add POLLIN and POLLOUT
* to handle the events at least in one active handler
*/
revents |= POLLIN|POLLOUT;
}
rev = c->read;
if ((revents & POLLIN) && rev->active) {
rev->ready = 1;
rev->available = -1;
if (flags & NGX_POST_EVENTS) {
queue = rev->accept ? &ngx_posted_accept_events
: &ngx_posted_events;
ngx_post_event(rev, queue);
} else {
instance = rev->instance;
rev->handler(rev);
if (c->fd == -1 || rev->instance != instance) {
continue;
}
}
}
wev = c->write;
if ((revents & POLLOUT) && wev->active) {
wev->ready = 1;
if (flags & NGX_POST_EVENTS) {
ngx_post_event(wev, &ngx_posted_events);
} else {
wev->handler(wev);
}
}
}
return NGX_OK;
}
static void *
ngx_devpoll_create_conf(ngx_cycle_t *cycle)
{
ngx_devpoll_conf_t *dpcf;
dpcf = ngx_palloc(cycle->pool, sizeof(ngx_devpoll_conf_t));
if (dpcf == NULL) {
return NULL;
}
dpcf->changes = NGX_CONF_UNSET;
dpcf->events = NGX_CONF_UNSET;
return dpcf;
}
static char *
ngx_devpoll_init_conf(ngx_cycle_t *cycle, void *conf)
{
ngx_devpoll_conf_t *dpcf = conf;
ngx_conf_init_uint_value(dpcf->changes, 32);
ngx_conf_init_uint_value(dpcf->events, 32);
return NGX_CONF_OK;
}
+1051
View File
@@ -0,0 +1,1051 @@
/*
* Copyright (C) Igor Sysoev
* Copyright (C) Nginx, Inc.
*/
#include <ngx_config.h>
#include <ngx_core.h>
#include <ngx_event.h>
#if (NGX_TEST_BUILD_EPOLL)
/* epoll declarations */
#define EPOLLIN 0x001
#define EPOLLPRI 0x002
#define EPOLLOUT 0x004
#define EPOLLERR 0x008
#define EPOLLHUP 0x010
#define EPOLLRDNORM 0x040
#define EPOLLRDBAND 0x080
#define EPOLLWRNORM 0x100
#define EPOLLWRBAND 0x200
#define EPOLLMSG 0x400
#define EPOLLRDHUP 0x2000
#define EPOLLEXCLUSIVE 0x10000000
#define EPOLLONESHOT 0x40000000
#define EPOLLET 0x80000000
#define EPOLL_CTL_ADD 1
#define EPOLL_CTL_DEL 2
#define EPOLL_CTL_MOD 3
typedef union epoll_data {
void *ptr;
int fd;
uint32_t u32;
uint64_t u64;
} epoll_data_t;
struct epoll_event {
uint32_t events;
epoll_data_t data;
};
int epoll_create(int size);
int epoll_create(int size)
{
return -1;
}
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)
{
return -1;
}
int epoll_wait(int epfd, struct epoll_event *events, int nevents, int timeout);
int epoll_wait(int epfd, struct epoll_event *events, int nevents, int timeout)
{
return -1;
}
#if (NGX_HAVE_EVENTFD)
#define SYS_eventfd 323
#endif
#if (NGX_HAVE_FILE_AIO)
#define SYS_io_setup 245
#define SYS_io_destroy 246
#define SYS_io_getevents 247
typedef u_int aio_context_t;
struct io_event {
uint64_t data; /* the data field from the iocb */
uint64_t obj; /* what iocb this event came from */
int64_t res; /* result code for this event */
int64_t res2; /* secondary result */
};
#endif
#endif /* NGX_TEST_BUILD_EPOLL */
typedef struct {
ngx_uint_t events;
ngx_uint_t aio_requests;
} ngx_epoll_conf_t;
static ngx_int_t ngx_epoll_init(ngx_cycle_t *cycle, ngx_msec_t timer);
#if (NGX_HAVE_EVENTFD)
static ngx_int_t ngx_epoll_notify_init(ngx_log_t *log);
static void ngx_epoll_notify_handler(ngx_event_t *ev);
#endif
#if (NGX_HAVE_EPOLLRDHUP)
static void ngx_epoll_test_rdhup(ngx_cycle_t *cycle);
#endif
static void ngx_epoll_done(ngx_cycle_t *cycle);
static ngx_int_t ngx_epoll_add_event(ngx_event_t *ev, ngx_int_t event,
ngx_uint_t flags);
static ngx_int_t ngx_epoll_del_event(ngx_event_t *ev, ngx_int_t event,
ngx_uint_t flags);
static ngx_int_t ngx_epoll_add_connection(ngx_connection_t *c);
static ngx_int_t ngx_epoll_del_connection(ngx_connection_t *c,
ngx_uint_t flags);
#if (NGX_HAVE_EVENTFD)
static ngx_int_t ngx_epoll_notify(ngx_event_handler_pt handler);
#endif
static ngx_int_t ngx_epoll_process_events(ngx_cycle_t *cycle, ngx_msec_t timer,
ngx_uint_t flags);
#if (NGX_HAVE_FILE_AIO)
static void ngx_epoll_eventfd_handler(ngx_event_t *ev);
#endif
static void *ngx_epoll_create_conf(ngx_cycle_t *cycle);
static char *ngx_epoll_init_conf(ngx_cycle_t *cycle, void *conf);
static int ep = -1;
static struct epoll_event *event_list;
static ngx_uint_t nevents;
#if (NGX_HAVE_EVENTFD)
static int notify_fd = -1;
static ngx_event_t notify_event;
static ngx_connection_t notify_conn;
#endif
#if (NGX_HAVE_FILE_AIO)
int ngx_eventfd = -1;
aio_context_t ngx_aio_ctx = 0;
static ngx_event_t ngx_eventfd_event;
static ngx_connection_t ngx_eventfd_conn;
#endif
#if (NGX_HAVE_EPOLLRDHUP)
ngx_uint_t ngx_use_epoll_rdhup;
#endif
static ngx_str_t epoll_name = ngx_string("epoll");
static ngx_command_t ngx_epoll_commands[] = {
{ ngx_string("epoll_events"),
NGX_EVENT_CONF|NGX_CONF_TAKE1,
ngx_conf_set_num_slot,
0,
offsetof(ngx_epoll_conf_t, events),
NULL },
{ ngx_string("worker_aio_requests"),
NGX_EVENT_CONF|NGX_CONF_TAKE1,
ngx_conf_set_num_slot,
0,
offsetof(ngx_epoll_conf_t, aio_requests),
NULL },
ngx_null_command
};
static ngx_event_module_t ngx_epoll_module_ctx = {
&epoll_name,
ngx_epoll_create_conf, /* create configuration */
ngx_epoll_init_conf, /* init configuration */
{
ngx_epoll_add_event, /* add an event */
ngx_epoll_del_event, /* delete an event */
ngx_epoll_add_event, /* enable an event */
ngx_epoll_del_event, /* disable an event */
ngx_epoll_add_connection, /* add an connection */
ngx_epoll_del_connection, /* delete an connection */
#if (NGX_HAVE_EVENTFD)
ngx_epoll_notify, /* trigger a notify */
#else
NULL, /* trigger a notify */
#endif
ngx_epoll_process_events, /* process the events */
ngx_epoll_init, /* init the events */
ngx_epoll_done, /* done the events */
}
};
ngx_module_t ngx_epoll_module = {
NGX_MODULE_V1,
&ngx_epoll_module_ctx, /* module context */
ngx_epoll_commands, /* module directives */
NGX_EVENT_MODULE, /* module type */
NULL, /* init master */
NULL, /* init module */
NULL, /* init process */
NULL, /* init thread */
NULL, /* exit thread */
NULL, /* exit process */
NULL, /* exit master */
NGX_MODULE_V1_PADDING
};
#if (NGX_HAVE_FILE_AIO)
/*
* We call io_setup(), io_destroy() io_submit(), and io_getevents() directly
* as syscalls instead of libaio usage, because the library header file
* supports eventfd() since 0.3.107 version only.
*/
static int
io_setup(u_int nr_reqs, aio_context_t *ctx)
{
return syscall(SYS_io_setup, nr_reqs, ctx);
}
static int
io_destroy(aio_context_t ctx)
{
return syscall(SYS_io_destroy, ctx);
}
static int
io_getevents(aio_context_t ctx, long min_nr, long nr, struct io_event *events,
struct timespec *tmo)
{
return syscall(SYS_io_getevents, ctx, min_nr, nr, events, tmo);
}
static void
ngx_epoll_aio_init(ngx_cycle_t *cycle, ngx_epoll_conf_t *epcf)
{
int n;
struct epoll_event ee;
#if (NGX_HAVE_SYS_EVENTFD_H)
ngx_eventfd = eventfd(0, 0);
#else
ngx_eventfd = syscall(SYS_eventfd, 0);
#endif
if (ngx_eventfd == -1) {
ngx_log_error(NGX_LOG_EMERG, cycle->log, ngx_errno,
"eventfd() failed");
ngx_file_aio = 0;
return;
}
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"eventfd: %d", ngx_eventfd);
n = 1;
if (ioctl(ngx_eventfd, FIONBIO, &n) == -1) {
ngx_log_error(NGX_LOG_EMERG, cycle->log, ngx_errno,
"ioctl(eventfd, FIONBIO) failed");
goto failed;
}
if (io_setup(epcf->aio_requests, &ngx_aio_ctx) == -1) {
ngx_log_error(NGX_LOG_EMERG, cycle->log, ngx_errno,
"io_setup() failed");
goto failed;
}
ngx_eventfd_event.data = &ngx_eventfd_conn;
ngx_eventfd_event.handler = ngx_epoll_eventfd_handler;
ngx_eventfd_event.log = cycle->log;
ngx_eventfd_event.active = 1;
ngx_eventfd_conn.fd = ngx_eventfd;
ngx_eventfd_conn.read = &ngx_eventfd_event;
ngx_eventfd_conn.log = cycle->log;
ee.events = EPOLLIN|EPOLLET;
ee.data.ptr = &ngx_eventfd_conn;
if (epoll_ctl(ep, EPOLL_CTL_ADD, ngx_eventfd, &ee) != -1) {
return;
}
ngx_log_error(NGX_LOG_EMERG, cycle->log, ngx_errno,
"epoll_ctl(EPOLL_CTL_ADD, eventfd) failed");
if (io_destroy(ngx_aio_ctx) == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"io_destroy() failed");
}
failed:
if (close(ngx_eventfd) == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"eventfd close() failed");
}
ngx_eventfd = -1;
ngx_aio_ctx = 0;
ngx_file_aio = 0;
}
#endif
static ngx_int_t
ngx_epoll_init(ngx_cycle_t *cycle, ngx_msec_t timer)
{
ngx_epoll_conf_t *epcf;
epcf = ngx_event_get_conf(cycle->conf_ctx, ngx_epoll_module);
if (ep == -1) {
ep = epoll_create(cycle->connection_n / 2);
if (ep == -1) {
ngx_log_error(NGX_LOG_EMERG, cycle->log, ngx_errno,
"epoll_create() failed");
return NGX_ERROR;
}
#if (NGX_HAVE_EVENTFD)
if (ngx_epoll_notify_init(cycle->log) != NGX_OK) {
ngx_epoll_module_ctx.actions.notify = NULL;
}
#endif
#if (NGX_HAVE_FILE_AIO)
ngx_epoll_aio_init(cycle, epcf);
#endif
#if (NGX_HAVE_EPOLLRDHUP)
ngx_epoll_test_rdhup(cycle);
#endif
}
if (nevents < epcf->events) {
if (event_list) {
ngx_free(event_list);
}
event_list = ngx_alloc(sizeof(struct epoll_event) * epcf->events,
cycle->log);
if (event_list == NULL) {
return NGX_ERROR;
}
}
nevents = epcf->events;
ngx_io = ngx_os_io;
ngx_event_actions = ngx_epoll_module_ctx.actions;
#if (NGX_HAVE_CLEAR_EVENT)
ngx_event_flags = NGX_USE_CLEAR_EVENT
#else
ngx_event_flags = NGX_USE_LEVEL_EVENT
#endif
|NGX_USE_GREEDY_EVENT
|NGX_USE_EPOLL_EVENT;
return NGX_OK;
}
#if (NGX_HAVE_EVENTFD)
static ngx_int_t
ngx_epoll_notify_init(ngx_log_t *log)
{
struct epoll_event ee;
#if (NGX_HAVE_SYS_EVENTFD_H)
notify_fd = eventfd(0, 0);
#else
notify_fd = syscall(SYS_eventfd, 0);
#endif
if (notify_fd == -1) {
ngx_log_error(NGX_LOG_EMERG, log, ngx_errno, "eventfd() failed");
return NGX_ERROR;
}
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, log, 0,
"notify eventfd: %d", notify_fd);
notify_event.handler = ngx_epoll_notify_handler;
notify_event.log = log;
notify_event.active = 1;
notify_conn.fd = notify_fd;
notify_conn.read = &notify_event;
notify_conn.log = log;
ee.events = EPOLLIN|EPOLLET;
ee.data.ptr = &notify_conn;
if (epoll_ctl(ep, EPOLL_CTL_ADD, notify_fd, &ee) == -1) {
ngx_log_error(NGX_LOG_EMERG, log, ngx_errno,
"epoll_ctl(EPOLL_CTL_ADD, eventfd) failed");
if (close(notify_fd) == -1) {
ngx_log_error(NGX_LOG_ALERT, log, ngx_errno,
"eventfd close() failed");
}
return NGX_ERROR;
}
return NGX_OK;
}
static void
ngx_epoll_notify_handler(ngx_event_t *ev)
{
ssize_t n;
uint64_t count;
ngx_err_t err;
ngx_event_handler_pt handler;
if (++ev->index == NGX_MAX_UINT32_VALUE) {
ev->index = 0;
n = read(notify_fd, &count, sizeof(uint64_t));
err = ngx_errno;
ngx_log_debug3(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"read() eventfd %d: %z count:%uL", notify_fd, n, count);
if ((size_t) n != sizeof(uint64_t)) {
ngx_log_error(NGX_LOG_ALERT, ev->log, err,
"read() eventfd %d failed", notify_fd);
}
}
handler = ev->data;
handler(ev);
}
#endif
#if (NGX_HAVE_EPOLLRDHUP)
static void
ngx_epoll_test_rdhup(ngx_cycle_t *cycle)
{
int s[2], events;
struct epoll_event ee;
if (socketpair(AF_UNIX, SOCK_STREAM, 0, s) == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"socketpair() failed");
return;
}
ee.events = EPOLLET|EPOLLIN|EPOLLRDHUP;
if (epoll_ctl(ep, EPOLL_CTL_ADD, s[0], &ee) == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"epoll_ctl() failed");
goto failed;
}
if (close(s[1]) == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"close() failed");
s[1] = -1;
goto failed;
}
s[1] = -1;
events = epoll_wait(ep, &ee, 1, 5000);
if (events == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"epoll_wait() failed");
goto failed;
}
if (events) {
ngx_use_epoll_rdhup = ee.events & EPOLLRDHUP;
} else {
ngx_log_error(NGX_LOG_ALERT, cycle->log, NGX_ETIMEDOUT,
"epoll_wait() timed out");
}
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"testing the EPOLLRDHUP flag: %s",
ngx_use_epoll_rdhup ? "success" : "fail");
failed:
if (s[1] != -1 && close(s[1]) == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"close() failed");
}
if (close(s[0]) == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"close() failed");
}
}
#endif
static void
ngx_epoll_done(ngx_cycle_t *cycle)
{
if (close(ep) == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"epoll close() failed");
}
ep = -1;
#if (NGX_HAVE_EVENTFD)
if (close(notify_fd) == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"eventfd close() failed");
}
notify_fd = -1;
#endif
#if (NGX_HAVE_FILE_AIO)
if (ngx_eventfd != -1) {
if (io_destroy(ngx_aio_ctx) == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"io_destroy() failed");
}
if (close(ngx_eventfd) == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"eventfd close() failed");
}
ngx_eventfd = -1;
}
ngx_aio_ctx = 0;
#endif
ngx_free(event_list);
event_list = NULL;
nevents = 0;
}
static ngx_int_t
ngx_epoll_add_event(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags)
{
int op;
uint32_t events, prev;
ngx_event_t *e;
ngx_connection_t *c;
struct epoll_event ee;
c = ev->data;
events = (uint32_t) event;
if (event == NGX_READ_EVENT) {
e = c->write;
prev = EPOLLOUT;
#if (NGX_READ_EVENT != EPOLLIN|EPOLLRDHUP)
events = EPOLLIN|EPOLLRDHUP;
#endif
} else {
e = c->read;
prev = EPOLLIN|EPOLLRDHUP;
#if (NGX_WRITE_EVENT != EPOLLOUT)
events = EPOLLOUT;
#endif
}
if (e->active) {
op = EPOLL_CTL_MOD;
events |= prev;
} else {
op = EPOLL_CTL_ADD;
}
#if (NGX_HAVE_EPOLLEXCLUSIVE && NGX_HAVE_EPOLLRDHUP)
if (flags & NGX_EXCLUSIVE_EVENT) {
events &= ~EPOLLRDHUP;
}
#endif
ee.events = events | (uint32_t) flags;
ee.data.ptr = (void *) ((uintptr_t) c | ev->instance);
ngx_log_debug3(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"epoll add event: fd:%d op:%d ev:%08XD",
c->fd, op, ee.events);
if (epoll_ctl(ep, op, c->fd, &ee) == -1) {
ngx_log_error(NGX_LOG_ALERT, ev->log, ngx_errno,
"epoll_ctl(%d, %d) failed", op, c->fd);
return NGX_ERROR;
}
ev->active = 1;
#if 0
ev->oneshot = (flags & NGX_ONESHOT_EVENT) ? 1 : 0;
#endif
return NGX_OK;
}
static ngx_int_t
ngx_epoll_del_event(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags)
{
int op;
uint32_t prev;
ngx_event_t *e;
ngx_connection_t *c;
struct epoll_event ee;
/*
* when the file descriptor is closed, the epoll automatically deletes
* it from its queue, so we do not need to delete explicitly the event
* before the closing the file descriptor
*/
if (flags & NGX_CLOSE_EVENT) {
ev->active = 0;
return NGX_OK;
}
c = ev->data;
if (event == NGX_READ_EVENT) {
e = c->write;
prev = EPOLLOUT;
} else {
e = c->read;
prev = EPOLLIN|EPOLLRDHUP;
}
if (e->active) {
op = EPOLL_CTL_MOD;
ee.events = prev | (uint32_t) flags;
ee.data.ptr = (void *) ((uintptr_t) c | ev->instance);
} else {
op = EPOLL_CTL_DEL;
ee.events = 0;
ee.data.ptr = NULL;
}
ngx_log_debug3(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"epoll del event: fd:%d op:%d ev:%08XD",
c->fd, op, ee.events);
if (epoll_ctl(ep, op, c->fd, &ee) == -1) {
ngx_log_error(NGX_LOG_ALERT, ev->log, ngx_errno,
"epoll_ctl(%d, %d) failed", op, c->fd);
return NGX_ERROR;
}
ev->active = 0;
return NGX_OK;
}
static ngx_int_t
ngx_epoll_add_connection(ngx_connection_t *c)
{
struct epoll_event ee;
ee.events = EPOLLIN|EPOLLOUT|EPOLLET|EPOLLRDHUP;
ee.data.ptr = (void *) ((uintptr_t) c | c->read->instance);
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, c->log, 0,
"epoll add connection: fd:%d ev:%08XD", c->fd, ee.events);
if (epoll_ctl(ep, EPOLL_CTL_ADD, c->fd, &ee) == -1) {
ngx_log_error(NGX_LOG_ALERT, c->log, ngx_errno,
"epoll_ctl(EPOLL_CTL_ADD, %d) failed", c->fd);
return NGX_ERROR;
}
c->read->active = 1;
c->write->active = 1;
return NGX_OK;
}
static ngx_int_t
ngx_epoll_del_connection(ngx_connection_t *c, ngx_uint_t flags)
{
int op;
struct epoll_event ee;
/*
* when the file descriptor is closed the epoll automatically deletes
* it from its queue so we do not need to delete explicitly the event
* before the closing the file descriptor
*/
if (flags & NGX_CLOSE_EVENT) {
c->read->active = 0;
c->write->active = 0;
return NGX_OK;
}
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, c->log, 0,
"epoll del connection: fd:%d", c->fd);
op = EPOLL_CTL_DEL;
ee.events = 0;
ee.data.ptr = NULL;
if (epoll_ctl(ep, op, c->fd, &ee) == -1) {
ngx_log_error(NGX_LOG_ALERT, c->log, ngx_errno,
"epoll_ctl(%d, %d) failed", op, c->fd);
return NGX_ERROR;
}
c->read->active = 0;
c->write->active = 0;
return NGX_OK;
}
#if (NGX_HAVE_EVENTFD)
static ngx_int_t
ngx_epoll_notify(ngx_event_handler_pt handler)
{
static uint64_t inc = 1;
notify_event.data = handler;
if ((size_t) write(notify_fd, &inc, sizeof(uint64_t)) != sizeof(uint64_t)) {
ngx_log_error(NGX_LOG_ALERT, notify_event.log, ngx_errno,
"write() to eventfd %d failed", notify_fd);
return NGX_ERROR;
}
return NGX_OK;
}
#endif
static ngx_int_t
ngx_epoll_process_events(ngx_cycle_t *cycle, ngx_msec_t timer, ngx_uint_t flags)
{
int events;
uint32_t revents;
ngx_int_t instance, i;
ngx_uint_t level;
ngx_err_t err;
ngx_event_t *rev, *wev;
ngx_queue_t *queue;
ngx_connection_t *c;
/* NGX_TIMER_INFINITE == INFTIM */
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"epoll timer: %M", timer);
events = epoll_wait(ep, event_list, (int) nevents, timer);
err = (events == -1) ? ngx_errno : 0;
if (flags & NGX_UPDATE_TIME || ngx_event_timer_alarm) {
ngx_time_update();
}
if (err) {
if (err == NGX_EINTR) {
if (ngx_event_timer_alarm) {
ngx_event_timer_alarm = 0;
return NGX_OK;
}
level = NGX_LOG_INFO;
} else {
level = NGX_LOG_ALERT;
}
ngx_log_error(level, cycle->log, err, "epoll_wait() failed");
return NGX_ERROR;
}
if (events == 0) {
if (timer != NGX_TIMER_INFINITE) {
return NGX_OK;
}
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
"epoll_wait() returned no events without timeout");
return NGX_ERROR;
}
for (i = 0; i < events; i++) {
c = event_list[i].data.ptr;
instance = (uintptr_t) c & 1;
c = (ngx_connection_t *) ((uintptr_t) c & (uintptr_t) ~1);
rev = c->read;
if (c->fd == -1 || rev->instance != instance) {
/*
* the stale event from a file descriptor
* that was just closed in this iteration
*/
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"epoll: stale event %p", c);
continue;
}
revents = event_list[i].events;
ngx_log_debug3(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"epoll: fd:%d ev:%04XD d:%p",
c->fd, revents, event_list[i].data.ptr);
if (revents & (EPOLLERR|EPOLLHUP)) {
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"epoll_wait() error on fd:%d ev:%04XD",
c->fd, revents);
/*
* if the error events were returned, add EPOLLIN and EPOLLOUT
* to handle the events at least in one active handler
*/
revents |= EPOLLIN|EPOLLOUT;
}
#if 0
if (revents & ~(EPOLLIN|EPOLLOUT|EPOLLERR|EPOLLHUP)) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
"strange epoll_wait() events fd:%d ev:%04XD",
c->fd, revents);
}
#endif
if ((revents & EPOLLIN) && rev->active) {
#if (NGX_HAVE_EPOLLRDHUP)
if (revents & EPOLLRDHUP) {
rev->pending_eof = 1;
}
#endif
rev->ready = 1;
rev->available = -1;
if (flags & NGX_POST_EVENTS) {
queue = rev->accept ? &ngx_posted_accept_events
: &ngx_posted_events;
ngx_post_event(rev, queue);
} else {
rev->handler(rev);
}
}
wev = c->write;
if ((revents & EPOLLOUT) && wev->active) {
if (c->fd == -1 || wev->instance != instance) {
/*
* the stale event from a file descriptor
* that was just closed in this iteration
*/
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"epoll: stale event %p", c);
continue;
}
wev->ready = 1;
#if (NGX_THREADS)
wev->complete = 1;
#endif
if (flags & NGX_POST_EVENTS) {
ngx_post_event(wev, &ngx_posted_events);
} else {
wev->handler(wev);
}
}
}
return NGX_OK;
}
#if (NGX_HAVE_FILE_AIO)
static void
ngx_epoll_eventfd_handler(ngx_event_t *ev)
{
int n, events;
long i;
uint64_t ready;
ngx_err_t err;
ngx_event_t *e;
ngx_event_aio_t *aio;
struct io_event event[64];
struct timespec ts;
ngx_log_debug0(NGX_LOG_DEBUG_EVENT, ev->log, 0, "eventfd handler");
n = read(ngx_eventfd, &ready, 8);
err = ngx_errno;
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, ev->log, 0, "eventfd: %d", n);
if (n != 8) {
if (n == -1) {
if (err == NGX_EAGAIN) {
return;
}
ngx_log_error(NGX_LOG_ALERT, ev->log, err, "read(eventfd) failed");
return;
}
ngx_log_error(NGX_LOG_ALERT, ev->log, 0,
"read(eventfd) returned only %d bytes", n);
return;
}
ts.tv_sec = 0;
ts.tv_nsec = 0;
while (ready) {
events = io_getevents(ngx_aio_ctx, 1, 64, event, &ts);
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"io_getevents: %d", events);
if (events > 0) {
ready -= events;
for (i = 0; i < events; i++) {
ngx_log_debug4(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"io_event: %XL %XL %L %L",
event[i].data, event[i].obj,
event[i].res, event[i].res2);
e = (ngx_event_t *) (uintptr_t) event[i].data;
e->complete = 1;
e->active = 0;
e->ready = 1;
aio = e->data;
aio->res = event[i].res;
ngx_post_event(e, &ngx_posted_events);
}
continue;
}
if (events == 0) {
return;
}
/* events == -1 */
ngx_log_error(NGX_LOG_ALERT, ev->log, ngx_errno,
"io_getevents() failed");
return;
}
}
#endif
static void *
ngx_epoll_create_conf(ngx_cycle_t *cycle)
{
ngx_epoll_conf_t *epcf;
epcf = ngx_palloc(cycle->pool, sizeof(ngx_epoll_conf_t));
if (epcf == NULL) {
return NULL;
}
epcf->events = NGX_CONF_UNSET;
epcf->aio_requests = NGX_CONF_UNSET;
return epcf;
}
static char *
ngx_epoll_init_conf(ngx_cycle_t *cycle, void *conf)
{
ngx_epoll_conf_t *epcf = conf;
ngx_conf_init_uint_value(epcf->events, 512);
ngx_conf_init_uint_value(epcf->aio_requests, 32);
return NGX_CONF_OK;
}
+650
View File
@@ -0,0 +1,650 @@
/*
* Copyright (C) Igor Sysoev
* Copyright (C) Nginx, Inc.
*/
#include <ngx_config.h>
#include <ngx_core.h>
#include <ngx_event.h>
#if (NGX_TEST_BUILD_EVENTPORT)
#define ushort_t u_short
#define uint_t u_int
#ifndef CLOCK_REALTIME
#define CLOCK_REALTIME 0
typedef int clockid_t;
typedef void * timer_t;
#elif (NGX_DARWIN)
typedef void * timer_t;
#endif
/* Solaris declarations */
#define PORT_SOURCE_AIO 1
#define PORT_SOURCE_TIMER 2
#define PORT_SOURCE_USER 3
#define PORT_SOURCE_FD 4
#define PORT_SOURCE_ALERT 5
#define PORT_SOURCE_MQ 6
#ifndef ETIME
#define ETIME 64
#endif
#define SIGEV_PORT 4
typedef struct {
int portev_events; /* event data is source specific */
ushort_t portev_source; /* event source */
ushort_t portev_pad; /* port internal use */
uintptr_t portev_object; /* source specific object */
void *portev_user; /* user cookie */
} port_event_t;
typedef struct port_notify {
int portnfy_port; /* bind request(s) to port */
void *portnfy_user; /* user defined */
} port_notify_t;
#if (__FreeBSD__ && __FreeBSD_version < 700005) || (NGX_DARWIN)
typedef struct itimerspec { /* definition per POSIX.4 */
struct timespec it_interval;/* timer period */
struct timespec it_value; /* timer expiration */
} itimerspec_t;
#endif
int port_create(void);
int port_create(void)
{
return -1;
}
int port_associate(int port, int source, uintptr_t object, int events,
void *user);
int port_associate(int port, int source, uintptr_t object, int events,
void *user)
{
return -1;
}
int port_dissociate(int port, int source, uintptr_t object);
int port_dissociate(int port, int source, uintptr_t object)
{
return -1;
}
int port_getn(int port, port_event_t list[], uint_t max, uint_t *nget,
struct timespec *timeout);
int port_getn(int port, port_event_t list[], uint_t max, uint_t *nget,
struct timespec *timeout)
{
return -1;
}
int port_send(int port, int events, void *user);
int port_send(int port, int events, void *user)
{
return -1;
}
int timer_create(clockid_t clock_id, struct sigevent *evp, timer_t *timerid);
int timer_create(clockid_t clock_id, struct sigevent *evp, timer_t *timerid)
{
return -1;
}
int timer_settime(timer_t timerid, int flags, const struct itimerspec *value,
struct itimerspec *ovalue);
int timer_settime(timer_t timerid, int flags, const struct itimerspec *value,
struct itimerspec *ovalue)
{
return -1;
}
int timer_delete(timer_t timerid);
int timer_delete(timer_t timerid)
{
return -1;
}
#endif
typedef struct {
ngx_uint_t events;
} ngx_eventport_conf_t;
static ngx_int_t ngx_eventport_init(ngx_cycle_t *cycle, ngx_msec_t timer);
static void ngx_eventport_done(ngx_cycle_t *cycle);
static ngx_int_t ngx_eventport_add_event(ngx_event_t *ev, ngx_int_t event,
ngx_uint_t flags);
static ngx_int_t ngx_eventport_del_event(ngx_event_t *ev, ngx_int_t event,
ngx_uint_t flags);
static ngx_int_t ngx_eventport_notify(ngx_event_handler_pt handler);
static ngx_int_t ngx_eventport_process_events(ngx_cycle_t *cycle,
ngx_msec_t timer, ngx_uint_t flags);
static void *ngx_eventport_create_conf(ngx_cycle_t *cycle);
static char *ngx_eventport_init_conf(ngx_cycle_t *cycle, void *conf);
static int ep = -1;
static port_event_t *event_list;
static ngx_uint_t nevents;
static timer_t event_timer = (timer_t) -1;
static ngx_event_t notify_event;
static ngx_str_t eventport_name = ngx_string("eventport");
static ngx_command_t ngx_eventport_commands[] = {
{ ngx_string("eventport_events"),
NGX_EVENT_CONF|NGX_CONF_TAKE1,
ngx_conf_set_num_slot,
0,
offsetof(ngx_eventport_conf_t, events),
NULL },
ngx_null_command
};
static ngx_event_module_t ngx_eventport_module_ctx = {
&eventport_name,
ngx_eventport_create_conf, /* create configuration */
ngx_eventport_init_conf, /* init configuration */
{
ngx_eventport_add_event, /* add an event */
ngx_eventport_del_event, /* delete an event */
ngx_eventport_add_event, /* enable an event */
ngx_eventport_del_event, /* disable an event */
NULL, /* add an connection */
NULL, /* delete an connection */
ngx_eventport_notify, /* trigger a notify */
ngx_eventport_process_events, /* process the events */
ngx_eventport_init, /* init the events */
ngx_eventport_done, /* done the events */
}
};
ngx_module_t ngx_eventport_module = {
NGX_MODULE_V1,
&ngx_eventport_module_ctx, /* module context */
ngx_eventport_commands, /* module directives */
NGX_EVENT_MODULE, /* module type */
NULL, /* init master */
NULL, /* init module */
NULL, /* init process */
NULL, /* init thread */
NULL, /* exit thread */
NULL, /* exit process */
NULL, /* exit master */
NGX_MODULE_V1_PADDING
};
static ngx_int_t
ngx_eventport_init(ngx_cycle_t *cycle, ngx_msec_t timer)
{
port_notify_t pn;
struct itimerspec its;
struct sigevent sev;
ngx_eventport_conf_t *epcf;
epcf = ngx_event_get_conf(cycle->conf_ctx, ngx_eventport_module);
if (ep == -1) {
ep = port_create();
if (ep == -1) {
ngx_log_error(NGX_LOG_EMERG, cycle->log, ngx_errno,
"port_create() failed");
return NGX_ERROR;
}
notify_event.active = 1;
notify_event.log = cycle->log;
}
if (nevents < epcf->events) {
if (event_list) {
ngx_free(event_list);
}
event_list = ngx_alloc(sizeof(port_event_t) * epcf->events,
cycle->log);
if (event_list == NULL) {
return NGX_ERROR;
}
}
ngx_event_flags = NGX_USE_EVENTPORT_EVENT;
if (timer) {
ngx_memzero(&pn, sizeof(port_notify_t));
pn.portnfy_port = ep;
ngx_memzero(&sev, sizeof(struct sigevent));
sev.sigev_notify = SIGEV_PORT;
sev.sigev_value.sival_ptr = &pn;
if (timer_create(CLOCK_REALTIME, &sev, &event_timer) == -1) {
ngx_log_error(NGX_LOG_EMERG, cycle->log, ngx_errno,
"timer_create() failed");
return NGX_ERROR;
}
its.it_interval.tv_sec = timer / 1000;
its.it_interval.tv_nsec = (timer % 1000) * 1000000;
its.it_value.tv_sec = timer / 1000;
its.it_value.tv_nsec = (timer % 1000) * 1000000;
if (timer_settime(event_timer, 0, &its, NULL) == -1) {
ngx_log_error(NGX_LOG_EMERG, cycle->log, ngx_errno,
"timer_settime() failed");
return NGX_ERROR;
}
ngx_event_flags |= NGX_USE_TIMER_EVENT;
}
nevents = epcf->events;
ngx_io = ngx_os_io;
ngx_event_actions = ngx_eventport_module_ctx.actions;
return NGX_OK;
}
static void
ngx_eventport_done(ngx_cycle_t *cycle)
{
if (event_timer != (timer_t) -1) {
if (timer_delete(event_timer) == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"timer_delete() failed");
}
event_timer = (timer_t) -1;
}
if (close(ep) == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"close() event port failed");
}
ep = -1;
ngx_free(event_list);
event_list = NULL;
nevents = 0;
}
static ngx_int_t
ngx_eventport_add_event(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags)
{
ngx_int_t events, prev;
ngx_event_t *e;
ngx_connection_t *c;
c = ev->data;
events = event;
if (event == NGX_READ_EVENT) {
e = c->write;
prev = POLLOUT;
#if (NGX_READ_EVENT != POLLIN)
events = POLLIN;
#endif
} else {
e = c->read;
prev = POLLIN;
#if (NGX_WRITE_EVENT != POLLOUT)
events = POLLOUT;
#endif
}
if (e->oneshot) {
events |= prev;
}
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"eventport add event: fd:%d ev:%04Xi", c->fd, events);
if (port_associate(ep, PORT_SOURCE_FD, c->fd, events,
(void *) ((uintptr_t) ev | ev->instance))
== -1)
{
ngx_log_error(NGX_LOG_ALERT, ev->log, ngx_errno,
"port_associate() failed");
return NGX_ERROR;
}
ev->active = 1;
ev->oneshot = 1;
return NGX_OK;
}
static ngx_int_t
ngx_eventport_del_event(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags)
{
ngx_event_t *e;
ngx_connection_t *c;
/*
* when the file descriptor is closed, the event port automatically
* dissociates it from the port, so we do not need to dissociate explicitly
* the event before the closing the file descriptor
*/
if (flags & NGX_CLOSE_EVENT) {
ev->active = 0;
ev->oneshot = 0;
return NGX_OK;
}
c = ev->data;
if (event == NGX_READ_EVENT) {
e = c->write;
event = POLLOUT;
} else {
e = c->read;
event = POLLIN;
}
if (e->oneshot) {
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"eventport change event: fd:%d ev:%04Xi", c->fd, event);
if (port_associate(ep, PORT_SOURCE_FD, c->fd, event,
(void *) ((uintptr_t) ev | ev->instance))
== -1)
{
ngx_log_error(NGX_LOG_ALERT, ev->log, ngx_errno,
"port_associate() failed");
return NGX_ERROR;
}
} else if (ev->active) {
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"eventport del event: fd:%d", c->fd);
if (port_dissociate(ep, PORT_SOURCE_FD, c->fd) == -1) {
ngx_log_error(NGX_LOG_ALERT, ev->log, ngx_errno,
"port_dissociate() failed");
return NGX_ERROR;
}
}
ev->active = 0;
ev->oneshot = 0;
return NGX_OK;
}
static ngx_int_t
ngx_eventport_notify(ngx_event_handler_pt handler)
{
notify_event.handler = handler;
if (port_send(ep, 0, &notify_event) != 0) {
ngx_log_error(NGX_LOG_ALERT, notify_event.log, ngx_errno,
"port_send() failed");
return NGX_ERROR;
}
return NGX_OK;
}
static ngx_int_t
ngx_eventport_process_events(ngx_cycle_t *cycle, ngx_msec_t timer,
ngx_uint_t flags)
{
int n, revents;
u_int events;
ngx_err_t err;
ngx_int_t instance;
ngx_uint_t i, level;
ngx_event_t *ev, *rev, *wev;
ngx_queue_t *queue;
ngx_connection_t *c;
struct timespec ts, *tp;
if (timer == NGX_TIMER_INFINITE) {
tp = NULL;
} else {
ts.tv_sec = timer / 1000;
ts.tv_nsec = (timer % 1000) * 1000000;
tp = &ts;
}
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"eventport timer: %M", timer);
events = 1;
n = port_getn(ep, event_list, (u_int) nevents, &events, tp);
err = ngx_errno;
if (flags & NGX_UPDATE_TIME) {
ngx_time_update();
}
if (n == -1) {
if (err == ETIME) {
if (timer != NGX_TIMER_INFINITE) {
return NGX_OK;
}
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
"port_getn() returned no events without timeout");
return NGX_ERROR;
}
level = (err == NGX_EINTR) ? NGX_LOG_INFO : NGX_LOG_ALERT;
ngx_log_error(level, cycle->log, err, "port_getn() failed");
return NGX_ERROR;
}
if (events == 0) {
if (timer != NGX_TIMER_INFINITE) {
return NGX_OK;
}
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
"port_getn() returned no events without timeout");
return NGX_ERROR;
}
for (i = 0; i < events; i++) {
if (event_list[i].portev_source == PORT_SOURCE_TIMER) {
ngx_time_update();
continue;
}
ev = event_list[i].portev_user;
switch (event_list[i].portev_source) {
case PORT_SOURCE_FD:
instance = (uintptr_t) ev & 1;
ev = (ngx_event_t *) ((uintptr_t) ev & (uintptr_t) ~1);
if (ev->closed || ev->instance != instance) {
/*
* the stale event from a file descriptor
* that was just closed in this iteration
*/
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"eventport: stale event %p", ev);
continue;
}
revents = event_list[i].portev_events;
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"eventport: fd:%d, ev:%04Xd",
(int) event_list[i].portev_object, revents);
if (revents & (POLLERR|POLLHUP|POLLNVAL)) {
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"port_getn() error fd:%d ev:%04Xd",
(int) event_list[i].portev_object, revents);
}
if (revents & ~(POLLIN|POLLOUT|POLLERR|POLLHUP|POLLNVAL)) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
"strange port_getn() events fd:%d ev:%04Xd",
(int) event_list[i].portev_object, revents);
}
if (revents & (POLLERR|POLLHUP|POLLNVAL)) {
/*
* if the error events were returned, add POLLIN and POLLOUT
* to handle the events at least in one active handler
*/
revents |= POLLIN|POLLOUT;
}
c = ev->data;
rev = c->read;
wev = c->write;
rev->active = 0;
wev->active = 0;
if (revents & POLLIN) {
rev->ready = 1;
rev->available = -1;
if (flags & NGX_POST_EVENTS) {
queue = rev->accept ? &ngx_posted_accept_events
: &ngx_posted_events;
ngx_post_event(rev, queue);
} else {
rev->handler(rev);
if (ev->closed || ev->instance != instance) {
continue;
}
}
if (rev->accept) {
if (ngx_use_accept_mutex) {
ngx_accept_events = 1;
continue;
}
if (port_associate(ep, PORT_SOURCE_FD, c->fd, POLLIN,
(void *) ((uintptr_t) ev | ev->instance))
== -1)
{
ngx_log_error(NGX_LOG_ALERT, ev->log, ngx_errno,
"port_associate() failed");
return NGX_ERROR;
}
}
}
if (revents & POLLOUT) {
wev->ready = 1;
if (flags & NGX_POST_EVENTS) {
ngx_post_event(wev, &ngx_posted_events);
} else {
wev->handler(wev);
}
}
continue;
case PORT_SOURCE_USER:
ev->handler(ev);
continue;
default:
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
"unexpected eventport object %d",
(int) event_list[i].portev_object);
continue;
}
}
return NGX_OK;
}
static void *
ngx_eventport_create_conf(ngx_cycle_t *cycle)
{
ngx_eventport_conf_t *epcf;
epcf = ngx_palloc(cycle->pool, sizeof(ngx_eventport_conf_t));
if (epcf == NULL) {
return NULL;
}
epcf->events = NGX_CONF_UNSET;
return epcf;
}
static char *
ngx_eventport_init_conf(ngx_cycle_t *cycle, void *conf)
{
ngx_eventport_conf_t *epcf = conf;
ngx_conf_init_uint_value(epcf->events, 32);
return NGX_CONF_OK;
}
+379
View File
@@ -0,0 +1,379 @@
/*
* Copyright (C) Igor Sysoev
* Copyright (C) Nginx, Inc.
*/
#include <ngx_config.h>
#include <ngx_core.h>
#include <ngx_event.h>
#include <ngx_iocp_module.h>
static ngx_int_t ngx_iocp_init(ngx_cycle_t *cycle, ngx_msec_t timer);
static ngx_thread_value_t __stdcall ngx_iocp_timer(void *data);
static void ngx_iocp_done(ngx_cycle_t *cycle);
static ngx_int_t ngx_iocp_add_event(ngx_event_t *ev, ngx_int_t event,
ngx_uint_t key);
static ngx_int_t ngx_iocp_del_connection(ngx_connection_t *c, ngx_uint_t flags);
static ngx_int_t ngx_iocp_process_events(ngx_cycle_t *cycle, ngx_msec_t timer,
ngx_uint_t flags);
static void *ngx_iocp_create_conf(ngx_cycle_t *cycle);
static char *ngx_iocp_init_conf(ngx_cycle_t *cycle, void *conf);
static ngx_str_t iocp_name = ngx_string("iocp");
static ngx_command_t ngx_iocp_commands[] = {
{ ngx_string("iocp_threads"),
NGX_EVENT_CONF|NGX_CONF_TAKE1,
ngx_conf_set_num_slot,
0,
offsetof(ngx_iocp_conf_t, threads),
NULL },
{ ngx_string("post_acceptex"),
NGX_EVENT_CONF|NGX_CONF_TAKE1,
ngx_conf_set_num_slot,
0,
offsetof(ngx_iocp_conf_t, post_acceptex),
NULL },
{ ngx_string("acceptex_read"),
NGX_EVENT_CONF|NGX_CONF_FLAG,
ngx_conf_set_flag_slot,
0,
offsetof(ngx_iocp_conf_t, acceptex_read),
NULL },
ngx_null_command
};
static ngx_event_module_t ngx_iocp_module_ctx = {
&iocp_name,
ngx_iocp_create_conf, /* create configuration */
ngx_iocp_init_conf, /* init configuration */
{
ngx_iocp_add_event, /* add an event */
NULL, /* delete an event */
NULL, /* enable an event */
NULL, /* disable an event */
NULL, /* add an connection */
ngx_iocp_del_connection, /* delete an connection */
NULL, /* trigger a notify */
ngx_iocp_process_events, /* process the events */
ngx_iocp_init, /* init the events */
ngx_iocp_done /* done the events */
}
};
ngx_module_t ngx_iocp_module = {
NGX_MODULE_V1,
&ngx_iocp_module_ctx, /* module context */
ngx_iocp_commands, /* module directives */
NGX_EVENT_MODULE, /* module type */
NULL, /* init master */
NULL, /* init module */
NULL, /* init process */
NULL, /* init thread */
NULL, /* exit thread */
NULL, /* exit process */
NULL, /* exit master */
NGX_MODULE_V1_PADDING
};
ngx_os_io_t ngx_iocp_io = {
ngx_overlapped_wsarecv,
NULL,
ngx_udp_overlapped_wsarecv,
NULL,
NULL,
NULL,
ngx_overlapped_wsasend_chain,
0
};
static HANDLE iocp;
static ngx_tid_t timer_thread;
static ngx_msec_t msec;
static ngx_int_t
ngx_iocp_init(ngx_cycle_t *cycle, ngx_msec_t timer)
{
ngx_iocp_conf_t *cf;
cf = ngx_event_get_conf(cycle->conf_ctx, ngx_iocp_module);
if (iocp == NULL) {
iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0,
cf->threads);
}
if (iocp == NULL) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"CreateIoCompletionPort() failed");
return NGX_ERROR;
}
ngx_io = ngx_iocp_io;
ngx_event_actions = ngx_iocp_module_ctx.actions;
ngx_event_flags = NGX_USE_IOCP_EVENT;
if (timer == 0) {
return NGX_OK;
}
/*
* The waitable timer could not be used, because
* GetQueuedCompletionStatus() does not set a thread to alertable state
*/
if (timer_thread == NULL) {
msec = timer;
if (ngx_create_thread(&timer_thread, ngx_iocp_timer, &msec, cycle->log)
!= 0)
{
return NGX_ERROR;
}
}
ngx_event_flags |= NGX_USE_TIMER_EVENT;
return NGX_OK;
}
static ngx_thread_value_t __stdcall
ngx_iocp_timer(void *data)
{
ngx_msec_t timer = *(ngx_msec_t *) data;
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ngx_cycle->log, 0,
"THREAD %p %p", &msec, data);
for ( ;; ) {
Sleep(timer);
ngx_time_update();
#if 1
ngx_log_debug0(NGX_LOG_DEBUG_EVENT, ngx_cycle->log, 0, "timer");
#endif
}
#if defined(__WATCOMC__) || defined(__GNUC__)
return 0;
#endif
}
static void
ngx_iocp_done(ngx_cycle_t *cycle)
{
if (CloseHandle(iocp) == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"iocp CloseHandle() failed");
}
iocp = NULL;
}
static ngx_int_t
ngx_iocp_add_event(ngx_event_t *ev, ngx_int_t event, ngx_uint_t key)
{
ngx_connection_t *c;
c = (ngx_connection_t *) ev->data;
c->read->active = 1;
c->write->active = 1;
ngx_log_debug3(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"iocp add: fd:%d k:%ui ov:%p", c->fd, key, &ev->ovlp);
if (CreateIoCompletionPort((HANDLE) c->fd, iocp, key, 0) == NULL) {
ngx_log_error(NGX_LOG_ALERT, c->log, ngx_errno,
"CreateIoCompletionPort() failed");
return NGX_ERROR;
}
return NGX_OK;
}
static ngx_int_t
ngx_iocp_del_connection(ngx_connection_t *c, ngx_uint_t flags)
{
#if 0
if (flags & NGX_CLOSE_EVENT) {
return NGX_OK;
}
if (CancelIo((HANDLE) c->fd) == 0) {
ngx_log_error(NGX_LOG_ALERT, c->log, ngx_errno, "CancelIo() failed");
return NGX_ERROR;
}
#endif
return NGX_OK;
}
static ngx_int_t
ngx_iocp_process_events(ngx_cycle_t *cycle, ngx_msec_t timer, ngx_uint_t flags)
{
int rc;
u_int key;
u_long bytes;
ngx_err_t err;
ngx_msec_t delta;
ngx_event_t *ev;
ngx_event_ovlp_t *ovlp;
if (timer == NGX_TIMER_INFINITE) {
timer = INFINITE;
}
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0, "iocp timer: %M", timer);
rc = GetQueuedCompletionStatus(iocp, &bytes, (PULONG_PTR) &key,
(LPOVERLAPPED *) &ovlp, (u_long) timer);
if (rc == 0) {
err = ngx_errno;
} else {
err = 0;
}
delta = ngx_current_msec;
if (flags & NGX_UPDATE_TIME) {
ngx_time_update();
}
ngx_log_debug4(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"iocp: %d b:%d k:%d ov:%p", rc, bytes, key, ovlp);
if (timer != INFINITE) {
delta = ngx_current_msec - delta;
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"iocp timer: %M, delta: %M", timer, delta);
}
if (err) {
if (ovlp == NULL) {
if (err != WAIT_TIMEOUT) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, err,
"GetQueuedCompletionStatus() failed");
return NGX_ERROR;
}
return NGX_OK;
}
ovlp->error = err;
}
if (ovlp == NULL) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
"GetQueuedCompletionStatus() returned no operation");
return NGX_ERROR;
}
ev = ovlp->event;
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, err, "iocp event:%p", ev);
if (err == ERROR_NETNAME_DELETED /* the socket was closed */
|| err == ERROR_OPERATION_ABORTED /* the operation was canceled */)
{
/*
* the WSA_OPERATION_ABORTED completion notification
* for a file descriptor that was closed
*/
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, err,
"iocp: aborted event %p", ev);
return NGX_OK;
}
if (err) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, err,
"GetQueuedCompletionStatus() returned operation error");
}
switch (key) {
case NGX_IOCP_ACCEPT:
if (bytes) {
ev->ready = 1;
}
break;
case NGX_IOCP_IO:
ev->complete = 1;
ev->ready = 1;
break;
case NGX_IOCP_CONNECT:
ev->ready = 1;
}
ev->available = bytes;
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"iocp event handler: %p", ev->handler);
ev->handler(ev);
return NGX_OK;
}
static void *
ngx_iocp_create_conf(ngx_cycle_t *cycle)
{
ngx_iocp_conf_t *cf;
cf = ngx_palloc(cycle->pool, sizeof(ngx_iocp_conf_t));
if (cf == NULL) {
return NULL;
}
cf->threads = NGX_CONF_UNSET;
cf->post_acceptex = NGX_CONF_UNSET;
cf->acceptex_read = NGX_CONF_UNSET;
return cf;
}
static char *
ngx_iocp_init_conf(ngx_cycle_t *cycle, void *conf)
{
ngx_iocp_conf_t *cf = conf;
ngx_conf_init_value(cf->threads, 0);
ngx_conf_init_value(cf->post_acceptex, 10);
ngx_conf_init_value(cf->acceptex_read, 1);
return NGX_CONF_OK;
}
+22
View File
@@ -0,0 +1,22 @@
/*
* Copyright (C) Igor Sysoev
* Copyright (C) Nginx, Inc.
*/
#ifndef _NGX_IOCP_MODULE_H_INCLUDED_
#define _NGX_IOCP_MODULE_H_INCLUDED_
typedef struct {
int threads;
int post_acceptex;
int acceptex_read;
} ngx_iocp_conf_t;
extern ngx_module_t ngx_iocp_module;
#endif /* _NGX_IOCP_MODULE_H_INCLUDED_ */
+722
View File
@@ -0,0 +1,722 @@
/*
* Copyright (C) Igor Sysoev
* Copyright (C) Nginx, Inc.
*/
#include <ngx_config.h>
#include <ngx_core.h>
#include <ngx_event.h>
typedef struct {
ngx_uint_t changes;
ngx_uint_t events;
} ngx_kqueue_conf_t;
static ngx_int_t ngx_kqueue_init(ngx_cycle_t *cycle, ngx_msec_t timer);
#ifdef EVFILT_USER
static ngx_int_t ngx_kqueue_notify_init(ngx_log_t *log);
#endif
static void ngx_kqueue_done(ngx_cycle_t *cycle);
static ngx_int_t ngx_kqueue_add_event(ngx_event_t *ev, ngx_int_t event,
ngx_uint_t flags);
static ngx_int_t ngx_kqueue_del_event(ngx_event_t *ev, ngx_int_t event,
ngx_uint_t flags);
static ngx_int_t ngx_kqueue_set_event(ngx_event_t *ev, ngx_int_t filter,
ngx_uint_t flags);
#ifdef EVFILT_USER
static ngx_int_t ngx_kqueue_notify(ngx_event_handler_pt handler);
#endif
static ngx_int_t ngx_kqueue_process_events(ngx_cycle_t *cycle, ngx_msec_t timer,
ngx_uint_t flags);
static ngx_inline void ngx_kqueue_dump_event(ngx_log_t *log,
struct kevent *kev);
static void *ngx_kqueue_create_conf(ngx_cycle_t *cycle);
static char *ngx_kqueue_init_conf(ngx_cycle_t *cycle, void *conf);
int ngx_kqueue = -1;
static struct kevent *change_list;
static struct kevent *event_list;
static ngx_uint_t max_changes, nchanges, nevents;
#ifdef EVFILT_USER
static ngx_event_t notify_event;
static struct kevent notify_kev;
#endif
static ngx_str_t kqueue_name = ngx_string("kqueue");
static ngx_command_t ngx_kqueue_commands[] = {
{ ngx_string("kqueue_changes"),
NGX_EVENT_CONF|NGX_CONF_TAKE1,
ngx_conf_set_num_slot,
0,
offsetof(ngx_kqueue_conf_t, changes),
NULL },
{ ngx_string("kqueue_events"),
NGX_EVENT_CONF|NGX_CONF_TAKE1,
ngx_conf_set_num_slot,
0,
offsetof(ngx_kqueue_conf_t, events),
NULL },
ngx_null_command
};
static ngx_event_module_t ngx_kqueue_module_ctx = {
&kqueue_name,
ngx_kqueue_create_conf, /* create configuration */
ngx_kqueue_init_conf, /* init configuration */
{
ngx_kqueue_add_event, /* add an event */
ngx_kqueue_del_event, /* delete an event */
ngx_kqueue_add_event, /* enable an event */
ngx_kqueue_del_event, /* disable an event */
NULL, /* add an connection */
NULL, /* delete an connection */
#ifdef EVFILT_USER
ngx_kqueue_notify, /* trigger a notify */
#else
NULL, /* trigger a notify */
#endif
ngx_kqueue_process_events, /* process the events */
ngx_kqueue_init, /* init the events */
ngx_kqueue_done /* done the events */
}
};
ngx_module_t ngx_kqueue_module = {
NGX_MODULE_V1,
&ngx_kqueue_module_ctx, /* module context */
ngx_kqueue_commands, /* module directives */
NGX_EVENT_MODULE, /* module type */
NULL, /* init master */
NULL, /* init module */
NULL, /* init process */
NULL, /* init thread */
NULL, /* exit thread */
NULL, /* exit process */
NULL, /* exit master */
NGX_MODULE_V1_PADDING
};
static ngx_int_t
ngx_kqueue_init(ngx_cycle_t *cycle, ngx_msec_t timer)
{
ngx_kqueue_conf_t *kcf;
struct timespec ts;
#if (NGX_HAVE_TIMER_EVENT)
struct kevent kev;
#endif
kcf = ngx_event_get_conf(cycle->conf_ctx, ngx_kqueue_module);
if (ngx_kqueue == -1) {
ngx_kqueue = kqueue();
if (ngx_kqueue == -1) {
ngx_log_error(NGX_LOG_EMERG, cycle->log, ngx_errno,
"kqueue() failed");
return NGX_ERROR;
}
#ifdef EVFILT_USER
if (ngx_kqueue_notify_init(cycle->log) != NGX_OK) {
return NGX_ERROR;
}
#endif
}
if (max_changes < kcf->changes) {
if (nchanges) {
ts.tv_sec = 0;
ts.tv_nsec = 0;
if (kevent(ngx_kqueue, change_list, (int) nchanges, NULL, 0, &ts)
== -1)
{
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"kevent() failed");
return NGX_ERROR;
}
nchanges = 0;
}
if (change_list) {
ngx_free(change_list);
}
change_list = ngx_alloc(kcf->changes * sizeof(struct kevent),
cycle->log);
if (change_list == NULL) {
return NGX_ERROR;
}
}
max_changes = kcf->changes;
if (nevents < kcf->events) {
if (event_list) {
ngx_free(event_list);
}
event_list = ngx_alloc(kcf->events * sizeof(struct kevent), cycle->log);
if (event_list == NULL) {
return NGX_ERROR;
}
}
ngx_event_flags = NGX_USE_ONESHOT_EVENT
|NGX_USE_KQUEUE_EVENT
|NGX_USE_VNODE_EVENT;
#if (NGX_HAVE_TIMER_EVENT)
if (timer) {
kev.ident = 0;
kev.filter = EVFILT_TIMER;
kev.flags = EV_ADD|EV_ENABLE;
kev.fflags = 0;
kev.data = timer;
kev.udata = 0;
ts.tv_sec = 0;
ts.tv_nsec = 0;
if (kevent(ngx_kqueue, &kev, 1, NULL, 0, &ts) == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"kevent(EVFILT_TIMER) failed");
return NGX_ERROR;
}
ngx_event_flags |= NGX_USE_TIMER_EVENT;
}
#endif
#if (NGX_HAVE_CLEAR_EVENT)
ngx_event_flags |= NGX_USE_CLEAR_EVENT;
#else
ngx_event_flags |= NGX_USE_LEVEL_EVENT;
#endif
#if (NGX_HAVE_LOWAT_EVENT)
ngx_event_flags |= NGX_USE_LOWAT_EVENT;
#endif
nevents = kcf->events;
ngx_io = ngx_os_io;
ngx_event_actions = ngx_kqueue_module_ctx.actions;
return NGX_OK;
}
#ifdef EVFILT_USER
static ngx_int_t
ngx_kqueue_notify_init(ngx_log_t *log)
{
notify_kev.ident = 0;
notify_kev.filter = EVFILT_USER;
notify_kev.data = 0;
notify_kev.flags = EV_ADD|EV_CLEAR;
notify_kev.fflags = 0;
notify_kev.udata = 0;
if (kevent(ngx_kqueue, &notify_kev, 1, NULL, 0, NULL) == -1) {
ngx_log_error(NGX_LOG_ALERT, log, ngx_errno,
"kevent(EVFILT_USER, EV_ADD) failed");
return NGX_ERROR;
}
notify_event.active = 1;
notify_event.log = log;
notify_kev.flags = 0;
notify_kev.fflags = NOTE_TRIGGER;
notify_kev.udata = NGX_KQUEUE_UDATA_T ((uintptr_t) &notify_event);
return NGX_OK;
}
#endif
static void
ngx_kqueue_done(ngx_cycle_t *cycle)
{
if (close(ngx_kqueue) == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
"kqueue close() failed");
}
ngx_kqueue = -1;
ngx_free(change_list);
ngx_free(event_list);
change_list = NULL;
event_list = NULL;
max_changes = 0;
nchanges = 0;
nevents = 0;
}
static ngx_int_t
ngx_kqueue_add_event(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags)
{
ngx_int_t rc;
#if 0
ngx_event_t *e;
ngx_connection_t *c;
#endif
ev->active = 1;
ev->disabled = 0;
ev->oneshot = (flags & NGX_ONESHOT_EVENT) ? 1 : 0;
#if 0
if (ev->index < nchanges
&& ((uintptr_t) change_list[ev->index].udata & (uintptr_t) ~1)
== (uintptr_t) ev)
{
if (change_list[ev->index].flags == EV_DISABLE) {
/*
* if the EV_DISABLE is still not passed to a kernel
* we will not pass it
*/
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"kevent activated: %d: ft:%i",
ngx_event_ident(ev->data), event);
if (ev->index < --nchanges) {
e = (ngx_event_t *)
((uintptr_t) change_list[nchanges].udata & (uintptr_t) ~1);
change_list[ev->index] = change_list[nchanges];
e->index = ev->index;
}
return NGX_OK;
}
c = ev->data;
ngx_log_error(NGX_LOG_ALERT, ev->log, 0,
"previous event on #%d were not passed in kernel", c->fd);
return NGX_ERROR;
}
#endif
rc = ngx_kqueue_set_event(ev, event, EV_ADD|EV_ENABLE|flags);
return rc;
}
static ngx_int_t
ngx_kqueue_del_event(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags)
{
ngx_int_t rc;
ngx_event_t *e;
ev->active = 0;
ev->disabled = 0;
if (ev->index < nchanges
&& ((uintptr_t) change_list[ev->index].udata & (uintptr_t) ~1)
== (uintptr_t) ev)
{
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"kevent deleted: %d: ft:%i",
ngx_event_ident(ev->data), event);
/* if the event is still not passed to a kernel we will not pass it */
nchanges--;
if (ev->index < nchanges) {
e = (ngx_event_t *)
((uintptr_t) change_list[nchanges].udata & (uintptr_t) ~1);
change_list[ev->index] = change_list[nchanges];
e->index = ev->index;
}
return NGX_OK;
}
/*
* when the file descriptor is closed the kqueue automatically deletes
* its filters so we do not need to delete explicitly the event
* before the closing the file descriptor.
*/
if (flags & NGX_CLOSE_EVENT) {
return NGX_OK;
}
if (flags & NGX_DISABLE_EVENT) {
ev->disabled = 1;
} else {
flags |= EV_DELETE;
}
rc = ngx_kqueue_set_event(ev, event, flags);
return rc;
}
static ngx_int_t
ngx_kqueue_set_event(ngx_event_t *ev, ngx_int_t filter, ngx_uint_t flags)
{
struct kevent *kev;
struct timespec ts;
ngx_connection_t *c;
c = ev->data;
ngx_log_debug3(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"kevent set event: %d: ft:%i fl:%04Xi",
c->fd, filter, flags);
if (nchanges >= max_changes) {
ngx_log_error(NGX_LOG_WARN, ev->log, 0,
"kqueue change list is filled up");
ts.tv_sec = 0;
ts.tv_nsec = 0;
if (kevent(ngx_kqueue, change_list, (int) nchanges, NULL, 0, &ts)
== -1)
{
ngx_log_error(NGX_LOG_ALERT, ev->log, ngx_errno, "kevent() failed");
return NGX_ERROR;
}
nchanges = 0;
}
kev = &change_list[nchanges];
kev->ident = c->fd;
kev->filter = (short) filter;
kev->flags = (u_short) flags;
kev->udata = NGX_KQUEUE_UDATA_T ((uintptr_t) ev | ev->instance);
if (filter == EVFILT_VNODE) {
kev->fflags = NOTE_DELETE|NOTE_WRITE|NOTE_EXTEND
|NOTE_ATTRIB|NOTE_RENAME
#if (__FreeBSD__ == 4 && __FreeBSD_version >= 430000) \
|| __FreeBSD_version >= 500018
|NOTE_REVOKE
#endif
;
kev->data = 0;
} else {
#if (NGX_HAVE_LOWAT_EVENT)
if (flags & NGX_LOWAT_EVENT) {
kev->fflags = NOTE_LOWAT;
kev->data = ev->available;
} else {
kev->fflags = 0;
kev->data = 0;
}
#else
kev->fflags = 0;
kev->data = 0;
#endif
}
ev->index = nchanges;
nchanges++;
if (flags & NGX_FLUSH_EVENT) {
ts.tv_sec = 0;
ts.tv_nsec = 0;
ngx_log_debug0(NGX_LOG_DEBUG_EVENT, ev->log, 0, "kevent flush");
if (kevent(ngx_kqueue, change_list, (int) nchanges, NULL, 0, &ts)
== -1)
{
ngx_log_error(NGX_LOG_ALERT, ev->log, ngx_errno, "kevent() failed");
return NGX_ERROR;
}
nchanges = 0;
}
return NGX_OK;
}
#ifdef EVFILT_USER
static ngx_int_t
ngx_kqueue_notify(ngx_event_handler_pt handler)
{
notify_event.handler = handler;
if (kevent(ngx_kqueue, &notify_kev, 1, NULL, 0, NULL) == -1) {
ngx_log_error(NGX_LOG_ALERT, notify_event.log, ngx_errno,
"kevent(EVFILT_USER, NOTE_TRIGGER) failed");
return NGX_ERROR;
}
return NGX_OK;
}
#endif
static ngx_int_t
ngx_kqueue_process_events(ngx_cycle_t *cycle, ngx_msec_t timer,
ngx_uint_t flags)
{
int events, n;
ngx_int_t i, instance;
ngx_uint_t level;
ngx_err_t err;
ngx_event_t *ev;
ngx_queue_t *queue;
struct timespec ts, *tp;
n = (int) nchanges;
nchanges = 0;
if (timer == NGX_TIMER_INFINITE) {
tp = NULL;
} else {
ts.tv_sec = timer / 1000;
ts.tv_nsec = (timer % 1000) * 1000000;
/*
* 64-bit Darwin kernel has the bug: kernel level ts.tv_nsec is
* the int32_t while user level ts.tv_nsec is the long (64-bit),
* so on the big endian PowerPC all nanoseconds are lost.
*/
#if (NGX_DARWIN_KEVENT_BUG)
ts.tv_nsec <<= 32;
#endif
tp = &ts;
}
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"kevent timer: %M, changes: %d", timer, n);
events = kevent(ngx_kqueue, change_list, n, event_list, (int) nevents, tp);
err = (events == -1) ? ngx_errno : 0;
if (flags & NGX_UPDATE_TIME || ngx_event_timer_alarm) {
ngx_time_update();
}
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"kevent events: %d", events);
if (err) {
if (err == NGX_EINTR) {
if (ngx_event_timer_alarm) {
ngx_event_timer_alarm = 0;
return NGX_OK;
}
level = NGX_LOG_INFO;
} else {
level = NGX_LOG_ALERT;
}
ngx_log_error(level, cycle->log, err, "kevent() failed");
return NGX_ERROR;
}
if (events == 0) {
if (timer != NGX_TIMER_INFINITE) {
return NGX_OK;
}
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
"kevent() returned no events without timeout");
return NGX_ERROR;
}
for (i = 0; i < events; i++) {
ngx_kqueue_dump_event(cycle->log, &event_list[i]);
if (event_list[i].flags & EV_ERROR) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, event_list[i].data,
"kevent() error on %d filter:%d flags:%04Xd",
(int) event_list[i].ident, event_list[i].filter,
event_list[i].flags);
continue;
}
#if (NGX_HAVE_TIMER_EVENT)
if (event_list[i].filter == EVFILT_TIMER) {
ngx_time_update();
continue;
}
#endif
ev = (ngx_event_t *) event_list[i].udata;
switch (event_list[i].filter) {
case EVFILT_READ:
case EVFILT_WRITE:
instance = (uintptr_t) ev & 1;
ev = (ngx_event_t *) ((uintptr_t) ev & (uintptr_t) ~1);
if (ev->closed || ev->instance != instance) {
/*
* the stale event from a file descriptor
* that was just closed in this iteration
*/
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"kevent: stale event %p", ev);
continue;
}
if (ev->log && (ev->log->log_level & NGX_LOG_DEBUG_CONNECTION)) {
ngx_kqueue_dump_event(ev->log, &event_list[i]);
}
if (ev->oneshot) {
ev->active = 0;
}
ev->available = event_list[i].data;
if (event_list[i].flags & EV_EOF) {
ev->pending_eof = 1;
ev->kq_errno = event_list[i].fflags;
}
ev->ready = 1;
break;
case EVFILT_VNODE:
ev->kq_vnode = 1;
break;
case EVFILT_AIO:
ev->complete = 1;
ev->ready = 1;
break;
#ifdef EVFILT_USER
case EVFILT_USER:
break;
#endif
default:
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
"unexpected kevent() filter %d",
event_list[i].filter);
continue;
}
if (flags & NGX_POST_EVENTS) {
queue = ev->accept ? &ngx_posted_accept_events
: &ngx_posted_events;
ngx_post_event(ev, queue);
continue;
}
ev->handler(ev);
}
return NGX_OK;
}
static ngx_inline void
ngx_kqueue_dump_event(ngx_log_t *log, struct kevent *kev)
{
if (kev->ident > 0x8000000 && kev->ident != (unsigned) -1) {
ngx_log_debug6(NGX_LOG_DEBUG_EVENT, log, 0,
"kevent: %p: ft:%d fl:%04Xd ff:%08Xd d:%d ud:%p",
(void *) kev->ident, kev->filter,
kev->flags, kev->fflags,
(int) kev->data, kev->udata);
} else {
ngx_log_debug6(NGX_LOG_DEBUG_EVENT, log, 0,
"kevent: %d: ft:%d fl:%04Xd ff:%08Xd d:%d ud:%p",
(int) kev->ident, kev->filter,
kev->flags, kev->fflags,
(int) kev->data, kev->udata);
}
}
static void *
ngx_kqueue_create_conf(ngx_cycle_t *cycle)
{
ngx_kqueue_conf_t *kcf;
kcf = ngx_palloc(cycle->pool, sizeof(ngx_kqueue_conf_t));
if (kcf == NULL) {
return NULL;
}
kcf->changes = NGX_CONF_UNSET;
kcf->events = NGX_CONF_UNSET;
return kcf;
}
static char *
ngx_kqueue_init_conf(ngx_cycle_t *cycle, void *conf)
{
ngx_kqueue_conf_t *kcf = conf;
ngx_conf_init_uint_value(kcf->changes, 512);
ngx_conf_init_uint_value(kcf->events, 512);
return NGX_CONF_OK;
}
+416
View File
@@ -0,0 +1,416 @@
/*
* Copyright (C) Igor Sysoev
* Copyright (C) Nginx, Inc.
*/
#include <ngx_config.h>
#include <ngx_core.h>
#include <ngx_event.h>
static ngx_int_t ngx_poll_init(ngx_cycle_t *cycle, ngx_msec_t timer);
static void ngx_poll_done(ngx_cycle_t *cycle);
static ngx_int_t ngx_poll_add_event(ngx_event_t *ev, ngx_int_t event,
ngx_uint_t flags);
static ngx_int_t ngx_poll_del_event(ngx_event_t *ev, ngx_int_t event,
ngx_uint_t flags);
static ngx_int_t ngx_poll_process_events(ngx_cycle_t *cycle, ngx_msec_t timer,
ngx_uint_t flags);
static char *ngx_poll_init_conf(ngx_cycle_t *cycle, void *conf);
static struct pollfd *event_list;
static ngx_uint_t nevents;
static ngx_str_t poll_name = ngx_string("poll");
static ngx_event_module_t ngx_poll_module_ctx = {
&poll_name,
NULL, /* create configuration */
ngx_poll_init_conf, /* init configuration */
{
ngx_poll_add_event, /* add an event */
ngx_poll_del_event, /* delete an event */
ngx_poll_add_event, /* enable an event */
ngx_poll_del_event, /* disable an event */
NULL, /* add an connection */
NULL, /* delete an connection */
NULL, /* trigger a notify */
ngx_poll_process_events, /* process the events */
ngx_poll_init, /* init the events */
ngx_poll_done /* done the events */
}
};
ngx_module_t ngx_poll_module = {
NGX_MODULE_V1,
&ngx_poll_module_ctx, /* module context */
NULL, /* module directives */
NGX_EVENT_MODULE, /* module type */
NULL, /* init master */
NULL, /* init module */
NULL, /* init process */
NULL, /* init thread */
NULL, /* exit thread */
NULL, /* exit process */
NULL, /* exit master */
NGX_MODULE_V1_PADDING
};
static ngx_int_t
ngx_poll_init(ngx_cycle_t *cycle, ngx_msec_t timer)
{
struct pollfd *list;
if (event_list == NULL) {
nevents = 0;
}
if (ngx_process >= NGX_PROCESS_WORKER
|| cycle->old_cycle == NULL
|| cycle->old_cycle->connection_n < cycle->connection_n)
{
list = ngx_alloc(sizeof(struct pollfd) * cycle->connection_n,
cycle->log);
if (list == NULL) {
return NGX_ERROR;
}
if (event_list) {
ngx_memcpy(list, event_list, sizeof(struct pollfd) * nevents);
ngx_free(event_list);
}
event_list = list;
}
ngx_io = ngx_os_io;
ngx_event_actions = ngx_poll_module_ctx.actions;
ngx_event_flags = NGX_USE_LEVEL_EVENT|NGX_USE_FD_EVENT;
return NGX_OK;
}
static void
ngx_poll_done(ngx_cycle_t *cycle)
{
ngx_free(event_list);
event_list = NULL;
}
static ngx_int_t
ngx_poll_add_event(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags)
{
ngx_event_t *e;
ngx_connection_t *c;
c = ev->data;
ev->active = 1;
if (ev->index != NGX_INVALID_INDEX) {
ngx_log_error(NGX_LOG_ALERT, ev->log, 0,
"poll event fd:%d ev:%i is already set", c->fd, event);
return NGX_OK;
}
if (event == NGX_READ_EVENT) {
e = c->write;
#if (NGX_READ_EVENT != POLLIN)
event = POLLIN;
#endif
} else {
e = c->read;
#if (NGX_WRITE_EVENT != POLLOUT)
event = POLLOUT;
#endif
}
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"poll add event: fd:%d ev:%i", c->fd, event);
if (e == NULL || e->index == NGX_INVALID_INDEX) {
event_list[nevents].fd = c->fd;
event_list[nevents].events = (short) event;
event_list[nevents].revents = 0;
ev->index = nevents;
nevents++;
} else {
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"poll add index: %i", e->index);
event_list[e->index].events |= (short) event;
ev->index = e->index;
}
return NGX_OK;
}
static ngx_int_t
ngx_poll_del_event(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags)
{
ngx_event_t *e;
ngx_connection_t *c;
c = ev->data;
ev->active = 0;
if (ev->index == NGX_INVALID_INDEX) {
ngx_log_error(NGX_LOG_ALERT, ev->log, 0,
"poll event fd:%d ev:%i is already deleted",
c->fd, event);
return NGX_OK;
}
if (event == NGX_READ_EVENT) {
e = c->write;
#if (NGX_READ_EVENT != POLLIN)
event = POLLIN;
#endif
} else {
e = c->read;
#if (NGX_WRITE_EVENT != POLLOUT)
event = POLLOUT;
#endif
}
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"poll del event: fd:%d ev:%i", c->fd, event);
if (e == NULL || e->index == NGX_INVALID_INDEX) {
nevents--;
if (ev->index < nevents) {
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"index: copy event %ui to %i", nevents, ev->index);
event_list[ev->index] = event_list[nevents];
c = ngx_cycle->files[event_list[nevents].fd];
if (c->fd == -1) {
ngx_log_error(NGX_LOG_ALERT, ev->log, 0,
"unexpected last event");
} else {
if (c->read->index == nevents) {
c->read->index = ev->index;
}
if (c->write->index == nevents) {
c->write->index = ev->index;
}
}
}
} else {
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"poll del index: %i", e->index);
event_list[e->index].events &= (short) ~event;
}
ev->index = NGX_INVALID_INDEX;
return NGX_OK;
}
static ngx_int_t
ngx_poll_process_events(ngx_cycle_t *cycle, ngx_msec_t timer, ngx_uint_t flags)
{
int ready, revents;
ngx_err_t err;
ngx_uint_t i, found, level;
ngx_event_t *ev;
ngx_queue_t *queue;
ngx_connection_t *c;
/* NGX_TIMER_INFINITE == INFTIM */
#if (NGX_DEBUG0)
if (cycle->log->log_level & NGX_LOG_DEBUG_ALL) {
for (i = 0; i < nevents; i++) {
ngx_log_debug3(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"poll: %ui: fd:%d ev:%04Xd",
i, event_list[i].fd, event_list[i].events);
}
}
#endif
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0, "poll timer: %M", timer);
ready = poll(event_list, (u_int) nevents, (int) timer);
err = (ready == -1) ? ngx_errno : 0;
if (flags & NGX_UPDATE_TIME || ngx_event_timer_alarm) {
ngx_time_update();
}
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"poll ready %d of %ui", ready, nevents);
if (err) {
if (err == NGX_EINTR) {
if (ngx_event_timer_alarm) {
ngx_event_timer_alarm = 0;
return NGX_OK;
}
level = NGX_LOG_INFO;
} else {
level = NGX_LOG_ALERT;
}
ngx_log_error(level, cycle->log, err, "poll() failed");
return NGX_ERROR;
}
if (ready == 0) {
if (timer != NGX_TIMER_INFINITE) {
return NGX_OK;
}
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
"poll() returned no events without timeout");
return NGX_ERROR;
}
for (i = 0; i < nevents && ready; i++) {
revents = event_list[i].revents;
#if 1
ngx_log_debug4(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"poll: %ui: fd:%d ev:%04Xd rev:%04Xd",
i, event_list[i].fd, event_list[i].events, revents);
#else
if (revents) {
ngx_log_debug4(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"poll: %ui: fd:%d ev:%04Xd rev:%04Xd",
i, event_list[i].fd, event_list[i].events, revents);
}
#endif
if (revents & POLLNVAL) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
"poll() error fd:%d ev:%04Xd rev:%04Xd",
event_list[i].fd, event_list[i].events, revents);
}
if (revents & ~(POLLIN|POLLOUT|POLLERR|POLLHUP|POLLNVAL)) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
"strange poll() events fd:%d ev:%04Xd rev:%04Xd",
event_list[i].fd, event_list[i].events, revents);
}
if (event_list[i].fd == -1) {
/*
* the disabled event, a workaround for our possible bug,
* see the comment below
*/
continue;
}
c = ngx_cycle->files[event_list[i].fd];
if (c->fd == -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0, "unexpected event");
/*
* it is certainly our fault and it should be investigated,
* in the meantime we disable this event to avoid a CPU spinning
*/
if (i == nevents - 1) {
nevents--;
} else {
event_list[i].fd = -1;
}
continue;
}
if (revents & (POLLERR|POLLHUP|POLLNVAL)) {
/*
* if the error events were returned, add POLLIN and POLLOUT
* to handle the events at least in one active handler
*/
revents |= POLLIN|POLLOUT;
}
found = 0;
if ((revents & POLLIN) && c->read->active) {
found = 1;
ev = c->read;
ev->ready = 1;
ev->available = -1;
queue = ev->accept ? &ngx_posted_accept_events
: &ngx_posted_events;
ngx_post_event(ev, queue);
}
if ((revents & POLLOUT) && c->write->active) {
found = 1;
ev = c->write;
ev->ready = 1;
ngx_post_event(ev, &ngx_posted_events);
}
if (found) {
ready--;
continue;
}
}
if (ready != 0) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0, "poll ready != events");
}
return NGX_OK;
}
static char *
ngx_poll_init_conf(ngx_cycle_t *cycle, void *conf)
{
ngx_event_conf_t *ecf;
ecf = ngx_event_get_conf(cycle->conf_ctx, ngx_event_core_module);
if (ecf->use != ngx_poll_module.ctx_index) {
return NGX_CONF_OK;
}
return NGX_CONF_OK;
}
+424
View File
@@ -0,0 +1,424 @@
/*
* Copyright (C) Igor Sysoev
* Copyright (C) Nginx, Inc.
*/
#include <ngx_config.h>
#include <ngx_core.h>
#include <ngx_event.h>
static ngx_int_t ngx_select_init(ngx_cycle_t *cycle, ngx_msec_t timer);
static void ngx_select_done(ngx_cycle_t *cycle);
static ngx_int_t ngx_select_add_event(ngx_event_t *ev, ngx_int_t event,
ngx_uint_t flags);
static ngx_int_t ngx_select_del_event(ngx_event_t *ev, ngx_int_t event,
ngx_uint_t flags);
static ngx_int_t ngx_select_process_events(ngx_cycle_t *cycle, ngx_msec_t timer,
ngx_uint_t flags);
static void ngx_select_repair_fd_sets(ngx_cycle_t *cycle);
static char *ngx_select_init_conf(ngx_cycle_t *cycle, void *conf);
static fd_set master_read_fd_set;
static fd_set master_write_fd_set;
static fd_set work_read_fd_set;
static fd_set work_write_fd_set;
static ngx_int_t max_fd;
static ngx_uint_t nevents;
static ngx_event_t **event_index;
static ngx_str_t select_name = ngx_string("select");
static ngx_event_module_t ngx_select_module_ctx = {
&select_name,
NULL, /* create configuration */
ngx_select_init_conf, /* init configuration */
{
ngx_select_add_event, /* add an event */
ngx_select_del_event, /* delete an event */
ngx_select_add_event, /* enable an event */
ngx_select_del_event, /* disable an event */
NULL, /* add an connection */
NULL, /* delete an connection */
NULL, /* trigger a notify */
ngx_select_process_events, /* process the events */
ngx_select_init, /* init the events */
ngx_select_done /* done the events */
}
};
ngx_module_t ngx_select_module = {
NGX_MODULE_V1,
&ngx_select_module_ctx, /* module context */
NULL, /* module directives */
NGX_EVENT_MODULE, /* module type */
NULL, /* init master */
NULL, /* init module */
NULL, /* init process */
NULL, /* init thread */
NULL, /* exit thread */
NULL, /* exit process */
NULL, /* exit master */
NGX_MODULE_V1_PADDING
};
static ngx_int_t
ngx_select_init(ngx_cycle_t *cycle, ngx_msec_t timer)
{
ngx_event_t **index;
if (event_index == NULL) {
FD_ZERO(&master_read_fd_set);
FD_ZERO(&master_write_fd_set);
nevents = 0;
}
if (ngx_process >= NGX_PROCESS_WORKER
|| cycle->old_cycle == NULL
|| cycle->old_cycle->connection_n < cycle->connection_n)
{
index = ngx_alloc(sizeof(ngx_event_t *) * 2 * cycle->connection_n,
cycle->log);
if (index == NULL) {
return NGX_ERROR;
}
if (event_index) {
ngx_memcpy(index, event_index, sizeof(ngx_event_t *) * nevents);
ngx_free(event_index);
}
event_index = index;
}
ngx_io = ngx_os_io;
ngx_event_actions = ngx_select_module_ctx.actions;
ngx_event_flags = NGX_USE_LEVEL_EVENT;
max_fd = -1;
return NGX_OK;
}
static void
ngx_select_done(ngx_cycle_t *cycle)
{
ngx_free(event_index);
event_index = NULL;
}
static ngx_int_t
ngx_select_add_event(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags)
{
ngx_connection_t *c;
c = ev->data;
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"select add event fd:%d ev:%i", c->fd, event);
if (ev->index != NGX_INVALID_INDEX) {
ngx_log_error(NGX_LOG_ALERT, ev->log, 0,
"select event fd:%d ev:%i is already set", c->fd, event);
return NGX_OK;
}
if ((event == NGX_READ_EVENT && ev->write)
|| (event == NGX_WRITE_EVENT && !ev->write))
{
ngx_log_error(NGX_LOG_ALERT, ev->log, 0,
"invalid select %s event fd:%d ev:%i",
ev->write ? "write" : "read", c->fd, event);
return NGX_ERROR;
}
if (event == NGX_READ_EVENT) {
FD_SET(c->fd, &master_read_fd_set);
} else if (event == NGX_WRITE_EVENT) {
FD_SET(c->fd, &master_write_fd_set);
}
if (max_fd != -1 && max_fd < c->fd) {
max_fd = c->fd;
}
ev->active = 1;
event_index[nevents] = ev;
ev->index = nevents;
nevents++;
return NGX_OK;
}
static ngx_int_t
ngx_select_del_event(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags)
{
ngx_event_t *e;
ngx_connection_t *c;
c = ev->data;
ev->active = 0;
if (ev->index == NGX_INVALID_INDEX) {
return NGX_OK;
}
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"select del event fd:%d ev:%i", c->fd, event);
if (event == NGX_READ_EVENT) {
FD_CLR(c->fd, &master_read_fd_set);
} else if (event == NGX_WRITE_EVENT) {
FD_CLR(c->fd, &master_write_fd_set);
}
if (max_fd == c->fd) {
max_fd = -1;
}
if (ev->index < --nevents) {
e = event_index[nevents];
event_index[ev->index] = e;
e->index = ev->index;
}
ev->index = NGX_INVALID_INDEX;
return NGX_OK;
}
static ngx_int_t
ngx_select_process_events(ngx_cycle_t *cycle, ngx_msec_t timer,
ngx_uint_t flags)
{
int ready, nready;
ngx_err_t err;
ngx_uint_t i, found;
ngx_event_t *ev;
ngx_queue_t *queue;
struct timeval tv, *tp;
ngx_connection_t *c;
if (max_fd == -1) {
for (i = 0; i < nevents; i++) {
c = event_index[i]->data;
if (max_fd < c->fd) {
max_fd = c->fd;
}
}
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"change max_fd: %i", max_fd);
}
#if (NGX_DEBUG)
if (cycle->log->log_level & NGX_LOG_DEBUG_ALL) {
for (i = 0; i < nevents; i++) {
ev = event_index[i];
c = ev->data;
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"select event: fd:%d wr:%d", c->fd, ev->write);
}
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"max_fd: %i", max_fd);
}
#endif
if (timer == NGX_TIMER_INFINITE) {
tp = NULL;
} else {
tv.tv_sec = (long) (timer / 1000);
tv.tv_usec = (long) ((timer % 1000) * 1000);
tp = &tv;
}
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"select timer: %M", timer);
work_read_fd_set = master_read_fd_set;
work_write_fd_set = master_write_fd_set;
ready = select(max_fd + 1, &work_read_fd_set, &work_write_fd_set, NULL, tp);
err = (ready == -1) ? ngx_errno : 0;
if (flags & NGX_UPDATE_TIME || ngx_event_timer_alarm) {
ngx_time_update();
}
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"select ready %d", ready);
if (err) {
ngx_uint_t level;
if (err == NGX_EINTR) {
if (ngx_event_timer_alarm) {
ngx_event_timer_alarm = 0;
return NGX_OK;
}
level = NGX_LOG_INFO;
} else {
level = NGX_LOG_ALERT;
}
ngx_log_error(level, cycle->log, err, "select() failed");
if (err == NGX_EBADF) {
ngx_select_repair_fd_sets(cycle);
}
return NGX_ERROR;
}
if (ready == 0) {
if (timer != NGX_TIMER_INFINITE) {
return NGX_OK;
}
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
"select() returned no events without timeout");
return NGX_ERROR;
}
nready = 0;
for (i = 0; i < nevents; i++) {
ev = event_index[i];
c = ev->data;
found = 0;
if (ev->write) {
if (FD_ISSET(c->fd, &work_write_fd_set)) {
found = 1;
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"select write %d", c->fd);
}
} else {
if (FD_ISSET(c->fd, &work_read_fd_set)) {
found = 1;
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"select read %d", c->fd);
}
}
if (found) {
ev->ready = 1;
ev->available = -1;
queue = ev->accept ? &ngx_posted_accept_events
: &ngx_posted_events;
ngx_post_event(ev, queue);
nready++;
}
}
if (ready != nready) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
"select ready != events: %d:%d", ready, nready);
ngx_select_repair_fd_sets(cycle);
}
return NGX_OK;
}
static void
ngx_select_repair_fd_sets(ngx_cycle_t *cycle)
{
int n;
socklen_t len;
ngx_err_t err;
ngx_socket_t s;
for (s = 0; s <= max_fd; s++) {
if (FD_ISSET(s, &master_read_fd_set) == 0) {
continue;
}
len = sizeof(int);
if (getsockopt(s, SOL_SOCKET, SO_TYPE, &n, &len) == -1) {
err = ngx_socket_errno;
ngx_log_error(NGX_LOG_ALERT, cycle->log, err,
"invalid descriptor #%d in read fd_set", s);
FD_CLR(s, &master_read_fd_set);
}
}
for (s = 0; s <= max_fd; s++) {
if (FD_ISSET(s, &master_write_fd_set) == 0) {
continue;
}
len = sizeof(int);
if (getsockopt(s, SOL_SOCKET, SO_TYPE, &n, &len) == -1) {
err = ngx_socket_errno;
ngx_log_error(NGX_LOG_ALERT, cycle->log, err,
"invalid descriptor #%d in write fd_set", s);
FD_CLR(s, &master_write_fd_set);
}
}
max_fd = -1;
}
static char *
ngx_select_init_conf(ngx_cycle_t *cycle, void *conf)
{
ngx_event_conf_t *ecf;
ecf = ngx_event_get_conf(cycle->conf_ctx, ngx_event_core_module);
if (ecf->use != ngx_select_module.ctx_index) {
return NGX_CONF_OK;
}
/* disable warning: the default FD_SETSIZE is 1024U in FreeBSD 5.x */
if (cycle->connection_n > FD_SETSIZE) {
ngx_log_error(NGX_LOG_EMERG, cycle->log, 0,
"the maximum number of files "
"supported by select() is %ud", FD_SETSIZE);
return NGX_CONF_ERROR;
}
return NGX_CONF_OK;
}
+436
View File
@@ -0,0 +1,436 @@
/*
* Copyright (C) Igor Sysoev
* Copyright (C) Maxim Dounin
* Copyright (C) Nginx, Inc.
*/
#include <ngx_config.h>
#include <ngx_core.h>
#include <ngx_event.h>
static ngx_int_t ngx_poll_init(ngx_cycle_t *cycle, ngx_msec_t timer);
static void ngx_poll_done(ngx_cycle_t *cycle);
static ngx_int_t ngx_poll_add_event(ngx_event_t *ev, ngx_int_t event,
ngx_uint_t flags);
static ngx_int_t ngx_poll_del_event(ngx_event_t *ev, ngx_int_t event,
ngx_uint_t flags);
static ngx_int_t ngx_poll_process_events(ngx_cycle_t *cycle, ngx_msec_t timer,
ngx_uint_t flags);
static char *ngx_poll_init_conf(ngx_cycle_t *cycle, void *conf);
static struct pollfd *event_list;
static ngx_connection_t **event_index;
static ngx_uint_t nevents;
static ngx_str_t poll_name = ngx_string("poll");
static ngx_event_module_t ngx_poll_module_ctx = {
&poll_name,
NULL, /* create configuration */
ngx_poll_init_conf, /* init configuration */
{
ngx_poll_add_event, /* add an event */
ngx_poll_del_event, /* delete an event */
ngx_poll_add_event, /* enable an event */
ngx_poll_del_event, /* disable an event */
NULL, /* add an connection */
NULL, /* delete an connection */
NULL, /* trigger a notify */
ngx_poll_process_events, /* process the events */
ngx_poll_init, /* init the events */
ngx_poll_done /* done the events */
}
};
ngx_module_t ngx_poll_module = {
NGX_MODULE_V1,
&ngx_poll_module_ctx, /* module context */
NULL, /* module directives */
NGX_EVENT_MODULE, /* module type */
NULL, /* init master */
NULL, /* init module */
NULL, /* init process */
NULL, /* init thread */
NULL, /* exit thread */
NULL, /* exit process */
NULL, /* exit master */
NGX_MODULE_V1_PADDING
};
static ngx_int_t
ngx_poll_init(ngx_cycle_t *cycle, ngx_msec_t timer)
{
struct pollfd *list;
ngx_connection_t **index;
if (event_list == NULL) {
nevents = 0;
}
if (ngx_process >= NGX_PROCESS_WORKER
|| cycle->old_cycle == NULL
|| cycle->old_cycle->connection_n < cycle->connection_n)
{
list = ngx_alloc(sizeof(struct pollfd) * cycle->connection_n,
cycle->log);
if (list == NULL) {
return NGX_ERROR;
}
if (event_list) {
ngx_memcpy(list, event_list, sizeof(struct pollfd) * nevents);
ngx_free(event_list);
}
event_list = list;
index = ngx_alloc(sizeof(ngx_connection_t *) * cycle->connection_n,
cycle->log);
if (index == NULL) {
return NGX_ERROR;
}
if (event_index) {
ngx_memcpy(index, event_index,
sizeof(ngx_connection_t *) * nevents);
ngx_free(event_index);
}
event_index = index;
}
ngx_io = ngx_os_io;
ngx_event_actions = ngx_poll_module_ctx.actions;
ngx_event_flags = NGX_USE_LEVEL_EVENT;
return NGX_OK;
}
static void
ngx_poll_done(ngx_cycle_t *cycle)
{
ngx_free(event_list);
ngx_free(event_index);
event_list = NULL;
event_index = NULL;
}
static ngx_int_t
ngx_poll_add_event(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags)
{
ngx_event_t *e;
ngx_connection_t *c;
c = ev->data;
ev->active = 1;
if (ev->index != NGX_INVALID_INDEX) {
ngx_log_error(NGX_LOG_ALERT, ev->log, 0,
"poll event fd:%d ev:%i is already set", c->fd, event);
return NGX_OK;
}
if (event == NGX_READ_EVENT) {
e = c->write;
#if (NGX_READ_EVENT != POLLIN)
event = POLLIN;
#endif
} else {
e = c->read;
#if (NGX_WRITE_EVENT != POLLOUT)
event = POLLOUT;
#endif
}
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"poll add event: fd:%d ev:%i", c->fd, event);
if (e == NULL || e->index == NGX_INVALID_INDEX) {
event_list[nevents].fd = c->fd;
event_list[nevents].events = (short) event;
event_list[nevents].revents = 0;
event_index[nevents] = c;
ev->index = nevents;
nevents++;
} else {
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"poll add index: %i", e->index);
event_list[e->index].events |= (short) event;
ev->index = e->index;
}
return NGX_OK;
}
static ngx_int_t
ngx_poll_del_event(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags)
{
ngx_event_t *e;
ngx_connection_t *c;
c = ev->data;
ev->active = 0;
if (ev->index == NGX_INVALID_INDEX) {
ngx_log_error(NGX_LOG_ALERT, ev->log, 0,
"poll event fd:%d ev:%i is already deleted",
c->fd, event);
return NGX_OK;
}
if (event == NGX_READ_EVENT) {
e = c->write;
#if (NGX_READ_EVENT != POLLIN)
event = POLLIN;
#endif
} else {
e = c->read;
#if (NGX_WRITE_EVENT != POLLOUT)
event = POLLOUT;
#endif
}
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"poll del event: fd:%d ev:%i", c->fd, event);
if (e == NULL || e->index == NGX_INVALID_INDEX) {
nevents--;
if (ev->index < nevents) {
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"index: copy event %ui to %i", nevents, ev->index);
event_list[ev->index] = event_list[nevents];
event_index[ev->index] = event_index[nevents];
c = event_index[ev->index];
if (c->fd == (ngx_socket_t) -1) {
ngx_log_error(NGX_LOG_ALERT, ev->log, 0,
"unexpected last event");
} else {
if (c->read->index == nevents) {
c->read->index = ev->index;
}
if (c->write->index == nevents) {
c->write->index = ev->index;
}
}
}
} else {
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"poll del index: %i", e->index);
event_list[e->index].events &= (short) ~event;
}
ev->index = NGX_INVALID_INDEX;
return NGX_OK;
}
static ngx_int_t
ngx_poll_process_events(ngx_cycle_t *cycle, ngx_msec_t timer, ngx_uint_t flags)
{
int ready, revents;
ngx_err_t err;
ngx_uint_t i, found;
ngx_event_t *ev;
ngx_queue_t *queue;
ngx_connection_t *c;
/* NGX_TIMER_INFINITE == INFTIM */
#if (NGX_DEBUG0)
if (cycle->log->log_level & NGX_LOG_DEBUG_ALL) {
for (i = 0; i < nevents; i++) {
ngx_log_debug3(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"poll: %ui: fd:%d ev:%04Xd",
i, event_list[i].fd, event_list[i].events);
}
}
#endif
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0, "poll timer: %M", timer);
ready = WSAPoll(event_list, (u_int) nevents, (int) timer);
err = (ready == -1) ? ngx_errno : 0;
if (flags & NGX_UPDATE_TIME || ngx_event_timer_alarm) {
ngx_time_update();
}
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"poll ready %d of %ui", ready, nevents);
if (err) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, err, "WSAPoll() failed");
return NGX_ERROR;
}
if (ready == 0) {
if (timer != NGX_TIMER_INFINITE) {
return NGX_OK;
}
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
"WSAPoll() returned no events without timeout");
return NGX_ERROR;
}
for (i = 0; i < nevents && ready; i++) {
revents = event_list[i].revents;
#if 1
ngx_log_debug4(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"poll: %ui: fd:%d ev:%04Xd rev:%04Xd",
i, event_list[i].fd, event_list[i].events, revents);
#else
if (revents) {
ngx_log_debug4(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"poll: %ui: fd:%d ev:%04Xd rev:%04Xd",
i, event_list[i].fd, event_list[i].events, revents);
}
#endif
if (revents & POLLNVAL) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
"poll() error fd:%d ev:%04Xd rev:%04Xd",
event_list[i].fd, event_list[i].events, revents);
}
if (revents & ~(POLLIN|POLLOUT|POLLERR|POLLHUP|POLLNVAL)) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
"strange poll() events fd:%d ev:%04Xd rev:%04Xd",
event_list[i].fd, event_list[i].events, revents);
}
if (event_list[i].fd == (ngx_socket_t) -1) {
/*
* the disabled event, a workaround for our possible bug,
* see the comment below
*/
continue;
}
c = event_index[i];
if (c->fd == (ngx_socket_t) -1) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0, "unexpected event");
/*
* it is certainly our fault and it should be investigated,
* in the meantime we disable this event to avoid a CPU spinning
*/
if (i == nevents - 1) {
nevents--;
} else {
event_list[i].fd = (ngx_socket_t) -1;
}
continue;
}
if (revents & (POLLERR|POLLHUP|POLLNVAL)) {
/*
* if the error events were returned, add POLLIN and POLLOUT
* to handle the events at least in one active handler
*/
revents |= POLLIN|POLLOUT;
}
found = 0;
if ((revents & POLLIN) && c->read->active) {
found = 1;
ev = c->read;
ev->ready = 1;
ev->available = -1;
queue = ev->accept ? &ngx_posted_accept_events
: &ngx_posted_events;
ngx_post_event(ev, queue);
}
if ((revents & POLLOUT) && c->write->active) {
found = 1;
ev = c->write;
ev->ready = 1;
ngx_post_event(ev, &ngx_posted_events);
}
if (found) {
ready--;
continue;
}
}
if (ready != 0) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0, "poll ready != events");
}
return NGX_OK;
}
static char *
ngx_poll_init_conf(ngx_cycle_t *cycle, void *conf)
{
ngx_event_conf_t *ecf;
ecf = ngx_event_get_conf(cycle->conf_ctx, ngx_event_core_module);
if (ecf->use != ngx_poll_module.ctx_index) {
return NGX_CONF_OK;
}
#if (NGX_LOAD_WSAPOLL)
if (!ngx_have_wsapoll) {
ngx_log_error(NGX_LOG_EMERG, cycle->log, 0,
"poll is not available on this platform");
return NGX_CONF_ERROR;
}
#endif
return NGX_CONF_OK;
}
+408
View File
@@ -0,0 +1,408 @@
/*
* Copyright (C) Igor Sysoev
* Copyright (C) Nginx, Inc.
*/
#include <ngx_config.h>
#include <ngx_core.h>
#include <ngx_event.h>
static ngx_int_t ngx_select_init(ngx_cycle_t *cycle, ngx_msec_t timer);
static void ngx_select_done(ngx_cycle_t *cycle);
static ngx_int_t ngx_select_add_event(ngx_event_t *ev, ngx_int_t event,
ngx_uint_t flags);
static ngx_int_t ngx_select_del_event(ngx_event_t *ev, ngx_int_t event,
ngx_uint_t flags);
static ngx_int_t ngx_select_process_events(ngx_cycle_t *cycle, ngx_msec_t timer,
ngx_uint_t flags);
static void ngx_select_repair_fd_sets(ngx_cycle_t *cycle);
static char *ngx_select_init_conf(ngx_cycle_t *cycle, void *conf);
static fd_set master_read_fd_set;
static fd_set master_write_fd_set;
static fd_set work_read_fd_set;
static fd_set work_write_fd_set;
static fd_set work_except_fd_set;
static ngx_uint_t max_read;
static ngx_uint_t max_write;
static ngx_uint_t nevents;
static ngx_event_t **event_index;
static ngx_str_t select_name = ngx_string("select");
static ngx_event_module_t ngx_select_module_ctx = {
&select_name,
NULL, /* create configuration */
ngx_select_init_conf, /* init configuration */
{
ngx_select_add_event, /* add an event */
ngx_select_del_event, /* delete an event */
ngx_select_add_event, /* enable an event */
ngx_select_del_event, /* disable an event */
NULL, /* add an connection */
NULL, /* delete an connection */
NULL, /* trigger a notify */
ngx_select_process_events, /* process the events */
ngx_select_init, /* init the events */
ngx_select_done /* done the events */
}
};
ngx_module_t ngx_select_module = {
NGX_MODULE_V1,
&ngx_select_module_ctx, /* module context */
NULL, /* module directives */
NGX_EVENT_MODULE, /* module type */
NULL, /* init master */
NULL, /* init module */
NULL, /* init process */
NULL, /* init thread */
NULL, /* exit thread */
NULL, /* exit process */
NULL, /* exit master */
NGX_MODULE_V1_PADDING
};
static ngx_int_t
ngx_select_init(ngx_cycle_t *cycle, ngx_msec_t timer)
{
ngx_event_t **index;
if (event_index == NULL) {
FD_ZERO(&master_read_fd_set);
FD_ZERO(&master_write_fd_set);
nevents = 0;
}
if (ngx_process >= NGX_PROCESS_WORKER
|| cycle->old_cycle == NULL
|| cycle->old_cycle->connection_n < cycle->connection_n)
{
index = ngx_alloc(sizeof(ngx_event_t *) * 2 * cycle->connection_n,
cycle->log);
if (index == NULL) {
return NGX_ERROR;
}
if (event_index) {
ngx_memcpy(index, event_index, sizeof(ngx_event_t *) * nevents);
ngx_free(event_index);
}
event_index = index;
}
ngx_io = ngx_os_io;
ngx_event_actions = ngx_select_module_ctx.actions;
ngx_event_flags = NGX_USE_LEVEL_EVENT;
max_read = 0;
max_write = 0;
return NGX_OK;
}
static void
ngx_select_done(ngx_cycle_t *cycle)
{
ngx_free(event_index);
event_index = NULL;
}
static ngx_int_t
ngx_select_add_event(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags)
{
ngx_connection_t *c;
c = ev->data;
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"select add event fd:%d ev:%i", c->fd, event);
if (ev->index != NGX_INVALID_INDEX) {
ngx_log_error(NGX_LOG_ALERT, ev->log, 0,
"select event fd:%d ev:%i is already set", c->fd, event);
return NGX_OK;
}
if ((event == NGX_READ_EVENT && ev->write)
|| (event == NGX_WRITE_EVENT && !ev->write))
{
ngx_log_error(NGX_LOG_ALERT, ev->log, 0,
"invalid select %s event fd:%d ev:%i",
ev->write ? "write" : "read", c->fd, event);
return NGX_ERROR;
}
if ((event == NGX_READ_EVENT && max_read >= FD_SETSIZE)
|| (event == NGX_WRITE_EVENT && max_write >= FD_SETSIZE))
{
ngx_log_error(NGX_LOG_ERR, ev->log, 0,
"maximum number of descriptors "
"supported by select() is %d", FD_SETSIZE);
return NGX_ERROR;
}
if (event == NGX_READ_EVENT) {
FD_SET(c->fd, &master_read_fd_set);
max_read++;
} else if (event == NGX_WRITE_EVENT) {
FD_SET(c->fd, &master_write_fd_set);
max_write++;
}
ev->active = 1;
event_index[nevents] = ev;
ev->index = nevents;
nevents++;
return NGX_OK;
}
static ngx_int_t
ngx_select_del_event(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags)
{
ngx_event_t *e;
ngx_connection_t *c;
c = ev->data;
ev->active = 0;
if (ev->index == NGX_INVALID_INDEX) {
return NGX_OK;
}
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0,
"select del event fd:%d ev:%i", c->fd, event);
if (event == NGX_READ_EVENT) {
FD_CLR(c->fd, &master_read_fd_set);
max_read--;
} else if (event == NGX_WRITE_EVENT) {
FD_CLR(c->fd, &master_write_fd_set);
max_write--;
}
if (ev->index < --nevents) {
e = event_index[nevents];
event_index[ev->index] = e;
e->index = ev->index;
}
ev->index = NGX_INVALID_INDEX;
return NGX_OK;
}
static ngx_int_t
ngx_select_process_events(ngx_cycle_t *cycle, ngx_msec_t timer,
ngx_uint_t flags)
{
int ready, nready;
ngx_err_t err;
ngx_uint_t i, found;
ngx_event_t *ev;
ngx_queue_t *queue;
struct timeval tv, *tp;
ngx_connection_t *c;
#if (NGX_DEBUG)
if (cycle->log->log_level & NGX_LOG_DEBUG_ALL) {
for (i = 0; i < nevents; i++) {
ev = event_index[i];
c = ev->data;
ngx_log_debug2(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"select event: fd:%d wr:%d", c->fd, ev->write);
}
}
#endif
if (timer == NGX_TIMER_INFINITE) {
tp = NULL;
} else {
tv.tv_sec = (long) (timer / 1000);
tv.tv_usec = (long) ((timer % 1000) * 1000);
tp = &tv;
}
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"select timer: %M", timer);
work_read_fd_set = master_read_fd_set;
work_write_fd_set = master_write_fd_set;
work_except_fd_set = master_write_fd_set;
if (max_read || max_write) {
ready = select(0, &work_read_fd_set, &work_write_fd_set,
&work_except_fd_set, tp);
} else {
/*
* Winsock select() requires that at least one descriptor set must be
* be non-null, and any non-null descriptor set must contain at least
* one handle to a socket. Otherwise select() returns WSAEINVAL.
*/
ngx_msleep(timer);
ready = 0;
}
err = (ready == -1) ? ngx_socket_errno : 0;
if (flags & NGX_UPDATE_TIME) {
ngx_time_update();
}
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"select ready %d", ready);
if (err) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, err, "select() failed");
if (err == WSAENOTSOCK) {
ngx_select_repair_fd_sets(cycle);
}
return NGX_ERROR;
}
if (ready == 0) {
if (timer != NGX_TIMER_INFINITE) {
return NGX_OK;
}
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
"select() returned no events without timeout");
return NGX_ERROR;
}
nready = 0;
for (i = 0; i < nevents; i++) {
ev = event_index[i];
c = ev->data;
found = 0;
if (ev->write) {
if (FD_ISSET(c->fd, &work_write_fd_set)) {
found++;
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"select write %d", c->fd);
}
if (FD_ISSET(c->fd, &work_except_fd_set)) {
found++;
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"select except %d", c->fd);
}
} else {
if (FD_ISSET(c->fd, &work_read_fd_set)) {
found++;
ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
"select read %d", c->fd);
}
}
if (found) {
ev->ready = 1;
ev->available = -1;
queue = ev->accept ? &ngx_posted_accept_events
: &ngx_posted_events;
ngx_post_event(ev, queue);
nready += found;
}
}
if (ready != nready) {
ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
"select ready != events: %d:%d", ready, nready);
ngx_select_repair_fd_sets(cycle);
}
return NGX_OK;
}
static void
ngx_select_repair_fd_sets(ngx_cycle_t *cycle)
{
int n;
u_int i;
socklen_t len;
ngx_err_t err;
ngx_socket_t s;
for (i = 0; i < master_read_fd_set.fd_count; i++) {
s = master_read_fd_set.fd_array[i];
len = sizeof(int);
if (getsockopt(s, SOL_SOCKET, SO_TYPE, (char *) &n, &len) == -1) {
err = ngx_socket_errno;
ngx_log_error(NGX_LOG_ALERT, cycle->log, err,
"invalid descriptor #%d in read fd_set", s);
FD_CLR(s, &master_read_fd_set);
}
}
for (i = 0; i < master_write_fd_set.fd_count; i++) {
s = master_write_fd_set.fd_array[i];
len = sizeof(int);
if (getsockopt(s, SOL_SOCKET, SO_TYPE, (char *) &n, &len) == -1) {
err = ngx_socket_errno;
ngx_log_error(NGX_LOG_ALERT, cycle->log, err,
"invalid descriptor #%d in write fd_set", s);
FD_CLR(s, &master_write_fd_set);
}
}
}
static char *
ngx_select_init_conf(ngx_cycle_t *cycle, void *conf)
{
ngx_event_conf_t *ecf;
ecf = ngx_event_get_conf(cycle->conf_ctx, ngx_event_core_module);
if (ecf->use != ngx_select_module.ctx_index) {
return NGX_CONF_OK;
}
return NGX_CONF_OK;
}