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

Using port rpc in router->master start worker request.

Browse Source
This commit is contained in:
Max Romanov
2017-08-02 13:20:53 +03:00
parent c93d2c7623
commit 82c0304ab8
6 changed files with 148 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
src/nxt_master_process.c
View File

@@ -135,9 +135,21 @@ static nxt_conf_map_t nxt_common_app_conf[] = {
static void
nxt_port_master_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
nxt_buf_t *b;
b = msg->buf;
nxt_debug(task, "master data: %*s", b->mem.free - b->mem.pos, b->mem.pos);
b->mem.pos = b->mem.free;
}
static void
nxt_port_master_start_worker_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
u_char *start;
size_t dump_size;
nxt_mp_t *mp;
nxt_int_t ret;
nxt_buf_t *b;
@@ -147,13 +159,9 @@ nxt_port_master_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
static nxt_str_t nobody = nxt_string("nobody");
b = msg->buf;
dump_size = b->mem.free - b->mem.pos;
if (dump_size > 300) {
dump_size = 300;
}
nxt_debug(task, "master data: %*s", dump_size, b->mem.pos);
nxt_debug(task, "master start worker: %*s", b->mem.free - b->mem.pos,
b->mem.pos);
mp = nxt_mp_create(1024, 128, 256, 32);
@@ -193,13 +201,14 @@ nxt_port_master_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
static nxt_port_handler_t nxt_master_process_port_handlers[] = {
NULL,
nxt_port_new_port_handler,
NULL,
nxt_port_mmap_handler,
NULL, /* NXT_PORT_MSG_QUIT */
NULL, /* NXT_PORT_MSG_NEW_PORT */
NULL, /* NXT_PORT_MSG_CHANGE_FILE */
NULL, /* NXT_PORT_MSG_MMAP */
nxt_port_master_data_handler,
NULL,
NULL, /* NXT_PORT_MSG_REMOVE_PID */
nxt_port_ready_handler,
nxt_port_master_start_worker_handler,
nxt_port_rpc_handler,
nxt_port_rpc_handler,
};
@@ -634,6 +643,7 @@ nxt_master_process_sigchld_handler(nxt_task_t *task, void *obj, void *data)
static void
nxt_master_cleanup_worker_process(nxt_task_t *task, nxt_pid_t pid)
{
nxt_buf_t *buf;
nxt_port_t *port;
nxt_runtime_t *rt;
nxt_process_t *process;
@@ -657,8 +667,11 @@ nxt_master_cleanup_worker_process(nxt_task_t *task, nxt_pid_t pid)
port = nxt_process_port_first(process);
buf = nxt_buf_mem_alloc(port->mem_pool, sizeof(pid), 0);
buf->mem.free = nxt_cpymem(buf->mem.free, &pid, sizeof(pid));
nxt_port_socket_write(task, port, NXT_PORT_MSG_REMOVE_PID,
-1, pid, 0, NULL);
-1, init->stream, 0, buf);
}
nxt_runtime_process_loop;
}

8
src/nxt_port.c
View File

@@ -413,14 +413,20 @@ nxt_port_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
void
nxt_port_remove_pid_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
nxt_buf_t *buf;
nxt_pid_t pid;
nxt_runtime_t *rt;
nxt_process_t *process;
nxt_debug(task, "port remove pid handler");
buf = msg->buf;
nxt_assert(nxt_buf_used_size(buf) == sizeof(pid));
nxt_memcpy(&pid, buf->mem.pos, sizeof(pid));
rt = task->thread->runtime;
pid = msg->port_msg.stream;
nxt_port_rpc_remove_peer(task, msg->port, pid);

3
src/nxt_port.h
View File

@@ -21,6 +21,7 @@ typedef enum {
_NXT_PORT_MSG_DATA,
_NXT_PORT_MSG_REMOVE_PID,
_NXT_PORT_MSG_READY,
_NXT_PORT_MSG_START_WORKER,
_NXT_PORT_MSG_RPC_READY,
_NXT_PORT_MSG_RPC_ERROR,
@@ -35,6 +36,8 @@ typedef enum {
NXT_PORT_MSG_DATA_LAST = _NXT_PORT_MSG_DATA | NXT_PORT_MSG_LAST,
NXT_PORT_MSG_REMOVE_PID = _NXT_PORT_MSG_REMOVE_PID | NXT_PORT_MSG_LAST,
NXT_PORT_MSG_READY = _NXT_PORT_MSG_READY | NXT_PORT_MSG_LAST,
NXT_PORT_MSG_START_WORKER = _NXT_PORT_MSG_START_WORKER |
NXT_PORT_MSG_LAST,
NXT_PORT_MSG_RPC_READY = _NXT_PORT_MSG_RPC_READY,
NXT_PORT_MSG_RPC_READY_LAST = _NXT_PORT_MSG_RPC_READY | NXT_PORT_MSG_LAST,
NXT_PORT_MSG_RPC_ERROR = _NXT_PORT_MSG_RPC_ERROR | NXT_PORT_MSG_LAST,

244
src/nxt_router.c
View File

@@ -24,10 +24,8 @@ typedef struct nxt_req_app_link_s nxt_req_app_link_t;
typedef struct nxt_start_worker_s nxt_start_worker_t;
struct nxt_start_worker_s {
uint32_t stream;
nxt_app_t *app;
nxt_req_app_link_t *ra;
nxt_mp_t *mem_pool;
nxt_work_t work;
};
@@ -120,11 +118,6 @@ static nxt_port_t * nxt_router_app_get_port(nxt_app_t *app, uint32_t req_id);
static void nxt_router_app_release_port(nxt_task_t *task, void *obj,
void *data);
static void nxt_router_sw_add(nxt_task_t *task, nxt_router_t *router,
nxt_start_worker_t *sw);
static nxt_start_worker_t *nxt_router_sw_find_remove(nxt_task_t *task,
nxt_router_t *router, uint32_t id);
static void nxt_router_conn_init(nxt_task_t *task, void *obj, void *data);
static void nxt_router_conn_http_header_parse(nxt_task_t *task, void *obj,
void *data);
@@ -174,28 +167,22 @@ nxt_router_start(nxt_task_t *task, void *data)
static nxt_start_worker_t *
nxt_router_sw_create(nxt_task_t *task, nxt_mp_t *mp, nxt_app_t *app,
nxt_req_app_link_t *ra)
nxt_router_sw_create(nxt_task_t *task, nxt_app_t *app, nxt_req_app_link_t *ra)
{
nxt_port_t *master_port;
nxt_runtime_t *rt;
nxt_start_worker_t *sw;
sw = nxt_mp_retain(mp, sizeof(nxt_start_worker_t));
sw = nxt_zalloc(sizeof(nxt_start_worker_t));
if (nxt_slow_path(sw == NULL)) {
return NULL;
}
nxt_memzero(sw, sizeof(nxt_start_worker_t));
sw->stream = nxt_random(&task->thread->random);
sw->mem_pool = mp;
sw->app = app;
sw->ra = ra;
nxt_debug(task, "sw #%uxD create, request #%uD, app '%V' %p", sw->stream,
nxt_debug(task, "sw %p create, request #%uxD, app '%V' %p", sw,
ra->req_id, &app->name, app);
rt = task->thread->runtime;
@@ -208,7 +195,7 @@ nxt_router_sw_create(nxt_task_t *task, nxt_mp_t *mp, nxt_app_t *app,
sw->work.next = NULL;
if (task->thread->engine != master_port->engine) {
nxt_debug(task, "post send sw %uxD to master engine %p", sw->stream,
nxt_debug(task, "sw %p post send to master engine %p", sw,
master_port->engine);
nxt_event_engine_post(master_port->engine, &sw->work);
@@ -221,30 +208,12 @@ nxt_router_sw_create(nxt_task_t *task, nxt_mp_t *mp, nxt_app_t *app,
}
static void
nxt_router_sw_release(nxt_task_t *task, void *obj, void *data)
nxt_inline void
nxt_router_sw_release(nxt_task_t *task, nxt_start_worker_t *sw)
{
nxt_event_engine_t *engine;
nxt_start_worker_t *sw;
nxt_debug(task, "sw %p release", sw);
sw = obj;
engine = data;
if (task->thread->engine != engine) {
sw->work.handler = nxt_router_sw_release;
sw->work.task = &engine->task;
sw->work.next = NULL;
nxt_debug(task, "sw #%uxD post release to %p", sw->stream, engine);
nxt_event_engine_post(engine, &sw->work);
return;
}
nxt_debug(task, "sw #%uxD release", sw->stream);
nxt_mp_release(sw->mem_pool, sw);
nxt_free(sw);
}
@@ -321,30 +290,17 @@ nxt_router_ra_release(nxt_task_t *task, void *obj, void *data)
void
nxt_router_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
nxt_start_worker_t *sw;
nxt_port_new_port_handler(task, msg);
if (msg->new_port == NULL || msg->new_port->type != NXT_PROCESS_WORKER) {
if (msg->port_msg.stream == 0) {
return;
}
sw = nxt_router_sw_find_remove(task, nxt_router, msg->port_msg.stream);
if (nxt_fast_path(sw != NULL)) {
msg->new_port->app = sw->app;
nxt_assert(sw->app->pending_workers != 0);
sw->app->pending_workers--;
sw->app->workers++;
nxt_debug(task, "sw #%uxD got port %p", sw->stream, msg->new_port);
nxt_router_app_release_port(task, msg->new_port, sw->app);
nxt_router_sw_release(task, sw, sw->work.data);
if (msg->new_port == NULL || msg->new_port->type != NXT_PROCESS_WORKER) {
msg->port_msg.type = _NXT_PORT_MSG_RPC_ERROR;
}
nxt_port_rpc_handler(task, msg);
}
@@ -392,6 +348,21 @@ nxt_router_conf_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
}
void
nxt_router_remove_pid_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
nxt_port_remove_pid_handler(task, msg);
if (msg->port_msg.stream == 0) {
return;
}
msg->port_msg.type = _NXT_PORT_MSG_RPC_ERROR;
nxt_port_rpc_handler(task, msg);
}
static nxt_router_temp_conf_t *
nxt_router_temp_conf(nxt_task_t *task)
{
@@ -1401,13 +1372,15 @@ static void
nxt_router_app_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
static nxt_port_handler_t nxt_router_app_port_handlers[] = {
NULL,
nxt_port_new_port_handler,
nxt_port_change_log_file_handler,
nxt_port_mmap_handler,
NULL, /* NXT_PORT_MSG_QUIT */
NULL, /* NXT_PORT_MSG_NEW_PORT */
NULL, /* NXT_PORT_MSG_CHANGE_FILE */
/* TODO: remove mmap_handler from app ports */
nxt_port_mmap_handler, /* NXT_PORT_MSG_MMAP */
nxt_router_app_data_handler,
nxt_port_remove_pid_handler,
NULL,
NULL, /* NXT_PORT_MSG_REMOVE_PID */
NULL, /* NXT_PORT_MSG_READY */
NULL, /* NXT_PORT_MSG_START_WORKER */
nxt_port_rpc_handler,
nxt_port_rpc_handler,
};
@@ -1937,13 +1910,55 @@ nxt_router_gen_error(nxt_task_t *task, nxt_conn_t *c, int code,
}
static void
nxt_router_sw_ready(nxt_task_t *task, nxt_port_recv_msg_t *msg, void *data)
{
nxt_start_worker_t *sw;
sw = data;
nxt_assert(sw != NULL);
nxt_assert(sw->app->pending_workers != 0);
msg->new_port->app = sw->app;
sw->app->pending_workers--;
sw->app->workers++;
nxt_debug(task, "sw %p got port %p", sw, msg->new_port);
nxt_router_app_release_port(task, msg->new_port, sw->app);
nxt_router_sw_release(task, sw);
}
static void
nxt_router_sw_error(nxt_task_t *task, nxt_port_recv_msg_t *msg, void *data)
{
nxt_start_worker_t *sw;
sw = data;
nxt_assert(sw != NULL);
nxt_assert(sw->app->pending_workers != 0);
sw->app->pending_workers--;
nxt_debug(task, "sw %p error, failed to start app '%V'", sw, &sw->app->name);
nxt_router_sw_release(task, sw);
}
static void
nxt_router_send_sw_request(nxt_task_t *task, void *obj, void *data)
{
size_t size;
uint32_t stream;
nxt_buf_t *b;
nxt_app_t *app;
nxt_port_t *port;
nxt_port_t *master_port, *router_port;
nxt_runtime_t *rt;
nxt_start_worker_t *sw;
@@ -1955,41 +1970,43 @@ nxt_router_send_sw_request(nxt_task_t *task, void *obj, void *data)
if (app->workers + app->pending_workers >= app->max_workers) {
nxt_debug(task, "app '%V' %p %uD/%uD running/penging workers, "
"post sw #%uxD release to %p", &app->name, app,
app->workers, app->pending_workers, sw->stream,
sw->work.data);
"sw %p release", &app->name, app,
app->workers, app->pending_workers, sw);
nxt_router_sw_release(task, sw, sw->work.data);
nxt_router_sw_release(task, sw);
return;
}
app->pending_workers++;
nxt_debug(task, "sw #%uxD send", sw->stream);
nxt_router_sw_add(task, nxt_router, sw);
nxt_debug(task, "sw %p send", sw);
rt = task->thread->runtime;
port = rt->port_by_type[NXT_PROCESS_MASTER];
master_port = rt->port_by_type[NXT_PROCESS_MASTER];
router_port = rt->port_by_type[NXT_PROCESS_ROUTER];
size = app->name.length + 1 + app->conf.length;
b = nxt_buf_mem_alloc(port->mem_pool, size, 0);
b = nxt_buf_mem_alloc(master_port->mem_pool, size, 0);
nxt_buf_cpystr(b, &app->name);
*b->mem.free++ = '\0';
nxt_buf_cpystr(b, &app->conf);
nxt_port_socket_write(task, port, NXT_PORT_MSG_DATA_LAST, -1,
sw->stream, 0, b);
stream = nxt_port_rpc_register_handler(task, router_port,
nxt_router_sw_ready,
nxt_router_sw_error,
master_port->pid, sw);
nxt_port_socket_write(task, master_port, NXT_PORT_MSG_START_WORKER, -1,
stream, router_port->id, b);
}
static nxt_bool_t
nxt_router_app_free(nxt_task_t *task, nxt_app_t *app)
{
nxt_port_t *master_port;
nxt_runtime_t *rt;
nxt_queue_link_t *lnk;
nxt_req_app_link_t *ra;
@@ -2015,10 +2032,7 @@ nxt_router_app_free(nxt_task_t *task, nxt_app_t *app)
ra = nxt_queue_link_data(lnk, nxt_req_app_link_t, link);
rt = task->thread->runtime;
master_port = rt->port_by_type[NXT_PROCESS_MASTER];
nxt_router_sw_create(task, master_port->mem_pool, app, ra);
nxt_router_sw_create(task, app, ra);
}
return 0;
@@ -2217,7 +2231,7 @@ nxt_router_app_port(nxt_task_t *task, nxt_req_app_link_t *ra)
return NXT_OK;
}
sw = nxt_router_sw_create(task, c->mem_pool, app, ra);
sw = nxt_router_sw_create(task, app, ra);
if (nxt_slow_path(sw == NULL)) {
nxt_router_gen_error(task, c, 500,
@@ -2627,71 +2641,3 @@ nxt_router_conn_timeout_value(nxt_conn_t *c, uintptr_t data)
return nxt_value_at(nxt_msec_t, joint->socket_conf, data);
}
static nxt_int_t
nxt_sw_test(nxt_lvlhsh_query_t *lhq, void *data)
{
return NXT_OK;
}
static const nxt_lvlhsh_proto_t lvlhsh_sw_proto nxt_aligned(64) = {
NXT_LVLHSH_DEFAULT,
nxt_sw_test,
nxt_lvlhsh_alloc,
nxt_lvlhsh_free,
};
static void
nxt_router_sw_add(nxt_task_t *task, nxt_router_t *router,
nxt_start_worker_t *sw)
{
nxt_lvlhsh_query_t lhq;
lhq.key_hash = nxt_murmur_hash2(&sw->stream, sizeof(sw->stream));
lhq.key.length = sizeof(sw->stream);
lhq.key.start = (u_char *) &sw->stream;
lhq.proto = &lvlhsh_sw_proto;
lhq.replace = 0;
lhq.value = sw;
lhq.pool = task->thread->runtime->mem_pool;
switch (nxt_lvlhsh_insert(&router->start_workers, &lhq)) {
case NXT_OK:
break;
default:
nxt_log_error(NXT_LOG_WARN, task->log, "stream %08uxD sw add failed",
sw->stream);
break;
}
}
static nxt_start_worker_t *
nxt_router_sw_find_remove(nxt_task_t *task, nxt_router_t *router, uint32_t id)
{
nxt_lvlhsh_query_t lhq;
lhq.key_hash = nxt_murmur_hash2(&id, sizeof(id));
lhq.key.length = sizeof(id);
lhq.key.start = (u_char *) &id;
lhq.proto = &lvlhsh_sw_proto;
lhq.pool = task->thread->runtime->mem_pool;
switch (nxt_lvlhsh_delete(&router->start_workers, &lhq)) {
case NXT_OK:
return lhq.value;
default:
nxt_log_error(NXT_LOG_WARN, task->log, "stream %08uxD sw remove failed",
id);
break;
}
return NULL;
}

3
src/nxt_router.h
View File

@@ -20,8 +20,6 @@ typedef struct {
nxt_queue_t sockets; /* of nxt_socket_conf_t */
nxt_queue_t apps; /* of nxt_app_t */
nxt_lvlhsh_t start_workers; /* stream to nxt_start_worker_t */
} nxt_router_t;
@@ -124,6 +122,7 @@ typedef struct {
void nxt_router_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
void nxt_router_conf_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
void nxt_router_remove_pid_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
nxt_bool_t nxt_router_app_remove_port(nxt_port_t *port);

14
src/nxt_worker_process.c
View File

@@ -29,7 +29,8 @@ nxt_port_handler_t nxt_controller_process_port_handlers[] = {
nxt_port_mmap_handler,
nxt_port_controller_data_handler,
nxt_port_remove_pid_handler,
NULL,
NULL, /* NXT_PORT_MSG_READY */
NULL, /* NXT_PORT_MSG_START_WORKER */
nxt_port_rpc_handler,
nxt_port_rpc_handler,
};
@@ -42,7 +43,8 @@ nxt_port_handler_t nxt_worker_process_port_handlers[] = {
nxt_port_mmap_handler,
nxt_port_data_handler,
nxt_port_remove_pid_handler,
NULL,
NULL, /* NXT_PORT_MSG_READY */
NULL, /* NXT_PORT_MSG_START_WORKER */
nxt_port_rpc_handler,
nxt_port_rpc_handler,
};
@@ -55,7 +57,8 @@ nxt_port_handler_t nxt_app_process_port_handlers[] = {
nxt_port_mmap_handler,
nxt_port_app_data_handler,
nxt_port_remove_pid_handler,
NULL,
NULL, /* NXT_PORT_MSG_READY */
NULL, /* NXT_PORT_MSG_START_WORKER */
nxt_port_rpc_handler,
nxt_port_rpc_handler,
};
@@ -67,8 +70,9 @@ nxt_port_handler_t nxt_router_process_port_handlers[] = {
nxt_port_change_log_file_handler,
nxt_port_mmap_handler,
nxt_router_conf_data_handler,
nxt_port_remove_pid_handler,
NULL,
nxt_router_remove_pid_handler,
NULL, /* NXT_PORT_MSG_READY */
NULL, /* NXT_PORT_MSG_START_WORKER */
nxt_port_rpc_handler,
nxt_port_rpc_handler,
};