Mercurial > hg > nginx

comparison src/http/modules/ngx_http_grpc_module.c @ 7349:f6047a579ca1

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
gRPC: improved keepalive handling. The code is now able to parse additional control frames after the response is received, and can send control frames as well. This fixes keepalive problems as observed with grpc-c, which can send window update and ping frames after the response, see http://mailman.nginx.org/pipermail/nginx/2018-August/056620.html.
author Maxim Dounin <mdounin@mdounin.ru>
date Mon, 03 Sep 2018 19:34:01 +0300
parents 45e513c3540d
children 67c6cb7f477c
comparison
equal deleted inserted replaced
7348:f6e7831a17d4 7349:f6047a579ca1
109 109
110 unsigned header_sent:1; 110 unsigned header_sent:1;
111 unsigned output_closed:1; 111 unsigned output_closed:1;
112 unsigned parsing_headers:1; 112 unsigned parsing_headers:1;
113 unsigned end_stream:1; 113 unsigned end_stream:1;
114 unsigned done:1;
114 unsigned status:1; 115 unsigned status:1;
115 116
116 ngx_http_request_t *request; 117 ngx_http_request_t *request;
117 } ngx_http_grpc_ctx_t; 118 } ngx_http_grpc_ctx_t;
118 119
1072 ctx->state = 0; 1073 ctx->state = 0;
1073 ctx->header_sent = 0; 1074 ctx->header_sent = 0;
1074 ctx->output_closed = 0; 1075 ctx->output_closed = 0;
1075 ctx->parsing_headers = 0; 1076 ctx->parsing_headers = 0;
1076 ctx->end_stream = 0; 1077 ctx->end_stream = 0;
1078 ctx->done = 0;
1077 ctx->status = 0; 1079 ctx->status = 0;
1078 ctx->connection = NULL; 1080 ctx->connection = NULL;
1079 1081
1080 return NGX_OK; 1082 return NGX_OK;
1081 } 1083 }
1091 size_t len, limit; 1093 size_t len, limit;
1092 ngx_buf_t *b; 1094 ngx_buf_t *b;
1093 ngx_int_t rc; 1095 ngx_int_t rc;
1094 ngx_uint_t next, last; 1096 ngx_uint_t next, last;
1095 ngx_chain_t *cl, *out, **ll; 1097 ngx_chain_t *cl, *out, **ll;
1098 ngx_http_upstream_t *u;
1096 ngx_http_grpc_ctx_t *ctx; 1099 ngx_http_grpc_ctx_t *ctx;
1097 ngx_http_grpc_frame_t *f; 1100 ngx_http_grpc_frame_t *f;
1098 1101
1099 ngx_log_debug0(NGX_LOG_DEBUG_HTTP, r->connection->log, 0, 1102 ngx_log_debug0(NGX_LOG_DEBUG_HTTP, r->connection->log, 0,
1100 "grpc output filter"); 1103 "grpc output filter");
1403 } 1406 }
1404 } 1407 }
1405 1408
1406 if (rc == NGX_OK && ctx->in) { 1409 if (rc == NGX_OK && ctx->in) {
1407 rc = NGX_AGAIN; 1410 rc = NGX_AGAIN;
1411 }
1412
1413 if (ctx->done) {
1414
1415 /*
1416 * We have already got the response and were sending some additional
1417 * control frames. Even if there is still something unsent, stop
1418 * here anyway.
1419 */
1420
1421 u = r->upstream;
1422 u->length = 0;
1423
1424 if (ctx->in == NULL
1425 && ctx->out == NULL
1426 && ctx->output_closed
1427 && ctx->state == ngx_http_grpc_st_start)
1428 {
1429 u->keepalive = 1;
1430 }
1431
1432 ngx_post_event(u->peer.connection->read, &ngx_posted_events);
1408 } 1433 }
1409 1434
1410 return rc; 1435 return rc;
1411 } 1436 }
1412 1437
1830 if (ctx->state < ngx_http_grpc_st_payload) { 1855 if (ctx->state < ngx_http_grpc_st_payload) {
1831 1856
1832 rc = ngx_http_grpc_parse_frame(r, ctx, b); 1857 rc = ngx_http_grpc_parse_frame(r, ctx, b);
1833 1858
1834 if (rc == NGX_AGAIN) { 1859 if (rc == NGX_AGAIN) {
1860
1861 if (ctx->done) {
1862
1863 /*
1864 * We have finished parsing the response and the
1865 * remaining control frames. If there are unsent
1866 * control frames, post a write event to send them.
1867 */
1868
1869 if (ctx->out) {
1870 ngx_post_event(u->peer.connection->write,
1871 &ngx_posted_events);
1872 return NGX_AGAIN;
1873 }
1874
1875 u->length = 0;
1876
1877 if (ctx->in == NULL
1878 && ctx->output_closed
1879 && ctx->state == ngx_http_grpc_st_start)
1880 {
1881 u->keepalive = 1;
1882 }
1883
1884 break;
1885 }
1886
1835 return NGX_AGAIN; 1887 return NGX_AGAIN;
1836 } 1888 }
1837 1889
1838 if (rc == NGX_ERROR) { 1890 if (rc == NGX_ERROR) {
1839 return NGX_ERROR; 1891 return NGX_ERROR;
1896 "upstream sent frame for unknown stream %ui", 1948 "upstream sent frame for unknown stream %ui",
1897 ctx->stream_id); 1949 ctx->stream_id);
1898 return NGX_ERROR; 1950 return NGX_ERROR;
1899 } 1951 }
1900 1952
1953 if (ctx->stream_id && ctx->done) {
1954 ngx_log_error(NGX_LOG_ERR, r->connection->log, 0,
1955 "upstream sent frame for closed stream %ui",
1956 ctx->stream_id);
1957 return NGX_ERROR;
1958 }
1959
1901 ctx->padding = 0; 1960 ctx->padding = 0;
1902 } 1961 }
1903 1962
1904 if (ctx->state == ngx_http_grpc_st_padding) { 1963 if (ctx->state == ngx_http_grpc_st_padding) {
1905 1964
1912 b->pos += ctx->rest; 1971 b->pos += ctx->rest;
1913 ctx->rest = 0; 1972 ctx->rest = 0;
1914 ctx->state = ngx_http_grpc_st_start; 1973 ctx->state = ngx_http_grpc_st_start;
1915 1974
1916 if (ctx->flags & NGX_HTTP_V2_END_STREAM_FLAG) { 1975 if (ctx->flags & NGX_HTTP_V2_END_STREAM_FLAG) {
1917 u->length = 0; 1976 ctx->done = 1;
1918
1919 if (ctx->in == NULL
1920 && ctx->out == NULL
1921 && ctx->output_closed
1922 && b->last == b->pos)
1923 {
1924 u->keepalive = 1;
1925 }
1926
1927 break;
1928 } 1977 }
1929 1978
1930 continue; 1979 continue;
1931 } 1980 }
1932 1981
2092 2141
2093 ngx_log_debug0(NGX_LOG_DEBUG_HTTP, r->connection->log, 0, 2142 ngx_log_debug0(NGX_LOG_DEBUG_HTTP, r->connection->log, 0,
2094 "grpc trailer done"); 2143 "grpc trailer done");
2095 2144
2096 if (ctx->end_stream) { 2145 if (ctx->end_stream) {
2097 u->length = 0; 2146 ctx->done = 1;
2098 2147 break;
2099 if (ctx->in == NULL
2100 && ctx->out == NULL
2101 && ctx->output_closed
2102 && b->last == b->pos)
2103 {
2104 u->keepalive = 1;
2105 }
2106
2107 return NGX_OK;
2108 } 2148 }
2109 2149
2110 ngx_log_error(NGX_LOG_ERR, r->connection->log, 0, 2150 ngx_log_error(NGX_LOG_ERR, r->connection->log, 0,
2111 "upstream sent trailer without " 2151 "upstream sent trailer without "
2112 "end stream flag"); 2152 "end stream flag");
2117 2157
2118 ngx_log_error(NGX_LOG_ERR, r->connection->log, 0, 2158 ngx_log_error(NGX_LOG_ERR, r->connection->log, 0,
2119 "upstream sent invalid trailer"); 2159 "upstream sent invalid trailer");
2120 2160
2121 return NGX_ERROR; 2161 return NGX_ERROR;
2162 }
2163
2164 if (rc == NGX_HTTP_PARSE_HEADER_DONE) {
2165 continue;
2122 } 2166 }
2123 2167
2124 /* rc == NGX_AGAIN */ 2168 /* rc == NGX_AGAIN */
2125 2169
2126 if (ctx->rest == 0) { 2170 if (ctx->rest == 0) {
2235 } 2279 }
2236 2280
2237 ctx->state = ngx_http_grpc_st_start; 2281 ctx->state = ngx_http_grpc_st_start;
2238 2282
2239 if (ctx->flags & NGX_HTTP_V2_END_STREAM_FLAG) { 2283 if (ctx->flags & NGX_HTTP_V2_END_STREAM_FLAG) {
2240 u->length = 0; 2284 ctx->done = 1;
2241
2242 if (ctx->in == NULL
2243 && ctx->out == NULL
2244 && ctx->output_closed
2245 && b->last == b->pos)
2246 {
2247 u->keepalive = 1;
2248 }
2249
2250 break;
2251 } 2285 }
2252 } 2286 }
2253 2287
2254 return NGX_OK; 2288 return NGX_OK;
2255 } 2289 }