From 3cbc22a6dc45abdeade4deb364601230ddca02c1 Mon Sep 17 00:00:00 2001 From: Max Romanov Date: Tue, 11 Aug 2020 19:20:10 +0300 Subject: Changing router to application port exchange protocol. The application process needs to request the port from the router instead of the latter pushing the port before sending a request to the application. This is required to simplify the communication between the router and the application and to prepare the router to use the application shared port and then the queue. --- src/nxt_port.h | 9 +++++++++ 1 file changed, 9 insertions(+) (limited to 'src/nxt_port.h') diff --git a/src/nxt_port.h b/src/nxt_port.h index 0e8707f3..838a7ffe 100644 --- a/src/nxt_port.h +++ b/src/nxt_port.h @@ -25,6 +25,7 @@ struct nxt_port_handlers_s { /* File descriptor exchange. */ nxt_port_handler_t change_file; nxt_port_handler_t new_port; + nxt_port_handler_t get_port; nxt_port_handler_t mmap; /* New process */ @@ -77,6 +78,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_MMAP = nxt_port_handler_idx(mmap), _NXT_PORT_MSG_PROCESS_CREATED = nxt_port_handler_idx(process_created), @@ -107,6 +109,7 @@ typedef enum { NXT_PORT_MSG_ACCESS_LOG = nxt_msg_last(_NXT_PORT_MSG_ACCESS_LOG), 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_MMAP = nxt_msg_last(_NXT_PORT_MSG_MMAP) | NXT_PORT_MSG_CLOSE_FD | NXT_PORT_MSG_SYNC, @@ -238,6 +241,12 @@ typedef struct { } nxt_port_msg_new_port_t; +typedef struct { + nxt_port_id_t id; + nxt_pid_t pid; +} nxt_port_msg_get_port_t; + + /* * nxt_port_data_t size is allocation size * which enables effective reuse of memory pool cache. -- cgit From 6e31d6cd39be9d3f4ee680fc13c3fe42f5cd39e7 Mon Sep 17 00:00:00 2001 From: Max Romanov Date: Tue, 11 Aug 2020 19:20:13 +0300 Subject: Changing router to application shared memory exchange protocol. The application process needs to request the shared memory segment from the router instead of the latter pushing the segment before sending a request to the application. This is required to simplify the communication between the router and the application and to prepare the router for using the application shared port and then the queue. --- src/nxt_port.h | 7 +++++++ 1 file changed, 7 insertions(+) (limited to 'src/nxt_port.h') diff --git a/src/nxt_port.h b/src/nxt_port.h index 838a7ffe..3a8a200a 100644 --- a/src/nxt_port.h +++ b/src/nxt_port.h @@ -27,6 +27,7 @@ struct nxt_port_handlers_s { nxt_port_handler_t new_port; nxt_port_handler_t get_port; nxt_port_handler_t mmap; + nxt_port_handler_t get_mmap; /* New process */ nxt_port_handler_t process_created; @@ -80,6 +81,7 @@ typedef enum { _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_MMAP = nxt_port_handler_idx(mmap), + _NXT_PORT_MSG_GET_MMAP = nxt_port_handler_idx(get_mmap), _NXT_PORT_MSG_PROCESS_CREATED = nxt_port_handler_idx(process_created), _NXT_PORT_MSG_PROCESS_READY = nxt_port_handler_idx(process_ready), @@ -247,6 +249,11 @@ typedef struct { } nxt_port_msg_get_port_t; +typedef struct { + uint32_t id; +} nxt_port_msg_get_mmap_t; + + /* * nxt_port_data_t size is allocation size * which enables effective reuse of memory pool cache. -- cgit From 83595606121a821f9e3cef0f0b7e7fe87eb1e50a Mon Sep 17 00:00:00 2001 From: Max Romanov Date: Tue, 11 Aug 2020 19:20:15 +0300 Subject: Introducing the shared application port. This is the port shared between all application processes which use it to pass requests for processing. Using it significantly simplifies the request processing code in the router. The drawback is 2 more file descriptors per each configured application and more complex libunit message wait/read code. --- src/nxt_port.h | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) (limited to 'src/nxt_port.h') diff --git a/src/nxt_port.h b/src/nxt_port.h index 3a8a200a..9a933e75 100644 --- a/src/nxt_port.h +++ b/src/nxt_port.h @@ -41,6 +41,7 @@ struct nxt_port_handlers_s { /* Request headers. */ nxt_port_handler_t req_headers; + nxt_port_handler_t req_headers_ack; /* Websocket frame. */ nxt_port_handler_t websocket_frame; @@ -89,6 +90,7 @@ typedef enum { _NXT_PORT_MSG_QUIT = nxt_port_handler_idx(quit), _NXT_PORT_MSG_REQ_HEADERS = nxt_port_handler_idx(req_headers), + _NXT_PORT_MSG_REQ_HEADERS_ACK = nxt_port_handler_idx(req_headers_ack), _NXT_PORT_MSG_WEBSOCKET = nxt_port_handler_idx(websocket_frame), _NXT_PORT_MSG_DATA = nxt_port_handler_idx(data), @@ -113,7 +115,8 @@ typedef enum { 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_MMAP = nxt_msg_last(_NXT_PORT_MSG_MMAP) - | NXT_PORT_MSG_CLOSE_FD | NXT_PORT_MSG_SYNC, + | NXT_PORT_MSG_SYNC, + NXT_PORT_MSG_GET_MMAP = nxt_msg_last(_NXT_PORT_MSG_GET_MMAP), NXT_PORT_MSG_PROCESS_CREATED = nxt_msg_last(_NXT_PORT_MSG_PROCESS_CREATED), NXT_PORT_MSG_PROCESS_READY = nxt_msg_last(_NXT_PORT_MSG_PROCESS_READY), @@ -193,6 +196,7 @@ struct nxt_port_s { nxt_queue_link_t app_link; /* for nxt_app_t.ports */ nxt_app_t *app; + nxt_port_t *main_app_port; nxt_queue_link_t idle_link; /* for nxt_app_t.idle_ports */ nxt_msec_t idle_start; @@ -205,11 +209,10 @@ struct nxt_port_s { /* Maximum interleave of message parts. */ uint32_t max_share; - uint32_t app_pending_responses; uint32_t app_responses; - nxt_queue_t pending_requests; - nxt_queue_t active_websockets; + uint32_t active_websockets; + uint32_t active_requests; nxt_port_handler_t handler; nxt_port_handler_t *data; -- cgit From a1e9df2aef5a3917728c6fd37280b03020d51123 Mon Sep 17 00:00:00 2001 From: Max Romanov Date: Tue, 11 Aug 2020 19:20:30 +0300 Subject: Port message extended to transfer 2 file descriptors. --- src/nxt_port.h | 2 ++ 1 file changed, 2 insertions(+) (limited to 'src/nxt_port.h') diff --git a/src/nxt_port.h b/src/nxt_port.h index 9a933e75..ab455f92 100644 --- a/src/nxt_port.h +++ b/src/nxt_port.h @@ -165,6 +165,7 @@ typedef struct { nxt_buf_t *buf; size_t share; nxt_fd_t fd; + nxt_fd_t fd2; nxt_port_msg_t port_msg; uint32_t tracking_msg[2]; uint8_t close_fd; /* 1 bit */ @@ -174,6 +175,7 @@ typedef struct { struct nxt_port_recv_msg_s { nxt_fd_t fd; + nxt_fd_t fd2; nxt_buf_t *buf; nxt_port_t *port; nxt_port_msg_t port_msg; -- cgit From e227fc9e6281c280c46139a81646ecd7b0510e2b Mon Sep 17 00:00:00 2001 From: Max Romanov Date: Tue, 11 Aug 2020 19:20:34 +0300 Subject: Introducing application and port shared memory queues. The goal is to minimize the number of syscalls needed to deliver a message. --- src/nxt_port.h | 25 ++++++++++++++++++++----- 1 file changed, 20 insertions(+), 5 deletions(-) (limited to 'src/nxt_port.h') diff --git a/src/nxt_port.h b/src/nxt_port.h index ab455f92..9fbf00b1 100644 --- a/src/nxt_port.h +++ b/src/nxt_port.h @@ -42,6 +42,7 @@ struct nxt_port_handlers_s { /* Request headers. */ nxt_port_handler_t req_headers; nxt_port_handler_t req_headers_ack; + nxt_port_handler_t req_body; /* Websocket frame. */ nxt_port_handler_t websocket_frame; @@ -51,6 +52,8 @@ struct nxt_port_handlers_s { nxt_port_handler_t oosm; nxt_port_handler_t shm_ack; + nxt_port_handler_t read_queue; + nxt_port_handler_t read_socket; }; @@ -91,12 +94,15 @@ typedef enum { _NXT_PORT_MSG_REQ_HEADERS = nxt_port_handler_idx(req_headers), _NXT_PORT_MSG_REQ_HEADERS_ACK = nxt_port_handler_idx(req_headers_ack), + _NXT_PORT_MSG_REQ_BODY = nxt_port_handler_idx(req_body), _NXT_PORT_MSG_WEBSOCKET = nxt_port_handler_idx(websocket_frame), _NXT_PORT_MSG_DATA = nxt_port_handler_idx(data), _NXT_PORT_MSG_OOSM = nxt_port_handler_idx(oosm), _NXT_PORT_MSG_SHM_ACK = nxt_port_handler_idx(shm_ack), + _NXT_PORT_MSG_READ_QUEUE = nxt_port_handler_idx(read_queue), + _NXT_PORT_MSG_READ_SOCKET = nxt_port_handler_idx(read_socket), NXT_PORT_MSG_MAX = sizeof(nxt_port_handlers_t) / sizeof(nxt_port_handler_t), @@ -124,6 +130,7 @@ typedef enum { NXT_PORT_MSG_REMOVE_PID = nxt_msg_last(_NXT_PORT_MSG_REMOVE_PID), NXT_PORT_MSG_REQ_HEADERS = _NXT_PORT_MSG_REQ_HEADERS, + NXT_PORT_MSG_REQ_BODY = _NXT_PORT_MSG_REQ_BODY, NXT_PORT_MSG_WEBSOCKET = _NXT_PORT_MSG_WEBSOCKET, NXT_PORT_MSG_WEBSOCKET_LAST = nxt_msg_last(_NXT_PORT_MSG_WEBSOCKET), @@ -132,6 +139,8 @@ typedef enum { NXT_PORT_MSG_OOSM = nxt_msg_last(_NXT_PORT_MSG_OOSM), NXT_PORT_MSG_SHM_ACK = nxt_msg_last(_NXT_PORT_MSG_SHM_ACK), + NXT_PORT_MSG_READ_QUEUE = _NXT_PORT_MSG_READ_QUEUE, + NXT_PORT_MSG_READ_SOCKET = _NXT_PORT_MSG_READ_SOCKET, } nxt_port_msg_type_t; @@ -236,6 +245,12 @@ struct nxt_port_s { nxt_atomic_t use_count; nxt_process_type_t type; + + nxt_fd_t queue_fd; + void *queue; + + void *socket_msg; + int from_socket; }; @@ -286,17 +301,17 @@ 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); void nxt_port_read_close(nxt_port_t *port); -nxt_int_t nxt_port_socket_twrite(nxt_task_t *task, nxt_port_t *port, - nxt_uint_t type, nxt_fd_t fd, uint32_t stream, nxt_port_id_t reply_port, - nxt_buf_t *b, void *tracking); +nxt_int_t nxt_port_socket_write2(nxt_task_t *task, nxt_port_t *port, + nxt_uint_t type, nxt_fd_t fd, nxt_fd_t fd2, uint32_t stream, + nxt_port_id_t reply_port, nxt_buf_t *b); nxt_inline nxt_int_t nxt_port_socket_write(nxt_task_t *task, nxt_port_t *port, nxt_uint_t type, nxt_fd_t fd, uint32_t stream, nxt_port_id_t reply_port, nxt_buf_t *b) { - return nxt_port_socket_twrite(task, port, type, fd, stream, reply_port, b, - NULL); + return nxt_port_socket_write2(task, port, type, fd, -1, stream, reply_port, + b); } void nxt_port_enable(nxt_task_t *task, nxt_port_t *port, -- cgit From f147943f6382c0e90a216615ff9bcf57a3db8c75 Mon Sep 17 00:00:00 2001 From: Max Romanov Date: Tue, 11 Aug 2020 21:48:27 +0300 Subject: Style fixes for 2 file descriptors transfer over port. Two consecutive fd and fd2 fields replaced with array. --- src/nxt_port.h | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) (limited to 'src/nxt_port.h') diff --git a/src/nxt_port.h b/src/nxt_port.h index 9fbf00b1..3ac8c735 100644 --- a/src/nxt_port.h +++ b/src/nxt_port.h @@ -173,8 +173,7 @@ typedef struct { nxt_queue_link_t link; nxt_buf_t *buf; size_t share; - nxt_fd_t fd; - nxt_fd_t fd2; + nxt_fd_t fd[2]; nxt_port_msg_t port_msg; uint32_t tracking_msg[2]; uint8_t close_fd; /* 1 bit */ @@ -183,8 +182,7 @@ typedef struct { struct nxt_port_recv_msg_s { - nxt_fd_t fd; - nxt_fd_t fd2; + nxt_fd_t fd[2]; nxt_buf_t *buf; nxt_port_t *port; nxt_port_msg_t port_msg; -- cgit