1 files changed, 237 insertions, 78 deletions

diff --git a/src/http/v2/ngx_http_v2.c b/src/http/v2/ngx_http_v2.c
index 3bef002bd..79c4f17c2 100644
--- a/src/http/v2/ngx_http_v2.c
+++ b/src/http/v2/ngx_http_v2.c
@@ -173,7 +173,7 @@ static ngx_int_t ngx_http_v2_construct_cookie_header(ngx_http_request_t *r);
 static void ngx_http_v2_run_request(ngx_http_request_t *r);
 static void ngx_http_v2_run_request_handler(ngx_event_t *ev);
 static ngx_int_t ngx_http_v2_process_request_body(ngx_http_request_t *r,
-    u_char *pos, size_t size, ngx_uint_t last);
+    u_char *pos, size_t size, ngx_uint_t last, ngx_uint_t flush);
 static ngx_int_t ngx_http_v2_filter_request_body(ngx_http_request_t *r);
 static void ngx_http_v2_read_client_request_body_handler(ngx_http_request_t *r);
 
@@ -1092,7 +1092,7 @@ static u_char *
 ngx_http_v2_state_read_data(ngx_http_v2_connection_t *h2c, u_char *pos,
     u_char *end)
 {
-    size_t                   size;
+    size_t                   size, window;
     ngx_buf_t               *buf;
     ngx_int_t                rc;
     ngx_connection_t        *fc;
@@ -1140,13 +1140,40 @@ ngx_http_v2_state_read_data(ngx_http_v2_connection_t *h2c, u_char *pos,
     h2c->payload_bytes += size;
 
     if (r->request_body) {
-        rc = ngx_http_v2_process_request_body(r, pos, size, stream->in_closed);
+        rc = ngx_http_v2_process_request_body(r, pos, size,
+                                              stream->in_closed, 0);
 
-        if (rc != NGX_OK) {
+        if (rc != NGX_OK && rc != NGX_AGAIN) {
             stream->skip_data = 1;
             ngx_http_finalize_request(r, rc);
         }
 
+        if (rc == NGX_AGAIN && !stream->no_flow_control) {
+            buf = r->request_body->buf;
+            window = buf->end - buf->last;
+
+            window -= h2c->state.length - size;
+
+            if (window < stream->recv_window) {
+                ngx_log_error(NGX_LOG_ALERT, h2c->connection->log, 0,
+                              "http2 negative window update");
+                return ngx_http_v2_connection_error(h2c,
+                                                    NGX_HTTP_V2_INTERNAL_ERROR);
+            }
+
+            if (window > stream->recv_window) {
+                if (ngx_http_v2_send_window_update(h2c, stream->node->id,
+                                                   window - stream->recv_window)
+                    == NGX_ERROR)
+                {
+                    return ngx_http_v2_connection_error(h2c,
+                                                    NGX_HTTP_V2_INTERNAL_ERROR);
+                }
+
+                stream->recv_window = window;
+            }
+        }
+
         ngx_http_run_posted_requests(fc);
 
     } else if (size) {
@@ -4027,16 +4054,30 @@ ngx_http_v2_read_request_body(ngx_http_request_t *r)
         return NGX_OK;
     }
 
+    rb->rest = 1;
+
+    /* set rb->filter_need_buffering */
+
+    rc = ngx_http_top_request_body_filter(r, NULL);
+
+    if (rc != NGX_OK) {
+        stream->skip_data = 1;
+        return rc;
+    }
+
     h2scf = ngx_http_get_module_srv_conf(r, ngx_http_v2_module);
     clcf = ngx_http_get_module_loc_conf(r, ngx_http_core_module);
 
     len = r->headers_in.content_length_n;
 
-    if (r->request_body_no_buffering && !stream->in_closed) {
+    if (len < 0 || len > (off_t) clcf->client_body_buffer_size) {
+        len = clcf->client_body_buffer_size;
 
-        if (len < 0 || len > (off_t) clcf->client_body_buffer_size) {
-            len = clcf->client_body_buffer_size;
-        }
+    } else {
+        len++;
+    }
+
+    if (r->request_body_no_buffering || rb->filter_need_buffering) {
 
         /*
          * We need a room to store data up to the stream's initial window size,
@@ -4050,57 +4091,54 @@ ngx_http_v2_read_request_body(ngx_http_request_t *r)
         if (len > NGX_HTTP_V2_MAX_WINDOW) {
             len = NGX_HTTP_V2_MAX_WINDOW;
         }
-
-        rb->buf = ngx_create_temp_buf(r->pool, (size_t) len);
-
-    } else if (len >= 0 && len <= (off_t) clcf->client_body_buffer_size
-               && !r->request_body_in_file_only)
-    {
-        rb->buf = ngx_create_temp_buf(r->pool, (size_t) len);
-
-    } else {
-        rb->buf = ngx_calloc_buf(r->pool);
-
-        if (rb->buf != NULL) {
-            rb->buf->sync = 1;
-        }
     }
 
+    rb->buf = ngx_create_temp_buf(r->pool, (size_t) len);
+
     if (rb->buf == NULL) {
         stream->skip_data = 1;
         return NGX_HTTP_INTERNAL_SERVER_ERROR;
     }
 
-    rb->rest = 1;
-
     buf = stream->preread;
 
     if (stream->in_closed) {
-        r->request_body_no_buffering = 0;
+        if (!rb->filter_need_buffering) {
+            r->request_body_no_buffering = 0;
+        }
 
         if (buf) {
             rc = ngx_http_v2_process_request_body(r, buf->pos,
-                                                  buf->last - buf->pos, 1);
+                                                  buf->last - buf->pos, 1, 0);
             ngx_pfree(r->pool, buf->start);
+
+        } else {
+            rc = ngx_http_v2_process_request_body(r, NULL, 0, 1, 0);
+        }
+
+        if (rc != NGX_AGAIN) {
             return rc;
         }
 
-        return ngx_http_v2_process_request_body(r, NULL, 0, 1);
+        r->read_event_handler = ngx_http_v2_read_client_request_body_handler;
+        r->write_event_handler = ngx_http_request_empty_handler;
+
+        return NGX_AGAIN;
     }
 
     if (buf) {
         rc = ngx_http_v2_process_request_body(r, buf->pos,
-                                              buf->last - buf->pos, 0);
+                                              buf->last - buf->pos, 0, 0);
 
         ngx_pfree(r->pool, buf->start);
 
-        if (rc != NGX_OK) {
+        if (rc != NGX_OK && rc != NGX_AGAIN) {
             stream->skip_data = 1;
             return rc;
         }
     }
 
-    if (r->request_body_no_buffering) {
+    if (r->request_body_no_buffering || rb->filter_need_buffering) {
         size = (size_t) len - h2scf->preread_size;
 
     } else {
@@ -4142,9 +4180,9 @@ ngx_http_v2_read_request_body(ngx_http_request_t *r)
 
 static ngx_int_t
 ngx_http_v2_process_request_body(ngx_http_request_t *r, u_char *pos,
-    size_t size, ngx_uint_t last)
+    size_t size, ngx_uint_t last, ngx_uint_t flush)
 {
-    ngx_buf_t                 *buf;
+    size_t                     n;
     ngx_int_t                  rc;
     ngx_connection_t          *fc;
     ngx_http_request_body_t   *rb;
@@ -4152,77 +4190,128 @@ ngx_http_v2_process_request_body(ngx_http_request_t *r, u_char *pos,
 
     fc = r->connection;
     rb = r->request_body;
-    buf = rb->buf;
 
-    if (size) {
-        if (buf->sync) {
-            buf->pos = buf->start = pos;
-            buf->last = buf->end = pos + size;
+    ngx_log_debug0(NGX_LOG_DEBUG_HTTP, fc->log, 0,
+                   "http2 process request body");
 
-            r->request_body_in_file_only = 1;
+    if (size == 0 && !last && !flush) {
+        return NGX_AGAIN;
+    }
 
-        } else {
-            if (size > (size_t) (buf->end - buf->last)) {
-                ngx_log_error(NGX_LOG_INFO, fc->log, 0,
-                              "client intended to send body data "
-                              "larger than declared");
+    for ( ;; ) {
+        for ( ;; ) {
+            if (rb->buf->last == rb->buf->end && size) {
 
-                return NGX_HTTP_BAD_REQUEST;
+                if (r->request_body_no_buffering) {
+
+                    /* should never happen due to flow control */
+
+                    ngx_log_error(NGX_LOG_ALERT, fc->log, 0,
+                                  "no space in http2 body buffer");
+
+                    return NGX_HTTP_INTERNAL_SERVER_ERROR;
+                }
+
+                /* update chains */
+
+                ngx_log_error(NGX_LOG_DEBUG, fc->log, 0,
+                              "http2 body update chains");
+
+                rc = ngx_http_v2_filter_request_body(r);
+
+                if (rc != NGX_OK) {
+                    return rc;
+                }
+
+                if (rb->busy != NULL) {
+                    ngx_log_error(NGX_LOG_ALERT, fc->log, 0,
+                                  "busy buffers after request body flush");
+                    return NGX_HTTP_INTERNAL_SERVER_ERROR;
+                }
+
+                rb->buf->pos = rb->buf->start;
+                rb->buf->last = rb->buf->start;
             }
 
-            buf->last = ngx_cpymem(buf->last, pos, size);
-        }
-    }
+            /* copy body data to the buffer */
 
-    if (last) {
-        rb->rest = 0;
+            n = rb->buf->end - rb->buf->last;
 
-        if (fc->read->timer_set) {
-            ngx_del_timer(fc->read);
-        }
+            if (n > size) {
+                n = size;
+            }
 
-        if (r->request_body_no_buffering) {
-            ngx_post_event(fc->read, &ngx_posted_events);
-            return NGX_OK;
-        }
+            if (n > 0) {
+                rb->buf->last = ngx_cpymem(rb->buf->last, pos, n);
+            }
 
-        rc = ngx_http_v2_filter_request_body(r);
+            ngx_log_debug1(NGX_LOG_DEBUG_HTTP, fc->log, 0,
+                           "http2 request body recv %uz", n);
 
-        if (rc != NGX_OK) {
-            return rc;
-        }
+            pos += n;
+            size -= n;
+
+            if (size == 0 && last) {
+                rb->rest = 0;
+            }
+
+            if (r->request_body_no_buffering) {
+                break;
+            }
+
+            /* pass buffer to request body filter chain */
+
+            rc = ngx_http_v2_filter_request_body(r);
+
+            if (rc != NGX_OK) {
+                return rc;
+            }
 
-        if (buf->sync) {
-            /* prevent reusing this buffer in the upstream module */
-            rb->buf = NULL;
+            if (rb->rest == 0) {
+                break;
+            }
+
+            if (size == 0) {
+                break;
+            }
         }
 
-        if (r->headers_in.chunked) {
-            r->headers_in.content_length_n = rb->received;
+        ngx_log_debug1(NGX_LOG_DEBUG_HTTP, fc->log, 0,
+                       "http2 request body rest %O", rb->rest);
+
+        if (rb->rest == 0 && rb->last_saved) {
+            break;
         }
 
-        r->read_event_handler = ngx_http_block_reading;
-        rb->post_handler(r);
+        if (size == 0) {
+            clcf = ngx_http_get_module_loc_conf(r, ngx_http_core_module);
+            ngx_add_timer(fc->read, clcf->client_body_timeout);
 
-        return NGX_OK;
-    }
+            if (r->request_body_no_buffering) {
+                ngx_post_event(fc->read, &ngx_posted_events);
+                return NGX_AGAIN;
+            }
 
-    if (size == 0) {
-        return NGX_OK;
+            return NGX_AGAIN;
+        }
     }
 
-    clcf = ngx_http_get_module_loc_conf(r, ngx_http_core_module);
-    ngx_add_timer(fc->read, clcf->client_body_timeout);
+    if (fc->read->timer_set) {
+        ngx_del_timer(fc->read);
+    }
 
     if (r->request_body_no_buffering) {
         ngx_post_event(fc->read, &ngx_posted_events);
         return NGX_OK;
     }
 
-    if (buf->sync) {
-        return ngx_http_v2_filter_request_body(r);
+    if (r->headers_in.chunked) {
+        r->headers_in.content_length_n = rb->received;
     }
 
+    r->read_event_handler = ngx_http_block_reading;
+    rb->post_handler(r);
+
     return NGX_OK;
 }
 
@@ -4239,7 +4328,7 @@ ngx_http_v2_filter_request_body(ngx_http_request_t *r)
     rb = r->request_body;
     buf = rb->buf;
 
-    if (buf->pos == buf->last && rb->rest) {
+    if (buf->pos == buf->last && (rb->rest || rb->last_sent)) {
         cl = NULL;
         goto update;
     }
@@ -4302,6 +4391,7 @@ ngx_http_v2_filter_request_body(ngx_http_request_t *r)
         }
 
         b->last_buf = 1;
+        rb->last_sent = 1;
     }
 
     b->tag = (ngx_buf_tag_t) &ngx_http_v2_filter_request_body;
@@ -4321,7 +4411,12 @@ update:
 static void
 ngx_http_v2_read_client_request_body_handler(ngx_http_request_t *r)
 {
-    ngx_connection_t  *fc;
+    size_t                     window;
+    ngx_buf_t                 *buf;
+    ngx_int_t                  rc;
+    ngx_connection_t          *fc;
+    ngx_http_v2_stream_t      *stream;
+    ngx_http_v2_connection_t  *h2c;
 
     fc = r->connection;
 
@@ -4347,6 +4442,63 @@ ngx_http_v2_read_client_request_body_handler(ngx_http_request_t *r)
         ngx_http_finalize_request(r, NGX_HTTP_CLIENT_CLOSED_REQUEST);
         return;
     }
+
+    rc = ngx_http_v2_process_request_body(r, NULL, 0, r->stream->in_closed, 1);
+
+    if (rc != NGX_OK && rc != NGX_AGAIN) {
+        r->stream->skip_data = 1;
+        ngx_http_finalize_request(r, rc);
+        return;
+    }
+
+    if (rc == NGX_OK) {
+        return;
+    }
+
+    if (r->request_body->rest == 0) {
+        return;
+    }
+
+    stream = r->stream;
+    h2c = stream->connection;
+
+    buf = r->request_body->buf;
+    window = buf->end - buf->start;
+
+    if (h2c->state.stream == stream) {
+        window -= h2c->state.length;
+    }
+
+    if (window <= stream->recv_window) {
+        if (window < stream->recv_window) {
+            ngx_log_error(NGX_LOG_ALERT, r->connection->log, 0,
+                          "http2 negative window update");
+
+            stream->skip_data = 1;
+
+            ngx_http_finalize_request(r, NGX_HTTP_INTERNAL_SERVER_ERROR);
+            return;
+        }
+
+        return;
+    }
+
+    if (ngx_http_v2_send_window_update(h2c, stream->node->id,
+                                       window - stream->recv_window)
+        == NGX_ERROR)
+    {
+        stream->skip_data = 1;
+        ngx_http_finalize_request(r, NGX_HTTP_INTERNAL_SERVER_ERROR);
+        return;
+    }
+
+    stream->recv_window = window;
+
+    if (ngx_http_v2_send_output_queue(h2c) == NGX_ERROR) {
+        stream->skip_data = 1;
+        ngx_http_finalize_request(r, NGX_HTTP_INTERNAL_SERVER_ERROR);
+        return;
+    }
 }
 
 
@@ -4364,6 +4516,9 @@ ngx_http_v2_read_unbuffered_request_body(ngx_http_request_t *r)
     stream = r->stream;
     fc = r->connection;
 
+    ngx_log_debug0(NGX_LOG_DEBUG_HTTP, fc->log, 0,
+                   "http2 read unbuffered request body");
+
     if (fc->read->timedout) {
         if (stream->recv_window) {
             stream->skip_data = 1;
@@ -4387,10 +4542,14 @@ ngx_http_v2_read_unbuffered_request_body(ngx_http_request_t *r)
         return rc;
     }
 
-    if (!r->request_body->rest) {
+    if (r->request_body->rest == 0 && r->request_body->last_saved) {
         return NGX_OK;
     }
 
+    if (r->request_body->rest == 0) {
+        return NGX_AGAIN;
+    }
+
     if (r->request_body->busy != NULL) {
         return NGX_AGAIN;
     }