Mercurial > hg > nginx

comparison src/http/ngx_http_upstream.c @ 6050:a08fad30aeac

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Request body: unbuffered reading. The r->request_body_no_buffering flag was introduced. It instructs client request body reading code to avoid reading the whole body, and to call post_handler early instead. The caller should use the ngx_http_read_unbuffered_request_body() function to read remaining parts of the body. Upstream module is now able to use this mode, if configured with the proxy_request_buffering directive.
author Maxim Dounin <mdounin@mdounin.ru>
date Mon, 23 Mar 2015 21:09:19 +0300
parents c8acea7c7041
children 643f2ce02f1c
comparison
equal deleted inserted replaced
6049:42d9beeb22db 6050:a08fad30aeac
34 static void ngx_http_upstream_connect(ngx_http_request_t *r, 34 static void ngx_http_upstream_connect(ngx_http_request_t *r,
35 ngx_http_upstream_t *u); 35 ngx_http_upstream_t *u);
36 static ngx_int_t ngx_http_upstream_reinit(ngx_http_request_t *r, 36 static ngx_int_t ngx_http_upstream_reinit(ngx_http_request_t *r,
37 ngx_http_upstream_t *u); 37 ngx_http_upstream_t *u);
38 static void ngx_http_upstream_send_request(ngx_http_request_t *r, 38 static void ngx_http_upstream_send_request(ngx_http_request_t *r,
39 ngx_http_upstream_t *u); 39 ngx_http_upstream_t *u, ngx_uint_t do_write);
40 static ngx_int_t ngx_http_upstream_send_request_body(ngx_http_request_t *r,
41 ngx_http_upstream_t *u, ngx_uint_t do_write);
40 static void ngx_http_upstream_send_request_handler(ngx_http_request_t *r, 42 static void ngx_http_upstream_send_request_handler(ngx_http_request_t *r,
41 ngx_http_upstream_t *u); 43 ngx_http_upstream_t *u);
44 static void ngx_http_upstream_read_request_handler(ngx_http_request_t *r);
42 static void ngx_http_upstream_process_header(ngx_http_request_t *r, 45 static void ngx_http_upstream_process_header(ngx_http_request_t *r,
43 ngx_http_upstream_t *u); 46 ngx_http_upstream_t *u);
44 static ngx_int_t ngx_http_upstream_test_next(ngx_http_request_t *r, 47 static ngx_int_t ngx_http_upstream_test_next(ngx_http_request_t *r,
45 ngx_http_upstream_t *u); 48 ngx_http_upstream_t *u);
46 static ngx_int_t ngx_http_upstream_intercept_errors(ngx_http_request_t *r, 49 static ngx_int_t ngx_http_upstream_intercept_errors(ngx_http_request_t *r,
566 569
567 u->output.alignment = clcf->directio_alignment; 570 u->output.alignment = clcf->directio_alignment;
568 u->output.pool = r->pool; 571 u->output.pool = r->pool;
569 u->output.bufs.num = 1; 572 u->output.bufs.num = 1;
570 u->output.bufs.size = clcf->client_body_buffer_size; 573 u->output.bufs.size = clcf->client_body_buffer_size;
571 u->output.output_filter = ngx_chain_writer; 574
572 u->output.filter_ctx = &u->writer; 575 if (u->output.output_filter == NULL) {
576 u->output.output_filter = ngx_chain_writer;
577 u->output.filter_ctx = &u->writer;
578 }
573 579
574 u->writer.pool = r->pool; 580 u->writer.pool = r->pool;
575 581
576 if (r->upstream_states == NULL) { 582 if (r->upstream_states == NULL) {
577 583
1430 return; 1436 return;
1431 } 1437 }
1432 1438
1433 #endif 1439 #endif
1434 1440
1435 ngx_http_upstream_send_request(r, u); 1441 ngx_http_upstream_send_request(r, u, 1);
1436 } 1442 }
1437 1443
1438 1444
1439 #if (NGX_HTTP_SSL) 1445 #if (NGX_HTTP_SSL)
1440 1446
1534 c->write->handler = ngx_http_upstream_handler; 1540 c->write->handler = ngx_http_upstream_handler;
1535 c->read->handler = ngx_http_upstream_handler; 1541 c->read->handler = ngx_http_upstream_handler;
1536 1542
1537 c = r->connection; 1543 c = r->connection;
1538 1544
1539 ngx_http_upstream_send_request(r, u); 1545 ngx_http_upstream_send_request(r, u, 1);
1540 1546
1541 ngx_http_run_posted_requests(c); 1547 ngx_http_run_posted_requests(c);
1542 return; 1548 return;
1543 } 1549 }
1544 1550
1722 return NGX_OK; 1728 return NGX_OK;
1723 } 1729 }
1724 1730
1725 1731
1726 static void 1732 static void
1727 ngx_http_upstream_send_request(ngx_http_request_t *r, ngx_http_upstream_t *u) 1733 ngx_http_upstream_send_request(ngx_http_request_t *r, ngx_http_upstream_t *u,
1734 ngx_uint_t do_write)
1728 { 1735 {
1729 ngx_int_t rc; 1736 ngx_int_t rc;
1730 ngx_connection_t *c; 1737 ngx_connection_t *c;
1731 1738
1732 c = u->peer.connection; 1739 c = u->peer.connection;
1739 return; 1746 return;
1740 } 1747 }
1741 1748
1742 c->log->action = "sending request to upstream"; 1749 c->log->action = "sending request to upstream";
1743 1750
1744 rc = ngx_output_chain(&u->output, u->request_sent ? NULL : u->request_bufs); 1751 rc = ngx_http_upstream_send_request_body(r, u, do_write);
1745
1746 u->request_sent = 1;
1747 1752
1748 if (rc == NGX_ERROR) { 1753 if (rc == NGX_ERROR) {
1749 ngx_http_upstream_next(r, u, NGX_HTTP_UPSTREAM_FT_ERROR); 1754 ngx_http_upstream_next(r, u, NGX_HTTP_UPSTREAM_FT_ERROR);
1750 return; 1755 return;
1751 } 1756 }
1752 1757
1753 if (c->write->timer_set) { 1758 if (rc >= NGX_HTTP_SPECIAL_RESPONSE) {
1754 ngx_del_timer(c->write); 1759 ngx_http_upstream_finalize_request(r, u, rc);
1760 return;
1755 } 1761 }
1756 1762
1757 if (rc == NGX_AGAIN) { 1763 if (rc == NGX_AGAIN) {
1758 ngx_add_timer(c->write, u->conf->send_timeout); 1764 if (!c->write->ready) {
1765 ngx_add_timer(c->write, u->conf->send_timeout);
1766
1767 } else if (c->write->timer_set) {
1768 ngx_del_timer(c->write);
1769 }
1759 1770
1760 if (ngx_handle_write_event(c->write, u->conf->send_lowat) != NGX_OK) { 1771 if (ngx_handle_write_event(c->write, u->conf->send_lowat) != NGX_OK) {
1761 ngx_http_upstream_finalize_request(r, u, 1772 ngx_http_upstream_finalize_request(r, u,
1762 NGX_HTTP_INTERNAL_SERVER_ERROR); 1773 NGX_HTTP_INTERNAL_SERVER_ERROR);
1763 return; 1774 return;
1765 1776
1766 return; 1777 return;
1767 } 1778 }
1768 1779
1769 /* rc == NGX_OK */ 1780 /* rc == NGX_OK */
1781
1782 if (c->write->timer_set) {
1783 ngx_del_timer(c->write);
1784 }
1770 1785
1771 if (c->tcp_nopush == NGX_TCP_NOPUSH_SET) { 1786 if (c->tcp_nopush == NGX_TCP_NOPUSH_SET) {
1772 if (ngx_tcp_push(c->fd) == NGX_ERROR) { 1787 if (ngx_tcp_push(c->fd) == NGX_ERROR) {
1773 ngx_log_error(NGX_LOG_CRIT, c->log, ngx_socket_errno, 1788 ngx_log_error(NGX_LOG_CRIT, c->log, ngx_socket_errno,
1774 ngx_tcp_push_n " failed"); 1789 ngx_tcp_push_n " failed");
1795 return; 1810 return;
1796 } 1811 }
1797 } 1812 }
1798 1813
1799 1814
1815 static ngx_int_t
1816 ngx_http_upstream_send_request_body(ngx_http_request_t *r,
1817 ngx_http_upstream_t *u, ngx_uint_t do_write)
1818 {
1819 int tcp_nodelay;
1820 ngx_int_t rc;
1821 ngx_chain_t *out, *cl, *ln;
1822 ngx_connection_t *c;
1823 ngx_http_core_loc_conf_t *clcf;
1824
1825 ngx_log_debug0(NGX_LOG_DEBUG_HTTP, r->connection->log, 0,
1826 "http upstream send request body");
1827
1828 if (!r->request_body_no_buffering) {
1829
1830 /* buffered request body */
1831
1832 if (!u->request_sent) {
1833 u->request_sent = 1;
1834 out = u->request_bufs;
1835
1836 } else {
1837 out = NULL;
1838 }
1839
1840 return ngx_output_chain(&u->output, out);
1841 }
1842
1843 if (!u->request_sent) {
1844 u->request_sent = 1;
1845 out = u->request_bufs;
1846
1847 if (r->request_body->bufs) {
1848 for (cl = out; cl->next; cl = out->next) { /* void */ }
1849 cl->next = r->request_body->bufs;
1850 r->request_body->bufs = NULL;
1851 }
1852
1853 c = u->peer.connection;
1854 clcf = ngx_http_get_module_loc_conf(r, ngx_http_core_module);
1855
1856 if (clcf->tcp_nodelay && c->tcp_nodelay == NGX_TCP_NODELAY_UNSET) {
1857 ngx_log_debug0(NGX_LOG_DEBUG_HTTP, c->log, 0, "tcp_nodelay");
1858
1859 tcp_nodelay = 1;
1860
1861 if (setsockopt(c->fd, IPPROTO_TCP, TCP_NODELAY,
1862 (const void *) &tcp_nodelay, sizeof(int)) == -1)
1863 {
1864 ngx_connection_error(c, ngx_socket_errno,
1865 "setsockopt(TCP_NODELAY) failed");
1866 return NGX_ERROR;
1867 }
1868
1869 c->tcp_nodelay = NGX_TCP_NODELAY_SET;
1870 }
1871
1872 r->read_event_handler = ngx_http_upstream_read_request_handler;
1873
1874 } else {
1875 out = NULL;
1876 }
1877
1878 for ( ;; ) {
1879
1880 if (do_write) {
1881 rc = ngx_output_chain(&u->output, out);
1882
1883 if (rc == NGX_ERROR) {
1884 return NGX_ERROR;
1885 }
1886
1887 while (out) {
1888 ln = out;
1889 out = out->next;
1890 ngx_free_chain(r->pool, ln);
1891 }
1892
1893 if (rc == NGX_OK && !r->reading_body) {
1894 break;
1895 }
1896 }
1897
1898 if (r->reading_body) {
1899 /* read client request body */
1900
1901 rc = ngx_http_read_unbuffered_request_body(r);
1902
1903 if (rc >= NGX_HTTP_SPECIAL_RESPONSE) {
1904 return rc;
1905 }
1906
1907 out = r->request_body->bufs;
1908 r->request_body->bufs = NULL;
1909 }
1910
1911 /* stop if there is nothing to send */
1912
1913 if (out == NULL) {
1914 rc = NGX_AGAIN;
1915 break;
1916 }
1917
1918 do_write = 1;
1919 }
1920
1921 if (!r->reading_body) {
1922 if (!u->store && !r->post_action && !u->conf->ignore_client_abort) {
1923 r->read_event_handler =
1924 ngx_http_upstream_rd_check_broken_connection;
1925 }
1926 }
1927
1928 return rc;
1929 }
1930
1931
1800 static void 1932 static void
1801 ngx_http_upstream_send_request_handler(ngx_http_request_t *r, 1933 ngx_http_upstream_send_request_handler(ngx_http_request_t *r,
1802 ngx_http_upstream_t *u) 1934 ngx_http_upstream_t *u)
1803 { 1935 {
1804 ngx_connection_t *c; 1936 ngx_connection_t *c;
1828 (void) ngx_handle_write_event(c->write, 0); 1960 (void) ngx_handle_write_event(c->write, 0);
1829 1961
1830 return; 1962 return;
1831 } 1963 }
1832 1964
1833 ngx_http_upstream_send_request(r, u); 1965 ngx_http_upstream_send_request(r, u, 1);
1966 }
1967
1968
1969 static void
1970 ngx_http_upstream_read_request_handler(ngx_http_request_t *r)
1971 {
1972 ngx_connection_t *c;
1973 ngx_http_upstream_t *u;
1974
1975 c = r->connection;
1976 u = r->upstream;
1977
1978 ngx_log_debug0(NGX_LOG_DEBUG_HTTP, r->connection->log, 0,
1979 "http upstream read request handler");
1980
1981 if (c->read->timedout) {
1982 c->timedout = 1;
1983 ngx_http_upstream_finalize_request(r, u, NGX_HTTP_REQUEST_TIME_OUT);
1984 return;
1985 }
1986
1987 ngx_http_upstream_send_request(r, u, 0);
1834 } 1988 }
1835 1989
1836 1990
1837 static void 1991 static void
1838 ngx_http_upstream_process_header(ngx_http_request_t *r, ngx_http_upstream_t *u) 1992 ngx_http_upstream_process_header(ngx_http_request_t *r, ngx_http_upstream_t *u)
3624 if (ft_type == NGX_HTTP_UPSTREAM_FT_TIMEOUT) { 3778 if (ft_type == NGX_HTTP_UPSTREAM_FT_TIMEOUT) {
3625 ngx_log_error(NGX_LOG_ERR, r->connection->log, NGX_ETIMEDOUT, 3779 ngx_log_error(NGX_LOG_ERR, r->connection->log, NGX_ETIMEDOUT,
3626 "upstream timed out"); 3780 "upstream timed out");
3627 } 3781 }
3628 3782
3629 if (u->peer.cached && ft_type == NGX_HTTP_UPSTREAM_FT_ERROR) { 3783 if (u->peer.cached && ft_type == NGX_HTTP_UPSTREAM_FT_ERROR
3784 && (!u->request_sent || !r->request_body_no_buffering))
3785 {
3630 status = 0; 3786 status = 0;
3631 3787
3632 /* TODO: inform balancer instead */ 3788 /* TODO: inform balancer instead */
3633 3789
3634 u->peer.tries++; 3790 u->peer.tries++;
3672 u->state->status = status; 3828 u->state->status = status;
3673 timeout = u->conf->next_upstream_timeout; 3829 timeout = u->conf->next_upstream_timeout;
3674 3830
3675 if (u->peer.tries == 0 3831 if (u->peer.tries == 0
3676 || !(u->conf->next_upstream & ft_type) 3832 || !(u->conf->next_upstream & ft_type)
3833 || (u->request_sent && r->request_body_no_buffering)
3677 || (timeout && ngx_current_msec - u->peer.start_time >= timeout)) 3834 || (timeout && ngx_current_msec - u->peer.start_time >= timeout))
3678 { 3835 {
3679 #if (NGX_HTTP_CACHE) 3836 #if (NGX_HTTP_CACHE)
3680 3837
3681 if (u->cache_status == NGX_HTTP_CACHE_EXPIRED 3838 if (u->cache_status == NGX_HTTP_CACHE_EXPIRED