Merged with the default branch.1.21.0-1

1 files changed, 52 insertions, 9 deletions

diff --git a/src/nxt_router.c b/src/nxt_router.c
index a3218047..9dd5c30e 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);
 }
 
 
@@ -1260,6 +1262,12 @@ static nxt_conf_map_t  nxt_router_http_conf[] = {
         NXT_CONF_MAP_STR,
         offsetof(nxt_socket_conf_t, body_temp_path),
     },
+
+    {
+        nxt_string("discard_unsafe_fields"),
+        NXT_CONF_MAP_INT8,
+        offsetof(nxt_socket_conf_t, discard_unsafe_fields),
+    },
 };
 
 
@@ -1647,6 +1655,7 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
             skcf->header_buffer_size = 2048;
             skcf->large_header_buffer_size = 8192;
             skcf->large_header_buffers = 4;
+            skcf->discard_unsafe_fields = 1;
             skcf->body_buffer_size = 16 * 1024;
             skcf->max_body_size = 8 * 1024 * 1024;
             skcf->proxy_header_buffer_size = 64 * 1024;
@@ -3722,6 +3731,7 @@ static void
 nxt_router_response_ready_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
     void *data)
 {
+    size_t                  b_size, count;
     nxt_int_t               ret;
     nxt_app_t               *app;
     nxt_buf_t               *b, *next;
@@ -3787,15 +3797,20 @@ nxt_router_response_ready_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
         nxt_http_request_send_body(task, r, NULL);
 
     } else {
-        size_t b_size = nxt_buf_mem_used_size(&b->mem);
+        b_size = nxt_buf_is_mem(b) ? nxt_buf_mem_used_size(&b->mem) : 0;
 
-        if (nxt_slow_path(b_size < sizeof(*resp))) {
+        if (nxt_slow_path(b_size < sizeof(nxt_unit_response_t))) {
+            nxt_alert(task, "response buffer too small: %z", b_size);
             goto fail;
         }
 
         resp = (void *) b->mem.pos;
-        if (nxt_slow_path(b_size < sizeof(*resp)
-              + resp->fields_count * sizeof(nxt_unit_field_t))) {
+        count = (b_size - sizeof(nxt_unit_response_t))
+                    / sizeof(nxt_unit_field_t);
+
+        if (nxt_slow_path(count < resp->fields_count)) {
+            nxt_alert(task, "response buffer too small for fields count: %D",
+                      resp->fields_count);
             goto fail;
         }
 
@@ -3904,6 +3919,7 @@ nxt_router_req_headers_ack_handler(nxt_task_t *task,
 {
     int                 res;
     nxt_app_t           *app;
+    nxt_buf_t           *b;
     nxt_bool_t          start_process, unlinked;
     nxt_port_t          *app_port, *main_app_port, *idle_port;
     nxt_queue_link_t    *idle_lnk;
@@ -4001,16 +4017,25 @@ nxt_router_req_headers_ack_handler(nxt_task_t *task,
 
     req_rpc_data->app_port = app_port;
 
-    if (req_rpc_data->msg_info.body_fd != -1) {
+    b = req_rpc_data->msg_info.buf;
+
+    if (b != NULL) {
+        /* First buffer is already sent.  Start from second. */
+        b = b->next;
+    }
+
+    if (req_rpc_data->msg_info.body_fd != -1 || b != NULL) {
         nxt_debug(task, "stream #%uD: send body fd %d", req_rpc_data->stream,
                   req_rpc_data->msg_info.body_fd);
 
-        lseek(req_rpc_data->msg_info.body_fd, 0, SEEK_SET);
+        if (req_rpc_data->msg_info.body_fd != -1) {
+            lseek(req_rpc_data->msg_info.body_fd, 0, SEEK_SET);
+        }
 
         res = nxt_port_socket_write(task, app_port, NXT_PORT_MSG_REQ_BODY,
                                     req_rpc_data->msg_info.body_fd,
                                     req_rpc_data->stream,
-                                    task->thread->engine->port->id, NULL);
+                                    task->thread->engine->port->id, b);
 
         if (nxt_slow_path(res != NXT_OK)) {
             nxt_http_request_error(task, r, NXT_HTTP_INTERNAL_SERVER_ERROR);
@@ -4667,6 +4692,25 @@ nxt_router_free_app(nxt_task_t *task, void *obj, void *data)
         nxt_port_use(task, port, -1);
     }
 
+    nxt_thread_mutex_lock(&app->mutex);
+
+    for ( ;; ) {
+        port = nxt_port_hash_retrieve(&app->port_hash);
+        if (port == NULL) {
+            break;
+        }
+
+        app->port_hash_count--;
+
+        port->app = NULL;
+
+        nxt_port_close(task, port);
+
+        nxt_port_use(task, port, -1);
+    }
+
+    nxt_thread_mutex_unlock(&app->mutex);
+
     nxt_assert(app->processes == 0);
     nxt_assert(app->active_requests == 0);
     nxt_assert(app->port_hash_count == 0);
@@ -5364,8 +5408,7 @@ nxt_router_oosm_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
     nxt_thread_mutex_unlock(&process->incoming.mutex);
 
     if (ack) {
-        (void) nxt_port_socket_write(task, msg->port, NXT_PORT_MSG_SHM_ACK,
-                                     -1, 0, 0, NULL);
+        nxt_process_broadcast_shm_ack(task, process);
     }
 }