nxt_router.c (198:6c83f6294893) nxt_router.c (206:86a529b2ea9b)
1
2/*
3 * Copyright (C) Igor Sysoev
4 * Copyright (C) Valentin V. Bartenev
5 * Copyright (C) NGINX, Inc.
6 */
7
8#include <nxt_router.h>

--- 114 unchanged lines hidden (view full) ---

123static nxt_bool_t nxt_router_app_free(nxt_task_t *task, nxt_app_t *app);
124static nxt_port_t * nxt_router_app_get_port(nxt_app_t *app, uint32_t req_id);
125static void nxt_router_app_release_port(nxt_task_t *task, void *obj,
126 void *data);
127
128static void nxt_router_conn_init(nxt_task_t *task, void *obj, void *data);
129static void nxt_router_conn_http_header_parse(nxt_task_t *task, void *obj,
130 void *data);
1
2/*
3 * Copyright (C) Igor Sysoev
4 * Copyright (C) Valentin V. Bartenev
5 * Copyright (C) NGINX, Inc.
6 */
7
8#include <nxt_router.h>

--- 114 unchanged lines hidden (view full) ---

123static nxt_bool_t nxt_router_app_free(nxt_task_t *task, nxt_app_t *app);
124static nxt_port_t * nxt_router_app_get_port(nxt_app_t *app, uint32_t req_id);
125static void nxt_router_app_release_port(nxt_task_t *task, void *obj,
126 void *data);
127
128static void nxt_router_conn_init(nxt_task_t *task, void *obj, void *data);
129static void nxt_router_conn_http_header_parse(nxt_task_t *task, void *obj,
130 void *data);
131static void nxt_router_conn_http_body_read(nxt_task_t *task, void *obj,
132 void *data);
131static void nxt_router_process_http_request(nxt_task_t *task,
132 nxt_conn_t *c, nxt_app_parse_ctx_t *ap);
133static void nxt_router_process_http_request_mp(nxt_task_t *task,
134 nxt_req_app_link_t *ra, nxt_port_t *port);
135static void nxt_router_conn_ready(nxt_task_t *task, void *obj, void *data);
136static void nxt_router_conn_close(nxt_task_t *task, void *obj, void *data);
137static void nxt_router_conn_free(nxt_task_t *task, void *obj, void *data);
138static void nxt_router_conn_error(nxt_task_t *task, void *obj, void *data);

--- 460 unchanged lines hidden (view full) ---

599
600 {
601 nxt_string("large_header_buffer_size"),
602 NXT_CONF_MAP_SIZE,
603 offsetof(nxt_socket_conf_t, large_header_buffer_size),
604 },
605
606 {
133static void nxt_router_process_http_request(nxt_task_t *task,
134 nxt_conn_t *c, nxt_app_parse_ctx_t *ap);
135static void nxt_router_process_http_request_mp(nxt_task_t *task,
136 nxt_req_app_link_t *ra, nxt_port_t *port);
137static void nxt_router_conn_ready(nxt_task_t *task, void *obj, void *data);
138static void nxt_router_conn_close(nxt_task_t *task, void *obj, void *data);
139static void nxt_router_conn_free(nxt_task_t *task, void *obj, void *data);
140static void nxt_router_conn_error(nxt_task_t *task, void *obj, void *data);

--- 460 unchanged lines hidden (view full) ---

601
602 {
603 nxt_string("large_header_buffer_size"),
604 NXT_CONF_MAP_SIZE,
605 offsetof(nxt_socket_conf_t, large_header_buffer_size),
606 },
607
608 {
609 nxt_string("large_header_buffers"),
610 NXT_CONF_MAP_SIZE,
611 offsetof(nxt_socket_conf_t, large_header_buffers),
612 },
613
614 {
615 nxt_string("body_buffer_size"),
616 NXT_CONF_MAP_SIZE,
617 offsetof(nxt_socket_conf_t, body_buffer_size),
618 },
619
620 {
621 nxt_string("max_body_size"),
622 NXT_CONF_MAP_SIZE,
623 offsetof(nxt_socket_conf_t, max_body_size),
624 },
625
626 {
607 nxt_string("header_read_timeout"),
608 NXT_CONF_MAP_MSEC,
609 offsetof(nxt_socket_conf_t, header_read_timeout),
610 },
627 nxt_string("header_read_timeout"),
628 NXT_CONF_MAP_MSEC,
629 offsetof(nxt_socket_conf_t, header_read_timeout),
630 },
631
632 {
633 nxt_string("body_read_timeout"),
634 NXT_CONF_MAP_MSEC,
635 offsetof(nxt_socket_conf_t, body_read_timeout),
636 },
611};
612
613
614static nxt_int_t
615nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
616 u_char *start, u_char *end)
617{
618 u_char *p;

--- 168 unchanged lines hidden (view full) ---

787 goto fail;
788 }
789
790 nxt_debug(task, "application: %V", &lscf.application);
791
792 // STUB, default values if http block is not defined.
793 skcf->header_buffer_size = 2048;
794 skcf->large_header_buffer_size = 8192;
637};
638
639
640static nxt_int_t
641nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
642 u_char *start, u_char *end)
643{
644 u_char *p;

--- 168 unchanged lines hidden (view full) ---

813 goto fail;
814 }
815
816 nxt_debug(task, "application: %V", &lscf.application);
817
818 // STUB, default values if http block is not defined.
819 skcf->header_buffer_size = 2048;
820 skcf->large_header_buffer_size = 8192;
821 skcf->large_header_buffers = 4;
822 skcf->body_buffer_size = 16 * 1024;
823 skcf->max_body_size = 2 * 1024 * 1024;
795 skcf->header_read_timeout = 5000;
824 skcf->header_read_timeout = 5000;
825 skcf->body_read_timeout = 5000;
796
797 if (http != NULL) {
798 ret = nxt_conf_map_object(http, nxt_router_http_conf,
799 nxt_nitems(nxt_router_http_conf), skcf);
800 if (ret != NXT_OK) {
801 nxt_log(task, NXT_LOG_CRIT, "http map error");
802 goto fail;
803 }

--- 998 unchanged lines hidden (view full) ---

1802 nxt_mp_destroy(engine->mem_pool);
1803
1804 nxt_event_engine_free(engine);
1805
1806 nxt_free(link);
1807}
1808
1809
826
827 if (http != NULL) {
828 ret = nxt_conf_map_object(http, nxt_router_http_conf,
829 nxt_nitems(nxt_router_http_conf), skcf);
830 if (ret != NXT_OK) {
831 nxt_log(task, NXT_LOG_CRIT, "http map error");
832 goto fail;
833 }

--- 998 unchanged lines hidden (view full) ---

1832 nxt_mp_destroy(engine->mem_pool);
1833
1834 nxt_event_engine_free(engine);
1835
1836 nxt_free(link);
1837}
1838
1839
1810static const nxt_conn_state_t nxt_router_conn_read_state
1840static const nxt_conn_state_t nxt_router_conn_read_header_state
1811 nxt_aligned(64) =
1812{
1813 .ready_handler = nxt_router_conn_http_header_parse,
1814 .close_handler = nxt_router_conn_close,
1815 .error_handler = nxt_router_conn_error,
1816
1817 .timer_handler = nxt_router_conn_timeout,
1818 .timer_value = nxt_router_conn_timeout_value,
1819 .timer_data = offsetof(nxt_socket_conf_t, header_read_timeout),
1820};
1821
1822
1841 nxt_aligned(64) =
1842{
1843 .ready_handler = nxt_router_conn_http_header_parse,
1844 .close_handler = nxt_router_conn_close,
1845 .error_handler = nxt_router_conn_error,
1846
1847 .timer_handler = nxt_router_conn_timeout,
1848 .timer_value = nxt_router_conn_timeout_value,
1849 .timer_data = offsetof(nxt_socket_conf_t, header_read_timeout),
1850};
1851
1852
1853static const nxt_conn_state_t nxt_router_conn_read_body_state
1854 nxt_aligned(64) =
1855{
1856 .ready_handler = nxt_router_conn_http_body_read,
1857 .close_handler = nxt_router_conn_close,
1858 .error_handler = nxt_router_conn_error,
1859
1860 .timer_handler = nxt_router_conn_timeout,
1861 .timer_value = nxt_router_conn_timeout_value,
1862 .timer_data = offsetof(nxt_socket_conf_t, body_read_timeout),
1863 .timer_autoreset = 1,
1864};
1865
1866
1823static void
1824nxt_router_conn_init(nxt_task_t *task, void *obj, void *data)
1825{
1826 size_t size;
1827 nxt_conn_t *c;
1828 nxt_event_engine_t *engine;
1829 nxt_socket_conf_joint_t *joint;
1830

--- 8 unchanged lines hidden (view full) ---

1839 c->read = nxt_buf_mem_alloc(c->mem_pool, size, 0);
1840
1841 c->socket.data = NULL;
1842
1843 engine = task->thread->engine;
1844 c->read_work_queue = &engine->fast_work_queue;
1845 c->write_work_queue = &engine->fast_work_queue;
1846
1867static void
1868nxt_router_conn_init(nxt_task_t *task, void *obj, void *data)
1869{
1870 size_t size;
1871 nxt_conn_t *c;
1872 nxt_event_engine_t *engine;
1873 nxt_socket_conf_joint_t *joint;
1874

--- 8 unchanged lines hidden (view full) ---

1883 c->read = nxt_buf_mem_alloc(c->mem_pool, size, 0);
1884
1885 c->socket.data = NULL;
1886
1887 engine = task->thread->engine;
1888 c->read_work_queue = &engine->fast_work_queue;
1889 c->write_work_queue = &engine->fast_work_queue;
1890
1847 c->read_state = &nxt_router_conn_read_state;
1891 c->read_state = &nxt_router_conn_read_header_state;
1848
1849 nxt_conn_read(engine, c);
1850}
1851
1852
1853static const nxt_conn_state_t nxt_router_conn_write_state
1854 nxt_aligned(64) =
1855{

--- 12 unchanged lines hidden (view full) ---

1868 nxt_req_conn_link_t *rc;
1869 nxt_event_engine_t *engine;
1870
1871 b = msg->buf;
1872 engine = task->thread->engine;
1873
1874 rc = nxt_event_engine_request_find(engine, msg->port_msg.stream);
1875 if (nxt_slow_path(rc == NULL)) {
1892
1893 nxt_conn_read(engine, c);
1894}
1895
1896
1897static const nxt_conn_state_t nxt_router_conn_write_state
1898 nxt_aligned(64) =
1899{

--- 12 unchanged lines hidden (view full) ---

1912 nxt_req_conn_link_t *rc;
1913 nxt_event_engine_t *engine;
1914
1915 b = msg->buf;
1916 engine = task->thread->engine;
1917
1918 rc = nxt_event_engine_request_find(engine, msg->port_msg.stream);
1919 if (nxt_slow_path(rc == NULL)) {
1876
1877 nxt_debug(task, "request id %08uxD not found", msg->port_msg.stream);
1878
1879 return;
1880 }
1881
1882 c = rc->conn;
1883
1884 dump_size = nxt_buf_used_size(b);

--- 20 unchanged lines hidden (view full) ---

1905
1906 nxt_buf_chain_add(&b, last);
1907
1908 if (rc->app_port != NULL) {
1909 nxt_router_app_release_port(task, rc->app_port, rc->app_port->app);
1910
1911 rc->app_port = NULL;
1912 }
1920 nxt_debug(task, "request id %08uxD not found", msg->port_msg.stream);
1921
1922 return;
1923 }
1924
1925 c = rc->conn;
1926
1927 dump_size = nxt_buf_used_size(b);

--- 20 unchanged lines hidden (view full) ---

1948
1949 nxt_buf_chain_add(&b, last);
1950
1951 if (rc->app_port != NULL) {
1952 nxt_router_app_release_port(task, rc->app_port, rc->app_port->app);
1953
1954 rc->app_port = NULL;
1955 }
1956
1957 rc->conn = NULL;
1913 }
1914
1915 if (b == NULL) {
1916 return;
1917 }
1918
1958 }
1959
1960 if (b == NULL) {
1961 return;
1962 }
1963
1919 /* Disable instant buffer completion/re-using by port. */
1920 msg->buf = NULL;
1964 if (msg->buf == b) {
1965 /* Disable instant buffer completion/re-using by port. */
1966 msg->buf = NULL;
1967 }
1921
1922 if (c->write == NULL) {
1923 c->write = b;
1924 c->write_state = &nxt_router_conn_write_state;
1925
1926 nxt_conn_write(task->thread->engine, c);
1927 } else {
1928 nxt_debug(task, "router data attach out bufs to existing chain");

--- 4 unchanged lines hidden (view full) ---

1933
1934nxt_inline const char *
1935nxt_router_text_by_code(int code)
1936{
1937 switch (code) {
1938 case 400: return "Bad request";
1939 case 404: return "Not found";
1940 case 403: return "Forbidden";
1968
1969 if (c->write == NULL) {
1970 c->write = b;
1971 c->write_state = &nxt_router_conn_write_state;
1972
1973 nxt_conn_write(task->thread->engine, c);
1974 } else {
1975 nxt_debug(task, "router data attach out bufs to existing chain");

--- 4 unchanged lines hidden (view full) ---

1980
1981nxt_inline const char *
1982nxt_router_text_by_code(int code)
1983{
1984 switch (code) {
1985 case 400: return "Bad request";
1986 case 404: return "Not found";
1987 case 403: return "Forbidden";
1988 case 408: return "Request Timeout";
1989 case 411: return "Length Required";
1990 case 413: return "Request Entity Too Large";
1941 case 500:
1942 default: return "Internal server error";
1943 }
1944}
1945
1946
1947static nxt_buf_t *
1948nxt_router_get_error_buf(nxt_task_t *task, nxt_mp_t *mp, int code,

--- 11 unchanged lines hidden (view full) ---

1960 "HTTP/1.0 %d %s\r\n"
1961 "Content-Type: text/plain\r\n"
1962 "Connection: close\r\n\r\n",
1963 code, nxt_router_text_by_code(code));
1964
1965 msg = (const char *) b->mem.free;
1966
1967 b->mem.free = nxt_vsprintf(b->mem.free, b->mem.end, fmt, args);
1991 case 500:
1992 default: return "Internal server error";
1993 }
1994}
1995
1996
1997static nxt_buf_t *
1998nxt_router_get_error_buf(nxt_task_t *task, nxt_mp_t *mp, int code,

--- 11 unchanged lines hidden (view full) ---

2010 "HTTP/1.0 %d %s\r\n"
2011 "Content-Type: text/plain\r\n"
2012 "Connection: close\r\n\r\n",
2013 code, nxt_router_text_by_code(code));
2014
2015 msg = (const char *) b->mem.free;
2016
2017 b->mem.free = nxt_vsprintf(b->mem.free, b->mem.end, fmt, args);
2018 b->mem.free[0] = '\0';
1968
1969 nxt_log_alert(task->log, "error %d: %s", code, msg);
1970
1971 last = nxt_buf_mem_ts_alloc(task, mp, 0);
1972
1973 if (nxt_slow_path(last == NULL)) {
1974 nxt_mp_release(mp, b);
1975 return NULL;

--- 15 unchanged lines hidden (view full) ---

1991{
1992 va_list args;
1993 nxt_buf_t *b;
1994
1995 va_start(args, fmt);
1996 b = nxt_router_get_error_buf(task, c->mem_pool, code, fmt, args);
1997 va_end(args);
1998
2019
2020 nxt_log_alert(task->log, "error %d: %s", code, msg);
2021
2022 last = nxt_buf_mem_ts_alloc(task, mp, 0);
2023
2024 if (nxt_slow_path(last == NULL)) {
2025 nxt_mp_release(mp, b);
2026 return NULL;

--- 15 unchanged lines hidden (view full) ---

2042{
2043 va_list args;
2044 nxt_buf_t *b;
2045
2046 va_start(args, fmt);
2047 b = nxt_router_get_error_buf(task, c->mem_pool, code, fmt, args);
2048 va_end(args);
2049
2050 if (c->socket.data != NULL) {
2051 nxt_mp_free(c->mem_pool, c->socket.data);
2052 c->socket.data = NULL;
2053 }
2054
1999 if (c->write == NULL) {
2000 c->write = b;
2001 c->write_state = &nxt_router_conn_write_state;
2002
2003 nxt_conn_write(task->thread->engine, c);
2004 } else {
2005 nxt_debug(task, "router data attach out bufs to existing chain");
2006

--- 333 unchanged lines hidden (view full) ---

2340
2341 return NXT_AGAIN;
2342}
2343
2344
2345static void
2346nxt_router_conn_http_header_parse(nxt_task_t *task, void *obj, void *data)
2347{
2055 if (c->write == NULL) {
2056 c->write = b;
2057 c->write_state = &nxt_router_conn_write_state;
2058
2059 nxt_conn_write(task->thread->engine, c);
2060 } else {
2061 nxt_debug(task, "router data attach out bufs to existing chain");
2062

--- 333 unchanged lines hidden (view full) ---

2396
2397 return NXT_AGAIN;
2398}
2399
2400
2401static void
2402nxt_router_conn_http_header_parse(nxt_task_t *task, void *obj, void *data)
2403{
2348 size_t size, preread;
2404 size_t size;
2349 nxt_int_t ret;
2405 nxt_int_t ret;
2350 nxt_buf_t *b;
2406 nxt_buf_t *buf;
2351 nxt_conn_t *c;
2352 nxt_app_parse_ctx_t *ap;
2407 nxt_conn_t *c;
2408 nxt_app_parse_ctx_t *ap;
2409 nxt_app_request_body_t *b;
2353 nxt_socket_conf_joint_t *joint;
2354 nxt_app_request_header_t *h;
2355
2356 c = obj;
2357 ap = data;
2410 nxt_socket_conf_joint_t *joint;
2411 nxt_app_request_header_t *h;
2412
2413 c = obj;
2414 ap = data;
2358 b = c->read;
2415 buf = c->read;
2416 joint = c->listen->socket.data;
2359
2360 nxt_debug(task, "router conn http header parse");
2361
2362 if (ap == NULL) {
2417
2418 nxt_debug(task, "router conn http header parse");
2419
2420 if (ap == NULL) {
2363 ap = nxt_mp_zget(c->mem_pool, sizeof(nxt_app_parse_ctx_t));
2421 ap = nxt_mp_zalloc(c->mem_pool, sizeof(nxt_app_parse_ctx_t));
2364 if (nxt_slow_path(ap == NULL)) {
2365 nxt_router_conn_close(task, c, data);
2366 return;
2367 }
2368
2369 ret = nxt_app_http_req_init(task, ap);
2370 if (nxt_slow_path(ret != NXT_OK)) {
2371 nxt_router_conn_close(task, c, data);
2372 return;
2373 }
2374
2375 c->socket.data = ap;
2376
2377 ap->r.remote.start = nxt_sockaddr_address(c->remote);
2378 ap->r.remote.length = c->remote->address_length;
2422 if (nxt_slow_path(ap == NULL)) {
2423 nxt_router_conn_close(task, c, data);
2424 return;
2425 }
2426
2427 ret = nxt_app_http_req_init(task, ap);
2428 if (nxt_slow_path(ret != NXT_OK)) {
2429 nxt_router_conn_close(task, c, data);
2430 return;
2431 }
2432
2433 c->socket.data = ap;
2434
2435 ap->r.remote.start = nxt_sockaddr_address(c->remote);
2436 ap->r.remote.length = c->remote->address_length;
2437
2438 ap->r.header.buf = buf;
2379 }
2380
2381 h = &ap->r.header;
2439 }
2440
2441 h = &ap->r.header;
2442 b = &ap->r.body;
2382
2443
2383 ret = nxt_app_http_req_parse(task, ap, b);
2444 ret = nxt_app_http_req_header_parse(task, ap, buf);
2384
2445
2385 nxt_debug(task, "http parse request: %d", ret);
2446 nxt_debug(task, "http parse request header: %d", ret);
2386
2387 switch (nxt_expect(NXT_DONE, ret)) {
2388
2389 case NXT_DONE:
2447
2448 switch (nxt_expect(NXT_DONE, ret)) {
2449
2450 case NXT_DONE:
2390 preread = nxt_buf_mem_used_size(&b->mem);
2391
2392 nxt_debug(task, "router request header parsing complete, "
2393 "content length: %O, preread: %uz",
2451 nxt_debug(task, "router request header parsing complete, "
2452 "content length: %O, preread: %uz",
2394 h->parsed_content_length, preread);
2453 h->parsed_content_length, nxt_buf_mem_used_size(&buf->mem));
2395
2454
2396 nxt_router_process_http_request(task, c, ap);
2397 return;
2455 if (b->done) {
2456 nxt_router_process_http_request(task, c, ap);
2398
2457
2458 return;
2459 }
2460
2461 if (joint->socket_conf->max_body_size > 0 &&
2462 (size_t) h->parsed_content_length >
2463 joint->socket_conf->max_body_size) {
2464
2465 nxt_router_gen_error(task, c, 413, "Content-Length too big");
2466 return;
2467 }
2468
2469 if (nxt_buf_mem_free_size(&buf->mem) == 0) {
2470 size = nxt_min(joint->socket_conf->body_buffer_size,
2471 (size_t) h->parsed_content_length);
2472
2473 buf->next = nxt_buf_mem_alloc(c->mem_pool, size, 0);
2474 if (nxt_slow_path(buf->next == NULL)) {
2475 nxt_router_gen_error(task, c, 500, "Failed to allocate "
2476 "buffer for request body");
2477 return;
2478 }
2479
2480 c->read = buf->next;
2481
2482 b->preread_size += nxt_buf_mem_used_size(&buf->mem);
2483 }
2484
2485 if (b->buf == NULL) {
2486 b->buf = c->read;
2487 }
2488
2489 c->read_state = &nxt_router_conn_read_body_state;
2490 break;
2491
2399 case NXT_ERROR:
2492 case NXT_ERROR:
2400 nxt_router_conn_close(task, c, data);
2493 nxt_router_gen_error(task, c, 400, "Request header parse error");
2401 return;
2402
2403 default: /* NXT_AGAIN */
2404
2494 return;
2495
2496 default: /* NXT_AGAIN */
2497
2405 if (h->done == 0) {
2498 if (c->read->mem.free == c->read->mem.end) {
2499 size = joint->socket_conf->large_header_buffer_size;
2406
2500
2407 if (c->read->mem.free == c->read->mem.end) {
2408 joint = c->listen->socket.data;
2409 size = joint->socket_conf->large_header_buffer_size;
2501 if (size <= (size_t) nxt_buf_mem_used_size(&buf->mem) ||
2502 ap->r.header.bufs >= joint->socket_conf->large_header_buffers) {
2503 nxt_router_gen_error(task, c, 413,
2504 "Too long request headers");
2505 return;
2506 }
2410
2507
2411 if (size > (size_t) nxt_buf_mem_size(&b->mem)) {
2412 b = nxt_buf_mem_alloc(c->mem_pool, size, 0);
2413 if (nxt_slow_path(b == NULL)) {
2414 nxt_router_conn_close(task, c, data);
2415 return;
2416 }
2508 buf->next = nxt_buf_mem_alloc(c->mem_pool, size, 0);
2509 if (nxt_slow_path(buf->next == NULL)) {
2510 nxt_router_gen_error(task, c, 500,
2511 "Failed to allocate large header "
2512 "buffer");
2513 return;
2514 }
2417
2515
2418 size = c->read->mem.free - c->read->mem.pos;
2516 ap->r.header.bufs++;
2419
2517
2420 c->read = nxt_buf_cpy(b, c->read->mem.pos, size);
2421 } else {
2422 nxt_router_gen_error(task, c, 400,
2423 "Too long request headers");
2424 return;
2425 }
2426 }
2518 size = c->read->mem.free - c->read->mem.pos;
2519
2520 c->read = nxt_buf_cpy(buf->next, c->read->mem.pos, size);
2427 }
2428
2521 }
2522
2429 if (ap->r.body.done == 0) {
2523 }
2430
2524
2431 preread = nxt_buf_mem_used_size(&b->mem);
2525 nxt_conn_read(task->thread->engine, c);
2526}
2432
2527
2433 if (h->parsed_content_length - preread >
2434 (size_t) nxt_buf_mem_free_size(&b->mem)) {
2435
2528
2436 b = nxt_buf_mem_alloc(c->mem_pool, h->parsed_content_length, 0);
2437 if (nxt_slow_path(b == NULL)) {
2438 nxt_router_gen_error(task, c, 500, "Failed to allocate "
2439 "buffer for request body");
2440 return;
2441 }
2529static void
2530nxt_router_conn_http_body_read(nxt_task_t *task, void *obj, void *data)
2531{
2532 size_t size;
2533 nxt_int_t ret;
2534 nxt_buf_t *buf;
2535 nxt_conn_t *c;
2536 nxt_app_parse_ctx_t *ap;
2537 nxt_app_request_body_t *b;
2538 nxt_socket_conf_joint_t *joint;
2539 nxt_app_request_header_t *h;
2442
2540
2443 c->read = nxt_buf_cpy(b, c->read->mem.pos, preread);
2444 }
2541 c = obj;
2542 ap = data;
2543 buf = c->read;
2445
2544
2446 nxt_debug(task, "router request body read again, rest: %uz",
2447 h->parsed_content_length - preread);
2545 nxt_debug(task, "router conn http body read");
2448
2546
2547 nxt_assert(ap != NULL);
2548
2549 b = &ap->r.body;
2550 h = &ap->r.header;
2551
2552 ret = nxt_app_http_req_body_read(task, ap, buf);
2553
2554 nxt_debug(task, "http read request body: %d", ret);
2555
2556 switch (nxt_expect(NXT_DONE, ret)) {
2557
2558 case NXT_DONE:
2559 nxt_router_process_http_request(task, c, ap);
2560 return;
2561
2562 case NXT_ERROR:
2563 nxt_router_gen_error(task, c, 500, "Read body error");
2564 return;
2565
2566 default: /* NXT_AGAIN */
2567
2568 if (nxt_buf_mem_free_size(&buf->mem) == 0) {
2569 joint = c->listen->socket.data;
2570
2571 b->preread_size += nxt_buf_mem_used_size(&buf->mem);
2572
2573 size = nxt_min(joint->socket_conf->body_buffer_size,
2574 (size_t) h->parsed_content_length - b->preread_size);
2575
2576 buf->next = nxt_buf_mem_alloc(c->mem_pool, size, 0);
2577 if (nxt_slow_path(buf->next == NULL)) {
2578 nxt_router_gen_error(task, c, 500, "Failed to allocate "
2579 "buffer for request body");
2580 return;
2581 }
2582
2583 c->read = buf->next;
2449 }
2450
2584 }
2585
2586 nxt_debug(task, "router request body read again, rest: %uz",
2587 h->parsed_content_length - b->preread_size);
2451 }
2452
2453 nxt_conn_read(task->thread->engine, c);
2454}
2455
2456
2457static void
2458nxt_router_process_http_request(nxt_task_t *task, nxt_conn_t *c,

--- 261 unchanged lines hidden (view full) ---

2720 nxt_timer_t *timer;
2721
2722 timer = obj;
2723
2724 nxt_debug(task, "router conn timeout");
2725
2726 c = nxt_read_timer_conn(timer);
2727
2588 }
2589
2590 nxt_conn_read(task->thread->engine, c);
2591}
2592
2593
2594static void
2595nxt_router_process_http_request(nxt_task_t *task, nxt_conn_t *c,

--- 261 unchanged lines hidden (view full) ---

2857 nxt_timer_t *timer;
2858
2859 timer = obj;
2860
2861 nxt_debug(task, "router conn timeout");
2862
2863 c = nxt_read_timer_conn(timer);
2864
2728 c->write_state = &nxt_router_conn_close_state;
2865 if (c->read_state == &nxt_router_conn_read_header_state) {
2866 nxt_router_gen_error(task, c, 408, "Read header timeout");
2729
2867
2730 nxt_conn_close(task->thread->engine, c);
2868 } else {
2869 nxt_router_gen_error(task, c, 408, "Read body timeout");
2870 }
2731}
2732
2733
2734static nxt_msec_t
2735nxt_router_conn_timeout_value(nxt_conn_t *c, uintptr_t data)
2736{
2737 nxt_socket_conf_joint_t *joint;
2738
2739 joint = c->listen->socket.data;
2740
2741 return nxt_value_at(nxt_msec_t, joint->socket_conf, data);
2742}
2871}
2872
2873
2874static nxt_msec_t
2875nxt_router_conn_timeout_value(nxt_conn_t *c, uintptr_t data)
2876{
2877 nxt_socket_conf_joint_t *joint;
2878
2879 joint = c->listen->socket.data;
2880
2881 return nxt_value_at(nxt_msec_t, joint->socket_conf, data);
2882}