summaryrefslogtreecommitdiffhomepage
path: root/src
diff options
context:
space:
mode:
authorMax Romanov <max.romanov@nginx.com>2020-03-12 17:54:29 +0300
committerMax Romanov <max.romanov@nginx.com>2020-03-12 17:54:29 +0300
commit5296be0b82784eb90abc86339e6c16841e9a9727 (patch)
tree442b315416f9c7ce00b9583ecf6daca07ae68290 /src
parent08b65721e25b1b94affc12078a623a11341525d1 (diff)
downloadunit-5296be0b82784eb90abc86339e6c16841e9a9727.tar.gz
unit-5296be0b82784eb90abc86339e6c16841e9a9727.tar.bz2
Using disk file to store large request body.
This closes #386 on GitHub.
Diffstat (limited to 'src')
-rw-r--r--src/nxt_conf_validation.c10
-rw-r--r--src/nxt_conn_write.c96
-rw-r--r--src/nxt_h1proto.c178
-rw-r--r--src/nxt_http_request.c8
-rw-r--r--src/nxt_router.c37
-rw-r--r--src/nxt_router.h2
-rw-r--r--src/nxt_runtime.c18
-rw-r--r--src/nxt_runtime.h1
-rw-r--r--src/nxt_unit.c126
-rw-r--r--src/nxt_unit.h1
10 files changed, 441 insertions, 36 deletions
diff --git a/src/nxt_conf_validation.c b/src/nxt_conf_validation.c
index 86c1dbcb..3a3654bd 100644
--- a/src/nxt_conf_validation.c
+++ b/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,
diff --git a/src/nxt_conn_write.c b/src/nxt_conn_write.c
index 298d8f75..d7a6a8da 100644
--- a/src/nxt_conn_write.c
+++ b/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)
diff --git a/src/nxt_h1proto.c b/src/nxt_h1proto.c
index c326ef30..35918bd8 100644
--- a/src/nxt_h1proto.c
+++ b/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 = body_length;
+ size = nxt_min(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", size);
+ nxt_debug(task, "h1p body rest: %uz", body_rest);
- if (size != 0) {
+ if (body_rest != 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);
-
- nxt_debug(task, "h1p body rest: %uz", size);
+ r = h1p->request;
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;
- size += nxt_buf_mem_used_size(&body->mem);
+ if (nxt_buf_is_file(r->body)) {
+ 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) {
- r = peer->request;
- r->state->ready_handler(task, r, peer);
+ if (c->write != NULL) {
+ nxt_conn_write(engine, c);
return;
}
- nxt_conn_write(engine, c);
+ r = peer->request;
+ r->state->ready_handler(task, r, peer);
}
diff --git a/src/nxt_http_request.c b/src/nxt_http_request.c
index 51553f0c..72aaa290 100644
--- a/src/nxt_http_request.c
+++ b/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;
diff --git a/src/nxt_router.c b/src/nxt_router.c
index 9138a9a3..a913284c 100644
--- a/src/nxt_router.c
+++ b/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,
- -1, req_app_link->stream, reply_port->id, buf,
+ body = req_app_link->request->body;
+ 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;
}
diff --git a/src/nxt_router.h b/src/nxt_router.h
index 85ef9a6c..08142ce3 100644
--- a/src/nxt_router.h
+++ b/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
diff --git a/src/nxt_runtime.c b/src/nxt_runtime.c
index 80b25c1b..f6d80ccb 100644
--- a/src/nxt_runtime.c
+++ b/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;
diff --git a/src/nxt_runtime.h b/src/nxt_runtime.h
index f8d19ec6..a364c38c 100644
--- a/src/nxt_runtime.h
+++ b/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;
diff --git a/src/nxt_unit.c b/src/nxt_unit.c
index 07717545..7a4124fb 100644
--- a/src/nxt_unit.c
+++ b/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,
- dst, size);
+ ssize_t buf_res, res;
+
+ 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;
- size_t l_size, b_size;
- nxt_unit_buf_t *b;
+ char *p;
+ size_t l_size, b_size;
+ 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;
diff --git a/src/nxt_unit.h b/src/nxt_unit.h
index 8e9e3015..900f3ac2 100644
--- a/src/nxt_unit.h
+++ b/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;
};