[PATCH 5 of 6] Upstream: don't wait for upstream close if we know length

[Maxim Dounin]

# HG changeset patch
# User Maxim Dounin <mdounin at mdounin.ru>
# Date 1225109046 -10800
# Node ID 9a6bf5cfcd63433a28dc00064af195a91b628d64
# Parent  aa6cd5a06ccc2037064686bcb183c71d8c65f44a
Upstream: don't wait for upstream close if we know length.

This change basically deligates right to control how many bytes may be
buffered by ngx_event_pipe to input filter by means of changing p->length.

Currently p->length is set to NGX_MAX_OFF_T_VALUE if content length isn't
known (and thus input must be terminated by upstream connection close), or
to content length if it's known.

Futher improvement may involve correct parsing of chunked transfer encoding -
by setting p->length in input filter as approprate.  E.g. set it to 2 if you
just parsed chunk end and expect other chunk to start (chunk takes at least
3 bytes, "0" CRLF, but RFC 2616 recomends to be tolerant and recognize bare
LF too).

diff --git a/src/event/ngx_event_pipe.c b/src/event/ngx_event_pipe.c
--- a/src/event/ngx_event_pipe.c
+++ b/src/event/ngx_event_pipe.c
@@ -328,22 +328,26 @@ ngx_event_pipe_read_upstream(ngx_event_p
 
             if (n >= size) {
                 cl->buf->last = cl->buf->end;
-
-                /* STUB */ cl->buf->num = p->num++;
-
-                if (p->input_filter(p, cl->buf) == NGX_ERROR) {
-                    return NGX_ABORT;
-                }
-
                 n -= size;
-                ln = cl;
-                cl = cl->next;
-                ngx_free_chain(p->pool, ln);
 
             } else {
                 cl->buf->last += n;
                 n = 0;
+
+                if (cl->buf->last - cl->buf->pos < p->length) {
+                    continue;
+                }
             }
+
+            /* STUB */ cl->buf->num = p->num++;
+
+            if (p->input_filter(p, cl->buf) == NGX_ERROR) {
+                return NGX_ABORT;
+            }
+
+            ln = cl;
+            cl = cl->next;
+            ngx_free_chain(p->pool, ln);
         }
 
         if (cl) {
@@ -409,6 +413,9 @@ flush:
                        cl->buf->file_pos,
                        cl->buf->file_last - cl->buf->file_pos);
     }
+
+    ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0,
+                   "pipe length: %O", p->length);
 
 #endif
 
@@ -856,6 +863,16 @@ ngx_event_pipe_copy_input_filter(ngx_eve
     }
     p->last_in = &cl->next;
 
+    if (p->length == NGX_MAX_OFF_T_VALUE) {
+        return NGX_OK;
+    }
+
+    p->length -= b->last - b->pos;
+
+    if (p->length <= 0) {
+        p->upstream_done = 1;
+    }
+
     return NGX_OK;
 }
 
diff --git a/src/event/ngx_event_pipe.h b/src/event/ngx_event_pipe.h
--- a/src/event/ngx_event_pipe.h
+++ b/src/event/ngx_event_pipe.h
@@ -65,6 +65,7 @@ struct ngx_event_pipe_s {
     ssize_t            busy_size;
 
     off_t              read_length;
+    off_t              length;
 
     off_t              max_temp_file_size;
     ssize_t            temp_file_write_size;
diff --git a/src/http/ngx_http_upstream.c b/src/http/ngx_http_upstream.c
--- a/src/http/ngx_http_upstream.c
+++ b/src/http/ngx_http_upstream.c
@@ -1710,6 +1710,13 @@ ngx_http_upstream_send_response(ngx_http
     p->pool = r->pool;
     p->log = c->log;
 
+    if (r->headers_out.content_length_n != -1) {
+        p->length = r->headers_out.content_length_n;
+
+    } else {
+        p->length = NGX_MAX_OFF_T_VALUE;
+    }
+
     p->cacheable = u->cacheable || u->store;
 
     p->temp_file = ngx_pcalloc(r->pool, sizeof(ngx_temp_file_t));