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

Introducing use counters for port and app. Thread safe port write.

Browse Source 
Use counter helps to simplify logic around port and application free.

Port 'post' function introduced to simplify post execution of particular
function to original port engine's thread.

Write message queue is protected by mutex which makes port write operation
thread safe.
This commit is contained in:
Max Romanov
2017-10-04 14:58:47 +03:00
parent 414d508e04
commit 6a64533fa3
9 changed files with 691 additions and 416 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
src/nxt_main_process.c
View File

@@ -260,14 +260,17 @@ nxt_main_process_port_create(nxt_task_t *task, nxt_runtime_t *rt)
return NXT_ERROR;
}
nxt_process_port_add(task, process, port);
ret = nxt_port_socket_init(task, port, 0);
if (nxt_slow_path(ret != NXT_OK)) {
nxt_port_use(task, port, -1);
return ret;
}
nxt_process_port_add(task, process, port);
nxt_runtime_port_add(task, port);
nxt_runtime_port_add(rt, port);
nxt_port_use(task, port, -1);
/*
* A main process port. A write port is not closed
@@ -508,7 +511,7 @@ nxt_main_create_worker_process(nxt_task_t *task, nxt_runtime_t *rt,
port = nxt_port_new(task, 0, 0, init->type);
if (nxt_slow_path(port == NULL)) {
nxt_runtime_process_remove(rt, process);
nxt_runtime_process_remove(task, process);
return NXT_ERROR;
}
@@ -516,12 +519,14 @@ nxt_main_create_worker_process(nxt_task_t *task, nxt_runtime_t *rt,
ret = nxt_port_socket_init(task, port, 0);
if (nxt_slow_path(ret != NXT_OK)) {
nxt_mp_release(port->mem_pool, port);
nxt_port_use(task, port, -1);
return ret;
}
pid = nxt_process_create(task, process);
nxt_port_use(task, port, -1);
switch (pid) {
case -1:
@@ -755,7 +760,7 @@ nxt_main_cleanup_worker_process(nxt_task_t *task, nxt_pid_t pid)
if (process) {
init = process->init;
nxt_runtime_process_remove(rt, process);
nxt_runtime_process_remove(task, process);
if (!nxt_exiting) {
nxt_runtime_process_each(rt, process) {

127
src/nxt_port.c
View File

@@ -27,13 +27,15 @@ nxt_port_mp_cleanup(nxt_task_t *task, void *obj, void *data)
nxt_assert(port->pair[0] == -1);
nxt_assert(port->pair[1] == -1);
nxt_assert(port->app_stream == 0);
nxt_assert(port->use_count == 0);
nxt_assert(port->app_link.next == NULL);
nxt_assert(nxt_queue_is_empty(&port->messages));
nxt_assert(nxt_lvlhsh_is_empty(&port->rpc_streams));
nxt_assert(nxt_lvlhsh_is_empty(&port->rpc_peers));
nxt_thread_mutex_destroy(&port->write_mutex);
nxt_mp_free(mp, port);
}
@@ -58,10 +60,12 @@ nxt_port_new(nxt_task_t *task, nxt_port_id_t id, nxt_pid_t pid,
port->pid = pid;
port->type = type;
port->mem_pool = mp;
port->use_count = 1;
nxt_mp_cleanup(mp, nxt_port_mp_cleanup, task, port, mp);
nxt_queue_init(&port->messages);
nxt_thread_mutex_create(&port->write_mutex);
} else {
nxt_mp_destroy(mp);
@@ -73,10 +77,10 @@ nxt_port_new(nxt_task_t *task, nxt_port_id_t id, nxt_pid_t pid,
}
nxt_bool_t
nxt_port_release(nxt_port_t *port)
void
nxt_port_close(nxt_task_t *task, nxt_port_t *port)
{
nxt_thread_log_debug("port %p %d:%d release, type %d", port, port->pid,
nxt_debug(task, "port %p %d:%d close, type %d", port, port->pid,
port->id, port->type);
if (port->pair[0] != -1) {
@@ -87,21 +91,31 @@ nxt_port_release(nxt_port_t *port)
if (port->pair[1] != -1) {
nxt_fd_close(port->pair[1]);
port->pair[1] = -1;
}
if (port->type == NXT_PROCESS_WORKER) {
if (nxt_router_app_remove_port(port) == 0) {
return 0;
if (port->app != NULL) {
nxt_router_app_port_close(task, port);
}
}
}
static void
nxt_port_release(nxt_task_t *task, nxt_port_t *port)
{
nxt_debug(task, "port %p %d:%d release, type %d", port, port->pid,
port->id, port->type);
if (port->app != NULL) {
nxt_router_app_use(task, port->app, -1);
port->app = NULL;
}
if (port->link.next != NULL) {
nxt_process_port_remove(port);
}
nxt_mp_release(port->mem_pool, NULL);
return 1;
}
@@ -263,7 +277,9 @@ nxt_port_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
port->socket.task = task;
nxt_runtime_port_add(rt, port);
nxt_runtime_port_add(task, port);
nxt_port_use(task, port, -1);
nxt_port_write_enable(task, port);
@@ -434,7 +450,7 @@ nxt_port_remove_pid_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
process = nxt_runtime_process_find(rt, pid);
if (process) {
nxt_runtime_process_remove(rt, process);
nxt_runtime_process_remove(task, process);
}
}
@@ -444,3 +460,90 @@ nxt_port_empty_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
nxt_debug(task, "port empty handler");
}
typedef struct {
nxt_work_t work;
nxt_port_t *port;
nxt_port_post_handler_t handler;
} nxt_port_work_t;
static void
nxt_port_post_handler(nxt_task_t *task, void *obj, void *data)
{
nxt_port_t *port;
nxt_port_work_t *pw;
nxt_port_post_handler_t handler;
pw = obj;
port = pw->port;
handler = pw->handler;
nxt_free(pw);
handler(task, port, data);
nxt_port_use(task, port, -1);
}
nxt_int_t
nxt_port_post(nxt_task_t *task, nxt_port_t *port,
nxt_port_post_handler_t handler, void *data)
{
nxt_port_work_t *pw;
if (task->thread->engine == port->engine) {
handler(task, port, data);
return NXT_OK;
}
pw = nxt_zalloc(sizeof(nxt_port_work_t));
if (nxt_slow_path(pw == NULL)) {
return NXT_ERROR;
}
nxt_atomic_fetch_add(&port->use_count, 1);
pw->work.handler = nxt_port_post_handler;
pw->work.task = &port->engine->task;
pw->work.obj = pw;
pw->work.data = data;
pw->port = port;
pw->handler = handler;
nxt_event_engine_post(port->engine, &pw->work);
return NXT_OK;
}
static void
nxt_port_release_handler(nxt_task_t *task, nxt_port_t *port, void *data)
{
/* no op */
}
void
nxt_port_use(nxt_task_t *task, nxt_port_t *port, int i)
{
int c;
c = nxt_atomic_fetch_add(&port->use_count, i);
if (i < 0 && c == -i) {
if (task->thread->engine == port->engine) {
nxt_port_release(task, port);
return;
}
nxt_port_post(task, port, nxt_port_release_handler, NULL);
}
}

17
src/nxt_port.h
View File

@@ -115,7 +115,6 @@ typedef struct {
size_t share;
nxt_fd_t fd;
nxt_bool_t close_fd;
nxt_bool_t opened;
nxt_port_msg_t port_msg;
nxt_work_t work;
@@ -145,12 +144,15 @@ struct nxt_port_s {
nxt_app_t *app;
nxt_queue_t messages; /* of nxt_port_send_msg_t */
nxt_thread_mutex_t write_mutex;
/* Maximum size of message part. */
uint32_t max_size;
/* Maximum interleave of message parts. */
uint32_t max_share;
uint32_t app_stream;
uint32_t app_requests;
uint32_t app_responses;
nxt_port_handler_t handler;
nxt_port_handler_t *data;
@@ -167,8 +169,9 @@ struct nxt_port_s {
nxt_lvlhsh_t rpc_streams; /* stream to nxt_port_rpc_reg_t */
nxt_lvlhsh_t rpc_peers; /* peer to queue of nxt_port_rpc_reg_t */
nxt_atomic_t use_count;
nxt_process_type_t type;
nxt_work_t work;
struct iovec *iov;
void *mmsg_buf;
@@ -194,9 +197,11 @@ typedef union {
} nxt_port_data_t;
typedef void (*nxt_port_post_handler_t)(nxt_task_t *task, nxt_port_t *port,
void *data);
nxt_port_t *nxt_port_new(nxt_task_t *task, nxt_port_id_t id, nxt_pid_t pid,
nxt_process_type_t type);
nxt_bool_t nxt_port_release(nxt_port_t *port);
nxt_port_id_t nxt_port_get_next_id(void);
void nxt_port_reset_next_id(void);
@@ -204,6 +209,7 @@ void nxt_port_reset_next_id(void);
nxt_int_t nxt_port_socket_init(nxt_task_t *task, nxt_port_t *port,
size_t max_size);
void nxt_port_destroy(nxt_port_t *port);
void nxt_port_close(nxt_task_t *task, nxt_port_t *port);
void nxt_port_write_enable(nxt_task_t *task, nxt_port_t *port);
void nxt_port_write_close(nxt_port_t *port);
void nxt_port_read_enable(nxt_task_t *task, nxt_port_t *port);
@@ -231,5 +237,8 @@ void 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);
void nxt_port_empty_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
nxt_int_t nxt_port_post(nxt_task_t *task, nxt_port_t *port,
nxt_port_post_handler_t handler, void *data);
void nxt_port_use(nxt_task_t *task, nxt_port_t *port, int i);
#endif /* _NXT_PORT_H_INCLUDED_ */

102
src/nxt_port_socket.c
View File

@@ -169,31 +169,6 @@ nxt_port_socket_write(nxt_task_t *task, nxt_port_t *port, nxt_uint_t type,
{
nxt_port_send_msg_t *msg;
nxt_queue_each(msg, &port->messages, nxt_port_send_msg_t, link) {
if ((type & NXT_PORT_MSG_SYNC) != 0) {
msg->opened = 0;
continue;
}
if (msg->port_msg.stream == stream
&& msg->port_msg.reply_port == reply_port
&& msg->port_msg.last == 0
&& msg->opened) {
/*
* An fd is ignored since a file descriptor
* must be sent only in the first message of a stream.
*/
nxt_buf_chain_add(&msg->buf, b);
msg->port_msg.last |= (type & NXT_PORT_MSG_LAST) != 0;
return NXT_OK;
}
} nxt_queue_loop;
msg = nxt_mp_retain(task->thread->engine->mem_pool,
sizeof(nxt_port_send_msg_t));
if (nxt_slow_path(msg == NULL)) {
@@ -207,7 +182,6 @@ nxt_port_socket_write(nxt_task_t *task, nxt_port_t *port, nxt_uint_t type,
msg->fd = fd;
msg->close_fd = (type & NXT_PORT_MSG_CLOSE_FD) != 0;
msg->share = 0;
msg->opened = 1;
msg->work.next = NULL;
msg->work.handler = nxt_port_release_send_msg;
@@ -225,8 +199,14 @@ nxt_port_socket_write(nxt_task_t *task, nxt_port_t *port, nxt_uint_t type,
msg->port_msg.last = (type & NXT_PORT_MSG_LAST) != 0;
msg->port_msg.mmap = 0;
nxt_thread_mutex_lock(&port->write_mutex);
nxt_queue_insert_tail(&port->messages, &msg->link);
nxt_thread_mutex_unlock(&port->write_mutex);
nxt_port_use(task, port, 1);
if (port->socket.write_ready) {
nxt_port_write_handler(task, &port->socket, NULL);
}
@@ -235,11 +215,27 @@ nxt_port_socket_write(nxt_task_t *task, nxt_port_t *port, nxt_uint_t type,
}
static void
nxt_port_fd_block_write(nxt_task_t *task, nxt_port_t *port, void *data)
{
nxt_fd_event_block_write(task->thread->engine, &port->socket);
}
static void
nxt_port_fd_enable_write(nxt_task_t *task, nxt_port_t *port, void *data)
{
nxt_fd_event_enable_write(task->thread->engine, &port->socket);
}
static void
nxt_port_write_handler(nxt_task_t *task, void *obj, void *data)
{
int use_delta;
size_t plain_size;
ssize_t n;
nxt_bool_t block_write, enable_write;
nxt_port_t *port;
struct iovec *iov;
nxt_work_queue_t *wq;
@@ -250,14 +246,20 @@ nxt_port_write_handler(nxt_task_t *task, void *obj, void *data)
port = nxt_container_of(obj, nxt_port_t, socket);
block_write = 0;
enable_write = 0;
use_delta = 0;
nxt_thread_mutex_lock(&port->write_mutex);
iov = port->iov;
do {
link = nxt_queue_first(&port->messages);
if (link == nxt_queue_tail(&port->messages)) {
nxt_fd_event_block_write(task->thread->engine, &port->socket);
return;
block_write = 1;
goto unlock_mutex;
}
msg = nxt_queue_link_data(link, nxt_port_send_msg_t, link);
@@ -334,6 +336,7 @@ nxt_port_write_handler(nxt_task_t *task, void *obj, void *data)
} else {
nxt_queue_remove(link);
use_delta--;
nxt_work_queue_add(wq, nxt_port_release_send_msg, task, msg,
msg->engine);
}
@@ -347,16 +350,33 @@ nxt_port_write_handler(nxt_task_t *task, void *obj, void *data)
} while (port->socket.write_ready);
if (nxt_fd_event_is_disabled(port->socket.write)) {
/* TODO task->thread->engine or port->engine ? */
nxt_fd_event_enable_write(task->thread->engine, &port->socket);
enable_write = 1;
}
return;
goto unlock_mutex;
fail:
use_delta++;
nxt_work_queue_add(&task->thread->engine->fast_work_queue,
nxt_port_error_handler, task, &port->socket, NULL);
nxt_port_error_handler, task, &port->socket,
&port->socket);
unlock_mutex:
nxt_thread_mutex_unlock(&port->write_mutex);
if (block_write && nxt_fd_event_is_active(port->socket.write)) {
nxt_port_post(task, port, nxt_port_fd_block_write, NULL);
}
if (enable_write) {
nxt_port_post(task, port, nxt_port_fd_enable_write, NULL);
}
if (use_delta != 0) {
nxt_port_use(task, port, use_delta);
}
}
@@ -541,6 +561,7 @@ nxt_port_buf_free(nxt_port_t *port, nxt_buf_t *b)
static void
nxt_port_error_handler(nxt_task_t *task, void *obj, void *data)
{
int use_delta;
nxt_buf_t *b;
nxt_port_t *port;
nxt_work_queue_t *wq;
@@ -551,10 +572,18 @@ nxt_port_error_handler(nxt_task_t *task, void *obj, void *data)
port = nxt_container_of(obj, nxt_port_t, socket);
nxt_queue_each(msg, &port->messages, nxt_port_send_msg_t, link) {
use_delta = 0;
if (obj == data) {
use_delta--;
}
wq = &task->thread->engine->fast_work_queue;
nxt_thread_mutex_lock(&port->write_mutex);
nxt_queue_each(msg, &port->messages, nxt_port_send_msg_t, link) {
for(b = msg->buf; b != NULL; b = b->next) {
if (nxt_buf_is_sync(b)) {
continue;
@@ -564,8 +593,15 @@ nxt_port_error_handler(nxt_task_t *task, void *obj, void *data)
}
nxt_queue_remove(&msg->link);
use_delta--;
nxt_work_queue_add(wq, nxt_port_release_send_msg, task, msg,
msg->engine);
} nxt_queue_loop;
nxt_thread_mutex_unlock(&port->write_mutex);
if (use_delta != 0) {
nxt_port_use(task, port, use_delta);
}
}

12
src/nxt_process.c
View File

@@ -58,7 +58,7 @@ nxt_process_create(nxt_task_t *task, nxt_process_t *process)
if (!p->ready) {
nxt_debug(task, "remove not ready process %PI", p->pid);
nxt_runtime_process_remove(rt, p);
nxt_runtime_process_remove(task, p);
} else {
nxt_port_mmaps_destroy(p->incoming, 0);
@@ -67,7 +67,7 @@ nxt_process_create(nxt_task_t *task, nxt_process_t *process)
} nxt_runtime_process_loop;
nxt_runtime_process_add(rt, process);
nxt_runtime_process_add(task, process);
nxt_process_start(task, process);
@@ -81,7 +81,7 @@ nxt_process_create(nxt_task_t *task, nxt_process_t *process)
process->pid = pid;
nxt_runtime_process_add(rt, process);
nxt_runtime_process_add(task, process);
break;
}
@@ -589,16 +589,14 @@ nxt_user_cred_set(nxt_task_t *task, nxt_user_cred_t *uc)
static void
nxt_process_port_mp_cleanup(nxt_task_t *task, void *obj, void *data)
{
nxt_runtime_t *rt;
nxt_process_t *process;
process = obj;
rt = data;
process->port_cleanups--;
if (process->port_cleanups == 0) {
nxt_runtime_process_remove(rt, process);
nxt_runtime_process_remove(task, process);
}
}
@@ -609,7 +607,7 @@ nxt_process_port_add(nxt_task_t *task, nxt_process_t *process, nxt_port_t *port)
nxt_queue_insert_tail(&process->ports, &port->link);
nxt_mp_cleanup(port->mem_pool, nxt_process_port_mp_cleanup, task, process,
task->thread->runtime);
NULL);
process->port_cleanups++;
}

766
src/nxt_router.c
View File

File diff suppressed because it is too large Load Diff

6
src/nxt_router.h
View File

@@ -87,6 +87,7 @@ struct nxt_app_s {
uint32_t pending_workers;
uint32_t workers;
uint32_t max_workers;
uint32_t max_pending_responses;
nxt_msec_t timeout;
@@ -97,6 +98,8 @@ struct nxt_app_s {
nxt_str_t conf;
nxt_app_prepare_msg_t prepare_msg;
nxt_atomic_t use_count;
};
@@ -141,6 +144,7 @@ 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);
void nxt_router_app_port_close(nxt_task_t *task, nxt_port_t *port);
void nxt_router_app_use(nxt_task_t *task, nxt_app_t *app, int i);
#endif /* _NXT_ROUTER_H_INCLUDED_ */

34
src/nxt_runtime.c
View File

@@ -553,7 +553,7 @@ nxt_runtime_exit(nxt_task_t *task, void *obj, void *data)
nxt_runtime_process_each(rt, process) {
nxt_runtime_process_remove(rt, process);
nxt_runtime_process_remove(task, process);
} nxt_runtime_process_loop;
@@ -1725,13 +1725,16 @@ nxt_runtime_process_get(nxt_runtime_t *rt, nxt_pid_t pid)
void
nxt_runtime_process_add(nxt_runtime_t *rt, nxt_process_t *process)
nxt_runtime_process_add(nxt_task_t *task, nxt_process_t *process)
{
nxt_port_t *port;
nxt_runtime_t *rt;
nxt_lvlhsh_query_t lhq;
nxt_assert(process->registered == 0);
rt = task->thread->runtime;
nxt_runtime_process_lhq_pid(&lhq, &process->pid);
lhq.replace = 0;
@@ -1753,7 +1756,7 @@ nxt_runtime_process_add(nxt_runtime_t *rt, nxt_process_t *process)
port->pid = process->pid;
nxt_runtime_port_add(rt, port);
nxt_runtime_port_add(task, port);
} nxt_process_port_loop;
@@ -1809,9 +1812,12 @@ nxt_runtime_process_remove_pid(nxt_runtime_t *rt, nxt_pid_t pid)
void
nxt_runtime_process_remove(nxt_runtime_t *rt, nxt_process_t *process)
nxt_runtime_process_remove(nxt_task_t *task, nxt_process_t *process)
{
nxt_port_t *port;
nxt_runtime_t *rt;
rt = task->thread->runtime;
if (process->port_cleanups == 0) {
if (process->registered == 1) {
@@ -1823,9 +1829,9 @@ nxt_runtime_process_remove(nxt_runtime_t *rt, nxt_process_t *process)
} else {
nxt_process_port_each(process, port) {
nxt_runtime_port_remove(rt, port);
nxt_port_close(task, port);
nxt_port_release(port);
nxt_runtime_port_remove(task, port);
} nxt_process_port_loop;
}
@@ -1851,22 +1857,34 @@ nxt_runtime_port_first(nxt_runtime_t *rt, nxt_lvlhsh_each_t *lhe)
void
nxt_runtime_port_add(nxt_runtime_t *rt, nxt_port_t *port)
nxt_runtime_port_add(nxt_task_t *task, nxt_port_t *port)
{
nxt_runtime_t *rt;
rt = task->thread->runtime;
nxt_port_hash_add(&rt->ports, port);
rt->port_by_type[port->type] = port;
nxt_port_use(task, port, 1);
}
void
nxt_runtime_port_remove(nxt_runtime_t *rt, nxt_port_t *port)
nxt_runtime_port_remove(nxt_task_t *task, nxt_port_t *port)
{
nxt_runtime_t *rt;
rt = task->thread->runtime;
nxt_port_hash_remove(&rt->ports, port);
if (rt->port_by_type[port->type] == port) {
rt->port_by_type[port->type] = NULL;
}
nxt_port_use(task, port, -1);
}

8
src/nxt_runtime.h
View File

@@ -102,11 +102,11 @@ nxt_process_t *nxt_runtime_process_new(nxt_runtime_t *rt);
nxt_process_t *nxt_runtime_process_get(nxt_runtime_t *rt, nxt_pid_t pid);
void nxt_runtime_process_add(nxt_runtime_t *rt, nxt_process_t *process);
void nxt_runtime_process_add(nxt_task_t *task, nxt_process_t *process);
nxt_process_t *nxt_runtime_process_find(nxt_runtime_t *rt, nxt_pid_t pid);
void nxt_runtime_process_remove(nxt_runtime_t *rt, nxt_process_t *process);
void nxt_runtime_process_remove(nxt_task_t *task, nxt_process_t *process);
nxt_process_t *nxt_runtime_process_first(nxt_runtime_t *rt,
nxt_lvlhsh_each_t *lhe);
@@ -115,9 +115,9 @@ nxt_process_t *nxt_runtime_process_first(nxt_runtime_t *rt,
nxt_lvlhsh_each(&rt->processes, lhe)
void nxt_runtime_port_add(nxt_runtime_t *rt, nxt_port_t *port);
void nxt_runtime_port_add(nxt_task_t *task, nxt_port_t *port);
void nxt_runtime_port_remove(nxt_runtime_t *rt, nxt_port_t *port);
void nxt_runtime_port_remove(nxt_task_t *task, nxt_port_t *port);
nxt_port_t *nxt_runtime_port_find(nxt_runtime_t *rt, nxt_pid_t pid,
nxt_port_id_t port_id);