Using disk file to store large request body.

This closes #386 on GitHub.

auto/help

@@ -20,6 +20,7 @@ cat << END
   --incdir=DIRECTORY   set includes directory name, default: "$NXT_INCDIR"
   --modules=DIRECTORY  set modules directory name, default: "$NXT_MODULES"
   --state=DIRECTORY    set state directory name, default: "$NXT_STATE"
+  --tmp=DIRECTORY      set tmp directory name, default: "$NXT_TMP"
 
   --pid=FILE           set pid filename, default: "$NXT_PID"
   --log=FILE           set log filename, default: "$NXT_LOG"

auto/options

@@ -58,6 +58,7 @@ do
         --incdir=*)                      NXT_INCDIR="$value"                 ;;
         --modules=*)                     NXT_MODULES="$value"                ;;
         --state=*)                       NXT_STATE="$value"                  ;;
+        --tmp=*)                         NXT_TMP="$value"                    ;;
 
         --pid=*)                         NXT_PID="$value"                    ;;
         --log=*)                         NXT_LOG="$value"                    ;;
@@ -149,6 +150,11 @@ case "$NXT_STATE" in
      *)  NXT_STATE="$NXT_PREFIX$NXT_STATE"  ;;
 esac
 
+case "$NXT_TMP" in
+    /*)  ;;
+     *)  NXT_TMP="$NXT_PREFIX$NXT_TMP"  ;;
+esac
+
 case "$NXT_PID" in
     /*)  ;;
      *)  NXT_PID="$NXT_PREFIX$NXT_PID"  ;;

auto/save

@@ -29,5 +29,6 @@ NXT_LIB_AUX_LIBS=
 NXT_LIB_UNIT_STATIC='$NXT_LIB_UNIT_STATIC'
 
 NXT_MODULES='$NXT_MODULES'
+NXT_TMP='$NXT_TMP'
 
 END

auto/summary

@@ -13,6 +13,7 @@ Unit configuration summary:
   include directory: ......... "$NXT_INCDIR"
   modules directory: ......... "$NXT_MODULES"
   state directory: ........... "$NXT_STATE"
+  tmp directory: ............. "$NXT_TMP"
 
   pid file: .................. "$NXT_PID"
   log file: .................. "$NXT_LOG"

configure

@@ -37,6 +37,7 @@ NXT_LIBDIR="lib"
 NXT_INCDIR="include"
 NXT_MODULES="modules"
 NXT_STATE="state"
+NXT_TMP="tmp"
 NXT_PID="unit.pid"
 NXT_LOG="unit.log"
 NXT_CONTROL="unix:control.unit.sock"
@@ -86,6 +87,7 @@ cat << END >> $NXT_AUTO_CONFIG_H
 #define NXT_LOG                "$NXT_LOG"
 #define NXT_MODULES            "$NXT_MODULES"
 #define NXT_STATE              "$NXT_STATE"
+#define NXT_TMP                "$NXT_TMP"
 
 #define NXT_CONTROL_SOCK       "$NXT_CONTROL"
 

pkg/deb/Makefile

@@ -129,6 +129,7 @@ CONFIGURE_ARGS=\
 	--control="unix:/var/run/control.unit.sock" \
 	--pid=/var/run/unit.pid \
 	--log=/var/log/unit.log \
+	--tmp=/var/tmp \
 	--tests \
 	--openssl
 

pkg/rpm/Makefile

@@ -144,6 +144,7 @@ CONFIGURE_ARGS=\
 	--control="unix:/var/run/unit/control.sock" \
 	--pid=/var/run/unit/unit.pid \
 	--log=/var/log/unit/unit.log \
+	--tmp=/var/tmp \
 	--tests \
 	--openssl
 

src/nxt_conf_validation.c

@@ -182,11 +182,21 @@ static nxt_conf_vldt_object_t  nxt_conf_vldt_http_members[] = {
       NULL,
       NULL },
 
+    { nxt_string("body_buffer_size"),
+      NXT_CONF_VLDT_INTEGER,
+      NULL,
+      NULL },
+
     { nxt_string("max_body_size"),
       NXT_CONF_VLDT_INTEGER,
       NULL,
       NULL },
 
+    { nxt_string("body_temp_path"),
+      NXT_CONF_VLDT_STRING,
+      NULL,
+      NULL },
+
     { nxt_string("websocket"),
       NXT_CONF_VLDT_OBJECT,
       &nxt_conf_vldt_object,

src/nxt_conn_write.c

@@ -9,6 +9,8 @@
 
 static void nxt_conn_write_timer_handler(nxt_task_t *task, void *obj,
     void *data);
+static ssize_t nxt_conn_io_sendfile(nxt_task_t *task, nxt_sendbuf_t *sb);
+static ssize_t nxt_sendfile(int fd, int s, off_t pos, size_t size);
 
 
 void
@@ -170,10 +172,104 @@ nxt_conn_io_sendbuf(nxt_task_t *task, nxt_sendbuf_t *sb)
         return 0;
     }
 
+    if (niov == 0 && nxt_buf_is_file(sb->buf)) {
+        return nxt_conn_io_sendfile(task, sb);
+    }
+
     return nxt_conn_io_writev(task, sb, iov, niov);
 }
 
 
+static ssize_t
+nxt_conn_io_sendfile(nxt_task_t *task, nxt_sendbuf_t *sb)
+{
+    size_t     size;
+    ssize_t    n;
+    nxt_buf_t  *b;
+    nxt_err_t  err;
+
+    b = sb->buf;
+
+    for ( ;; ) {
+        size = b->file_end - b->file_pos;
+
+        n = nxt_sendfile(b->file->fd, sb->socket, b->file_pos, size);
+
+        err = (n == -1) ? nxt_errno : 0;
+
+        nxt_debug(task, "sendfile(%FD, %d, @%O, %uz): %z",
+                  b->file->fd, sb->socket, b->file_pos, size, n);
+
+        if (n > 0) {
+            if (n < (ssize_t) size) {
+                sb->ready = 0;
+            }
+
+            return n;
+        }
+
+        if (nxt_slow_path(n == 0)) {
+            nxt_alert(task, "sendfile() reported that file was truncated at %O",
+                      b->file_pos);
+
+            return NXT_ERROR;
+        }
+
+        /* n == -1 */
+
+        switch (err) {
+
+        case NXT_EAGAIN:
+            sb->ready = 0;
+            nxt_debug(task, "sendfile() %E", err);
+
+            return NXT_AGAIN;
+
+        case NXT_EINTR:
+            nxt_debug(task, "sendfile() %E", err);
+            continue;
+
+        default:
+            sb->error = err;
+            nxt_log(task, nxt_socket_error_level(err),
+                    "sendfile(%FD, %d, @%O, %uz) failed %E",
+                    b->file->fd, sb->socket, b->file_pos, size, err);
+
+            return NXT_ERROR;
+        }
+    }
+}
+
+
+static ssize_t
+nxt_sendfile(int fd, int s, off_t pos, size_t size)
+{
+    ssize_t  res;
+
+#ifdef NXT_HAVE_MACOSX_SENDFILE
+    off_t sent = size;
+
+    int rc = sendfile(fd, s, pos, &sent, NULL, 0);
+
+    res = (rc == 0 || sent > 0) ? sent : -1;
+#endif
+
+#ifdef NXT_HAVE_FREEBSD_SENDFILE
+    off_t sent = 0;
+
+    int rc = sendfile(fd, s, pos, size, NULL, &sent, 0);
+
+    res = (rc == 0 || sent > 0) ? sent : -1;
+#endif
+
+#ifdef NXT_HAVE_LINUX_SENDFILE
+    res = sendfile(s, fd, &pos, size);
+#endif
+
+    return res;
+}
+
+
 ssize_t
 nxt_conn_io_writev(nxt_task_t *task, nxt_sendbuf_t *sb, struct iovec *iov,
     nxt_uint_t niov)

src/nxt_h1proto.c

@@ -817,12 +817,16 @@ nxt_h1p_transfer_encoding(void *ctx, nxt_http_field_t *field, uintptr_t data)
 static void
 nxt_h1p_request_body_read(nxt_task_t *task, nxt_http_request_t *r)
 {
-    size_t             size, body_length;
+    size_t             size, body_length, body_buffer_size, body_rest;
+    ssize_t            res;
+    nxt_str_t          *tmp_path, tmp_name;
     nxt_buf_t          *in, *b;
     nxt_conn_t         *c;
     nxt_h1proto_t      *h1p;
     nxt_http_status_t  status;
 
+    static const nxt_str_t tmp_name_pattern = nxt_string("/req-XXXXXXXX");
+
     h1p = r->proto.h1;
 
     nxt_debug(task, "h1p request body read %O te:%d",
@@ -849,36 +853,95 @@ nxt_h1p_request_body_read(nxt_task_t *task, nxt_http_request_t *r)
 
     body_length = (size_t) r->content_length_n;
 
-    b = r->body;
+    body_buffer_size = nxt_min(r->conf->socket_conf->body_buffer_size,
+                               body_length);
+
+    if (body_length > body_buffer_size) {
+        tmp_path = &r->conf->socket_conf->body_temp_path;
+
+        tmp_name.length = tmp_path->length + tmp_name_pattern.length;
+
+        b = nxt_buf_file_alloc(r->mem_pool,
+                               body_buffer_size + sizeof(nxt_file_t)
+                               + tmp_name.length + 1, 0);
+
+    } else {
+        /* This initialization required for CentOS 6, gcc 4.4.7. */
+        tmp_path = NULL;
+        tmp_name.length = 0;
+
+        b = nxt_buf_mem_alloc(r->mem_pool, body_buffer_size, 0);
+    }
+
+    if (nxt_slow_path(b == NULL)) {
+        status = NXT_HTTP_INTERNAL_SERVER_ERROR;
+        goto error;
+    }
+
+    r->body = b;
+
+    if (body_length > body_buffer_size) {
+        tmp_name.start = nxt_pointer_to(b->mem.start, sizeof(nxt_file_t));
+
+        memcpy(tmp_name.start, tmp_path->start, tmp_path->length);
+        memcpy(tmp_name.start + tmp_path->length, tmp_name_pattern.start,
+               tmp_name_pattern.length);
+        tmp_name.start[tmp_name.length] = '\0';
+
+        b->file = (nxt_file_t *) b->mem.start;
+        nxt_memzero(b->file, sizeof(nxt_file_t));
+        b->file->fd = -1;
+        b->file->size = body_length;
+
+        b->mem.start += sizeof(nxt_file_t) + tmp_name.length + 1;
+        b->mem.pos = b->mem.start;
+        b->mem.free = b->mem.start;
+
+        b->file->fd = mkstemp((char *) tmp_name.start);
+        if (nxt_slow_path(b->file->fd == -1)) {
+            nxt_log(task, NXT_LOG_ERR, "mkstemp() failed %E", nxt_errno);
 
-    if (b == NULL) {
-        b = nxt_buf_mem_alloc(r->mem_pool, body_length, 0);
-        if (nxt_slow_path(b == NULL)) {
             status = NXT_HTTP_INTERNAL_SERVER_ERROR;
             goto error;
         }
 
-        r->body = b;
+        nxt_debug(task, "create body tmp file \"%V\", %d",
+                  &tmp_name, b->file->fd);
+
+        unlink((char *) tmp_name.start);
     }
 
+    body_rest = body_length;
+
     in = h1p->conn->read;
 
     size = nxt_buf_mem_used_size(&in->mem);
 
     if (size != 0) {
-        if (size > body_length) {
+        size = nxt_min(size, body_length);
-            size = body_length;
+
+        if (nxt_buf_is_file(b)) {
+            res = nxt_fd_write(b->file->fd, in->mem.pos, size);
+            if (nxt_slow_path(res < (ssize_t) size)) {
+                status = NXT_HTTP_INTERNAL_SERVER_ERROR;
+                goto error;
+            }
+
+            b->file_end += size;
+
+        } else {
+            size = nxt_min(body_buffer_size, size);
+            b->mem.free = nxt_cpymem(b->mem.free, in->mem.pos, size);
+            body_buffer_size -= size;
         }
 
-        b->mem.free = nxt_cpymem(b->mem.free, in->mem.pos, size);
         in->mem.pos += size;
+        body_rest -= size;
     }
 
-    size = nxt_buf_mem_free_size(&b->mem);
+    nxt_debug(task, "h1p body rest: %uz", body_rest);
 
-    nxt_debug(task, "h1p body rest: %uz", size);
+    if (body_rest != 0) {
-
-    if (size != 0) {
         in->next = h1p->buffers;
         h1p->buffers = in;
         h1p->nbuffers++;
@@ -891,6 +954,13 @@ nxt_h1p_request_body_read(nxt_task_t *task, nxt_http_request_t *r)
         return;
     }
 
+    if (nxt_buf_is_file(b)) {
+        b->mem.start = NULL;
+        b->mem.end = NULL;
+        b->mem.pos = NULL;
+        b->mem.free = NULL;
+    }
+
 ready:
 
     r->state->ready_handler(task, r, NULL);
@@ -922,7 +992,9 @@ static const nxt_conn_state_t  nxt_h1p_read_body_state
 static void
 nxt_h1p_conn_request_body_read(nxt_task_t *task, void *obj, void *data)
 {
-    size_t              size;
+    size_t              size, body_rest;
+    ssize_t             res;
+    nxt_buf_t           *b;
     nxt_conn_t          *c;
     nxt_h1proto_t       *h1p;
     nxt_http_request_t  *r;
@@ -933,18 +1005,59 @@ nxt_h1p_conn_request_body_read(nxt_task_t *task, void *obj, void *data)
 
     nxt_debug(task, "h1p conn request body read");
 
-    size = nxt_buf_mem_free_size(&c->read->mem);
+    r = h1p->request;
-
-    nxt_debug(task, "h1p body rest: %uz", size);
 
     engine = task->thread->engine;
 
-    if (size != 0) {
+    b = c->read;
+
+    if (nxt_buf_is_file(b)) {
+        body_rest = b->file->size - b->file_end;
+
+        size = nxt_buf_mem_used_size(&b->mem);
+        size = nxt_min(size, body_rest);
+
+        res = nxt_fd_write(b->file->fd, b->mem.pos, size);
+        if (nxt_slow_path(res < (ssize_t) size)) {
+            nxt_h1p_request_error(task, h1p, r);
+            return;
+        }
+
+        b->file_end += size;
+        body_rest -= res;
+
+        b->mem.pos += size;
+
+        if (b->mem.pos == b->mem.free) {
+            if (body_rest >= (size_t) nxt_buf_mem_size(&b->mem)) {
+                b->mem.free = b->mem.start;
+
+            } else {
+                /* This required to avoid reading next request. */
+                b->mem.free = b->mem.end - body_rest;
+            }
+
+            b->mem.pos = b->mem.free;
+        }
+
+    } else {
+        body_rest = nxt_buf_mem_free_size(&c->read->mem);
+    }
+
+    nxt_debug(task, "h1p body rest: %uz", body_rest);
+
+    if (body_rest != 0) {
         nxt_conn_read(engine, c);
 
     } else {
+        if (nxt_buf_is_file(b)) {
+            b->mem.start = NULL;
+            b->mem.end = NULL;
+            b->mem.pos = NULL;
+            b->mem.free = NULL;
+        }
+
         c->read = NULL;
-        r = h1p->request;
 
         r->state->ready_handler(task, r, NULL);
     }
@@ -2140,7 +2253,13 @@ nxt_h1p_peer_header_send(nxt_task_t *task, nxt_http_peer_t *peer)
     c->write_state = &nxt_h1p_peer_header_send_state;
 
     if (r->body != NULL) {
-        body = nxt_buf_mem_alloc(r->mem_pool, 0, 0);
+        if (nxt_buf_is_file(r->body)) {
+            body = nxt_buf_file_alloc(r->mem_pool, 0, 0);
+
+        } else {
+            body = nxt_buf_mem_alloc(r->mem_pool, 0, 0);
+        }
+
         if (nxt_slow_path(body == NULL)) {
             r->state->error_handler(task, r, peer);
             return;
@@ -2148,8 +2267,15 @@ nxt_h1p_peer_header_send(nxt_task_t *task, nxt_http_peer_t *peer)
 
         header->next = body;
 
-        body->mem = r->body->mem;
+        if (nxt_buf_is_file(r->body)) {
-        size += nxt_buf_mem_used_size(&body->mem);
+            body->file = r->body->file;
+            body->file_end = r->body->file_end;
+
+        } else {
+            body->mem = r->body->mem;
+        }
+
+        size += nxt_buf_used_size(body);
 
 //        nxt_mp_retain(r->mem_pool);
     }
@@ -2205,13 +2331,13 @@ nxt_h1p_peer_header_sent(nxt_task_t *task, void *obj, void *data)
 
     c->write = nxt_sendbuf_completion(task, &engine->fast_work_queue, c->write);
 
-    if (c->write == NULL) {
+    if (c->write != NULL) {
-        r = peer->request;
+        nxt_conn_write(engine, c);
-        r->state->ready_handler(task, r, peer);
         return;
     }
 
-    nxt_conn_write(engine, c);
+    r = peer->request;
+    r->state->ready_handler(task, r, peer);
 }
 
 

src/nxt_http_request.c

@@ -569,6 +569,14 @@ nxt_http_request_close_handler(nxt_task_t *task, void *obj, void *data)
 
     r->proto.any = NULL;
 
+    if (r->body != NULL && nxt_buf_is_file(r->body)
+        && r->body->file->fd != -1)
+    {
+        nxt_fd_close(r->body->file->fd);
+
+        r->body->file->fd = -1;
+    }
+
     if (nxt_fast_path(proto.any != NULL)) {
         protocol = r->protocol;
 

src/nxt_router.c

@@ -1361,6 +1361,12 @@ static nxt_conf_map_t  nxt_router_http_conf[] = {
         NXT_CONF_MAP_MSEC,
         offsetof(nxt_socket_conf_t, send_timeout),
     },
+
+    {
+        nxt_string("body_temp_path"),
+        NXT_CONF_MAP_STR,
+        offsetof(nxt_socket_conf_t, body_temp_path),
+    },
 };
 
 
@@ -1397,6 +1403,7 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
     nxt_int_t                   ret;
     nxt_str_t                   name, path;
     nxt_app_t                   *app, *prev;
+    nxt_str_t                   *t;
     nxt_router_t                *router;
     nxt_app_joint_t             *app_joint;
     nxt_conf_value_t            *conf, *http, *value, *websocket;
@@ -1698,6 +1705,8 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
             skcf->websocket_conf.read_timeout = 60 * 1000;
             skcf->websocket_conf.keepalive_interval = 30 * 1000;
 
+            nxt_str_null(&skcf->body_temp_path);
+
             if (http != NULL) {
                 ret = nxt_conf_map_object(mp, http, nxt_router_http_conf,
                                           nxt_nitems(nxt_router_http_conf),
@@ -1719,6 +1728,13 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
                 }
             }
 
+            t = &skcf->body_temp_path;
+
+            if (t->length == 0) {
+                t->start = (u_char *) task->thread->runtime->tmp;
+                t->length = nxt_strlen(t->start);
+            }
+
 #if (NXT_TLS)
             value = nxt_conf_get_path(listener, &certificate_path);
 
@@ -4758,7 +4774,8 @@ static void
 nxt_router_app_prepare_request(nxt_task_t *task,
     nxt_request_app_link_t *req_app_link)
 {
-    nxt_buf_t         *buf;
+    nxt_fd_t          fd;
+    nxt_buf_t         *buf, *body;
     nxt_int_t         res;
     nxt_port_t        *port, *c_port, *reply_port;
     nxt_apr_action_t  apr_action;
@@ -4817,8 +4834,14 @@ nxt_router_app_prepare_request(nxt_task_t *task,
         goto release_port;
     }
 
-    res = nxt_port_socket_twrite(task, port, NXT_PORT_MSG_REQ_HEADERS,
+    body = req_app_link->request->body;
-                                 -1, req_app_link->stream, reply_port->id, buf,
+    fd = (body != NULL && nxt_buf_is_file(body)) ? body->file->fd : -1;
+
+    res = nxt_port_socket_twrite(task, port,
+                                 NXT_PORT_MSG_REQ_HEADERS
+                                    | NXT_PORT_MSG_CLOSE_FD,
+                                 fd,
+                                 req_app_link->stream, reply_port->id, buf,
                                  &req_app_link->msg_info.tracking);
 
     if (nxt_slow_path(res != NXT_OK)) {
@@ -4827,6 +4850,10 @@ 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);
@@ -5151,6 +5178,10 @@ 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;
 }
 

src/nxt_router.h

@@ -187,6 +187,8 @@ typedef struct {
 
     nxt_websocket_conf_t   websocket_conf;
 
+    nxt_str_t              body_temp_path;
+
 #if (NXT_TLS)
     nxt_tls_conf_t         *tls;
 #endif

src/nxt_runtime.c

@@ -693,6 +693,7 @@ nxt_runtime_conf_init(nxt_task_t *task, nxt_runtime_t *rt)
     rt->modules = NXT_MODULES;
     rt->state = NXT_STATE;
     rt->control = NXT_CONTROL_SOCK;
+    rt->tmp = NXT_TMP;
 
     nxt_memzero(&rt->capabilities, sizeof(nxt_capabilities_t));
 
@@ -835,6 +836,7 @@ nxt_runtime_conf_read_cmd(nxt_task_t *task, nxt_runtime_t *rt)
     static const char  no_modules[] =
                        "option \"--modules\" requires directory\n";
     static const char  no_state[] = "option \"--state\" requires directory\n";
+    static const char  no_tmp[] = "option \"--tmp\" requires directory\n";
 
     static const char  help[] =
         "\n"
@@ -859,6 +861,9 @@ nxt_runtime_conf_read_cmd(nxt_task_t *task, nxt_runtime_t *rt)
         "  --state DIRECTORY    set state directory name\n"
         "                       default: \"" NXT_STATE "\"\n"
         "\n"
+        "  --tmp DIRECTORY      set tmp directory name\n"
+        "                       default: \"" NXT_TMP "\"\n"
+        "\n"
         "  --user USER          set non-privileged processes to run"
                                 " as specified user\n"
         "                       default: \"" NXT_USER "\"\n"
@@ -966,6 +971,19 @@ nxt_runtime_conf_read_cmd(nxt_task_t *task, nxt_runtime_t *rt)
             continue;
         }
 
+        if (nxt_strcmp(p, "--tmp") == 0) {
+            if (*argv == NULL) {
+                write(STDERR_FILENO, no_tmp, nxt_length(no_tmp));
+                return NXT_ERROR;
+            }
+
+            p = *argv++;
+
+            rt->tmp = p;
+
+            continue;
+        }
+
         if (nxt_strcmp(p, "--no-daemon") == 0) {
             rt->daemon = 0;
             continue;

src/nxt_runtime.h

@@ -68,6 +68,7 @@ struct nxt_runtime_s {
     const char             *conf;
     const char             *conf_tmp;
     const char             *control;
+    const char             *tmp;
 
     nxt_str_t              certs;
 

src/nxt_unit.c

@@ -76,6 +76,8 @@ static nxt_unit_read_buf_t *nxt_unit_read_buf_get_impl(
     nxt_unit_ctx_impl_t *ctx_impl);
 static void nxt_unit_read_buf_release(nxt_unit_ctx_t *ctx,
     nxt_unit_read_buf_t *rbuf);
+static nxt_unit_mmap_buf_t *nxt_unit_request_preread(
+    nxt_unit_request_info_t *req, size_t size);
 static ssize_t nxt_unit_buf_read(nxt_unit_buf_t **b, uint64_t *len, void *dst,
     size_t size);
 static nxt_port_mmap_header_t *nxt_unit_mmap_get(nxt_unit_ctx_t *ctx,
@@ -961,6 +963,9 @@ nxt_unit_process_req_headers(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg)
     req_impl->incoming_buf->prev = &req_impl->incoming_buf;
     recv_msg->incoming_buf = NULL;
 
+    req->content_fd = recv_msg->fd;
+    recv_msg->fd = -1;
+
     req->response_max_fields = 0;
     req_impl->state = NXT_UNIT_RS_START;
     req_impl->websocket = 0;
@@ -1178,6 +1183,12 @@ nxt_unit_request_info_release(nxt_unit_request_info_t *req)
         nxt_unit_mmap_buf_free(req_impl->incoming_buf);
     }
 
+    if (req->content_fd != -1) {
+        close(req->content_fd);
+
+        req->content_fd = -1;
+    }
+
     /*
      * Process release should go after buffers release to guarantee mmap
      * existence.
@@ -2423,17 +2434,46 @@ nxt_unit_response_write_cb(nxt_unit_request_info_t *req,
 ssize_t
 nxt_unit_request_read(nxt_unit_request_info_t *req, void *dst, size_t size)
 {
-    return nxt_unit_buf_read(&req->content_buf, &req->content_length,
+    ssize_t  buf_res, res;
-                             dst, size);
+
+    buf_res = nxt_unit_buf_read(&req->content_buf, &req->content_length,
+                                dst, size);
+
+    if (buf_res < (ssize_t) size && req->content_fd != -1) {
+        res = read(req->content_fd, dst, size);
+        if (res < 0) {
+            nxt_unit_req_alert(req, "failed to read content: %s (%d)",
+                               strerror(errno), errno);
+
+            return res;
+        }
+
+        if (res < (ssize_t) size) {
+            close(req->content_fd);
+
+            req->content_fd = -1;
+        }
+
+        req->content_length -= res;
+        size -= res;
+
+        dst = nxt_pointer_to(dst, res);
+
+    } else {
+        res = 0;
+    }
+
+    return buf_res + res;
 }
 
 
 ssize_t
 nxt_unit_request_readline_size(nxt_unit_request_info_t *req, size_t max_size)
 {
-    char            *p;
+    char                 *p;
-    size_t          l_size, b_size;
+    size_t               l_size, b_size;
-    nxt_unit_buf_t  *b;
+    nxt_unit_buf_t       *b;
+    nxt_unit_mmap_buf_t  *mmap_buf, *preread_buf;
 
     if (req->content_length == 0) {
         return 0;
@@ -2459,6 +2499,19 @@ nxt_unit_request_readline_size(nxt_unit_request_info_t *req, size_t max_size)
             break;
         }
 
+        mmap_buf = nxt_container_of(b, nxt_unit_mmap_buf_t, buf);
+        if (mmap_buf->next == NULL
+            && req->content_fd != -1
+            && l_size < req->content_length)
+        {
+            preread_buf = nxt_unit_request_preread(req, 16384);
+            if (nxt_slow_path(preread_buf == NULL)) {
+                return -1;
+            }
+
+            nxt_unit_mmap_buf_insert(&mmap_buf->next, preread_buf);
+        }
+
         b = nxt_unit_buf_next(b);
     }
 
@@ -2466,19 +2519,78 @@ nxt_unit_request_readline_size(nxt_unit_request_info_t *req, size_t max_size)
 }
 
 
+static nxt_unit_mmap_buf_t *
+nxt_unit_request_preread(nxt_unit_request_info_t *req, size_t size)
+{
+    ssize_t              res;
+    nxt_unit_mmap_buf_t  *mmap_buf;
+
+    if (req->content_fd == -1) {
+        nxt_unit_req_alert(req, "preread: content_fd == -1");
+        return NULL;
+    }
+
+    mmap_buf = nxt_unit_mmap_buf_get(req->ctx);
+    if (nxt_slow_path(mmap_buf == NULL)) {
+        nxt_unit_req_alert(req, "preread: failed to allocate buf");
+        return NULL;
+    }
+
+    mmap_buf->free_ptr = malloc(size);
+    if (nxt_slow_path(mmap_buf->free_ptr == NULL)) {
+        nxt_unit_req_alert(req, "preread: failed to allocate buf memory");
+        nxt_unit_mmap_buf_release(mmap_buf);
+        return NULL;
+    }
+
+    mmap_buf->plain_ptr = mmap_buf->free_ptr;
+
+    mmap_buf->hdr = NULL;
+    mmap_buf->buf.start = mmap_buf->free_ptr;
+    mmap_buf->buf.free = mmap_buf->buf.start;
+    mmap_buf->buf.end = mmap_buf->buf.start + size;
+    mmap_buf->process = NULL;
+
+    res = read(req->content_fd, mmap_buf->free_ptr, size);
+    if (res < 0) {
+        nxt_unit_req_alert(req, "failed to read content: %s (%d)",
+                           strerror(errno), errno);
+
+        nxt_unit_mmap_buf_free(mmap_buf);
+
+        return NULL;
+    }
+
+    if (res < (ssize_t) size) {
+        close(req->content_fd);
+
+        req->content_fd = -1;
+    }
+
+    nxt_unit_req_debug(req, "preread: read %d", (int) res);
+
+    mmap_buf->buf.end = mmap_buf->buf.free + res;
+
+    return mmap_buf;
+}
+
+
 static ssize_t
 nxt_unit_buf_read(nxt_unit_buf_t **b, uint64_t *len, void *dst, size_t size)
 {
     u_char          *p;
     size_t          rest, copy, read;
-    nxt_unit_buf_t  *buf;
+    nxt_unit_buf_t  *buf, *last_buf;
 
     p = dst;
     rest = size;
 
     buf = *b;
+    last_buf = buf;
 
     while (buf != NULL) {
+        last_buf = buf;
+
         copy = buf->end - buf->free;
         copy = nxt_min(rest, copy);
 
@@ -2498,7 +2610,7 @@ nxt_unit_buf_read(nxt_unit_buf_t **b, uint64_t *len, void *dst, size_t size)
         buf = nxt_unit_buf_next(buf);
     }
 
-    *b = buf;
+    *b = last_buf;
 
     read = size - rest;
 

src/nxt_unit.h

@@ -103,6 +103,7 @@ struct nxt_unit_request_info_s {
 
     nxt_unit_buf_t        *content_buf;
     uint64_t              content_length;
+    int                   content_fd;
 
     void                  *data;
 };

test/unit/main.py

@@ -219,6 +219,7 @@ class TestUnit(unittest.TestCase):
                     '--pid',      self.testdir + '/unit.pid',
                     '--log',      self.testdir + '/unit.log',
                     '--control',  'unix:' + self.testdir + '/control.unit.sock',
+                    '--tmp',      self.testdir,
                 ],
                 stderr=log,
             )