summaryrefslogtreecommitdiffhomepage
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/nxt_main_process.c15
-rw-r--r--src/nxt_port.c129
-rw-r--r--src/nxt_port.h17
-rw-r--r--src/nxt_port_socket.c104
-rw-r--r--src/nxt_process.c12
-rw-r--r--src/nxt_router.c762
-rw-r--r--src/nxt_router.h6
-rw-r--r--src/nxt_runtime.c36
-rw-r--r--src/nxt_runtime.h8
9 files changed, 682 insertions, 407 deletions
diff --git a/src/nxt_main_process.c b/src/nxt_main_process.c
index 90c06a05..9c27a89b 100644
--- a/src/nxt_main_process.c
+++ b/src/nxt_main_process.c
@@ -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) {
diff --git a/src/nxt_port.c b/src/nxt_port.c
index d7f42012..3f7dc411 100644
--- a/src/nxt_port.c
+++ b/src/nxt_port.c
@@ -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,11 +77,11 @@ 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,
- port->id, port->type);
+ nxt_debug(task, "port %p %d:%d close, type %d", port, port->pid,
+ port->id, port->type);
if (port->pair[0] != -1) {
nxt_fd_close(port->pair[0]);
@@ -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);
+ }
+}
diff --git a/src/nxt_port.h b/src/nxt_port.h
index c5b2b40e..f6679bb2 100644
--- a/src/nxt_port.h
+++ b/src/nxt_port.h
@@ -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_ */
diff --git a/src/nxt_port_socket.c b/src/nxt_port_socket.c
index 75706459..d5ed493d 100644
--- a/src/nxt_port_socket.c
+++ b/src/nxt_port_socket.c
@@ -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);
}
@@ -236,10 +216,26 @@ 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,9 +572,17 @@ 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;
- 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)) {
@@ -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);
+ }
}
diff --git a/src/nxt_process.c b/src/nxt_process.c
index 95e701f8..d3ec36ed 100644
--- a/src/nxt_process.c
+++ b/src/nxt_process.c
@@ -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++;
}
diff --git a/src/nxt_router.c b/src/nxt_router.c
index df3caf82..5eb95b59 100644
--- a/src/nxt_router.c
+++ b/src/nxt_router.c
@@ -24,19 +24,12 @@ typedef struct {
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 {
- nxt_app_t *app;
- nxt_req_app_link_t *ra;
-
- nxt_work_t work;
-};
typedef struct {
uint32_t stream;
nxt_conn_t *conn;
+ nxt_app_t *app;
nxt_port_t *app_port;
nxt_req_app_link_t *ra;
@@ -72,6 +65,8 @@ typedef struct {
} nxt_remove_pid_msg_t;
+static nxt_int_t nxt_router_start_worker(nxt_task_t *task, nxt_app_t *app);
+
static void nxt_router_worker_remove_pid_handler(nxt_task_t *task, void *obj,
void *data);
static void nxt_router_worker_remove_pid_done(nxt_task_t *task, void *obj,
@@ -124,7 +119,7 @@ static nxt_int_t nxt_router_threads_create(nxt_task_t *task, nxt_runtime_t *rt,
nxt_router_temp_conf_t *tmcf);
static nxt_int_t nxt_router_thread_create(nxt_task_t *task, nxt_runtime_t *rt,
nxt_event_engine_t *engine);
-static void nxt_router_apps_sort(nxt_router_t *router,
+static void nxt_router_apps_sort(nxt_task_t *task, nxt_router_t *router,
nxt_router_temp_conf_t *tmcf);
static void nxt_router_engines_post(nxt_router_t *router,
@@ -150,12 +145,14 @@ static void nxt_router_thread_exit_handler(nxt_task_t *task, void *obj,
static void nxt_router_conf_release(nxt_task_t *task,
nxt_socket_conf_joint_t *joint);
-static void nxt_router_send_sw_request(nxt_task_t *task, void *obj,
- void *data);
-static nxt_bool_t nxt_router_app_free(nxt_task_t *task, nxt_app_t *app);
-static nxt_port_t * nxt_router_app_get_port(nxt_app_t *app, uint32_t stream);
-static void nxt_router_app_release_port(nxt_task_t *task, void *obj,
- void *data);
+static void nxt_router_app_port_ready(nxt_task_t *task,
+ nxt_port_recv_msg_t *msg, void *data);
+static void nxt_router_app_port_error(nxt_task_t *task,
+ nxt_port_recv_msg_t *msg, void *data);
+
+static nxt_port_t * nxt_router_app_get_idle_port(nxt_app_t *app);
+static void nxt_router_app_port_release(nxt_task_t *task, nxt_port_t *port,
+ uint32_t request_failed, uint32_t got_response);
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,
@@ -165,7 +162,7 @@ static void nxt_router_conn_http_body_read(nxt_task_t *task, void *obj,
static void nxt_router_process_http_request(nxt_task_t *task,
nxt_conn_t *c, nxt_app_parse_ctx_t *ap);
static void nxt_router_process_http_request_mp(nxt_task_t *task,
- nxt_req_app_link_t *ra, nxt_port_t *port);
+ nxt_req_app_link_t *ra);
static nxt_int_t nxt_python_prepare_msg(nxt_task_t *task, nxt_app_request_t *r,
nxt_app_wmsg_t *wmsg);
static nxt_int_t nxt_php_prepare_msg(nxt_task_t *task, nxt_app_request_t *r,
@@ -222,68 +219,91 @@ nxt_router_start(nxt_task_t *task, void *data)
}
-static nxt_start_worker_t *
-nxt_router_sw_create(nxt_task_t *task, nxt_app_t *app, nxt_req_app_link_t *ra)
+static void
+nxt_router_start_worker_handler(nxt_task_t *task, nxt_port_t *port, void *data)
{
- nxt_port_t *main_port;
- nxt_runtime_t *rt;
- nxt_start_worker_t *sw;
+ size_t size;
+ uint32_t stream;
+ nxt_app_t *app;
+ nxt_buf_t *b;
+ nxt_port_t *main_port;
+ nxt_runtime_t *rt;
- sw = nxt_zalloc(sizeof(nxt_start_worker_t));
+ app = data;
- if (nxt_slow_path(sw == NULL)) {
- return NULL;
- }
+ rt = task->thread->runtime;
+ main_port = rt->port_by_type[NXT_PROCESS_MAIN];
- sw->app = app;
- sw->ra = ra;
+ nxt_debug(task, "app '%V' %p start worker", &app->name, app);
- nxt_debug(task, "sw %p create, stream #%uD, app '%V' %p", sw,
- ra->stream, &app->name, app);
+ size = app->name.length + 1 + app->conf.length;
- rt = task->thread->runtime;
- main_port = rt->port_by_type[NXT_PROCESS_MAIN];
+ b = nxt_buf_mem_ts_alloc(task, task->thread->engine->mem_pool, size);
- sw->work.handler = nxt_router_send_sw_request;
- sw->work.task = &main_port->engine->task;
- sw->work.obj = sw;
- sw->work.data = task->thread->engine;
- sw->work.next = NULL;
+ if (nxt_slow_path(b == NULL)) {
+ goto failed;
+ }
+
+ nxt_buf_cpystr(b, &app->name);
+ *b->mem.free++ = '\0';
+ nxt_buf_cpystr(b, &app->conf);
- if (task->thread->engine != main_port->engine) {
- nxt_debug(task, "sw %p post send to main engine %p", sw,
- main_port->engine);
+ stream = nxt_port_rpc_register_handler(task, port,
+ nxt_router_app_port_ready,
+ nxt_router_app_port_error,
+ -1, app);
- nxt_event_engine_post(main_port->engine, &sw->work);
+ if (nxt_slow_path(stream == 0)) {
+ nxt_mp_release(b->data, b);
- } else {
- nxt_router_send_sw_request(task, sw, sw->work.data);
+ goto failed;
}
- return sw;
-}
+ nxt_port_socket_write(task, main_port, NXT_PORT_MSG_START_WORKER, -1,
+ stream, port->id, b);
+
+ return;
+failed:
-nxt_inline void
-nxt_router_sw_release(nxt_task_t *task, nxt_start_worker_t *sw)
-{
- nxt_debug(task, "sw %p release", sw);
+ nxt_thread_mutex_lock(&app->mutex);
+
+ app->pending_workers--;
- nxt_free(sw);
+ nxt_thread_mutex_unlock(&app->mutex);
+
+ nxt_router_app_use(task, app, -1);
}
-nxt_inline void
-nxt_router_rc_unlink(nxt_req_conn_link_t *rc)
+static nxt_int_t
+nxt_router_start_worker(nxt_task_t *task, nxt_app_t *app)
{
- nxt_queue_remove(&rc->link);
+ nxt_int_t res;
+ nxt_port_t *router_port;
+ nxt_runtime_t *rt;
- if (rc->ra != NULL) {
- rc->ra->rc = NULL;
- rc->ra = NULL;
+ rt = task->thread->runtime;
+ router_port = rt->port_by_type[NXT_PROCESS_ROUTER];
+
+ nxt_router_app_use(task, app, 1);
+
+ res = nxt_port_post(task, router_port, nxt_router_start_worker_handler,
+ app);
+
+ if (res == NXT_OK) {
+ return res;
}
- rc->conn = NULL;
+ nxt_thread_mutex_lock(&app->mutex);
+
+ app->pending_workers--;
+
+ nxt_thread_mutex_unlock(&app->mutex);
+
+ nxt_router_app_use(task, app, -1);
+
+ return NXT_ERROR;
}
@@ -327,36 +347,18 @@ nxt_router_ra_create(nxt_task_t *task, nxt_req_conn_link_t *rc)
static void
nxt_router_ra_release(nxt_task_t *task, void *obj, void *data)
{
- nxt_port_t *app_port;
- nxt_req_app_link_t *ra;
- nxt_event_engine_t *engine;
+ nxt_req_app_link_t *ra;
+ nxt_event_engine_t *engine;
+ nxt_req_conn_link_t *rc;
ra = obj;
engine = data;
- if (ra->app_port != NULL) {
-
- app_port = ra->app_port;
- ra->app_port = NULL;
-
- if (task->thread->engine != engine) {
- ra->app_pid = app_port->pid;
- }
-
- nxt_router_app_release_port(task, app_port, app_port->app);
-
-#if 0
- /* Uncomment to hold app port until complete response received. */
- if (ra->rc != NULL) {
- ra->rc->app_port = ra->app_port;
-
- } else {
- nxt_router_app_release_port(task, ra->app_port, ra->app_port->app);
+ if (task->thread->engine != engine) {
+ if (ra->app_port != NULL) {
+ ra->app_pid = ra->app_port->pid;
}
-#endif
- }
- if (task->thread->engine != engine) {
ra->work.handler = nxt_router_ra_release;
ra->work.task = &engine->task;
ra->work.next = NULL;
@@ -369,11 +371,27 @@ nxt_router_ra_release(nxt_task_t *task, void *obj, void *data)
return;
}
- if (ra->rc != NULL && ra->app_pid != -1) {
- nxt_port_rpc_ex_set_peer(task, engine->port, ra->rc, ra->app_pid);
+ nxt_debug(task, "ra stream #%uD release", ra->stream);
+
+ rc = ra->rc;
+
+ if (rc != NULL) {
+ if (ra->app_pid != -1) {
+ nxt_port_rpc_ex_set_peer(task, engine->port, rc, ra->app_pid);
+ }
+
+ rc->app_port = ra->app_port;
+
+ ra->app_port = NULL;
+ rc->ra = NULL;
+ ra->rc = NULL;
}
- nxt_debug(task, "ra stream #%uD release", ra->stream);
+ if (ra->app_port != NULL) {
+ nxt_router_app_port_release(task, ra->app_port, 0, 1);
+
+ ra->app_port = NULL;
+ }
nxt_mp_release(ra->mem_pool, ra);
}
@@ -382,9 +400,10 @@ nxt_router_ra_release(nxt_task_t *task, void *obj, void *data)
static void
nxt_router_ra_abort(nxt_task_t *task, void *obj, void *data)
{
- nxt_conn_t *c;
- nxt_req_app_link_t *ra;
- nxt_event_engine_t *engine;
+ nxt_conn_t *c;
+ nxt_req_app_link_t *ra;
+ nxt_req_conn_link_t *rc;
+ nxt_event_engine_t *engine;
ra = obj;
engine = data;
@@ -403,17 +422,75 @@ nxt_router_ra_abort(nxt_task_t *task, void *obj, void *data)
nxt_debug(task, "ra stream #%uD abort", ra->stream);
- if (ra->rc != NULL) {
- c = ra->rc->conn;
+ rc = ra->rc;
+
+ if (rc != NULL) {
+ c = rc->conn;
nxt_router_gen_error(task, c, 500,
"Failed to start application worker");
+
+ rc->ra = NULL;
+ ra->rc = NULL;
+ }
+
+ if (ra->app_port != NULL) {
+ nxt_router_app_port_release(task, ra->app_port, 0, 1);
+
+ ra->app_port = NULL;
}
nxt_mp_release(ra->mem_pool, ra);
}
+nxt_inline void
+nxt_router_rc_unlink(nxt_task_t *task, nxt_req_conn_link_t *rc)
+{
+ nxt_req_app_link_t *ra;
+
+ if (rc->app_port != NULL) {
+ nxt_router_app_port_release(task, rc->app_port, 0, 1);
+
+ rc->app_port = NULL;
+ }
+
+ ra = rc->ra;
+
+ if (ra != NULL) {
+ rc->ra = NULL;
+ ra->rc = NULL;
+
+ nxt_thread_mutex_lock(&rc->app->mutex);
+
+ if (ra->link.next != NULL) {
+ nxt_queue_remove(&ra->link);
+
+ ra->link.next = NULL;
+
+ } else {
+ ra = NULL;
+ }
+
+ nxt_thread_mutex_unlock(&rc->app->mutex);
+ }
+
+ if (ra != NULL) {
+ nxt_router_ra_release(task, ra, ra->work.data);
+ }
+
+ if (rc->app != NULL) {
+ nxt_router_app_use(task, rc->app, -1);
+
+ rc->app = NULL;
+ }
+
+ nxt_queue_remove(&rc->link);
+
+ rc->conn = NULL;
+}
+
+
void
nxt_router_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
@@ -526,12 +603,20 @@ nxt_router_remove_pid_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
static void
nxt_router_worker_remove_pid_handler(nxt_task_t *task, void *obj, void *data)
{
+ nxt_pid_t pid;
+ nxt_buf_t *buf;
nxt_event_engine_t *engine;
nxt_remove_pid_msg_t *rp;
rp = obj;
- nxt_port_remove_pid_handler(task, &rp->msg);
+ buf = rp->msg.buf;
+
+ nxt_assert(nxt_buf_used_size(buf) == sizeof(pid));
+
+ nxt_memcpy(&pid, buf->mem.pos, sizeof(pid));
+
+ nxt_port_rpc_remove_peer(task, rp->msg.port, pid);
engine = rp->work.data;
@@ -658,7 +743,7 @@ nxt_router_conf_apply(nxt_task_t *task, void *obj, void *data)
goto fail;
}
- nxt_router_apps_sort(router, tmcf);
+ nxt_router_apps_sort(task, router, tmcf);
nxt_router_engines_post(router, tmcf);
@@ -1005,9 +1090,12 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
app->max_workers = apcf.workers;
app->timeout = apcf.timeout;
app->live = 1;
+ app->max_pending_responses = 2;
app->prepare_msg = nxt_app_prepare_msg[type];
nxt_queue_insert_tail(&tmcf->apps, &app->link);
+
+ nxt_router_app_use(task, app, 1);
}
http = nxt_conf_get_path(conf, &http_path);
@@ -1695,42 +1783,36 @@ nxt_router_thread_create(nxt_task_t *task, nxt_runtime_t *rt,
static void
-nxt_router_apps_sort(nxt_router_t *router, nxt_router_temp_conf_t *tmcf)
+nxt_router_apps_sort(nxt_task_t *task, nxt_router_t *router,
+ nxt_router_temp_conf_t *tmcf)
{
- nxt_app_t *app;
- nxt_port_t *port;
+ nxt_app_t *app;
+ nxt_port_t *port;
nxt_queue_each(app, &router->apps, nxt_app_t, link) {
nxt_queue_remove(&app->link);
- nxt_thread_log_debug("about to remove app '%V' %p", &app->name, app);
+ nxt_debug(task, "about to free app '%V' %p", &app->name, app);
app->live = 0;
- if (nxt_router_app_free(NULL, app) != 0) {
- continue;
- }
-
- if (!nxt_queue_is_empty(&app->requests)) {
-
- nxt_thread_log_debug("app '%V' %p pending requests found",
- &app->name, app);
- continue;
- }
-
do {
- port = nxt_router_app_get_port(app, 0);
+ port = nxt_router_app_get_idle_port(app);
if (port == NULL) {
break;
}
- nxt_thread_log_debug("port %p send quit", port);
+ nxt_debug(task, "port %p send quit", port);
+
+ nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT, -1, 0, 0,
+ NULL);
- nxt_port_socket_write(&port->engine->task, port,
- NXT_PORT_MSG_QUIT, -1, 0, 0, NULL);
+ nxt_port_use(task, port, -1);
} while (1);
+ nxt_router_app_use(task, app, -1);
+
} nxt_queue_loop;
nxt_queue_add(&router->apps, &tmcf->previous);
@@ -1833,7 +1915,7 @@ nxt_router_thread_start(void *data)
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;
}
@@ -2124,8 +2206,9 @@ nxt_router_thread_exit_handler(nxt_task_t *task, void *obj, void *data)
// TODO notify all apps
+ port->engine = task->thread->engine;
nxt_mp_thread_adopt(port->mem_pool);
- nxt_port_release(port);
+ nxt_port_use(task, port, -1);
nxt_mp_thread_adopt(engine->mem_pool);
nxt_mp_destroy(engine->mem_pool);
@@ -2240,13 +2323,7 @@ nxt_router_response_ready_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
nxt_buf_chain_add(&b, last);
- if (rc->app_port != NULL) {
- nxt_router_app_release_port(task, rc->app_port, rc->app_port->app);
-
- rc->app_port = NULL;
- }
-
- nxt_router_rc_unlink(rc);
+ nxt_router_rc_unlink(task, rc);
}
if (b == NULL) {
@@ -2283,7 +2360,7 @@ nxt_router_response_error_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
nxt_router_gen_error(task, rc->conn, 500,
"Application terminated unexpectedly");
- nxt_router_rc_unlink(rc);
+ nxt_router_rc_unlink(task, rc);
}
@@ -2383,194 +2460,153 @@ 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_router_app_port_ready(nxt_task_t *task, nxt_port_recv_msg_t *msg,
+ void *data)
{
- nxt_start_worker_t *sw;
+ nxt_app_t *app;
+ nxt_port_t *port;
+
+ app = data;
+ port = msg->new_port;
- sw = data;
+ nxt_assert(app != NULL);
+ nxt_assert(port != NULL);
- nxt_assert(sw != NULL);
- nxt_assert(sw->app->pending_workers != 0);
+ port->app = app;
- msg->new_port->app = sw->app;
+ nxt_thread_mutex_lock(&app->mutex);
- sw->app->pending_workers--;
- sw->app->workers++;
+ nxt_assert(app->pending_workers != 0);
- nxt_debug(task, "sw %p got port %p", sw, msg->new_port);
+ app->pending_workers--;
+ app->workers++;
+
+ nxt_thread_mutex_unlock(&app->mutex);
- nxt_router_app_release_port(task, msg->new_port, sw->app);
+ nxt_debug(task, "app '%V' %p new port ready", &app->name, app);
- nxt_router_sw_release(task, sw);
+ nxt_router_app_port_release(task, port, 0, 0);
}
static void
-nxt_router_sw_error(nxt_task_t *task, nxt_port_recv_msg_t *msg, void *data)
+nxt_router_app_port_error(nxt_task_t *task, nxt_port_recv_msg_t *msg,
+ void *data)
{
nxt_app_t *app;
nxt_queue_link_t *lnk;
nxt_req_app_link_t *ra;
- nxt_start_worker_t *sw;
- sw = data;
+ app = data;
- nxt_assert(sw != NULL);
- nxt_assert(sw->app != NULL);
- nxt_assert(sw->app->pending_workers != 0);
+ nxt_assert(app != NULL);
- app = sw->app;
+ nxt_debug(task, "app '%V' %p start error", &app->name, app);
+
+ nxt_thread_mutex_lock(&app->mutex);
- sw->app->pending_workers--;
+ nxt_assert(app->pending_workers != 0);
- nxt_debug(task, "sw %p error, failed to start app '%V'",
- sw, &app->name);
+ app->pending_workers--;
if (!nxt_queue_is_empty(&app->requests)) {
lnk = nxt_queue_last(&app->requests);
nxt_queue_remove(lnk);
+ lnk->next = NULL;
ra = nxt_queue_link_data(lnk, nxt_req_app_link_t, link);
+ } else {
+ ra = NULL;
+ }
+
+ nxt_thread_mutex_unlock(&app->mutex);
+
+ if (ra != NULL) {
nxt_debug(task, "app '%V' %p abort next stream #%uD",
&app->name, app, ra->stream);
nxt_router_ra_abort(task, ra, ra->work.data);
}
- nxt_router_sw_release(task, sw);
+ nxt_router_app_use(task, app, -1);
}
-static void
-nxt_router_send_sw_request(nxt_task_t *task, void *obj, void *data)
+void
+nxt_router_app_use(nxt_task_t *task, nxt_app_t *app, int i)
{
- size_t size;
- uint32_t stream;
- nxt_buf_t *b;
- nxt_app_t *app;
- nxt_port_t *main_port, *router_port, *app_port;
- nxt_runtime_t *rt;
- nxt_start_worker_t *sw;
- nxt_req_app_link_t *ra;
-
- sw = obj;
- app = sw->app;
+ int c;
- if (nxt_queue_is_empty(&app->requests)) {
- ra = sw->ra;
- app_port = nxt_router_app_get_port(app, ra->stream);
+ c = nxt_atomic_fetch_add(&app->use_count, i);
- if (app_port != NULL) {
- nxt_debug(task, "app '%V' %p process stream #%uD",
- &app->name, app, ra->stream);
+ if (i < 0 && c == -i) {
- ra->app_port = app_port;
+ nxt_assert(app->live == 0);
+ nxt_assert(app->workers == 0);
+ nxt_assert(app->pending_workers == 0);
+ nxt_assert(nxt_queue_is_empty(&app->requests) != 0);
+ nxt_assert(nxt_queue_is_empty(&app->ports) != 0);
- nxt_router_process_http_request_mp(task, ra, app_port);
-
- nxt_router_ra_release(task, ra, ra->work.data);
- nxt_router_sw_release(task, sw);
-
- return;
- }
- }
-
- nxt_queue_insert_tail(&app->requests, &sw->ra->link);
-
- if (app->workers + app->pending_workers >= app->max_workers) {
- nxt_debug(task, "app '%V' %p %uD/%uD running/pending workers, "
- "max_workers (%uD) reached", &app->name, app,
- app->workers, app->pending_workers, app->max_workers);
-
- nxt_router_sw_release(task, sw);
-
- return;
+ nxt_thread_mutex_destroy(&app->mutex);
+ nxt_free(app);
}
-
- app->pending_workers++;
-
- nxt_debug(task, "sw %p send", sw);
-
- rt = task->thread->runtime;
- main_port = rt->port_by_type[NXT_PROCESS_MAIN];
- router_port = rt->port_by_type[NXT_PROCESS_ROUTER];
-
- size = app->name.length + 1 + app->conf.length;
-
- b = nxt_buf_mem_ts_alloc(task, task->thread->engine->mem_pool, size);
-
- nxt_buf_cpystr(b, &app->name);
- *b->mem.free++ = '\0';
- nxt_buf_cpystr(b, &app->conf);
-
- stream = nxt_port_rpc_register_handler(task, router_port,
- nxt_router_sw_ready,
- nxt_router_sw_error,
- main_port->pid, sw);
-
- nxt_port_socket_write(task, main_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_inline nxt_port_t *
+nxt_router_app_get_port_unsafe(nxt_app_t *app, int *use_delta)
{
- nxt_queue_link_t *lnk;
- nxt_req_app_link_t *ra;
+ nxt_port_t *port;
+ nxt_queue_link_t *lnk;
- nxt_thread_log_debug("app '%V' %p state: %d/%uD/%uD/%d", &app->name, app,
- app->live, app->workers, app->pending_workers,
- nxt_queue_is_empty(&app->requests));
+ lnk = nxt_queue_first(&app->ports);
+ nxt_queue_remove(lnk);
- if (app->live == 0
- && app->workers == 0
- && app->pending_workers == 0
- && nxt_queue_is_empty(&app->requests))
- {
- nxt_thread_mutex_destroy(&app->mutex);
- nxt_free(app);
+ port = nxt_queue_link_data(lnk, nxt_port_t, app_link);
- return 1;
- }
+ port->app_requests++;
- if (app->live == 1
- && nxt_queue_is_empty(&app->requests) == 0
- && app->workers + app->pending_workers < app->max_workers)
+ if (app->live &&
+ (app->max_pending_responses == 0 ||
+ (port->app_requests - port->app_responses) <
+ app->max_pending_responses) )
{
- lnk = nxt_queue_first(&app->requests);
- nxt_queue_remove(lnk);
+ nxt_queue_insert_tail(&app->ports, lnk);
- ra = nxt_queue_link_data(lnk, nxt_req_app_link_t, link);
+ } else {
+ lnk->next = NULL;
- nxt_router_sw_create(task, app, ra);
+ (*use_delta)--;
}
- return 0;
+ return port;
}
static nxt_port_t *
-nxt_router_app_get_port(nxt_app_t *app, uint32_t stream)
+nxt_router_app_get_idle_port(nxt_app_t *app)
{
- nxt_port_t *port;
- nxt_queue_link_t *lnk;
+ nxt_port_t *port;
port = NULL;
nxt_thread_mutex_lock(&app->mutex);
- if (!nxt_queue_is_empty(&app->ports)) {
- lnk = nxt_queue_first(&app->ports);
- nxt_queue_remove(lnk);
+ nxt_queue_each(port, &app->ports, nxt_port_t, app_link) {
- lnk->next = NULL;
+ if (port->app_requests > port->app_responses) {
+ port = NULL;
- port = nxt_queue_link_data(lnk, nxt_port_t, app_link);
+ continue;
+ }
- port->app_stream = stream;
- }
+ nxt_queue_remove(&port->app_link);
+ port->app_link.next = NULL;
+
+ break;
+
+ } nxt_queue_loop;
nxt_thread_mutex_unlock(&app->mutex);
@@ -2579,151 +2615,175 @@ nxt_router_app_get_port(nxt_app_t *app, uint32_t stream)
static void
-nxt_router_app_release_port(nxt_task_t *task, void *obj, void *data)
+nxt_router_app_process_request(nxt_task_t *task, void *obj, void *data)
{
- nxt_app_t *app;
- nxt_port_t *port;
- nxt_work_t *work;
- nxt_queue_link_t *lnk;
- nxt_req_app_link_t *ra;
+ nxt_app_t *app;
+ nxt_req_app_link_t *ra;
- port = obj;
- app = data;
+ app = obj;
+ ra = data;
nxt_assert(app != NULL);
- nxt_assert(app == port->app);
- nxt_assert(port->app_link.next == NULL);
+ nxt_assert(ra != NULL);
+ nxt_assert(ra->app_port != NULL);
+ nxt_debug(task, "app '%V' %p process next stream #%uD",
+ &app->name, app, ra->stream);
- if (task->thread->engine != port->engine) {
- work = &port->work;
+ nxt_router_process_http_request_mp(task, ra);
+}
- nxt_debug(task, "post release port to engine %p", port->engine);
- work->next = NULL;
- work->handler = nxt_router_app_release_port;
- work->task = &port->engine->task;
- work->obj = port;
- work->data = app;
+static void
+nxt_router_app_port_release(nxt_task_t *task, nxt_port_t *port,
+ uint32_t request_failed, uint32_t got_response)
+{
+ int use_delta, ra_use_delta;
+ nxt_app_t *app;
+ nxt_queue_link_t *lnk;
+ nxt_req_app_link_t *ra;
- nxt_event_engine_post(port->engine, work);
+ nxt_assert(port != NULL);
+ nxt_assert(port->app != NULL);
- return;
+ app = port->app;
+
+ use_delta = (request_failed == 0 && got_response == 0) ? 0 : -1;
+
+ nxt_thread_mutex_lock(&app->mutex);
+
+ port->app_requests -= request_failed;
+ port->app_responses += got_response;
+
+ if (app->live != 0 &&
+ port->pair[1] != -1 &&
+ port->app_link.next == NULL &&
+ (app->max_pending_responses == 0 ||
+ (port->app_requests - port->app_responses) <
+ app->max_pending_responses) )
+ {
+ nxt_queue_insert_tail(&app->ports, &port->app_link);
+ use_delta++;
}
- if (!nxt_queue_is_empty(&app->requests)) {
+ if (app->live != 0 &&
+ !nxt_queue_is_empty(&app->ports) &&
+ !nxt_queue_is_empty(&app->requests))
+ {
lnk = nxt_queue_first(&app->requests);
nxt_queue_remove(lnk);
+ lnk->next = NULL;
ra = nxt_queue_link_data(lnk, nxt_req_app_link_t, link);
- nxt_debug(task, "app '%V' %p process next stream #%uD",
- &app->name, app, ra->stream);
+ ra_use_delta = 1;
+ ra->app_port = nxt_router_app_get_port_unsafe(app, &ra_use_delta);
- ra->app_port = port;
- port->app_stream = ra->stream;
+ } else {
+ ra = NULL;
+ ra_use_delta = 0;
+ }
- nxt_router_process_http_request_mp(task, ra, port);
+ nxt_thread_mutex_unlock(&app->mutex);
- nxt_router_ra_release(task, ra, ra->work.data);
+ if (ra != NULL) {
+ nxt_work_queue_add(&task->thread->engine->fast_work_queue,
+ nxt_router_app_process_request,
+ &task->thread->engine->task, app, ra);
- return;
+ goto adjust_use;
}
- port->app_stream = 0;
-
+ /* ? */
if (port->pair[1] == -1) {
- nxt_debug(task, "app '%V' %p port already closed (pid %PI dead?)",
- &app->name, app, port->pid);
-
- app->workers--;
- nxt_router_app_free(task, app);
+ nxt_debug(task, "app '%V' %p port %p already closed (pid %PI dead?)",
+ &app->name, app, port, port->pid);
- port->app = NULL;
-
- nxt_port_release(port);
-
- return;
+ goto adjust_use;
}
- if (!app->live) {
+ if (app->live == 0) {
nxt_debug(task, "app '%V' %p is not alive, send QUIT to port",
&app->name, app);
nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT,
-1, 0, 0, NULL);
- return;
+ goto adjust_use;
}
nxt_debug(task, "app '%V' %p requests queue is empty, keep the port",
&app->name, app);
- nxt_thread_mutex_lock(&app->mutex);
+adjust_use:
- nxt_queue_insert_head(&app->ports, &port->app_link);
+ if (use_delta != 0) {
+ nxt_port_use(task, port, use_delta);
+ }
- nxt_thread_mutex_unlock(&app->mutex);
+ if (ra_use_delta != 0) {
+ nxt_port_use(task, ra->app_port, ra_use_delta);
+ }
}
-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)
{
nxt_app_t *app;
- nxt_bool_t busy;
+ nxt_bool_t unchain, start_worker;
app = port->app;
- busy = port->app_stream != 0;
-
- if (app == NULL) {
- nxt_thread_log_debug("port %p app remove, no app", port);
-
- nxt_assert(port->app_link.next == NULL);
- return 1;
- }
+ nxt_assert(app != NULL);
nxt_thread_mutex_lock(&app->mutex);
- if (port->app_link.next != NULL) {
+ unchain = port->app_link.next != NULL;
+ if (unchain) {
nxt_queue_remove(&port->app_link);
port->app_link.next = NULL;
+ }
+ app->workers--;
+
+ start_worker = app->live != 0 &&
+ nxt_queue_is_empty(&app->requests) == 0 &&
+ app->workers + app->pending_workers < app->max_workers;
+
+ if (start_worker) {
+ app->pending_workers++;
}
nxt_thread_mutex_unlock(&app->mutex);
- if (busy == 0) {
- nxt_thread_log_debug("port %p app remove, free, app '%V' %p", port,
- &app->name, app);
-
- app->workers--;
- nxt_router_app_free(&port->engine->task, app);
+ nxt_debug(task, "app '%V' %p port %p close", &app->name, app, port);
- return 1;
+ if (unchain) {
+ nxt_port_use(task, port, -1);
}
- nxt_thread_log_debug("port %p app remove, busy, app '%V' %p, "
- "app stream #%uD", port, &app->name, app,
- port->app_stream);
-
- return 0;
+ if (start_worker) {
+ nxt_router_start_worker(task, app);
+ }
}
static nxt_int_t
nxt_router_app_port(nxt_task_t *task, nxt_req_app_link_t *ra)
{
+ int use_delta;
+ nxt_int_t res;
nxt_app_t *app;
+ nxt_bool_t can_start_worker;
nxt_conn_t *c;
nxt_port_t *port;
nxt_event_engine_t *engine;
- nxt_start_worker_t *sw;
nxt_socket_conf_joint_t *joint;
port = NULL;
+ use_delta = 1;
c = ra->rc->conn;
joint = c->listen->socket.data;
@@ -2735,6 +2795,10 @@ nxt_router_app_port(nxt_task_t *task, nxt_req_app_link_t *ra)
return NXT_ERROR;
}
+ ra->rc->app = app;
+
+ nxt_router_app_use(task, app, 1);
+
engine = task->thread->engine;
nxt_timer_disable(engine, &c->read_timer);
@@ -2744,20 +2808,50 @@ nxt_router_app_port(nxt_task_t *task, nxt_req_app_link_t *ra)
nxt_timer_add(engine, &c->read_timer, app->timeout);
}
- port = nxt_router_app_get_port(app, ra->stream);
+ nxt_thread_mutex_lock(&app->mutex);
+
+ if (!nxt_queue_is_empty(&app->ports)) {
+ port = nxt_router_app_get_port_unsafe(app, &use_delta);
+
+ can_start_worker = 0;
+
+ } else {
+ nxt_queue_insert_tail(&app->requests, &ra->link);
+
+ can_start_worker = (app->workers + app->pending_workers) <
+ app->max_workers;
+ if (can_start_worker) {
+ app->pending_workers++;
+ }
+
+ port = NULL;
+ }
+
+ nxt_thread_mutex_unlock(&app->mutex);
if (port != NULL) {
- nxt_debug(task, "already have port for app '%V'", &app->name);
+ nxt_debug(task, "already have port for app '%V' %p ", &app->name, app);
ra->app_port = port;
+
+ if (use_delta != 0) {
+ nxt_port_use(task, port, use_delta);
+ }
return NXT_OK;
}
- sw = nxt_router_sw_create(task, app, ra);
+ if (!can_start_worker) {
+ nxt_debug(task, "app '%V' %p too many running or pending workers",
+ &app->name, app);
+
+ return NXT_AGAIN;
+ }
+
+ res = nxt_router_start_worker(task, app);
+
+ if (nxt_slow_path(res != NXT_OK)) {
+ nxt_router_gen_error(task, c, 500, "Failed to start worker");
- if (nxt_slow_path(sw == NULL)) {
- nxt_router_gen_error(task, c, 500,
- "Failed to allocate start worker struct");
return NXT_ERROR;
}
@@ -3011,18 +3105,16 @@ nxt_router_process_http_request(nxt_task_t *task, nxt_conn_t *c,
nxt_port_rpc_ex_set_peer(task, engine->port, rc, port->pid);
- nxt_router_process_http_request_mp(task, ra, port);
-
- nxt_router_ra_release(task, ra, ra->work.data);
+ nxt_router_process_http_request_mp(task, ra);
}
static void
-nxt_router_process_http_request_mp(nxt_task_t *task, nxt_req_app_link_t *ra,
- nxt_port_t *port)
+nxt_router_process_http_request_mp(nxt_task_t *task, nxt_req_app_link_t *ra)
{
+ uint32_t request_failed;
nxt_int_t res;
- nxt_port_t *c_port, *reply_port;
+ nxt_port_t *port, *c_port, *reply_port;
nxt_conn_t *c;
nxt_app_wmsg_t wmsg;
nxt_app_parse_ctx_t *ap;
@@ -3030,11 +3122,15 @@ nxt_router_process_http_request_mp(nxt_task_t *task, nxt_req_app_link_t *ra,
/* TODO: it is unsafe to use ra->rc and ra->rc->conn in main thread */
nxt_assert(ra->rc != NULL);
+ nxt_assert(ra->app_port != NULL);
+ port = ra->app_port;
reply_port = ra->reply_port;
ap = ra->ap;
c = ra->rc->conn;
+ request_failed = 1;
+
c_port = nxt_process_connected_port_find(port->process, reply_port->pid,
reply_port->id);
if (nxt_slow_path(c_port != reply_port)) {
@@ -3043,7 +3139,7 @@ nxt_router_process_http_request_mp(nxt_task_t *task, nxt_req_app_link_t *ra,
if (nxt_slow_path(res != NXT_OK)) {
nxt_router_gen_error(task, c, 500,
"Failed to send reply port to application");
- return;
+ goto release_port;
}
nxt_process_connected_port_add(port->process, reply_port);
@@ -3059,21 +3155,33 @@ nxt_router_process_http_request_mp(nxt_task_t *task, nxt_req_app_link_t *ra,
if (nxt_slow_path(res != NXT_OK)) {
nxt_router_gen_error(task, c, 500,
"Failed to prepare message for application");
- return;
+ goto release_port;
}
nxt_debug(task, "about to send %d bytes buffer to worker port %d",
nxt_buf_used_size(wmsg.write),
wmsg.port->socket.fd);
+ request_failed = 0;
+
res = nxt_port_socket_write(task, wmsg.port, NXT_PORT_MSG_DATA,
-1, ra->stream, reply_port->id, wmsg.write);
if (nxt_slow_path(res != NXT_OK)) {
nxt_router_gen_error(task, c, 500,
"Failed to send message to application");
- return;
+ goto release_port;
}
+
+release_port:
+
+ if (request_failed != 0) {
+ ra->app_port = 0;
+ }
+
+ nxt_router_app_port_release(task, port, request_failed, 0);
+
+ nxt_router_ra_release(task, ra, ra->work.data);
}
@@ -3452,13 +3560,7 @@ nxt_router_conn_free(nxt_task_t *task, void *obj, void *data)
nxt_debug(task, "conn %p close, stream #%uD", c, rc->stream);
- if (rc->app_port != NULL) {
- nxt_router_app_release_port(task, rc->app_port, rc->app_port->app);
-
- rc->app_port = NULL;
- }
-
- nxt_router_rc_unlink(rc);
+ nxt_router_rc_unlink(task, rc);
nxt_port_rpc_cancel(task, task->thread->engine->port, rc->stream);
diff --git a/src/nxt_router.h b/src/nxt_router.h
index f5c5f7aa..5056021e 100644
--- a/src/nxt_router.h
+++ b/src/nxt_router.h
@@ -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_ */
diff --git a/src/nxt_runtime.c b/src/nxt_runtime.c
index deab980e..a96f4cea 100644
--- a/src/nxt_runtime.c
+++ b/src/nxt_runtime.c
@@ -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_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);
}
diff --git a/src/nxt_runtime.h b/src/nxt_runtime.h
index d0decfd2..5aa897dc 100644
--- a/src/nxt_runtime.h
+++ b/src/nxt_runtime.h
@@ -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);
generated by cgit (git 2.34.1) at 2024-11-17 17:26:26 +0000