technocloud-public/nginx-unit
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

nxt_event_timer has been renamed to nxt_timer.

Browse Source
This commit is contained in:
Igor Sysoev
2017-01-30 16:47:50 +03:00
parent 952291c93c
commit bb87fa11ca
26 changed files with 521 additions and 532 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/nxt_cache.h
View File

@@ -109,7 +109,7 @@ struct nxt_cache_query_s {
nxt_time_t now;
nxt_msec_t timeout;
nxt_event_timer_t timer;
nxt_timer_t timer;
};

2
src/nxt_cycle.h
View File

@@ -66,7 +66,7 @@ struct nxt_cycle_s {
void **core_ctx;
nxt_event_timer_t timer;
nxt_timer_t timer;
uint8_t daemon;
uint8_t batch;

2
src/nxt_epoll.c
View File

@@ -1136,7 +1136,7 @@ nxt_epoll_edge_event_conn_connected(nxt_task_t *task, void *obj, void *data)
c->socket.write = NXT_EVENT_BLOCKED;
if (c->write_state->autoreset_timer) {
nxt_event_timer_disable(&c->write_timer);
nxt_timer_disable(&c->write_timer);
}
nxt_event_conn_io_handle(task->thread, c->write_work_queue,

8
src/nxt_event_conn.c
View File

@@ -145,8 +145,8 @@ nxt_event_conn_close(nxt_task_t *task, nxt_event_conn_t *c)
engine = thr->engine;
nxt_event_timer_delete(engine, &c->read_timer);
nxt_event_timer_delete(engine, &c->write_timer);
nxt_timer_delete(engine, &c->read_timer);
nxt_timer_delete(engine, &c->write_timer);
nxt_event_fd_close(engine, &c->socket);
engine->connections--;
@@ -203,7 +203,7 @@ nxt_event_conn_close_socket(nxt_task_t *task, void *obj, void *data)
void
nxt_event_conn_timer(nxt_event_engine_t *engine, nxt_event_conn_t *c,
const nxt_event_conn_state_t *state, nxt_event_timer_t *tev)
const nxt_event_conn_state_t *state, nxt_timer_t *tev)
{
nxt_msec_t timer;
@@ -212,7 +212,7 @@ nxt_event_conn_timer(nxt_event_engine_t *engine, nxt_event_conn_t *c,
if (timer != 0) {
tev->handler = state->timer_handler;
nxt_event_timer_add(engine, tev, timer);
nxt_timer_add(engine, tev, timer);
}
}
}

17
src/nxt_event_conn.h
View File

@@ -110,13 +110,13 @@ struct nxt_event_conn_s {
nxt_buf_t *read;
const nxt_event_conn_state_t *read_state;
nxt_work_queue_t *read_work_queue;
nxt_event_timer_t read_timer;
nxt_timer_t read_timer;
nxt_buf_t *write;
const nxt_event_conn_state_t *write_state;
nxt_work_queue_t *write_work_queue;
nxt_event_write_rate_t *rate;
nxt_event_timer_t write_timer;
nxt_timer_t write_timer;
nxt_off_t sent;
uint32_t max_chunk;
@@ -182,7 +182,7 @@ typedef struct {
nxt_listen_socket_t *listen;
nxt_event_timer_t timer;
nxt_timer_t timer;
nxt_queue_link_t link;
} nxt_event_conn_listen_t;
@@ -205,19 +205,19 @@ nxt_event_conn_timer_init(ev, c, wq) \
do { \
(ev)->work_queue = (wq); \
(ev)->log = &(c)->log; \
(ev)->precision = NXT_EVENT_TIMER_DEFAULT_PRECISION; \
nxt_event_timer_ident((ev), (c)->socket.fd); \
(ev)->precision = NXT_TIMER_DEFAULT_PRECISION; \
nxt_timer_ident((ev), (c)->socket.fd); \
} while (0)
#define \
nxt_event_read_timer_conn(ev) \
nxt_event_timer_data(ev, nxt_event_conn_t, read_timer)
nxt_timer_data(ev, nxt_event_conn_t, read_timer)
#define \
nxt_event_write_timer_conn(ev) \
nxt_event_timer_data(ev, nxt_event_conn_t, write_timer)
nxt_timer_data(ev, nxt_event_conn_t, write_timer)
#if (NXT_HAVE_UNIX_DOMAIN)
@@ -258,8 +258,7 @@ void nxt_event_conn_io_shutdown(nxt_task_t *task, void *obj, void *data);
NXT_EXPORT void nxt_event_conn_close(nxt_task_t *task, nxt_event_conn_t *c);
NXT_EXPORT void nxt_event_conn_timer(nxt_event_engine_t *engine,
nxt_event_conn_t *c, const nxt_event_conn_state_t *state,
nxt_event_timer_t *tev);
nxt_event_conn_t *c, const nxt_event_conn_state_t *state, nxt_timer_t *tev);
NXT_EXPORT void nxt_event_conn_work_queue_set(nxt_event_conn_t *c,
nxt_work_queue_t *wq);

12
src/nxt_event_conn_accept.c
View File

@@ -66,7 +66,7 @@ nxt_event_conn_listen(nxt_task_t *task, nxt_listen_socket_t *ls)
cls->timer.handler = nxt_event_conn_listen_timer_handler;
cls->timer.log = &nxt_main_log;
nxt_event_timer_ident(&cls->timer, cls->socket.fd);
nxt_timer_ident(&cls->timer, cls->socket.fd);
cls->task.thread = task->thread;
cls->task.log = &nxt_main_log;
@@ -202,8 +202,8 @@ nxt_event_conn_accept(nxt_task_t *task, nxt_event_conn_listen_t *cls,
{
nxt_event_conn_t *next;
nxt_event_timer_ident(&c->read_timer, c->socket.fd);
nxt_event_timer_ident(&c->write_timer, c->socket.fd);
nxt_timer_ident(&c->read_timer, c->socket.fd);
nxt_timer_ident(&c->write_timer, c->socket.fd);
/* This allocation cannot fail. */
(void) nxt_sockaddr_text(c->mem_pool, c->remote, 0);
@@ -287,7 +287,7 @@ nxt_event_conn_accept_close_idle(nxt_task_t *task, nxt_event_conn_listen_t *cls)
}
}
nxt_event_timer_add(task->thread->engine, &cls->timer, 1000);
nxt_timer_add(task->thread->engine, &cls->timer, 1000);
nxt_event_fd_disable_read(task->thread->engine, &cls->socket);
@@ -340,13 +340,13 @@ nxt_event_conn_accept_error(nxt_task_t *task, nxt_event_conn_listen_t *cls,
static void
nxt_event_conn_listen_timer_handler(nxt_task_t *task, void *obj, void *data)
{
nxt_timer_t *ev;
nxt_event_conn_t *c;
nxt_event_timer_t *ev;
nxt_event_conn_listen_t *cls;
ev = obj;
cls = nxt_event_timer_data(ev, nxt_event_conn_listen_t, timer);
cls = nxt_timer_data(ev, nxt_event_conn_listen_t, timer);
c = cls->socket.data;
if (c == NULL) {

6
src/nxt_event_conn_connect.c
View File

@@ -124,8 +124,8 @@ nxt_event_conn_socket(nxt_task_t *task, nxt_event_conn_t *c)
#endif
c->socket.fd = s;
nxt_event_timer_ident(&c->read_timer, s);
nxt_event_timer_ident(&c->write_timer, s);
nxt_timer_ident(&c->read_timer, s);
nxt_timer_ident(&c->write_timer, s);
c->socket.task = task;
c->read_timer.task = task;
@@ -156,7 +156,7 @@ nxt_event_conn_connect_test(nxt_task_t *task, void *obj, void *data)
nxt_event_fd_block_write(task->thread->engine, &c->socket);
if (c->write_state->autoreset_timer) {
nxt_event_timer_disable(&c->write_timer);
nxt_timer_disable(&c->write_timer);
}
err = 0;

10
src/nxt_event_conn_job_sendfile.c
View File

@@ -84,8 +84,8 @@ nxt_event_conn_job_sendfile_start(nxt_task_t *task, void *obj, void *data)
c->blocked = 1;
if (c->write_timer.state != NXT_EVENT_TIMER_DISABLED) {
c->write_timer.state = NXT_EVENT_TIMER_BLOCKED;
if (c->write_timer.state != NXT_TIMER_DISABLED) {
c->write_timer.state = NXT_TIMER_BLOCKED;
}
nxt_job_start(task, &jbs->job, nxt_event_conn_job_sendfile_handler);
@@ -210,10 +210,10 @@ nxt_event_conn_job_sendfile_return(nxt_task_t *task, void *obj, void *data)
}
if (sent != 0 && c->write_state->autoreset_timer) {
nxt_event_timer_disable(&c->write_timer);
nxt_timer_disable(&c->write_timer);
} else if (c->write_timer.state == NXT_EVENT_TIMER_BLOCKED) {
c->write_timer.state = NXT_EVENT_TIMER_ACTIVE;
} else if (c->write_timer.state == NXT_TIMER_BLOCKED) {
c->write_timer.state = NXT_TIMER_ACTIVE;
}
if (c->socket.error == 0

16
src/nxt_event_conn_proxy.c
View File

@@ -797,8 +797,8 @@ nxt_event_conn_proxy_error(nxt_task_t *task, void *obj, void *data)
static void
nxt_event_conn_proxy_read_timeout(nxt_task_t *task, void *obj, void *data)
{
nxt_event_conn_t *c;
nxt_event_timer_t *ev;
nxt_timer_t *ev;
nxt_event_conn_t *c;
ev = obj;
@@ -815,8 +815,8 @@ nxt_event_conn_proxy_read_timeout(nxt_task_t *task, void *obj, void *data)
static void
nxt_event_conn_proxy_write_timeout(nxt_task_t *task, void *obj, void *data)
{
nxt_event_conn_t *c;
nxt_event_timer_t *ev;
nxt_timer_t *ev;
nxt_event_conn_t *c;
ev = obj;
@@ -870,16 +870,16 @@ nxt_event_conn_proxy_refused(nxt_task_t *task, void *obj, void *data)
p->delayed = 1;
peer->write_timer.handler = nxt_event_conn_proxy_reconnect_handler;
nxt_event_timer_add(task->thread->engine, &peer->write_timer,
p->reconnect_timeout);
nxt_timer_add(task->thread->engine, &peer->write_timer,
p->reconnect_timeout);
}
static void
nxt_event_conn_proxy_reconnect_handler(nxt_task_t *task, void *obj, void *data)
{
nxt_timer_t *ev;
nxt_event_conn_t *peer;
nxt_event_timer_t *ev;
nxt_event_conn_proxy_t *p;
ev = obj;
@@ -1009,7 +1009,7 @@ nxt_event_conn_proxy_complete(nxt_task_t *task, nxt_event_conn_proxy_t *p)
} else if (p->delayed) {
nxt_queue_remove(&p->peer->link);
nxt_event_timer_delete(task->thread->engine, &p->peer->write_timer);
nxt_timer_delete(task->thread->engine, &p->peer->write_timer);
}
nxt_mem_free(p->client->mem_pool, p->client_buffer);

6
src/nxt_event_conn_read.c
View File

@@ -86,7 +86,7 @@ nxt_event_conn_io_read(nxt_task_t *task, void *obj, void *data)
if (n != NXT_AGAIN) {
nxt_event_fd_block_read(engine, &c->socket);
nxt_event_timer_disable(&c->read_timer);
nxt_timer_disable(&c->read_timer);
if (n == 0) {
handler = state->close_handler;
@@ -107,7 +107,7 @@ nxt_event_conn_io_read(nxt_task_t *task, void *obj, void *data)
c->socket.read_handler = c->io->read;
c->socket.error_handler = state->error_handler;
if (c->read_timer.state == NXT_EVENT_TIMER_DISABLED
if (c->read_timer.state == NXT_TIMER_DISABLED
|| nxt_event_fd_is_disabled(c->socket.read))
{
/* Timer may be set or reset. */
@@ -125,7 +125,7 @@ ready:
nxt_event_fd_block_read(engine, &c->socket);
if (state->autoreset_timer) {
nxt_event_timer_disable(&c->read_timer);
nxt_timer_disable(&c->read_timer);
}
handler = state->ready_handler;

8
src/nxt_event_conn_write.c
View File

@@ -90,7 +90,7 @@ nxt_event_conn_io_write(nxt_task_t *task, void *obj, void *data)
if (sent != 0) {
if (c->write_state->autoreset_timer) {
nxt_event_timer_disable(&c->write_timer);
nxt_timer_disable(&c->write_timer);
}
}
@@ -103,7 +103,7 @@ nxt_event_conn_io_write(nxt_task_t *task, void *obj, void *data)
* process other recevied events and to get new events.
*/
c->write_timer.handler = nxt_event_conn_write_timer_handler;
nxt_event_timer_add(engine, &c->write_timer, 0);
nxt_timer_add(engine, &c->write_timer, 0);
} else if (ret == NXT_AGAIN) {
/*
@@ -204,7 +204,7 @@ nxt_event_conn_write_delayed(nxt_event_engine_t *engine, nxt_event_conn_t *c,
nxt_event_fd_block_write(engine, &c->socket);
c->write_timer.handler = nxt_event_conn_write_timer_handler;
nxt_event_timer_add(engine, &c->write_timer, timer);
nxt_timer_add(engine, &c->write_timer, timer);
return 1;
}
@@ -288,8 +288,8 @@ nxt_event_conn_exponential_approximation(double x)
static void
nxt_event_conn_write_timer_handler(nxt_task_t *task, void *obj, void *data)
{
nxt_timer_t *ev;
nxt_event_conn_t *c;
nxt_event_timer_t *ev;
ev = obj;

6
src/nxt_event_engine.c
View File

@@ -109,7 +109,7 @@ nxt_event_engine_create(nxt_thread_t *thr, const nxt_event_set_ops_t *event_set,
goto post_fail;
}
if (nxt_event_timers_init(&engine->timers, 4 * events) != NXT_OK) {
if (nxt_timers_init(&engine->timers, 4 * events) != NXT_OK) {
goto timers_fail;
}
@@ -521,7 +521,7 @@ nxt_event_engine_start(nxt_event_engine_t *engine)
/* Attach some event engine work queues in preferred order. */
timeout = nxt_event_timer_find(engine);
timeout = nxt_timer_find(engine);
engine->event->poll(&engine->task, engine->event_set, timeout);
@@ -535,7 +535,7 @@ nxt_event_engine_start(nxt_event_engine_t *engine)
now = nxt_thread_monotonic_time(thr) / 1000000;
if (timeout == 0 || now != engine->timers.now) {
nxt_event_timer_expire(thr, now);
nxt_timer_expire(thr, now);
}
}
}

2
src/nxt_event_engine.h
View File

@@ -21,7 +21,7 @@ struct nxt_event_engine_s {
const nxt_event_set_ops_t *event;
nxt_event_set_t *event_set;
nxt_event_timers_t timers;
nxt_timers_t timers;
nxt_task_t task;
/* The engine ID, the main engine has ID 0. */

319
src/nxt_event_timer.c
View File

@@ -1,319 +0,0 @@
/*
* Copyright (C) Igor Sysoev
* Copyright (C) NGINX, Inc.
*/
#include <nxt_main.h>
/*
* Timer operations are batched to improve instruction and data
* cache locality of rbtree operations.
*
* nxt_event_timer_add() adds a timer to the changes array to add or to
* modify the timer. The changes are processed by nxt_event_timer_find().
*
* nxt_event_timer_disable() disables a timer. The disabled timer may
* however present in rbtree for a long time and may be eventually removed
* by nxt_event_timer_find() or nxt_event_timer_expire().
*
* nxt_event_timer_delete() removes a timer at once from both the rbtree and
* the changes array and should be used only if the timer memory must be freed.
*/
static intptr_t nxt_event_timer_rbtree_compare(nxt_rbtree_node_t *node1,
nxt_rbtree_node_t *node2);
static void nxt_event_timer_change(nxt_event_timers_t *timers,
nxt_event_timer_t *ev, nxt_msec_t time);
static void nxt_event_commit_timer_changes(nxt_event_timers_t *timers);
static void nxt_event_timer_drop_changes(nxt_event_timers_t *timers,
nxt_event_timer_t *ev);
nxt_int_t
nxt_event_timers_init(nxt_event_timers_t *timers, nxt_uint_t mchanges)
{
nxt_rbtree_init(&timers->tree, nxt_event_timer_rbtree_compare);
timers->mchanges = mchanges;
timers->changes = nxt_malloc(sizeof(nxt_event_timer_change_t) * mchanges);
if (nxt_fast_path(timers->changes != NULL)) {
return NXT_OK;
}
return NXT_ERROR;
}
static intptr_t
nxt_event_timer_rbtree_compare(nxt_rbtree_node_t *node1,
nxt_rbtree_node_t *node2)
{
nxt_event_timer_t *ev1, *ev2;
ev1 = (nxt_event_timer_t *) node1;
ev2 = (nxt_event_timer_t *) node2;
/*
* Timer values are distributed in small range, usually several minutes
* and overflow every 49 days if nxt_msec_t is stored in 32 bits.
* This signed comparison takes into account that overflow.
*/
/* ev1->time < ev2->time */
return nxt_msec_diff(ev1->time, ev2->time);
}
void
nxt_event_timer_add(nxt_event_engine_t *engine, nxt_event_timer_t *ev,
nxt_msec_t timer)
{
int32_t diff;
uint32_t time;
time = engine->timers.now + timer;
if (nxt_event_timer_is_in_tree(ev)) {
diff = nxt_msec_diff(time, ev->time);
/*
* Use the previous timer if difference between it and the
* new timer is less than required precision milliseconds:
* this decreases rbtree operations for fast connections.
*/
if (nxt_abs(diff) < ev->precision) {
nxt_log_debug(ev->log, "event timer previous: %D: %d:%M",
ev->ident, ev->state, time);
if (ev->state == NXT_EVENT_TIMER_DISABLED) {
ev->state = NXT_EVENT_TIMER_ACTIVE;
}
return;
}
nxt_log_debug(ev->log, "event timer change: %D: %d:%M",
ev->ident, ev->state, ev->time);
} else {
/*
* The timer's time is updated here just to log a correct
* value by debug logging in nxt_event_timer_disable().
* It could be updated only in nxt_event_commit_timer_changes()
* just before nxt_rbtree_insert().
*/
ev->time = time;
nxt_log_debug(ev->log, "event timer add: %D: %M:%M",
ev->ident, timer, time);
}
nxt_event_timer_change(&engine->timers, ev, time);
}
static void
nxt_event_timer_change(nxt_event_timers_t *timers, nxt_event_timer_t *ev,
nxt_msec_t time)
{
nxt_event_timer_change_t *ch;
if (timers->nchanges >= timers->mchanges) {
nxt_event_commit_timer_changes(timers);
}
ev->state = NXT_EVENT_TIMER_ACTIVE;
ch = &timers->changes[timers->nchanges];
ch->time = time;
ch->event = ev;
timers->nchanges++;
}
#if (NXT_DEBUG)
void
nxt_event_timer_disable(nxt_event_timer_t *ev)
{
nxt_debug(ev->task, "event timer disable: %D: %d:%M",
ev->ident, ev->state, ev->time);
ev->state = NXT_EVENT_TIMER_DISABLED;
}
#endif
void
nxt_event_timer_delete(nxt_event_engine_t *engine, nxt_event_timer_t *ev)
{
if (nxt_event_timer_is_in_tree(ev)) {
nxt_log_debug(ev->log, "event timer delete: %D: %d:%M",
ev->ident, ev->state, ev->time);
nxt_rbtree_delete(&engine->timers.tree, &ev->node);
nxt_event_timer_in_tree_clear(ev);
ev->state = NXT_EVENT_TIMER_DISABLED;
}
nxt_event_timer_drop_changes(&engine->timers, ev);
}
static void
nxt_event_timer_drop_changes(nxt_event_timers_t *timers, nxt_event_timer_t *ev)
{
nxt_event_timer_change_t *dst, *src, *end;
dst = timers->changes;
end = dst + timers->nchanges;
for (src = dst; src < end; src++) {
if (src->event == ev) {
continue;
}
if (dst != src) {
*dst = *src;
}
dst++;
}
timers->nchanges -= end - dst;
}
static void
nxt_event_commit_timer_changes(nxt_event_timers_t *timers)
{
nxt_event_timer_t *ev;
nxt_event_timer_change_t *ch, *end;
nxt_thread_log_debug("event timers changes: %ui", timers->nchanges);
ch = timers->changes;
end = ch + timers->nchanges;
while (ch < end) {
ev = ch->event;
if (ev->state != NXT_EVENT_TIMER_DISABLED) {
if (nxt_event_timer_is_in_tree(ev)) {
nxt_log_debug(ev->log, "event timer delete: %D: %d:%M",
ev->ident, ev->state, ev->time);
nxt_rbtree_delete(&timers->tree, &ev->node);
ev->time = ch->time;
}
nxt_log_debug(ev->log, "event timer add: %D: %M",
ev->ident, ev->time);
nxt_rbtree_insert(&timers->tree, &ev->node);
nxt_event_timer_in_tree_set(ev);
}
ch++;
}
timers->nchanges = 0;
}
nxt_msec_t
nxt_event_timer_find(nxt_event_engine_t *engine)
{
int32_t time;
nxt_rbtree_node_t *node, *next;
nxt_event_timer_t *ev;
if (engine->timers.nchanges != 0) {
nxt_event_commit_timer_changes(&engine->timers);
}
for (node = nxt_rbtree_min(&engine->timers.tree);
nxt_rbtree_is_there_successor(&engine->timers.tree, node);
node = next)
{
next = nxt_rbtree_node_successor(&engine->timers.tree, node);
ev = (nxt_event_timer_t *) node;
if (ev->state != NXT_EVENT_TIMER_DISABLED) {
if (ev->state == NXT_EVENT_TIMER_BLOCKED) {
nxt_log_debug(ev->log, "event timer blocked: %D: %M",
ev->ident, ev->time);
continue;
}
time = nxt_msec_diff(ev->time, engine->timers.now);
return (nxt_msec_t) nxt_max(time, 0);
}
/* Delete disabled timer. */
nxt_log_debug(ev->log, "event timer delete: %D: 0:%M",
ev->ident, ev->time);
nxt_rbtree_delete(&engine->timers.tree, &ev->node);
nxt_event_timer_in_tree_clear(ev);
}
return NXT_INFINITE_MSEC;
}
void
nxt_event_timer_expire(nxt_thread_t *thr, nxt_msec_t now)
{
nxt_rbtree_t *tree;
nxt_rbtree_node_t *node, *next;
nxt_event_timer_t *ev;
thr->engine->timers.now = now;
tree = &thr->engine->timers.tree;
for (node = nxt_rbtree_min(tree);
nxt_rbtree_is_there_successor(tree, node);
node = next)
{
ev = (nxt_event_timer_t *) node;
/* ev->time > now */
if (nxt_msec_diff(ev->time, now) > 0) {
return;
}
next = nxt_rbtree_node_successor(tree, node);
if (ev->state == NXT_EVENT_TIMER_BLOCKED) {
nxt_log_debug(ev->log, "event timer blocked: %D: %M",
ev->ident, ev->time);
continue;
}
nxt_log_debug(ev->log, "event timer delete: %D: %d:%M",
ev->ident, ev->state, ev->time);
nxt_rbtree_delete(tree, &ev->node);
nxt_event_timer_in_tree_clear(ev);
if (ev->state != NXT_EVENT_TIMER_DISABLED) {
ev->state = NXT_EVENT_TIMER_DISABLED;
nxt_work_queue_add(ev->work_queue, ev->handler, ev->task, ev, NULL);
}
}
}

147
src/nxt_event_timer.h
View File

@@ -1,147 +0,0 @@
/*
* Copyright (C) Igor Sysoev
* Copyright (C) NGINX, Inc.
*/
#ifndef _NXT_EVENT_TIMER_H_INCLUDED_
#define _NXT_EVENT_TIMER_H_INCLUDED_
/* Valid values are between 1ms to 255ms. */
#define NXT_EVENT_TIMER_DEFAULT_PRECISION 100
//#define NXT_EVENT_TIMER_DEFAULT_PRECISION 1
#if (NXT_DEBUG)
#define NXT_EVENT_TIMER { NXT_RBTREE_NODE_INIT, 0, 0, 0, \
NULL, NULL, NULL, NULL, -1 }
#else
#define NXT_EVENT_TIMER { NXT_RBTREE_NODE_INIT, 0, 0, 0, \
NULL, NULL, NULL, NULL }
#endif
typedef struct {
/* The rbtree node must be the first field. */
NXT_RBTREE_NODE (node);
uint8_t state;
uint8_t precision;
nxt_msec_t time;
nxt_work_queue_t *work_queue;
nxt_work_handler_t handler;
nxt_task_t *task;
nxt_log_t *log;
#if (NXT_DEBUG)
int32_t ident;
#endif
} nxt_event_timer_t;
typedef struct {
nxt_msec_t time;
nxt_event_timer_t *event;
} nxt_event_timer_change_t;
typedef struct {
nxt_rbtree_t tree;
/* An overflown milliseconds counter. */
nxt_msec_t now;
nxt_uint_t mchanges;
nxt_uint_t nchanges;
nxt_event_timer_change_t *changes;
} nxt_event_timers_t;
#define \
nxt_event_timer_data(ev, type, timer) \
nxt_container_of(ev, type, timer)
/*
* When timer resides in rbtree all links of its node are not NULL.
* A parent link is the nearst to other timer flags.
*/
#define \
nxt_event_timer_is_in_tree(ev) \
((ev)->node.parent != NULL)
#define \
nxt_event_timer_in_tree_set(ev)
/* Noop, because rbtree insertion sets a node's parent link. */
#define \
nxt_event_timer_in_tree_clear(ev) \
(ev)->node.parent = NULL
#define NXT_EVENT_TIMER_DISABLED 0
#define NXT_EVENT_TIMER_BLOCKED 1
#define NXT_EVENT_TIMER_ACTIVE 2
#if (NXT_DEBUG)
#define \
nxt_event_timer_ident(ev, val) \
(ev)->ident = (val)
#else
#define \
nxt_event_timer_ident(ev, val)
#endif
nxt_inline nxt_event_timer_t *
nxt_event_timer_create(int32_t ident)
{
nxt_event_timer_t *ev;
ev = nxt_zalloc(sizeof(nxt_event_timer_t));
if (ev == NULL) {
return NULL;
}
ev->precision = NXT_EVENT_TIMER_DEFAULT_PRECISION;
#if (NXT_DEBUG)
ev->ident = ident;
#endif
return ev;
}
nxt_int_t nxt_event_timers_init(nxt_event_timers_t *timers,
nxt_uint_t mchanges);
NXT_EXPORT void nxt_event_timer_add(nxt_event_engine_t *engine,
nxt_event_timer_t *ev, nxt_msec_t timer);
NXT_EXPORT void nxt_event_timer_delete(nxt_event_engine_t *engine,
nxt_event_timer_t *ev);
nxt_msec_t nxt_event_timer_find(nxt_event_engine_t *engine);
void nxt_event_timer_expire(nxt_thread_t *thr, nxt_msec_t now);
#if (NXT_DEBUG)
NXT_EXPORT void nxt_event_timer_disable(nxt_event_timer_t *ev);
#else
#define \
nxt_event_timer_disable(ev) \
(ev)->state = NXT_EVENT_TIMER_DISABLED
#endif
#endif /* _NXT_EVENT_TIMER_H_INCLUDED_ */

8
src/nxt_fiber.c
View File

@@ -398,7 +398,7 @@ nxt_fiber_sleep(nxt_task_t *task, nxt_msec_t timeout)
task = &fib->task;
nxt_event_timer_add(task->thread->engine, &fib->timer, timeout);
nxt_timer_add(task->thread->engine, &fib->timer, timeout);
if (_setjmp(fib->jmp) == 0) {
@@ -416,14 +416,14 @@ nxt_fiber_sleep(nxt_task_t *task, nxt_msec_t timeout)
static void
nxt_fiber_timer_handler(nxt_task_t *task, void *obj, void *data)
{
nxt_fiber_t *fib;
nxt_event_timer_t *ev;
nxt_fiber_t *fib;
nxt_timer_t *ev;
ev = obj;
nxt_debug(task, "fiber timer handler");
fib = nxt_event_timer_data(ev, nxt_fiber_t, timer);
fib = nxt_timer_data(ev, nxt_fiber_t, timer);
nxt_fiber_switch(task, fib);

2
src/nxt_fiber.h
View File

@@ -32,7 +32,7 @@ struct nxt_fiber_s {
nxt_fiber_main_t *main;
nxt_fiber_t *next;
nxt_event_timer_t timer;
nxt_timer_t timer;
};

2
src/nxt_kqueue.c
View File

@@ -934,7 +934,7 @@ nxt_kqueue_event_conn_connected(nxt_task_t *task, void *obj, void *data)
c->socket.write = NXT_EVENT_BLOCKED;
if (c->write_state->autoreset_timer) {
nxt_event_timer_disable(&c->write_timer);
nxt_timer_disable(&c->write_timer);
}
nxt_work_queue_add(c->write_work_queue, c->write_state->ready_handler,

6
src/nxt_log_moderation.c
View File

@@ -62,7 +62,7 @@ nxt_log_moderate_allow(nxt_log_moderation_t *mod)
mod->timer.handler = nxt_log_moderate_timer_handler;
mod->timer.log = &nxt_main_log;
nxt_event_timer_add(thr->engine, &mod->timer, 1000);
nxt_timer_add(thr->engine, &mod->timer, 1000);
}
return allow;
@@ -73,12 +73,12 @@ static void
nxt_log_moderate_timer_handler(nxt_task_t *task, void *obj, void *data)
{
nxt_bool_t msg;
nxt_timer_t *ev;
nxt_atomic_uint_t n;
nxt_event_timer_t *ev;
nxt_log_moderation_t *mod;
ev = obj;
mod = nxt_event_timer_data(ev, nxt_log_moderation_t, timer);
mod = nxt_timer_data(ev, nxt_log_moderation_t, timer);
nxt_thread_spin_lock(&mod->lock);

4
src/nxt_log_moderation.h
View File

@@ -16,11 +16,11 @@ typedef struct {
nxt_pid_t pid;
nxt_uint_t count;
nxt_time_t last;
nxt_event_timer_t timer;
nxt_timer_t timer;
} nxt_log_moderation_t;
#define NXT_LOG_MODERATION 0, -1, 0, 0, NXT_EVENT_TIMER
#define NXT_LOG_MODERATION 0, -1, 0, 0, NXT_TIMER
#define \

2
src/nxt_main.h
View File

@@ -71,7 +71,7 @@ typedef struct {
#include <nxt_thread_time.h>
typedef struct nxt_event_engine_s nxt_event_engine_t;
#include <nxt_event_timer.h>
#include <nxt_timer.h>
#include <nxt_fiber.h>
typedef struct nxt_thread_pool_s nxt_thread_pool_t;

4
src/nxt_master_process.c
View File

@@ -226,11 +226,11 @@ nxt_master_process_new_cycle(nxt_task_t *task, nxt_cycle_t *cycle)
*/
cycle->timer.handler = nxt_master_stop_previous_worker_processes;
cycle->timer.log = &nxt_main_log;
nxt_event_timer_ident(&cycle->timer, -1);
nxt_timer_ident(&cycle->timer, -1);
cycle->timer.work_queue = &thr->engine->fast_work_queue;
nxt_event_timer_add(thr->engine, &cycle->timer, 500);
nxt_timer_add(thr->engine, &cycle->timer, 500);
return;

2
src/nxt_openssl.c
View File

@@ -574,7 +574,7 @@ nxt_openssl_conn_io_shutdown(nxt_task_t *task, void *obj, void *data)
if (n != NXT_ERROR) { /* n == NXT_AGAIN */
c->socket.error_handler = c->read_state->error_handler;
nxt_event_timer_add(task->thread->engine, &c->read_timer, 5000);
nxt_timer_add(task->thread->engine, &c->read_timer, 5000);
return;
}

317
src/nxt_timer.c Normal file
View File

@@ -0,0 +1,317 @@
/*
* Copyright (C) Igor Sysoev
* Copyright (C) NGINX, Inc.
*/
#include <nxt_main.h>
/*
* Timer operations are batched to improve instruction and data
* cache locality of rbtree operations.
*
* nxt_timer_add() adds a timer to the changes array to add or to
* modify the timer. The changes are processed by nxt_timer_find().
*
* nxt_timer_disable() disables a timer. The disabled timer may
* however present in rbtree for a long time and may be eventually removed
* by nxt_timer_find() or nxt_timer_expire().
*
* nxt_timer_delete() removes a timer at once from both the rbtree and
* the changes array and should be used only if the timer memory must be freed.
*/
static intptr_t nxt_timer_rbtree_compare(nxt_rbtree_node_t *node1,
nxt_rbtree_node_t *node2);
static void nxt_timer_change(nxt_timers_t *timers, nxt_timer_t *timer,
nxt_msec_t time);
static void nxt_commit_timer_changes(nxt_timers_t *timers);
static void nxt_timer_drop_changes(nxt_timers_t *timers, nxt_timer_t *timer);
nxt_int_t
nxt_timers_init(nxt_timers_t *timers, nxt_uint_t mchanges)
{
nxt_rbtree_init(&timers->tree, nxt_timer_rbtree_compare);
timers->mchanges = mchanges;
timers->changes = nxt_malloc(sizeof(nxt_timer_change_t) * mchanges);
if (nxt_fast_path(timers->changes != NULL)) {
return NXT_OK;
}
return NXT_ERROR;
}
static intptr_t
nxt_timer_rbtree_compare(nxt_rbtree_node_t *node1, nxt_rbtree_node_t *node2)
{
nxt_timer_t *timer1, *timer2;
timer1 = (nxt_timer_t *) node1;
timer2 = (nxt_timer_t *) node2;
/*
* Timer values are distributed in small range, usually several minutes
* and overflow every 49 days if nxt_msec_t is stored in 32 bits.
* This signed comparison takes into account that overflow.
*/
/* timer1->time < timer2->time */
return nxt_msec_diff(timer1->time, timer2->time);
}
void
nxt_timer_add(nxt_event_engine_t *engine, nxt_timer_t *timer,
nxt_msec_t timeout)
{
int32_t diff;
uint32_t time;
time = engine->timers.now + timeout;
if (nxt_timer_is_in_tree(timer)) {
diff = nxt_msec_diff(time, timer->time);
/*
* Use the previous timer if difference between it and the
* new timer is less than required precision milliseconds:
* this decreases rbtree operations for fast connections.
*/
if (nxt_abs(diff) < timer->precision) {
nxt_log_debug(timer->log, "timer previous: %D: %d:%M",
timer->ident, timer->state, time);
if (timer->state == NXT_TIMER_DISABLED) {
timer->state = NXT_TIMER_ACTIVE;
}
return;
}
nxt_log_debug(timer->log, "timer change: %D: %d:%M",
timer->ident, timer->state, timer->time);
} else {
/*
* The timer's time is updated here just to log a correct
* value by debug logging in nxt_timer_disable().
* It could be updated only in nxt_commit_timer_changes()
* just before nxt_rbtree_insert().
*/
timer->time = time;
nxt_log_debug(timer->log, "timer add: %D: %M:%M",
timer->ident, timeout, time);
}
nxt_timer_change(&engine->timers, timer, time);
}
static void
nxt_timer_change(nxt_timers_t *timers, nxt_timer_t *timer, nxt_msec_t time)
{
nxt_timer_change_t *ch;
if (timers->nchanges >= timers->mchanges) {
nxt_commit_timer_changes(timers);
}
timer->state = NXT_TIMER_ACTIVE;
ch = &timers->changes[timers->nchanges];
ch->time = time;
ch->timer = timer;
timers->nchanges++;
}
#if (NXT_DEBUG)
void
nxt_timer_disable(nxt_timer_t *timer)
{
nxt_debug(timer->task, "timer disable: %D: %d:%M",
timer->ident, timer->state, timer->time);
timer->state = NXT_TIMER_DISABLED;
}
#endif
void
nxt_timer_delete(nxt_event_engine_t *engine, nxt_timer_t *timer)
{
if (nxt_timer_is_in_tree(timer)) {
nxt_log_debug(timer->log, "timer delete: %D: %d:%M",
timer->ident, timer->state, timer->time);
nxt_rbtree_delete(&engine->timers.tree, &timer->node);
nxt_timer_in_tree_clear(timer);
timer->state = NXT_TIMER_DISABLED;
}
nxt_timer_drop_changes(&engine->timers, timer);
}
static void
nxt_timer_drop_changes(nxt_timers_t *timers, nxt_timer_t *timer)
{
nxt_timer_change_t *dst, *src, *end;
dst = timers->changes;
end = dst + timers->nchanges;
for (src = dst; src < end; src++) {
if (src->timer == timer) {
continue;
}
if (dst != src) {
*dst = *src;
}
dst++;
}
timers->nchanges -= end - dst;
}
static void
nxt_commit_timer_changes(nxt_timers_t *timers)
{
nxt_timer_t *timer;
nxt_timer_change_t *ch, *end;
nxt_thread_log_debug("timers changes: %ui", timers->nchanges);
ch = timers->changes;
end = ch + timers->nchanges;
while (ch < end) {
timer = ch->timer;
if (timer->state != NXT_TIMER_DISABLED) {
if (nxt_timer_is_in_tree(timer)) {
nxt_log_debug(timer->log, "timer delete: %D: %d:%M",
timer->ident, timer->state, timer->time);
nxt_rbtree_delete(&timers->tree, &timer->node);
timer->time = ch->time;
}
nxt_log_debug(timer->log, "timer add: %D: %M",
timer->ident, timer->time);
nxt_rbtree_insert(&timers->tree, &timer->node);
nxt_timer_in_tree_set(timer);
}
ch++;
}
timers->nchanges = 0;
}
nxt_msec_t
nxt_timer_find(nxt_event_engine_t *engine)
{
int32_t time;
nxt_timer_t *timer;
nxt_rbtree_node_t *node, *next;
if (engine->timers.nchanges != 0) {
nxt_commit_timer_changes(&engine->timers);
}
for (node = nxt_rbtree_min(&engine->timers.tree);
nxt_rbtree_is_there_successor(&engine->timers.tree, node);
node = next)
{
next = nxt_rbtree_node_successor(&engine->timers.tree, node);
timer = (nxt_timer_t *) node;
if (timer->state != NXT_TIMER_DISABLED) {
if (timer->state == NXT_TIMER_BLOCKED) {
nxt_log_debug(timer->log, "timer blocked: %D: %M",
timer->ident, timer->time);
continue;
}
time = nxt_msec_diff(timer->time, engine->timers.now);
return (nxt_msec_t) nxt_max(time, 0);
}
/* Delete disabled timer. */
nxt_log_debug(timer->log, "timer delete: %D: 0:%M",
timer->ident, timer->time);
nxt_rbtree_delete(&engine->timers.tree, &timer->node);
nxt_timer_in_tree_clear(timer);
}
return NXT_INFINITE_MSEC;
}
void
nxt_timer_expire(nxt_thread_t *thr, nxt_msec_t now)
{
nxt_timer_t *timer;
nxt_rbtree_t *tree;
nxt_rbtree_node_t *node, *next;
thr->engine->timers.now = now;
tree = &thr->engine->timers.tree;
for (node = nxt_rbtree_min(tree);
nxt_rbtree_is_there_successor(tree, node);
node = next)
{
timer = (nxt_timer_t *) node;
/* timer->time > now */
if (nxt_msec_diff(timer->time, now) > 0) {
return;
}
next = nxt_rbtree_node_successor(tree, node);
if (timer->state == NXT_TIMER_BLOCKED) {
nxt_log_debug(timer->log, "timer blocked: %D: %M",
timer->ident, timer->time);
continue;
}
nxt_log_debug(timer->log, "timer delete: %D: %d:%M",
timer->ident, timer->state, timer->time);
nxt_rbtree_delete(tree, &timer->node);
nxt_timer_in_tree_clear(timer);
if (timer->state != NXT_TIMER_DISABLED) {
timer->state = NXT_TIMER_DISABLED;
nxt_work_queue_add(timer->work_queue, timer->handler, timer->task,
timer, NULL);
}
}
}

139
src/nxt_timer.h Normal file
View File

@@ -0,0 +1,139 @@
/*
* Copyright (C) Igor Sysoev
* Copyright (C) NGINX, Inc.
*/
#ifndef _NXT_TIMER_H_INCLUDED_
#define _NXT_TIMER_H_INCLUDED_
/* Valid values are between 1ms to 255ms. */
#define NXT_TIMER_DEFAULT_PRECISION 100
//#define NXT_TIMER_DEFAULT_PRECISION 1
#if (NXT_DEBUG)
#define NXT_TIMER { NXT_RBTREE_NODE_INIT, 0, 0, 0, \
NULL, NULL, NULL, NULL, -1 }
#else
#define NXT_TIMER { NXT_RBTREE_NODE_INIT, 0, 0, 0, \
NULL, NULL, NULL, NULL }
#endif
typedef struct {
/* The rbtree node must be the first field. */
NXT_RBTREE_NODE (node);
uint8_t state;
uint8_t precision;
nxt_msec_t time;
nxt_work_queue_t *work_queue;
nxt_work_handler_t handler;
nxt_task_t *task;
nxt_log_t *log;
#if (NXT_DEBUG)
int32_t ident;
#endif
} nxt_timer_t;
typedef struct {
nxt_msec_t time;
nxt_timer_t *timer;
} nxt_timer_change_t;
typedef struct {
nxt_rbtree_t tree;
/* An overflown milliseconds counter. */
nxt_msec_t now;
nxt_uint_t mchanges;
nxt_uint_t nchanges;
nxt_timer_change_t *changes;
} nxt_timers_t;
#define nxt_timer_data(obj, type, timer) \
nxt_container_of(obj, type, timer)
/*
* When timer resides in rbtree all links of its node are not NULL.
* A parent link is the nearst to other timer flags.
*/
#define nxt_timer_is_in_tree(timer) \
((timer)->node.parent != NULL)
#define nxt_timer_in_tree_set(timer)
/* Noop, because rbtree insertion sets a node's parent link. */
#define nxt_timer_in_tree_clear(timer) \
(timer)->node.parent = NULL
#define NXT_TIMER_DISABLED 0
#define NXT_TIMER_BLOCKED 1
#define NXT_TIMER_ACTIVE 2
#if (NXT_DEBUG)
#define nxt_timer_ident(timer, val) \
(timer)->ident = (val)
#else
#define nxt_timer_ident(timer, val)
#endif
nxt_inline nxt_timer_t *
nxt_timer_create(int32_t ident)
{
nxt_timer_t *timer;
timer = nxt_zalloc(sizeof(nxt_timer_t));
if (timer == NULL) {
return NULL;
}
timer->precision = NXT_TIMER_DEFAULT_PRECISION;
#if (NXT_DEBUG)
timer->ident = ident;
#endif
return timer;
}
nxt_int_t nxt_timers_init(nxt_timers_t *timers, nxt_uint_t mchanges);
NXT_EXPORT void nxt_timer_add(nxt_event_engine_t *engine, nxt_timer_t *timer,
nxt_msec_t timeout);
NXT_EXPORT void nxt_timer_delete(nxt_event_engine_t *engine,
nxt_timer_t *timer);
nxt_msec_t nxt_timer_find(nxt_event_engine_t *engine);
void nxt_timer_expire(nxt_thread_t *thr, nxt_msec_t now);
#if (NXT_DEBUG)
NXT_EXPORT void nxt_timer_disable(nxt_timer_t *timer);
#else
#define nxt_timer_disable(timer) \
(timer)->state = NXT_TIMER_DISABLED
#endif
#endif /* _NXT_TIMER_H_INCLUDED_ */