diff options
Diffstat (limited to 'src/nxt_router.c')
| -rw-r--r-- | src/nxt_router.c | 234 |
1 files changed, 103 insertions, 131 deletions
diff --git a/src/nxt_router.c b/src/nxt_router.c index 103ba78c..f637c864 100644 --- a/src/nxt_router.c +++ b/src/nxt_router.c @@ -23,6 +23,13 @@ typedef struct { } nxt_router_listener_conf_t; +typedef struct nxt_msg_info_s { + nxt_buf_t *buf; + nxt_port_mmap_tracking_t tracking; + nxt_work_handler_t completion_handler; +} nxt_msg_info_t; + + typedef struct nxt_req_app_link_s nxt_req_app_link_t; @@ -32,6 +39,7 @@ typedef struct { nxt_app_t *app; nxt_port_t *app_port; nxt_app_parse_ctx_t *ap; + nxt_msg_info_t msg_info; nxt_req_app_link_t *ra; nxt_queue_link_t link; /* for nxt_conn_t.requests */ @@ -44,6 +52,7 @@ struct nxt_req_app_link_s { nxt_pid_t app_pid; nxt_port_t *reply_port; nxt_app_parse_ctx_t *ap; + nxt_msg_info_t msg_info; nxt_req_conn_link_t *rc; nxt_queue_link_t link; /* for nxt_app_t.requests */ @@ -64,9 +73,6 @@ typedef struct { static nxt_int_t nxt_router_start_worker(nxt_task_t *task, nxt_app_t *app); -static void nxt_router_ra_error(nxt_task_t *task, nxt_req_app_link_t *ra, - int code, const char* str); - static nxt_router_temp_conf_t *nxt_router_temp_conf(nxt_task_t *task); static void nxt_router_conf_apply(nxt_task_t *task, void *obj, void *data); static void nxt_router_conf_ready(nxt_task_t *task, @@ -352,93 +358,87 @@ nxt_router_ra_create(nxt_task_t *task, nxt_req_app_link_t *ra_src) } -static void -nxt_router_ra_release(nxt_task_t *task, void *obj, void *data) +nxt_inline nxt_bool_t +nxt_router_msg_cancel(nxt_task_t *task, nxt_msg_info_t *msg_info, + uint32_t stream) { - nxt_req_app_link_t *ra; - nxt_event_engine_t *engine; - nxt_req_conn_link_t *rc; + nxt_buf_t *b, *next; + nxt_bool_t cancelled; - ra = obj; - engine = data; - - if (task->thread->engine != engine) { - if (ra->app_port != NULL) { - ra->app_pid = ra->app_port->pid; - } - - ra->work.handler = nxt_router_ra_release; - ra->work.task = &engine->task; - ra->work.next = NULL; + if (msg_info->buf == NULL) { + return 0; + } - nxt_debug(task, "ra stream #%uD post release to %p", - ra->stream, engine); + cancelled = nxt_port_mmap_tracking_cancel(task, &msg_info->tracking, + stream); - nxt_event_engine_post(engine, &ra->work); - - return; + if (cancelled) { + nxt_debug(task, "stream #%uD: cancelled by router", stream); } - nxt_debug(task, "ra stream #%uD release", ra->stream); + for (b = msg_info->buf; b != NULL; b = next) { + next = b->next; - rc = ra->rc; + b->completion_handler = msg_info->completion_handler; - if (rc != NULL) { - if (ra->app_pid != -1) { - nxt_port_rpc_ex_set_peer(task, engine->port, rc, ra->app_pid); + if (b->is_port_mmap_sent) { + b->is_port_mmap_sent = cancelled == 0; + b->completion_handler(task, b, b->parent); } - - rc->app_port = ra->app_port; - - ra->app_port = NULL; - 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; - } + msg_info->buf = NULL; - if (ra->mem_pool != NULL) { - nxt_mp_release(ra->mem_pool, ra); - } + return cancelled; } static void -nxt_router_ra_abort(nxt_task_t *task, void *obj, void *data) +nxt_router_ra_release(nxt_task_t *task, void *obj, void *data) { - nxt_conn_t *c; nxt_req_app_link_t *ra; - nxt_req_conn_link_t *rc; nxt_event_engine_t *engine; + nxt_req_conn_link_t *rc; ra = obj; engine = data; if (task->thread->engine != engine) { - ra->work.handler = nxt_router_ra_abort; + if (ra->app_port != NULL) { + ra->app_pid = ra->app_port->pid; + } + + ra->work.handler = nxt_router_ra_release; ra->work.task = &engine->task; ra->work.next = NULL; - nxt_debug(task, "ra stream #%uD post abort to %p", ra->stream, engine); + nxt_debug(task, "ra stream #%uD post release to %p", + ra->stream, engine); nxt_event_engine_post(engine, &ra->work); return; } - nxt_debug(task, "ra stream #%uD abort", ra->stream); + nxt_debug(task, "ra stream #%uD release", ra->stream); rc = ra->rc; if (rc != NULL) { - c = rc->conn; + if (ra->app_pid != -1) { + nxt_port_rpc_ex_set_peer(task, engine->port, rc, ra->app_pid); + } - nxt_router_gen_error(task, c, 500, - "Failed to start application worker"); + if (nxt_slow_path(ra->err_code != 0)) { + nxt_router_gen_error(task, rc->conn, ra->err_code, ra->err_str); + + } else { + rc->app_port = ra->app_port; + rc->msg_info = ra->msg_info; + + ra->app_port = NULL; + ra->msg_info.buf = NULL; + } rc->ra = NULL; ra->rc = NULL; @@ -450,70 +450,20 @@ nxt_router_ra_abort(nxt_task_t *task, void *obj, void *data) ra->app_port = NULL; } + nxt_router_msg_cancel(task, &ra->msg_info, ra->stream); + if (ra->mem_pool != NULL) { nxt_mp_release(ra->mem_pool, ra); } } -static void -nxt_router_ra_error_handler(nxt_task_t *task, void *obj, void *data) -{ - nxt_req_app_link_t *ra; - - ra = obj; - - nxt_router_ra_error(task, ra, ra->err_code, ra->err_str); -} - - -static void -nxt_router_ra_error(nxt_task_t *task, nxt_req_app_link_t *ra, int code, - const char* str) +nxt_inline void +nxt_router_ra_error(nxt_req_app_link_t *ra, int code, const char* str) { - nxt_conn_t *c; - nxt_req_conn_link_t *rc; - nxt_event_engine_t *engine; - - engine = ra->work.data; - - if (task->thread->engine != engine) { - ra->err_code = code; - ra->err_str = str; - - ra->work.handler = nxt_router_ra_error_handler; - ra->work.task = &engine->task; - ra->work.next = NULL; - - nxt_debug(task, "ra stream #%uD post error to %p", ra->stream, engine); - - nxt_event_engine_post(engine, &ra->work); - - return; - } - - nxt_debug(task, "ra stream #%uD error", ra->stream); - - rc = ra->rc; - - if (rc != NULL) { - c = rc->conn; - - nxt_router_gen_error(task, c, code, str); - - 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; - } - - if (ra->mem_pool != NULL) { - nxt_mp_release(ra->mem_pool, ra); - } + ra->app_port = NULL; + ra->err_code = code; + ra->err_str = str; } @@ -528,6 +478,8 @@ nxt_router_rc_unlink(nxt_task_t *task, nxt_req_conn_link_t *rc) rc->app_port = NULL; } + nxt_router_msg_cancel(task, &rc->msg_info, rc->stream); + ra = rc->ra; if (ra != NULL) { @@ -601,6 +553,10 @@ nxt_router_conf_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg) return; } + nxt_debug(task, "nxt_router_conf_data_handler(%d): %*s", + nxt_buf_used_size(msg->buf), + nxt_buf_used_size(msg->buf), msg->buf->mem.pos); + b = nxt_buf_chk_make_plain(tmcf->conf->mem_pool, msg->buf, msg->size); nxt_assert(b != NULL); @@ -2417,8 +2373,6 @@ nxt_router_get_error_buf(nxt_task_t *task, nxt_mp_t *mp, int code, b->mem.free = nxt_cpymem(b->mem.free, str, nxt_strlen(str)); - nxt_log_alert(task->log, "error %d: %s", code, str); - last = nxt_buf_sync_alloc(mp, NXT_BUF_SYNC_LAST); if (nxt_slow_path(last == NULL)) { @@ -2442,13 +2396,16 @@ nxt_router_gen_error(nxt_task_t *task, nxt_conn_t *c, int code, /* TODO: fix when called in the middle of response */ + nxt_log_alert(task->log, "error %d: %s", code, str); + + if (c->socket.fd == -1) { + return; + } + mp = c->mem_pool; b = nxt_router_get_error_buf(task, mp, code, str); - - if (c->socket.fd == -1) { - nxt_mp_free(mp, b->next); - nxt_mp_free(mp, b); + if (nxt_slow_path(b == NULL)) { return; } @@ -2533,7 +2490,8 @@ nxt_router_app_port_error(nxt_task_t *task, nxt_port_recv_msg_t *msg, 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_ra_error(ra, 500, "Failed to start application worker"); + nxt_router_ra_release(task, ra, ra->work.data); } nxt_router_app_use(task, app, -1); @@ -3198,9 +3156,16 @@ nxt_router_process_http_request(nxt_task_t *task, nxt_conn_t *c, static void +nxt_router_dummy_buf_completion(nxt_task_t *task, void *obj, void *data) +{ +} + + +static void nxt_router_process_http_request_mp(nxt_task_t *task, nxt_req_app_link_t *ra) { uint32_t request_failed; + nxt_buf_t *b; nxt_int_t res; nxt_port_t *port, *c_port, *reply_port; nxt_app_wmsg_t wmsg; @@ -3220,9 +3185,8 @@ nxt_router_process_http_request_mp(nxt_task_t *task, nxt_req_app_link_t *ra) res = nxt_port_send_port(task, port, reply_port, 0); if (nxt_slow_path(res != NXT_OK)) { - nxt_router_ra_error(task, ra, 500, + nxt_router_ra_error(ra, 500, "Failed to send reply port to application"); - ra = NULL; goto release_port; } @@ -3237,9 +3201,8 @@ nxt_router_process_http_request_mp(nxt_task_t *task, nxt_req_app_link_t *ra) res = port->app->prepare_msg(task, &ap->r, &wmsg); if (nxt_slow_path(res != NXT_OK)) { - nxt_router_ra_error(task, ra, 500, + nxt_router_ra_error(ra, 500, "Failed to prepare message for application"); - ra = NULL; goto release_port; } @@ -3249,13 +3212,28 @@ nxt_router_process_http_request_mp(nxt_task_t *task, nxt_req_app_link_t *ra) request_failed = 0; - res = nxt_port_socket_write(task, wmsg.port, NXT_PORT_MSG_DATA, - -1, ra->stream, reply_port->id, wmsg.write); + ra->msg_info.buf = wmsg.write; + ra->msg_info.completion_handler = wmsg.write->completion_handler; + + for (b = wmsg.write; b != NULL; b = b->next) { + b->completion_handler = nxt_router_dummy_buf_completion; + } + + res = nxt_port_mmap_get_tracking(task, port, &ra->msg_info.tracking, + ra->stream); + if (nxt_slow_path(res != NXT_OK)) { + nxt_router_ra_error(ra, 500, + "Failed to get tracking area"); + goto release_port; + } + + res = nxt_port_socket_twrite(task, wmsg.port, NXT_PORT_MSG_DATA, + -1, ra->stream, reply_port->id, wmsg.write, + &ra->msg_info.tracking); if (nxt_slow_path(res != NXT_OK)) { - nxt_router_ra_error(task, ra, 500, + nxt_router_ra_error(ra, 500, "Failed to send message to application"); - ra = NULL; goto release_port; } @@ -3263,13 +3241,7 @@ release_port: nxt_router_app_port_release(task, port, request_failed, 0); - if (ra != NULL) { - if (request_failed != 0) { - ra->app_port = 0; - } - - nxt_router_ra_release(task, ra, ra->work.data); - } + nxt_router_ra_release(task, ra, ra->work.data); } |