Make ngx_pagespeed behave like mod_pagespeed with regard to Flushing (#1217)

This change makes ngx_pagespeed listen to the FollowFlushes option.
When set to on (=default), ngx_pagespeed will forward incoming flushes
to ProxyFetch. When writing output, we'll now also set the flush flag on
the buffers we are about to send downstream.

Companion to mps commit: https://github.com/pagespeed/mod_pagespeed/commit/02de03e825bbd1f8d4ad4e1a1bef5263a16f3857

src/ngx_base_fetch.cc

@@ -53,6 +53,7 @@ NgxBaseFetch::NgxBaseFetch(StringPiece url,
       request_(r),
       server_context_(server_context),
       options_(options),
+      need_flush_(false),
       done_called_(false),
       last_buf_sent_(false),
       references_(2),
@@ -227,8 +228,11 @@ ngx_int_t NgxBaseFetch::CopyBufferToNginx(ngx_chain_t** link_ptr) {
     return NGX_AGAIN;
   }
 
-  int rc = string_piece_to_buffer_chain(
-      request_->pool, buffer_, link_ptr, done_called_ /* send_last_buf */);
+  int rc = string_piece_to_buffer_chain(request_->pool, buffer_, link_ptr,
+                                        done_called_ /* send_last_buf */,
+                                        need_flush_);
+  need_flush_ = false;
+
   if (rc != NGX_OK) {
     return rc;
   }
@@ -313,6 +317,9 @@ void NgxBaseFetch::HandleHeadersComplete() {
 }
 
 bool NgxBaseFetch::HandleFlush(MessageHandler* handler) {
+  Lock();
+  need_flush_ = true;
+  Unlock();
   RequestCollection(kFlush);  // A new part of the response body is available
   return true;
 }

src/ngx_base_fetch.h

@@ -169,6 +169,7 @@ class NgxBaseFetch : public AsyncFetch {
   GoogleString buffer_;
   NgxServerContext* server_context_;
   const RewriteOptions* options_;
+  bool need_flush_;
   bool done_called_;
   bool last_buf_sent_;
   // How many active references there are to this fetch. Starts at two,

src/ngx_pagespeed.cc

@@ -124,7 +124,7 @@ char* string_piece_to_pool_string(ngx_pool_t* pool, StringPiece sp) {
 // (potentially) longer string to nginx and want it to take ownership.
 ngx_int_t string_piece_to_buffer_chain(
     ngx_pool_t* pool, StringPiece sp, ngx_chain_t** link_ptr,
-    bool send_last_buf) {
+    bool send_last_buf, bool send_flush) {
   // Below, *link_ptr will be NULL if we're starting the chain, and the head
   // chain link.
   *link_ptr = NULL;
@@ -198,6 +198,9 @@ ngx_int_t string_piece_to_buffer_chain(
 
 
   CHECK(tail_link != NULL);
+  if (send_flush) {
+    tail_link->buf->flush = true;
+  }
   if (send_last_buf) {
     tail_link->buf->last_buf = true;
   }
@@ -1875,6 +1878,7 @@ ngx_int_t ps_resource_handler(ngx_http_request_t* r,
     ctx->r = r;
     ctx->html_rewrite = false;
     ctx->in_place = false;
+    ctx->follow_flushes = options->follow_flushes();
     ctx->preserve_caching_headers = kDontPreserveHeaders;
 
     // See build_context_for_request() in mod_instaweb.cc
@@ -2095,7 +2099,6 @@ void ps_send_to_pagespeed(ngx_http_request_t* r,
   int last_buf = 0;
   for (cur = in; cur != NULL; cur = cur->next) {
     last_buf = cur->buf->last_buf;
-
     // Buffers are not really the last buffer until they've been through
     // pagespeed.
     cur->buf->last_buf = 0;
@@ -2120,6 +2123,19 @@ void ps_send_to_pagespeed(ngx_http_request_t* r,
         }
       }
     }
+    if (cur->buf->flush && ctx->follow_flushes) {
+      // Calling ctx->proxy_fetch->Flush(cfg_s->handler) will be a no-op here,
+      // unless we have follow_flushes or flush_html enabled. Note that PSOL
+      // might aggregate multiple flushes into 1, and actually flush a little bit
+      // later due to html parser state and earlier scheduled operations.
+      // Also, unless we also set the flush flag on the nginx buffers we won't
+      // actually flush.
+      // Note that too many flushes could harm optimization over larger html
+      // fragments as PSOL gets less context to work with, e.g. it can't combine
+      // two css files if a flush happens in between.
+      ctx->proxy_fetch->Flush(cfg_s->handler);
+    }
+
     // We're done with buffers as we pass them through, so mark them as sent as
     // we go.
     cur->buf->pos = cur->buf->last;
@@ -2128,9 +2144,6 @@ void ps_send_to_pagespeed(ngx_http_request_t* r,
   if (last_buf) {
     ctx->proxy_fetch->Done(true /* success */);
     ctx->proxy_fetch = NULL;  // ProxyFetch deletes itself on Done().
-  } else {
-    // TODO(jefftk): Decide whether Flush() is warranted here.
-    ctx->proxy_fetch->Flush(cfg_s->handler);
   }
 }
 
@@ -2583,7 +2596,7 @@ ngx_int_t send_out_headers_and_body(
   // Send the body.
   ngx_chain_t* out;
   rc = string_piece_to_buffer_chain(
-      r->pool, output, &out, true /* send_last_buf */);
+      r->pool, output, &out, true /* send_last_buf */, false);
   if (rc == NGX_ERROR) {
     return NGX_ERROR;
   }

src/ngx_pagespeed.h

@@ -52,7 +52,7 @@ class InPlaceResourceRecorder;
 // NGX_DECLINED immediately unless send_last_buf.
 ngx_int_t string_piece_to_buffer_chain(
     ngx_pool_t* pool, StringPiece sp,
-    ngx_chain_t** link_ptr, bool send_last_buf);
+    ngx_chain_t** link_ptr, bool send_last_buf, bool send_flush);
 
 StringPiece str_to_string_piece(ngx_str_t s);
 
@@ -109,6 +109,7 @@ typedef struct {
   // Location: headers that start with '/' without regarding X-Forwarded-Proto.
   bool location_field_set;
   bool psol_vary_accept_only;
+  bool follow_flushes;
 } ps_request_ctx_t;
 
 ps_request_ctx_t* ps_get_request_context(ngx_http_request_t* r);

test/nginx_system_test.sh

@@ -1309,6 +1309,12 @@ OUT=$(http_proxy=$SECONDARY_HOSTNAME $WGET_DUMP -O /dev/null -S $URL 2>&1)
 MATCHES=$(echo "$OUT" | grep -c "Vary: Accept-Encoding") || true
 check [ $MATCHES -eq 1 ]
 
+start_test Follow flushes can be turned off.
+echo "Check that FollowFlushes off outputs a single chunk"
+URL="http://noflush.example.com/mod_pagespeed_test/slow_flushing_html_response.php"
+check_flushing "curl -N --raw --silent --proxy $SECONDARY_HOSTNAME $URL" \
+  5.4 1
+
 start_test Shutting down.
 
 # Fire up some heavy load if ab is available to test a stressed shutdown

test/pagespeed_test.conf.template

@@ -1522,7 +1522,45 @@ http {
     pagespeed InPlaceResourceOptimization on;
     pagespeed FileCachePath "@@FILE_CACHE@@";
   }
+  server {
+    pagespeed on;
+    listen @@SECONDARY_PORT@@;
+    listen [::]:@@SECONDARY_PORT@@;
+    server_name flush.example.com;
+    pagespeed FileCachePath "@@FILE_CACHE@@";
+    pagespeed RewriteLevel PassThrough;
+    pagespeed RewriteDeadlinePerFlushMs 1;
 
+    location ~ \.php$ {
+      fastcgi_param SCRIPT_FILENAME $request_filename;
+      fastcgi_param QUERY_STRING $query_string;
+      fastcgi_param REQUEST_METHOD $request_method;
+      fastcgi_param CONTENT_TYPE $content_type;
+      fastcgi_param CONTENT_LENGTH $content_length;
+      fastcgi_pass 127.0.0.1:9000;
+      fastcgi_buffering off;
+    }
+  }
+  server {
+    pagespeed on;
+    listen @@SECONDARY_PORT@@;
+    listen [::]:@@SECONDARY_PORT@@;
+    server_name noflush.example.com;
+    pagespeed FileCachePath "@@FILE_CACHE@@";
+    pagespeed RewriteLevel PassThrough;
+    pagespeed RewriteDeadlinePerFlushMs 1;
+    pagespeed FollowFlushes off;
+
+    location ~ \.php$ {
+      fastcgi_param SCRIPT_FILENAME $request_filename;
+      fastcgi_param QUERY_STRING $query_string;
+      fastcgi_param REQUEST_METHOD $request_method;
+      fastcgi_param CONTENT_TYPE $content_type;
+      fastcgi_param CONTENT_LENGTH $content_length;
+      fastcgi_pass 127.0.0.1:9000;
+      fastcgi_buffering off;
+    }
+  }
   server {
     listen @@PRIMARY_PORT@@;
     listen [::]:@@PRIMARY_PORT@@;