Fixing body fd access racing condition.

To avoid closing the body fd prematurely, the fd value is moved from
the request struct to the app link.  The body fd should not be closed
immediately after the request is sent to the application due to possible
request rescheduling.

3 files changed, 32 insertions, 18 deletions

diff --git a/src/nxt_router.c b/src/nxt_router.c
index a913284c..d4f25d7e 100644
--- a/src/nxt_router.c
+++ b/src/nxt_router.c
@@ -462,6 +462,7 @@ nxt_inline void
 nxt_request_app_link_init(nxt_task_t *task,
     nxt_request_app_link_t *req_app_link, nxt_request_rpc_data_t *req_rpc_data)
 {
+    nxt_buf_t           *body;
     nxt_event_engine_t  *engine;
 
     engine = task->thread->engine;
@@ -480,6 +481,17 @@ nxt_request_app_link_init(nxt_task_t *task,
     req_app_link->work.task = &engine->task;
     req_app_link->work.obj = req_app_link;
     req_app_link->work.data = engine;
+
+    body = req_rpc_data->request->body;
+
+    if (body != NULL && nxt_buf_is_file(body)) {
+        req_app_link->body_fd = body->file->fd;
+
+        body->file->fd = -1;
+
+    } else {
+        req_app_link->body_fd = -1;
+    }
 }
 
 
@@ -513,6 +525,10 @@ nxt_request_app_link_alloc(nxt_task_t *task,
 
     nxt_request_app_link_init(task, req_app_link, req_rpc_data);
 
+    if (ra_src != NULL) {
+        req_app_link->body_fd = ra_src->body_fd;
+    }
+
     req_app_link->mem_pool = mp;
 
     return req_app_link;
@@ -654,6 +670,12 @@ nxt_request_app_link_release(nxt_task_t *task,
         req_app_link->app_port = NULL;
     }
 
+    if (req_app_link->body_fd != -1) {
+        nxt_fd_close(req_app_link->body_fd);
+
+        req_app_link->body_fd = -1;
+    }
+
     nxt_router_msg_cancel(task, &req_app_link->msg_info, req_app_link->stream);
 
     mp = req_app_link->mem_pool;
@@ -4774,8 +4796,7 @@ static void
 nxt_router_app_prepare_request(nxt_task_t *task,
     nxt_request_app_link_t *req_app_link)
 {
-    nxt_fd_t          fd;
-    nxt_buf_t         *buf, *body;
+    nxt_buf_t         *buf;
     nxt_int_t         res;
     nxt_port_t        *port, *c_port, *reply_port;
     nxt_apr_action_t  apr_action;
@@ -4834,13 +4855,15 @@ nxt_router_app_prepare_request(nxt_task_t *task,
         goto release_port;
     }
 
-    body = req_app_link->request->body;
-    fd = (body != NULL && nxt_buf_is_file(body)) ? body->file->fd : -1;
+    if (req_app_link->body_fd != -1) {
+        nxt_debug(task, "stream #%uD: send body fd %d", req_app_link->stream,
+                  req_app_link->body_fd);
 
-    res = nxt_port_socket_twrite(task, port,
-                                 NXT_PORT_MSG_REQ_HEADERS
-                                    | NXT_PORT_MSG_CLOSE_FD,
-                                 fd,
+        lseek(req_app_link->body_fd, 0, SEEK_SET);
+    }
+
+    res = nxt_port_socket_twrite(task, port, NXT_PORT_MSG_REQ_HEADERS,
+                                 req_app_link->body_fd,
                                  req_app_link->stream, reply_port->id, buf,
                                  &req_app_link->msg_info.tracking);
 
@@ -4850,10 +4873,6 @@ nxt_router_app_prepare_request(nxt_task_t *task,
         goto release_port;
     }
 
-    if (fd != -1) {
-        body->file->fd = -1;
-    }
-
 release_port:
 
     nxt_router_app_port_release(task, port, apr_action);
@@ -5178,10 +5197,6 @@ nxt_router_prepare_msg(nxt_task_t *task, nxt_http_request_t *r,
         }
     }
 
-    if (r->body != NULL && nxt_buf_is_file(r->body)) {
-        lseek(r->body->file->fd, 0, SEEK_SET);
-    }
-
     return out;
 }
 
diff --git a/src/nxt_router_request.h b/src/nxt_router_request.h
index c3d5767e..a38980ee 100644
--- a/src/nxt_router_request.h
+++ b/src/nxt_router_request.h
@@ -50,6 +50,7 @@ struct nxt_request_app_link_s {
     nxt_http_request_t      *request;
     nxt_msg_info_t          msg_info;
     nxt_request_rpc_data_t  *req_rpc_data;
+    nxt_fd_t                body_fd;
 
     nxt_nsec_t              res_time;
 
diff --git a/test/test_proxy.py b/test/test_proxy.py
index 5568550a..74bd0873 100644
--- a/test/test_proxy.py
+++ b/test/test_proxy.py
@@ -196,8 +196,6 @@ Content-Length: 10
         self.assertEqual(resp['body'], payload, 'body')
 
     def test_proxy_parallel(self):
-        self.skip_alerts.append(r'close\(\d+\) failed')
-
         payload = 'X' * 4096 * 257
         buff_size = 4096 * 258