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

Router: using joint jobs to pass listening socket handlers to

Browse Source 
worker engines.
This commit is contained in:
Igor Sysoev
2017-07-14 17:17:15 +03:00
parent d669045b75
commit 668aabac3c
2 changed files with 75 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

132
src/nxt_router.c
View File

@@ -359,11 +359,11 @@ nxt_router_conf_new(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
static void static void
nxt_router_conf_wait(nxt_task_t *task, void *obj, void *data) nxt_router_conf_wait(nxt_task_t *task, void *obj, void *data)
{ {
nxt_router_temp_conf_t *tmcf; nxt_joint_job_t *job;
tmcf = obj; job = obj;
nxt_router_conf_success(task, tmcf); nxt_router_conf_success(task, job->tmcf);
} }
@@ -948,7 +948,8 @@ nxt_router_engine_conf_create(nxt_router_temp_conf_t *tmcf,
nxt_int_t ret; nxt_int_t ret;
nxt_thread_spinlock_t *lock; nxt_thread_spinlock_t *lock;
recf->creating = nxt_array_create(tmcf->mem_pool, 4, sizeof(nxt_work_t)); recf->creating = nxt_array_create(tmcf->mem_pool, 4,
sizeof(nxt_joint_job_t));
if (nxt_slow_path(recf->creating == NULL)) { if (nxt_slow_path(recf->creating == NULL)) {
return NXT_ERROR; return NXT_ERROR;
} }
@@ -988,7 +989,8 @@ nxt_router_engine_conf_update(nxt_router_temp_conf_t *tmcf,
nxt_int_t ret; nxt_int_t ret;
nxt_thread_spinlock_t *lock; nxt_thread_spinlock_t *lock;
recf->creating = nxt_array_create(tmcf->mem_pool, 4, sizeof(nxt_work_t)); recf->creating = nxt_array_create(tmcf->mem_pool, 4,
sizeof(nxt_joint_job_t));
if (nxt_slow_path(recf->creating == NULL)) { if (nxt_slow_path(recf->creating == NULL)) {
return NXT_ERROR; return NXT_ERROR;
} }
@@ -1001,7 +1003,8 @@ nxt_router_engine_conf_update(nxt_router_temp_conf_t *tmcf,
return ret; return ret;
} }
recf->updating = nxt_array_create(tmcf->mem_pool, 4, sizeof(nxt_work_t)); recf->updating = nxt_array_create(tmcf->mem_pool, 4,
sizeof(nxt_joint_job_t));
if (nxt_slow_path(recf->updating == NULL)) { if (nxt_slow_path(recf->updating == NULL)) {
return NXT_ERROR; return NXT_ERROR;
} }
@@ -1012,7 +1015,8 @@ nxt_router_engine_conf_update(nxt_router_temp_conf_t *tmcf,
return ret; return ret;
} }
recf->deleting = nxt_array_create(tmcf->mem_pool, 4, sizeof(nxt_work_t)); recf->deleting = nxt_array_create(tmcf->mem_pool, 4,
sizeof(nxt_joint_job_t));
if (nxt_slow_path(recf->deleting == NULL)) { if (nxt_slow_path(recf->deleting == NULL)) {
return NXT_ERROR; return NXT_ERROR;
} }
@@ -1040,7 +1044,8 @@ nxt_router_engine_conf_delete(nxt_router_temp_conf_t *tmcf,
{ {
nxt_int_t ret; nxt_int_t ret;
recf->deleting = nxt_array_create(tmcf->mem_pool, 4, sizeof(nxt_work_t)); recf->deleting = nxt_array_create(tmcf->mem_pool,
4, sizeof(nxt_joint_job_t));
if (nxt_slow_path(recf->deleting == NULL)) { if (nxt_slow_path(recf->deleting == NULL)) {
return NXT_ERROR; return NXT_ERROR;
} }
@@ -1059,7 +1064,7 @@ nxt_router_engine_joints_create(nxt_mp_t *mp, nxt_router_temp_conf_t *tmcf,
nxt_router_engine_conf_t *recf, nxt_queue_t *sockets, nxt_array_t *array, nxt_router_engine_conf_t *recf, nxt_queue_t *sockets, nxt_array_t *array,
nxt_work_handler_t handler) nxt_work_handler_t handler)
{ {
nxt_work_t *work, *back; nxt_joint_job_t *job;
nxt_queue_link_t *qlk; nxt_queue_link_t *qlk;
nxt_socket_conf_joint_t *joint; nxt_socket_conf_joint_t *joint;
@@ -1067,33 +1072,24 @@ nxt_router_engine_joints_create(nxt_mp_t *mp, nxt_router_temp_conf_t *tmcf,
qlk != nxt_queue_tail(sockets); qlk != nxt_queue_tail(sockets);
qlk = nxt_queue_next(qlk)) qlk = nxt_queue_next(qlk))
{ {
back = nxt_mp_get(tmcf->mem_pool, sizeof(nxt_work_t)); job = nxt_array_add(array);
if (nxt_slow_path(back == NULL)) { if (nxt_slow_path(job == NULL)) {
return NXT_ERROR; return NXT_ERROR;
} }
back->next = NULL; job->task = tmcf->engine->task;
back->handler = nxt_router_conf_wait; job->work.next = NULL;
back->task = &tmcf->engine->task; job->work.handler = handler;
back->obj = tmcf; job->work.task = &job->task;
back->data = NULL; job->work.obj = job;
job->tmcf = tmcf;
work = nxt_array_add(array);
if (nxt_slow_path(work == NULL)) {
return NXT_ERROR;
}
work->next = NULL;
work->handler = handler;
work->task = &recf->engine->task;
work->obj = back;
joint = nxt_mp_alloc(mp, sizeof(nxt_socket_conf_joint_t)); joint = nxt_mp_alloc(mp, sizeof(nxt_socket_conf_joint_t));
if (nxt_slow_path(joint == NULL)) { if (nxt_slow_path(joint == NULL)) {
return NXT_ERROR; return NXT_ERROR;
} }
work->data = joint; job->work.data = joint;
joint->count = 1; joint->count = 1;
joint->socket_conf = nxt_queue_link_data(qlk, nxt_socket_conf_t, link); joint->socket_conf = nxt_queue_link_data(qlk, nxt_socket_conf_t, link);
@@ -1126,34 +1122,25 @@ static nxt_int_t
nxt_router_engine_joints_delete(nxt_router_temp_conf_t *tmcf, nxt_router_engine_joints_delete(nxt_router_temp_conf_t *tmcf,
nxt_router_engine_conf_t *recf, nxt_queue_t *sockets) nxt_router_engine_conf_t *recf, nxt_queue_t *sockets)
{ {
nxt_work_t *work, *back; nxt_joint_job_t *job;
nxt_queue_link_t *qlk; nxt_queue_link_t *qlk;
for (qlk = nxt_queue_first(sockets); for (qlk = nxt_queue_first(sockets);
qlk != nxt_queue_tail(sockets); qlk != nxt_queue_tail(sockets);
qlk = nxt_queue_next(qlk)) qlk = nxt_queue_next(qlk))
{ {
back = nxt_mp_get(tmcf->mem_pool, sizeof(nxt_work_t)); job = nxt_array_add(recf->deleting);
if (nxt_slow_path(back == NULL)) { if (nxt_slow_path(job == NULL)) {
return NXT_ERROR; return NXT_ERROR;
} }
back->next = NULL; job->task = tmcf->engine->task;
back->handler = nxt_router_conf_wait; job->work.next = NULL;
back->task = &tmcf->engine->task; job->work.handler = nxt_router_listen_socket_delete;
back->obj = tmcf; job->work.task = &job->task;
back->data = NULL; job->work.obj = job;
job->work.data = nxt_queue_link_data(qlk, nxt_socket_conf_t, link);
work = nxt_array_add(recf->deleting); job->tmcf = tmcf;
if (nxt_slow_path(work == NULL)) {
return NXT_ERROR;
}
work->next = NULL;
work->handler = nxt_router_listen_socket_delete;
work->task = &recf->engine->task;
work->obj = back;
work->data = nxt_queue_link_data(qlk, nxt_socket_conf_t, link);
} }
return NXT_OK; return NXT_OK;
@@ -1255,34 +1242,34 @@ nxt_router_engine_post(nxt_router_temp_conf_t *tmcf,
nxt_router_engine_conf_t *recf) nxt_router_engine_conf_t *recf)
{ {
nxt_uint_t n; nxt_uint_t n;
nxt_work_t *work; nxt_joint_job_t *job;
if (recf->creating != NULL) { if (recf->creating != NULL) {
work = recf->creating->elts; job = recf->creating->elts;
for (n = recf->creating->nelts; n != 0; n--) { for (n = recf->creating->nelts; n != 0; n--) {
nxt_event_engine_post(recf->engine, work); nxt_event_engine_post(recf->engine, &job->work);
work++; job++;
tmcf->count++; tmcf->count++;
} }
} }
if (recf->updating != NULL) { if (recf->updating != NULL) {
work = recf->updating->elts; job = recf->updating->elts;
for (n = recf->updating->nelts; n != 0; n--) { for (n = recf->updating->nelts; n != 0; n--) {
nxt_event_engine_post(recf->engine, work); nxt_event_engine_post(recf->engine, &job->work);
work++; job++;
tmcf->count++; tmcf->count++;
} }
} }
if (recf->deleting != NULL) { if (recf->deleting != NULL) {
work = recf->deleting->elts; job = recf->deleting->elts;
for (n = recf->deleting->nelts; n != 0; n--) { for (n = recf->deleting->nelts; n != 0; n--) {
nxt_event_engine_post(recf->engine, work); nxt_event_engine_post(recf->engine, &job->work);
work++; job++;
tmcf->count++; tmcf->count++;
} }
} }
@@ -1355,13 +1342,12 @@ nxt_router_thread_start(void *data)
static void static void
nxt_router_listen_socket_create(nxt_task_t *task, void *obj, void *data) nxt_router_listen_socket_create(nxt_task_t *task, void *obj, void *data)
{ {
nxt_work_t *work; nxt_joint_job_t *job;
nxt_listen_event_t *listen; nxt_listen_event_t *listen;
nxt_listen_socket_t *ls; nxt_listen_socket_t *ls;
nxt_router_temp_conf_t *tmcf;
nxt_socket_conf_joint_t *joint; nxt_socket_conf_joint_t *joint;
work = obj; job = obj;
joint = data; joint = data;
ls = &joint->socket_conf->listen; ls = &joint->socket_conf->listen;
@@ -1374,8 +1360,10 @@ nxt_router_listen_socket_create(nxt_task_t *task, void *obj, void *data)
listen->socket.data = joint; listen->socket.data = joint;
tmcf = work->obj; job->work.next = NULL;
nxt_event_engine_post(tmcf->engine, work); job->work.handler = nxt_router_conf_wait;
nxt_event_engine_post(job->tmcf->engine, &job->work);
} }
@@ -1407,13 +1395,12 @@ nxt_router_listen_event(nxt_queue_t *listen_connections,
static void static void
nxt_router_listen_socket_update(nxt_task_t *task, void *obj, void *data) nxt_router_listen_socket_update(nxt_task_t *task, void *obj, void *data)
{ {
nxt_work_t *work; nxt_joint_job_t *job;
nxt_event_engine_t *engine; nxt_event_engine_t *engine;
nxt_listen_event_t *listen; nxt_listen_event_t *listen;
nxt_router_temp_conf_t *tmcf;
nxt_socket_conf_joint_t *joint, *old; nxt_socket_conf_joint_t *joint, *old;
work = obj; job = obj;
joint = data; joint = data;
engine = task->thread->engine; engine = task->thread->engine;
@@ -1424,8 +1411,10 @@ nxt_router_listen_socket_update(nxt_task_t *task, void *obj, void *data)
old = listen->socket.data; old = listen->socket.data;
listen->socket.data = joint; listen->socket.data = joint;
tmcf = work->obj; job->work.next = NULL;
nxt_event_engine_post(tmcf->engine, work); job->work.handler = nxt_router_conf_wait;
nxt_event_engine_post(job->tmcf->engine, &job->work);
nxt_router_conf_release(task, old); nxt_router_conf_release(task, old);
} }
@@ -1434,13 +1423,12 @@ nxt_router_listen_socket_update(nxt_task_t *task, void *obj, void *data)
static void static void
nxt_router_listen_socket_delete(nxt_task_t *task, void *obj, void *data) nxt_router_listen_socket_delete(nxt_task_t *task, void *obj, void *data)
{ {
nxt_work_t *work; nxt_joint_job_t *job;
nxt_socket_conf_t *skcf; nxt_socket_conf_t *skcf;
nxt_listen_event_t *listen; nxt_listen_event_t *listen;
nxt_event_engine_t *engine; nxt_event_engine_t *engine;
nxt_router_temp_conf_t *tmcf;
work = obj; job = obj;
skcf = data; skcf = data;
engine = task->thread->engine; engine = task->thread->engine;
@@ -1454,8 +1442,10 @@ nxt_router_listen_socket_delete(nxt_task_t *task, void *obj, void *data)
nxt_timer_add(engine, &listen->timer, 0); nxt_timer_add(engine, &listen->timer, 0);
tmcf = work->obj; job->work.next = NULL;
nxt_event_engine_post(tmcf->engine, work); job->work.handler = nxt_router_conf_wait;
nxt_event_engine_post(job->tmcf->engine, &job->work);
} }

13
src/nxt_router.h
View File

@@ -35,9 +35,9 @@ typedef struct {
typedef struct { typedef struct {
nxt_event_engine_t *engine; nxt_event_engine_t *engine;
nxt_array_t *creating; /* of nxt_work_t */ nxt_array_t *creating; /* of nxt_joint_job_t */
nxt_array_t *updating; /* of nxt_work_t */ nxt_array_t *updating; /* of nxt_joint_job_t */
nxt_array_t *deleting; /* of nxt_work_t */ nxt_array_t *deleting; /* of nxt_joint_job_t */
} nxt_router_engine_conf_t; } nxt_router_engine_conf_t;
@@ -63,6 +63,13 @@ typedef struct {
} nxt_router_temp_conf_t; } nxt_router_temp_conf_t;
typedef struct {
nxt_task_t task;
nxt_work_t work;
nxt_router_temp_conf_t *tmcf;
} nxt_joint_job_t;
typedef struct nxt_app_module_s nxt_app_module_t; typedef struct nxt_app_module_s nxt_app_module_t;
typedef struct nxt_app_s nxt_app_t; typedef struct nxt_app_s nxt_app_t;