diff options
| -rw-r--r-- | src/nxt_port.h | 3 | ||||
| -rw-r--r-- | src/nxt_router.c | 2 | ||||
| -rw-r--r-- | src/nxt_unit.c | 37 | ||||
| -rw-r--r-- | src/nxt_unit.h | 2 |
4 files changed, 39 insertions, 5 deletions
diff --git a/src/nxt_port.h b/src/nxt_port.h index 3ac8c735..5ece3bfa 100644 --- a/src/nxt_port.h +++ b/src/nxt_port.h @@ -26,6 +26,7 @@ struct nxt_port_handlers_s { nxt_port_handler_t change_file; nxt_port_handler_t new_port; nxt_port_handler_t get_port; + nxt_port_handler_t port_ack; nxt_port_handler_t mmap; nxt_port_handler_t get_mmap; @@ -84,6 +85,7 @@ typedef enum { _NXT_PORT_MSG_CHANGE_FILE = nxt_port_handler_idx(change_file), _NXT_PORT_MSG_NEW_PORT = nxt_port_handler_idx(new_port), _NXT_PORT_MSG_GET_PORT = nxt_port_handler_idx(get_port), + _NXT_PORT_MSG_PORT_ACK = nxt_port_handler_idx(port_ack), _NXT_PORT_MSG_MMAP = nxt_port_handler_idx(mmap), _NXT_PORT_MSG_GET_MMAP = nxt_port_handler_idx(get_mmap), @@ -120,6 +122,7 @@ typedef enum { NXT_PORT_MSG_CHANGE_FILE = nxt_msg_last(_NXT_PORT_MSG_CHANGE_FILE), NXT_PORT_MSG_NEW_PORT = nxt_msg_last(_NXT_PORT_MSG_NEW_PORT), NXT_PORT_MSG_GET_PORT = nxt_msg_last(_NXT_PORT_MSG_GET_PORT), + NXT_PORT_MSG_PORT_ACK = nxt_msg_last(_NXT_PORT_MSG_PORT_ACK), NXT_PORT_MSG_MMAP = nxt_msg_last(_NXT_PORT_MSG_MMAP) | NXT_PORT_MSG_SYNC, NXT_PORT_MSG_GET_MMAP = nxt_msg_last(_NXT_PORT_MSG_GET_MMAP), diff --git a/src/nxt_router.c b/src/nxt_router.c index cf627746..fbc9a4c8 100644 --- a/src/nxt_router.c +++ b/src/nxt_router.c @@ -688,6 +688,8 @@ nxt_router_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg) port->app = app; port->main_app_port = main_app_port; + + nxt_port_socket_write(task, port, NXT_PORT_MSG_PORT_ACK, -1, 0, 0, NULL); } diff --git a/src/nxt_unit.c b/src/nxt_unit.c index 7d2bf2c7..d35a3307 100644 --- a/src/nxt_unit.c +++ b/src/nxt_unit.c @@ -57,6 +57,7 @@ static int nxt_unit_ready(nxt_unit_ctx_t *ctx, int ready_fd, uint32_t stream, static int nxt_unit_process_msg(nxt_unit_ctx_t *ctx, nxt_unit_read_buf_t *rbuf); static int nxt_unit_process_new_port(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg); +static int nxt_unit_ctx_ready(nxt_unit_ctx_t *ctx); static int nxt_unit_process_req_headers(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg); static int nxt_unit_process_req_body(nxt_unit_ctx_t *ctx, @@ -306,6 +307,7 @@ struct nxt_unit_ctx_impl_s { nxt_queue_t free_rbuf; int online; + int ready; nxt_unit_mmap_buf_t ctx_buf[2]; nxt_unit_read_buf_t ctx_read_buf; @@ -624,6 +626,7 @@ nxt_unit_ctx_init(nxt_unit_impl_t *lib, nxt_unit_ctx_impl_t *ctx_impl, ctx_impl->use_count = 1; ctx_impl->wait_items = 0; ctx_impl->online = 1; + ctx_impl->ready = 0; nxt_queue_init(&ctx_impl->free_req); nxt_queue_init(&ctx_impl->free_ws); @@ -996,6 +999,10 @@ nxt_unit_process_msg(nxt_unit_ctx_t *ctx, nxt_unit_read_buf_t *rbuf) rc = nxt_unit_process_new_port(ctx, &recv_msg); break; + case _NXT_PORT_MSG_PORT_ACK: + rc = nxt_unit_ctx_ready(ctx); + break; + case _NXT_PORT_MSG_CHANGE_FILE: nxt_unit_debug(ctx, "#%"PRIu32": change_file: fd %d", port_msg->stream, recv_msg.fd[0]); @@ -1169,8 +1176,28 @@ nxt_unit_process_new_port(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg) if (new_port_msg->id == NXT_UNIT_SHARED_PORT_ID) { lib->shared_port = port; - } else { - nxt_unit_port_release(port); + return nxt_unit_ctx_ready(ctx); + } + + nxt_unit_port_release(port); + + return NXT_UNIT_OK; +} + + +static int +nxt_unit_ctx_ready(nxt_unit_ctx_t *ctx) +{ + nxt_unit_impl_t *lib; + nxt_unit_ctx_impl_t *ctx_impl; + + lib = nxt_container_of(ctx->unit, nxt_unit_impl_t, unit); + ctx_impl = nxt_container_of(ctx, nxt_unit_ctx_impl_t, ctx); + + ctx_impl->ready = 1; + + if (lib->callbacks.ready_handler) { + return lib->callbacks.ready_handler(ctx); } return NXT_UNIT_OK; @@ -4495,17 +4522,17 @@ nxt_unit_read_buf(nxt_unit_ctx_t *ctx, nxt_unit_read_buf_t *rbuf) nxt_unit_port_impl_t *port_impl; struct pollfd fds[2]; - lib = nxt_container_of(ctx->unit, nxt_unit_impl_t, unit); ctx_impl = nxt_container_of(ctx, nxt_unit_ctx_impl_t, ctx); - if (ctx_impl->wait_items > 0 || lib->shared_port == NULL) { - + if (ctx_impl->wait_items > 0 || ctx_impl->ready == 0) { return nxt_unit_ctx_port_recv(ctx, ctx_impl->read_port, rbuf); } port_impl = nxt_container_of(ctx_impl->read_port, nxt_unit_port_impl_t, port); + lib = nxt_container_of(ctx->unit, nxt_unit_impl_t, unit); + retry: if (port_impl->from_socket == 0) { diff --git a/src/nxt_unit.h b/src/nxt_unit.h index 303d5aa1..cb78e862 100644 --- a/src/nxt_unit.h +++ b/src/nxt_unit.h @@ -154,6 +154,8 @@ struct nxt_unit_callbacks_s { /* Receive data on port id. Optional. */ ssize_t (*port_recv)(nxt_unit_ctx_t *, nxt_unit_port_t *port, void *buf, size_t buf_size, void *oob, size_t oob_size); + + int (*ready_handler)(nxt_unit_ctx_t *); }; |