config: whitelist a few more directives for script variables

When configuring downstream caching if you have an origin with multiple caches in front of it you might want to choose how you handle requests in response to which cache you're getting them from.  Whitelist DownstreamCachePurgeLocationPrefix, DownstreamCachePurgeMethod, and DownstreamCacheRewrittenPercentageThreshold for scripting so this is possible.

.gitignore

@@ -1,4 +1,4 @@
 test/tmp
 psol/
 psol-*.tar.gz
-
+*.*.*.*.tar.gz

config

@@ -27,8 +27,8 @@ if [ "$mod_pagespeed_dir" = "unset" ] ; then
     echo "   You need to separately download the pagespeed library:"
     echo ""
     echo "     $ cd /path/to/ngx_pagespeed"
-    echo "     $ wget https://dl.google.com/dl/page-speed/psol/1.8.31.2.tar.gz"
+    echo "     $ wget https://dl.google.com/dl/page-speed/psol/1.9.32.4.tar.gz"
-    echo "     $ tar -xzvf 1.8.31.2.tar.gz # expands to psol/"
+    echo "     $ tar -xzvf 1.9.32.4.tar.gz # expands to psol/"
     echo ""
     echo "   Or see the installation instructions:"
     echo "     https://github.com/pagespeed/ngx_pagespeed#how-to-build"
@@ -39,31 +39,6 @@ else
   build_from_source=true
 fi
 
-psol_binary="${PSOL_BINARY:-unset}"
-if [ "$psol_binary" = "unset" ] ; then
-  if $build_from_source ; then
-    psol_binary="\
-        $mod_pagespeed_dir/net/instaweb/automatic/pagespeed_automatic.a"
-  else
-    psol_library_dir="$ngx_addon_dir/psol/lib/$buildtype/$os_name/$arch_name"
-    psol_binary="$psol_library_dir/pagespeed_automatic.a"
-  fi
-fi
-
-echo "mod_pagespeed_dir=$mod_pagespeed_dir"
-echo "build_from_source=$build_from_source"
-
-ngx_feature="psol"
-ngx_feature_name=""
-ngx_feature_run=no
-ngx_feature_incs="
-#include \"net/instaweb/htmlparse/public/html_parse.h\"
-#include \"net/instaweb/htmlparse/public/html_writer_filter.h\"
-#include \"net/instaweb/util/public/string.h\"
-#include \"net/instaweb/util/public/string_writer.h\"
-#include \"net/instaweb/util/public/null_message_handler.h\"
-"
-
 os_name='unknown_os'
 arch_name='unknown_arch'
 uname_os=`uname`
@@ -124,6 +99,31 @@ if [ "$WNO_ERROR" = "YES" ]; then
     CFLAGS="$CFLAGS -Wno-error"
 fi
 
+psol_binary="${PSOL_BINARY:-unset}"
+if [ "$psol_binary" = "unset" ] ; then
+  if $build_from_source ; then
+    psol_binary="\
+        $mod_pagespeed_dir/net/instaweb/automatic/pagespeed_automatic.a"
+  else
+    psol_library_dir="$ngx_addon_dir/psol/lib/$buildtype/$os_name/$arch_name"
+    psol_binary="$psol_library_dir/pagespeed_automatic.a"
+  fi
+fi
+
+echo "mod_pagespeed_dir=$mod_pagespeed_dir"
+echo "build_from_source=$build_from_source"
+
+ngx_feature="psol"
+ngx_feature_name=""
+ngx_feature_run=no
+ngx_feature_incs="
+#include \"pagespeed/kernel/base/string.h\"
+#include \"pagespeed/kernel/base/string_writer.h\"
+#include \"pagespeed/kernel/base/null_message_handler.h\"
+#include \"pagespeed/kernel/html/html_parse.h\"
+#include \"pagespeed/kernel/html/html_writer_filter.h\"
+"
+
 pagespeed_include="\
   $mod_pagespeed_dir \
   $mod_pagespeed_dir/third_party/chromium/src \
@@ -205,7 +205,7 @@ if [ $ngx_found = yes ]; then
 else
   cat << END
 $0: error: module ngx_pagespeed requires the pagespeed optimization library.
-Look in obj/autoconf.err for more details.
+Look in objs/autoconf.err for more details.
 END
   exit 1
 fi

scripts/pagespeed_libraries_generator.sh

@@ -13,9 +13,9 @@
 # Author: vid@zippykid.com (Vid Luther)
 #         jefftk@google.com (Jeff Kaufman)
 
-URL="https://modpagespeed.googlecode.com/svn/trunk/src/"
+URL="https://github.com/pagespeed/mod_pagespeed/raw/master/"
 URL+="net/instaweb/genfiles/conf/pagespeed_libraries.conf"
-curl -s "$URL" \
+curl -L -s -S "$URL" \
     | grep ModPagespeedLibrary \
     | while read library size hash url ; do
   echo "  pagespeed Library $size $hash $url;"

src/ngx_base_fetch.cc

@@ -39,7 +39,7 @@ NgxBaseFetch::NgxBaseFetch(ngx_http_request_t* r, int pipe_fd,
       last_buf_sent_(false),
       pipe_fd_(pipe_fd),
       references_(2),
-      handle_error_(true),
+      ipro_lookup_(false),
       preserve_caching_headers_(preserve_caching_headers) {
   if (pthread_mutex_init(&mutex_, NULL)) CHECK(0);
 }
@@ -137,14 +137,20 @@ void NgxBaseFetch::HandleHeadersComplete() {
   int status_code = response_headers()->status_code();
   bool status_ok = (status_code != 0) && (status_code < 400);
 
-  if (status_ok || handle_error_) {
+  if (!ipro_lookup_ || status_ok) {
     // If this is a 404 response we need to count it in the stats.
     if (response_headers()->status_code() == HttpStatus::kNotFound) {
       server_context_->rewrite_stats()->resource_404_count()->Add(1);
     }
   }
 
+  // For the IPRO lookup, suppress notification of the nginx side here.
+  // If we send both this event and the one from done, nasty stuff will happen
+  // if we loose the race with with the nginx side destructing this base fetch
+  // instance (and thereby clearing the byte and its pending extraneous event.
+  if (!ipro_lookup_) {
     RequestCollection();  // Headers available.
+  }
 }
 
 bool NgxBaseFetch::HandleFlush(MessageHandler* handler) {

src/ngx_base_fetch.h

@@ -79,7 +79,7 @@ class NgxBaseFetch : public AsyncFetch {
 
   // Called by nginx when it's done with us.
   void Release();
-  void set_handle_error(bool x) { handle_error_ = x; }
+  void set_ipro_lookup(bool x) { ipro_lookup_ = x; }
 
  private:
   virtual bool HandleWrite(const StringPiece& sp, MessageHandler* handler);
@@ -117,7 +117,7 @@ class NgxBaseFetch : public AsyncFetch {
   // decremented once when Done() is called and once when Release() is called.
   int references_;
   pthread_mutex_t mutex_;
-  bool handle_error_;
+  bool ipro_lookup_;
   PreserveCachingHeaders preserve_caching_headers_;
 
   DISALLOW_COPY_AND_ASSIGN(NgxBaseFetch);

src/ngx_fetch.cc

@@ -24,6 +24,14 @@
 //  - The read handler parses the response. Add the response to the buffer at
 //    last.
 
+// TODO(oschaaf): Currently the first applicable connection is picked from the
+// pool when re-using connections. Perhaps it would be worth it to pick the one
+// that was active the longest time ago to keep a larger pool available.
+// TODO(oschaaf): style: reindent namespace according to google C++ style guide
+// TODO(oschaaf): Retry mechanism for failures on a re-used k-a connection.
+// Currently we don't think it's going to be an issue, see the comments at
+// https://github.com/pagespeed/ngx_pagespeed/pull/781.
+
 extern "C" {
 #include <nginx.h>
 }
@@ -34,6 +42,7 @@ extern "C" {
 
 #include <algorithm>
 #include <string>
+#include <typeinfo>
 #include <vector>
 
 #include "net/instaweb/util/public/scoped_ptr.h"
@@ -48,6 +57,7 @@ extern "C" {
 #include "net/instaweb/util/public/message_handler.h"
 #include "net/instaweb/util/public/pool.h"
 #include "net/instaweb/util/public/pool_element.h"
+#include "net/instaweb/util/public/pthread_mutex.h"
 #include "net/instaweb/util/public/statistics.h"
 #include "net/instaweb/util/public/string_writer.h"
 #include "net/instaweb/util/public/string_util.h"
@@ -56,10 +66,230 @@ extern "C" {
 #include "net/instaweb/util/public/writer.h"
 
 namespace net_instaweb {
-  NgxFetch::NgxFetch(const GoogleString& url,
+
+class NgxConnection : public PoolElement<NgxConnection> {
+ public:
+  NgxConnection(MessageHandler* handler, int max_keepalive_requests);
+  ~NgxConnection();
+  void SetSock(u_char *sockaddr, socklen_t socklen) {
+    socklen_ = socklen;
+    ngx_memcpy(&sockaddr_, sockaddr, socklen);
+  }
+  // Close ensures that NgxConnection deletes itself at the appropriate time,
+  // which can be after receiving a non-keepalive response, or when the remote
+  // server closes the connection when the NgxConnection is pooled and idle.
+  void Close();
+
+  // Once keepalive is disabled, it can't be toggled back on.
+  void set_keepalive(bool k) { keepalive_ = keepalive_ && k; }
+  bool keepalive() { return keepalive_; }
+
+  typedef Pool<NgxConnection> NgxConnectionPool;
+
+  static NgxConnection* Connect(ngx_peer_connection_t* pc,
+                                MessageHandler* handler,
+                                int max_keepalive_requests);
+  static void IdleWriteHandler(ngx_event_t* ev);
+  static void IdleReadHandler(ngx_event_t* ev);
+
+  static NgxConnectionPool connection_pool;
+  static PthreadMutex connection_pool_mutex;
+
+  // c_ is owned by NgxConnection and freed in ::Close()
+  ngx_connection_t* c_;
+  static const int64 keepalive_timeout_ms;
+  static const GoogleString ka_header;
+
+ private:
+  int max_keepalive_requests_;
+  bool keepalive_;
+  socklen_t socklen_;
+  u_char sockaddr_[NGX_SOCKADDRLEN];
+  MessageHandler* handler_;
+
+  DISALLOW_COPY_AND_ASSIGN(NgxConnection);
+};
+
+NgxConnection::NgxConnectionPool NgxConnection::connection_pool;
+PthreadMutex NgxConnection::connection_pool_mutex;
+// Default keepalive 60s.
+const int64 NgxConnection::keepalive_timeout_ms = 60000;
+const GoogleString NgxConnection::ka_header =
+    StrCat("keep-alive ",
+           Integer64ToString(NgxConnection::keepalive_timeout_ms));
+
+NgxConnection::NgxConnection(MessageHandler* handler,
+                             int max_keepalive_requests) {
+  c_ = NULL;
+  max_keepalive_requests_ = max_keepalive_requests;
+  handler_ = handler;
+  // max_keepalive_requests specifies the number of http requests that are
+  // allowed to be performed over a single connection. So, a
+  // max_keepalive_requests of 1 effectively disables keepalive.
+  keepalive_ = max_keepalive_requests_ > 1;
+}
+
+NgxConnection::~NgxConnection() {
+  CHECK(c_ == NULL) << "NgxFetch: Underlying connection should be NULL";
+}
+
+NgxConnection* NgxConnection::Connect(ngx_peer_connection_t* pc,
+                                      MessageHandler* handler,
+                                      int max_keepalive_requests) {
+  NgxConnection* nc;
+  {
+    ScopedMutex lock(&NgxConnection::connection_pool_mutex);
+
+    for (NgxConnectionPool::iterator p = connection_pool.begin();
+         p != connection_pool.end(); ++p) {
+      nc = *p;
+
+      if (ngx_memn2cmp(static_cast<u_char*>(nc->sockaddr_),
+                       reinterpret_cast<u_char*>(pc->sockaddr),
+                       nc->socklen_, pc->socklen) == 0) {
+        CHECK(nc->c_->idle) << "Pool should only contain idle connections!";
+
+        nc->c_->idle = 0;
+        nc->c_->log = pc->log;
+        nc->c_->read->log = pc->log;
+        nc->c_->write->log = pc->log;
+        if (nc->c_->pool != NULL) {
+          nc->c_->pool->log = pc->log;
+        }
+
+        if (nc->c_->read->timer_set) {
+          ngx_del_timer(nc->c_->read);
+        }
+        connection_pool.Remove(nc);
+
+        ngx_log_error(NGX_LOG_DEBUG, pc->log, 0,
+                      "NgxFetch: re-using connection %p (pool size: %l)\n",
+                      nc, connection_pool.size());
+        return nc;
+      }
+    }
+  }
+
+  int rc = ngx_event_connect_peer(pc);
+  if (rc == NGX_ERROR || rc == NGX_DECLINED || rc == NGX_BUSY) {
+    return NULL;
+  }
+
+  // NgxConnection deletes itself if NgxConnection::Close()
+  nc = new NgxConnection(handler, max_keepalive_requests);
+  nc->SetSock(reinterpret_cast<u_char*>(pc->sockaddr), pc->socklen);
+  nc->c_ = pc->connection;
+  return nc;
+}
+
+void NgxConnection::Close() {
+  bool removed_from_pool = false;
+
+  {
+    ScopedMutex lock(&NgxConnection::connection_pool_mutex);
+    for (NgxConnectionPool::iterator p = connection_pool.begin();
+         p != connection_pool.end(); ++p) {
+      if (*p == this) {
+        // When we get here, that means that the connection either has timed
+        // out or has been closed remotely.
+        connection_pool.Remove(this);
+        ngx_log_error(NGX_LOG_DEBUG, c_->log, 0,
+                      "NgxFetch: removed connection %p (pool size: %l)\n",
+                      this, connection_pool.size());
+        removed_from_pool = true;
+        break;
+      }
+    }
+  }
+
+  max_keepalive_requests_--;
+
+  if (c_->read->timer_set) {
+    ngx_del_timer(c_->read);
+  }
+
+  if (c_->write->timer_set) {
+    ngx_del_timer(c_->write);
+  }
+
+  if (!keepalive_ || max_keepalive_requests_ <= 0 || removed_from_pool) {
+    ngx_close_connection(c_);
+    c_ = NULL;
+    delete this;
+    return;
+  }
+
+  ngx_add_timer(c_->read, static_cast<ngx_msec_t>(
+      NgxConnection::keepalive_timeout_ms));
+
+  c_->data = this;
+  c_->read->handler = NgxConnection::IdleReadHandler;
+  c_->write->handler = NgxConnection::IdleWriteHandler;
+  c_->idle = 1;
+
+  // This connection should not be associated with current fetch.
+  c_->log = ngx_cycle->log;
+  c_->read->log = ngx_cycle->log;
+  c_->write->log = ngx_cycle->log;
+  if (c_->pool != NULL) {
+    c_->pool->log = ngx_cycle->log;
+  }
+
+  // Allow this connection to be re-used, by adding it to the connection pool.
+  {
+    ScopedMutex lock(&NgxConnection::connection_pool_mutex);
+    connection_pool.Add(this);
+    ngx_log_error(NGX_LOG_DEBUG, c_->log, 0,
+                  "NgxFetch: Added connection %p (pool size: %l - "
+                  " max_keepalive_requests_ %d)\n",
+                  this, connection_pool.size(), max_keepalive_requests_);
+  }
+}
+
+void NgxConnection::IdleWriteHandler(ngx_event_t* ev) {
+  ngx_connection_t* c = static_cast<ngx_connection_t*>(ev->data);
+  u_char buf[1];
+  int n = c->recv(c, buf, 1);
+  if (c->write->timedout) {
+    DCHECK(false) << "NgxFetch: write timeout not expected." << n;
+  }
+  if (n == NGX_AGAIN) {
+    return;
+  }
+}
+
+void NgxConnection::IdleReadHandler(ngx_event_t* ev) {
+  ngx_connection_t* c = static_cast<ngx_connection_t*>(ev->data);
+  NgxConnection* nc = static_cast<NgxConnection*>(c->data);
+
+  if (c->read->timedout) {
+    nc->set_keepalive(false);
+    nc->Close();
+    return;
+  }
+
+  char buf[1];
+  int n;
+
+  // not a timeout event, we should check connection
+  n = recv(c->fd, buf, 1, MSG_PEEK);
+  if (n == -1 && ngx_socket_errno == NGX_EAGAIN) {
+    if (ngx_handle_read_event(c->read, 0) != NGX_OK) {
+      nc->set_keepalive(false);
+      nc->Close();
+      return;
+    }
+
+    return;
+  }
+
+  nc->set_keepalive(false);
+  nc->Close();
+}
+
+NgxFetch::NgxFetch(const GoogleString& url,
                    AsyncFetch* async_fetch,
                    MessageHandler* message_handler,
-                     ngx_msec_t timeout_ms,
                    ngx_log_t* log)
     : str_url_(url),
       fetcher_(NULL),
@@ -78,31 +308,38 @@ namespace net_instaweb {
   pool_ = NULL;
   timeout_event_ = NULL;
   connection_ = NULL;
-  }
+}
 
-  NgxFetch::~NgxFetch() {
+NgxFetch::~NgxFetch() {
   if (timeout_event_ != NULL && timeout_event_->timer_set) {
     ngx_del_timer(timeout_event_);
   }
   if (connection_ != NULL) {
-      ngx_close_connection(connection_);
+    connection_->Close();
+    connection_ = NULL;
   }
   if (pool_ != NULL) {
     ngx_destroy_pool(pool_);
+    pool_ = NULL;
   }
-  }
+}
 
-  // This function is called by NgxUrlAsyncFetcher::StartFetch.
+// This function is called by NgxUrlAsyncFetcher::StartFetch.
-  bool NgxFetch::Start(NgxUrlAsyncFetcher* fetcher) {
+bool NgxFetch::Start(NgxUrlAsyncFetcher* fetcher) {
   fetcher_ = fetcher;
-    return Init();
+  bool ok = Init();
-  }
+  if (ok) {
+    ngx_log_error(NGX_LOG_DEBUG, log_, 0, "NgxFetch %p: initialized\n",
+                  this);
+  }  // else Init() will have emitted a reason
+  return ok;
+}
 
-  // Create the pool, parse the url, add the timeout event and
+// Create the pool, parse the url, add the timeout event and
-  // hook the DNS resolver if needed. Else we connect directly.
+// hook the DNS resolver if needed. Else we connect directly.
-  // When this returns false, our caller (NgxUrlAsyncFetcher::StartFetch)
+// When this returns false, our caller (NgxUrlAsyncFetcher::StartFetch)
-  // will call fetch->CallbackDone()
+// will call fetch->CallbackDone()
-  bool NgxFetch::Init() {
+bool NgxFetch::Init() {
   pool_ = ngx_create_pool(12288, log_);
   if (pool_ == NULL) {
     message_handler_->Message(kError, "NgxFetch: ngx_create_pool failed");
@@ -121,20 +358,23 @@ namespace net_instaweb {
                               "NgxFetch: ngx_pcalloc failed for timeout");
     return false;
   }
+
   timeout_event_->data = this;
-    timeout_event_->handler = NgxFetchTimeout;
+  timeout_event_->handler = NgxFetch::TimeoutHandler;
   timeout_event_->log = log_;
 
   ngx_add_timer(timeout_event_, fetcher_->fetch_timeout_);
-    r_ = static_cast<ngx_http_request_t*>(ngx_pcalloc(pool_,
+  r_ = static_cast<ngx_http_request_t*>(
-                                           sizeof(ngx_http_request_t)));
+      ngx_pcalloc(pool_, sizeof(ngx_http_request_t)));
+
   if (r_ == NULL) {
     message_handler_->Message(kError,
                               "NgxFetch: ngx_pcalloc failed for timer");
     return false;
   }
-    status_ = static_cast<ngx_http_status_t*>(ngx_pcalloc(pool_,
+  status_ = static_cast<ngx_http_status_t*>(
-                                              sizeof(ngx_http_status_t)));
+      ngx_pcalloc(pool_, sizeof(ngx_http_status_t)));
+
   if (status_ == NULL) {
     message_handler_->Message(kError,
                               "NgxFetch: ngx_pcalloc failed for status");
@@ -147,7 +387,7 @@ namespace net_instaweb {
 
   // Maybe we have a Proxy.
   ngx_url_t* tmp_url = &url_;
-    if (0 != fetcher_->proxy_.url.len) {
+  if (fetcher_->proxy_.url.len != 0) {
     tmp_url = &fetcher_->proxy_;
   }
 
@@ -172,6 +412,10 @@ namespace net_instaweb {
           kError, "NgxFetch: Couldn't start resolving, "
           "is there a proper resolver configured in nginx.conf?");
       return false;
+    } else {
+      ngx_log_error(NGX_LOG_DEBUG, log_, 0,
+                    "NgxFetch %p: start resolve for: %s\n",
+                    this, s_ipaddress.c_str());
     }
 
     resolver_ctx_->data = this;
@@ -182,7 +426,7 @@ namespace net_instaweb {
     resolver_ctx_->type = NGX_RESOLVE_A;
 #endif
 
-      resolver_ctx_->handler = NgxFetchResolveDone;
+    resolver_ctx_->handler = NgxFetch::ResolveDoneHandler;
     resolver_ctx_->timeout = fetcher_->resolver_timeout_;
 
     if (ngx_resolve_name(resolver_ctx_) != NGX_OK) {
@@ -197,15 +441,18 @@ namespace net_instaweb {
     }
   }
   return true;
-  }
+}
 
-  const char* NgxFetch::str_url() {
+const char* NgxFetch::str_url() {
   return str_url_.c_str();
-  }
+}
+
+// This function should be called only once. The only argument is success or
+// not.
+void NgxFetch::CallbackDone(bool success) {
+  ngx_log_error(NGX_LOG_DEBUG, log_, 0, "NgxFetch %p: CallbackDone: %s\n",
+                this, success ? "OK":"FAIL");
 
-  // This function should be called only once. The only argument is sucess or
-  // not.
-  void NgxFetch::CallbackDone(bool success) {
   if (async_fetch_ == NULL) {
     LOG(FATAL)
         << "BUG: NgxFetch callback called more than once on same fetch"
@@ -220,11 +467,36 @@ namespace net_instaweb {
     ngx_del_timer(timeout_event_);
     timeout_event_ = NULL;
   }
-    if (connection_) {
+
-      ngx_close_connection(connection_);
+  if (connection_ != NULL) {
+    // Connection will be re-used only on responses that specify
+    // 'Connection: keep-alive' in their headers.
+    bool keepalive = false;
+
+    if (success) {
+      ConstStringStarVector v;
+      if (async_fetch_->response_headers()->Lookup(
+              StringPiece(HttpAttributes::kConnection), &v)) {
+        for (size_t i = 0; i < v.size(); i++) {
+          if (*v[i] == "keep-alive") {
+            keepalive = true;
+            break;
+          } else if (*v[i] == "close") {
+            break;
+          }
+        }
+      }
+      ngx_log_error(NGX_LOG_DEBUG, log_, 0,
+                    "NgxFetch %p: connection %p attempt keep-alive: %s\n",
+                    this, connection_, keepalive ? "Yes":"No");
+    }
+
+    connection_->set_keepalive(keepalive);
+    connection_->Close();
     connection_ = NULL;
   }
 
+  // TODO(oschaaf): see https://github.com/pagespeed/ngx_pagespeed/pull/755
   async_fetch_->Done(success);
 
   if (fetcher_ != NULL) {
@@ -238,36 +510,36 @@ namespace net_instaweb {
   }
 
   async_fetch_ = NULL;
-  }
+}
 
-  size_t NgxFetch::bytes_received() {
+size_t NgxFetch::bytes_received() {
   return bytes_received_;
-  }
+}
-  void NgxFetch::bytes_received_add(int64 x) {
+void NgxFetch::bytes_received_add(int64 x) {
   bytes_received_ += x;
-  }
+}
 
-  int64 NgxFetch::fetch_start_ms() {
+int64 NgxFetch::fetch_start_ms() {
   return fetch_start_ms_;
-  }
+}
 
-  void NgxFetch::set_fetch_start_ms(int64 start_ms) {
+void NgxFetch::set_fetch_start_ms(int64 start_ms) {
   fetch_start_ms_ = start_ms;
-  }
+}
 
-  int64 NgxFetch::fetch_end_ms() {
+int64 NgxFetch::fetch_end_ms() {
   return fetch_end_ms_;
-  }
+}
 
-  void NgxFetch::set_fetch_end_ms(int64 end_ms) {
+void NgxFetch::set_fetch_end_ms(int64 end_ms) {
   fetch_end_ms_ = end_ms;
-  }
+}
 
-  MessageHandler* NgxFetch::message_handler() {
+MessageHandler* NgxFetch::message_handler() {
   return message_handler_;
-  }
+}
 
-  bool NgxFetch::ParseUrl() {
+bool NgxFetch::ParseUrl() {
   url_.url.len = str_url_.length();
   url_.url.data = static_cast<u_char*>(ngx_palloc(pool_, url_.url.len));
   if (url_.url.data == NULL) {
@@ -276,32 +548,64 @@ namespace net_instaweb {
   str_url_.copy(reinterpret_cast<char*>(url_.url.data), str_url_.length(), 0);
 
   return NgxUrlAsyncFetcher::ParseUrl(&url_, pool_);
-  }
+}
 
-  // Issue a request after the resolver is done
+// Issue a request after the resolver is done
-  void NgxFetch::NgxFetchResolveDone(ngx_resolver_ctx_t* resolver_ctx) {
+void NgxFetch::ResolveDoneHandler(ngx_resolver_ctx_t* resolver_ctx) {
   NgxFetch* fetch = static_cast<NgxFetch*>(resolver_ctx->data);
   NgxUrlAsyncFetcher* fetcher = fetch->fetcher_;
+
   if (resolver_ctx->state != NGX_OK) {
     if (fetch->timeout_event() != NULL && fetch->timeout_event()->timer_set) {
       ngx_del_timer(fetch->timeout_event());
       fetch->set_timeout_event(NULL);
     }
     fetch->message_handler()->Message(
-          kWarning, "NgxFetch: failed to resolve host [%.*s]",
+        kWarning, "NgxFetch %p: failed to resolve host [%.*s]", fetch,
         static_cast<int>(resolver_ctx->name.len), resolver_ctx->name.data);
     fetch->CallbackDone(false);
     return;
   }
+
+  ngx_uint_t i;
+  // Find the first ipv4 address. We don't support ipv6 yet.
+  for (i = 0; i < resolver_ctx->naddrs; i++) {
+    // Old versions of nginx and tengine have a different definition of addrs,
+    // work around to make sure we are using the right type (ngx_addr_t*).
+    ngx_addr_t* ngx_addrs = reinterpret_cast<ngx_addr_t*>(resolver_ctx->addrs);
+    if (typeid(*ngx_addrs) == typeid(*resolver_ctx->addrs)) {
+      if (reinterpret_cast<struct sockaddr_in*>(ngx_addrs[i].sockaddr)
+              ->sin_family == AF_INET) {
+        break;
+      }
+    } else {
+      // We're using an old version that uses in_addr_t* for addrs.
+      break;
+    }
+  }
+
+  // If no suitable ipv4 address was found, we fail.
+  if (i == resolver_ctx->naddrs) {
+    if (fetch->timeout_event() != NULL && fetch->timeout_event()->timer_set) {
+      ngx_del_timer(fetch->timeout_event());
+      fetch->set_timeout_event(NULL);
+    }
+    fetch->message_handler()->Message(
+        kWarning, "NgxFetch %p: no suitable address for host [%.*s]", fetch,
+        static_cast<int>(resolver_ctx->name.len), resolver_ctx->name.data);
+    fetch->CallbackDone(false);
+  }
+
   ngx_memzero(&fetch->sin_, sizeof(fetch->sin_));
 
 #if (nginx_version < 1005008)
-    fetch->sin_.sin_addr.s_addr = resolver_ctx->addrs[0];
+  fetch->sin_.sin_addr.s_addr = resolver_ctx->addrs[i];
 #else
-
   struct sockaddr_in* sin;
+
   sin = reinterpret_cast<struct sockaddr_in*>(
-        resolver_ctx->addrs[0].sockaddr);
+      resolver_ctx->addrs[i].sockaddr);
+
   fetch->sin_.sin_family = sin->sin_family;
   fetch->sin_.sin_addr.s_addr = sin->sin_addr.s_addr;
 #endif
@@ -316,10 +620,9 @@ namespace net_instaweb {
 
   char* ip_address = inet_ntoa(fetch->sin_.sin_addr);
 
-    fetch->message_handler()->Message(
+  ngx_log_error(NGX_LOG_DEBUG, fetch->log_, 0,
-        kInfo, "NgxFetch: Resolved host [%.*s] to [%s]",
+                "NgxFetch %p: Resolved host [%V] to [%s]", fetch,
-        static_cast<int>(resolver_ctx->name.len), resolver_ctx->name.data,
+                &resolver_ctx->name, ip_address);
-        ip_address);
 
   fetch->release_resolver();
 
@@ -327,10 +630,10 @@ namespace net_instaweb {
     fetch->message_handler()->Message(kError, "NgxFetch: InitRequest failed");
     fetch->CallbackDone(false);
   }
-  }
+}
 
-  // prepare the send data for this fetch, and hook write event.
+// Prepare the request data for this fetch, and hook the write event.
-  int NgxFetch::InitRequest() {
+int NgxFetch::InitRequest() {
   in_ = ngx_create_temp_buf(pool_, 4096);
   if (in_ == NULL) {
     return NGX_ERROR;
@@ -344,6 +647,13 @@ namespace net_instaweb {
   bool have_host = false;
   GoogleString port;
 
+  response_handler = NgxFetch::HandleStatusLine;
+  int rc = Connect();
+  if (rc == NGX_AGAIN || rc == NGX_OK) {
+    if (connection_->keepalive()) {
+      request_headers->Add(HttpAttributes::kConnection,
+                           NgxConnection::ka_header);
+    }
     const char* method = request_headers->method_string();
     size_t method_len = strlen(method);
 
@@ -361,8 +671,7 @@ namespace net_instaweb {
 
       // name: value\r\n
       size += request_headers->Name(i).length()
-           + request_headers->Value(i).length()
+          + request_headers->Value(i).length() + 4;  // 4 for ": \r\n"
-           + 4;  // for ": \r\n"
     }
 
     if (!have_host) {
@@ -402,20 +711,18 @@ namespace net_instaweb {
     }
     *(out_->last++) = CR;
     *(out_->last++) = LF;
-
-    response_handler = NgxFetchHandleStatusLine;
-    int rc = Connect();
     if (rc == NGX_AGAIN) {
       return NGX_OK;
+    }
   } else if (rc < NGX_OK) {
     return rc;
   }
-
+  CHECK(rc == NGX_OK);
-    NgxFetchWrite(connection_->write);
+  NgxFetch::ConnectionWriteHandler(connection_->c_->write);
   return NGX_OK;
-  }
+}
 
-  int NgxFetch::Connect() {
+int NgxFetch::Connect() {
   ngx_peer_connection_t pc;
   ngx_memzero(&pc, sizeof(pc));
   pc.sockaddr = (struct sockaddr*)&sin_;
@@ -428,29 +735,38 @@ namespace net_instaweb {
   pc.log = fetcher_->log_;
   pc.rcvbuf = -1;
 
-    int rc = ngx_event_connect_peer(&pc);
+
-    if (rc == NGX_ERROR || rc == NGX_DECLINED || rc == NGX_BUSY) {
+  connection_ = NgxConnection::Connect(&pc, message_handler(),
-      return rc;
+                                       fetcher_->max_keepalive_requests_);
+  ngx_log_error(NGX_LOG_DEBUG, fetcher_->log_, 0,
+                "NgxFetch %p Connect() connection %p for [%s]\n",
+                this, connection_, str_url());
+
+  if (connection_ == NULL) {
+    return NGX_ERROR;
   }
 
-    connection_ = pc.connection;
+  connection_->c_->write->handler = NgxFetch::ConnectionWriteHandler;
-    connection_->write->handler = NgxFetchWrite;
+  connection_->c_->read->handler = NgxFetch::ConnectionReadHandler;
-    connection_->read->handler = NgxFetchRead;
+  connection_->c_->data = this;
-    connection_->data = this;
 
   // Timer set in Init() is still in effect.
-    return rc;
+  return NGX_OK;
-  }
+}
 
-  // When the fetch sends the request completely, it will hook the read event,
+// When the fetch sends the request completely, it will hook the read event,
-  // and prepare to parse the response.
+// and prepare to parse the response.
-  void NgxFetch::NgxFetchWrite(ngx_event_t* wev) {
+void NgxFetch::ConnectionWriteHandler(ngx_event_t* wev) {
   ngx_connection_t* c = static_cast<ngx_connection_t*>(wev->data);
   NgxFetch* fetch = static_cast<NgxFetch*>(c->data);
   ngx_buf_t* out = fetch->out_;
 
   while (out->pos < out->last) {
     int n = c->send(c, out->pos, out->last - out->pos);
+    ngx_log_error(NGX_LOG_DEBUG, fetch->log_, 0,
+                  "NgxFetch %p: ConnectionWriteHandler "
+                  "send result %d", fetch, n);
+
     if (n >= 0) {
       out->pos += n;
     } else if (n == NGX_AGAIN) {
@@ -472,9 +788,9 @@ namespace net_instaweb {
   }
 
   return;
-  }
+}
 
-  void NgxFetch::NgxFetchRead(ngx_event_t* rev) {
+void NgxFetch::ConnectionReadHandler(ngx_event_t* rev) {
   ngx_connection_t* c = static_cast<ngx_connection_t*>(rev->data);
   NgxFetch* fetch = static_cast<NgxFetch*>(c->data);
 
@@ -482,6 +798,10 @@ namespace net_instaweb {
     int n = c->recv(
         c, fetch->in_->start, fetch->in_->end - fetch->in_->start);
 
+    ngx_log_error(NGX_LOG_DEBUG, fetch->log_, 0,
+                  "NgxFetch %p: ConnectionReadHandler "
+                  "recv result %d", fetch, n);
+
     if (n == NGX_AGAIN) {
       break;
     }
@@ -527,11 +847,14 @@ namespace net_instaweb {
   }
 
   // Timer set in Init() is still in effect.
-  }
+}
 
-  // Parse the status line: "HTTP/1.1 200 OK\r\n"
+// Parse the status line: "HTTP/1.1 200 OK\r\n"
-  bool NgxFetch::NgxFetchHandleStatusLine(ngx_connection_t* c) {
+bool NgxFetch::HandleStatusLine(ngx_connection_t* c) {
   NgxFetch* fetch = static_cast<NgxFetch*>(c->data);
+  ngx_log_error(NGX_LOG_DEBUG, fetch->log_, 0,
+                "NgxFetch %p: Handle status line\n", fetch);
+
   // This function only works after Nginx-1.1.4. Before nginx-1.1.4,
   // ngx_http_parse_status_line didn't save http_version.
   ngx_int_t n = ngx_http_parse_status_line(fetch->r_, fetch->in_,
@@ -549,17 +872,21 @@ namespace net_instaweb {
       static_cast<HttpStatus::Code>(fetch->get_status_code()));
   response_headers->set_major_version(fetch->get_major_version());
   response_headers->set_minor_version(fetch->get_minor_version());
-    fetch->set_response_handler(NgxFetchHandleHeader);
+  fetch->set_response_handler(NgxFetch::HandleHeader);
   return fetch->response_handler(c);
-  }
+}
 
-  // Parse the HTTP headers
+// Parse the HTTP headers
-  bool NgxFetch::NgxFetchHandleHeader(ngx_connection_t* c) {
+bool NgxFetch::HandleHeader(ngx_connection_t* c) {
   NgxFetch* fetch = static_cast<NgxFetch*>(c->data);
   char* data = reinterpret_cast<char*>(fetch->in_->pos);
   size_t size = fetch->in_->last - fetch->in_->pos;
   size_t n = fetch->parser_.ParseChunk(StringPiece(data, size),
                                        fetch->message_handler_);
+
+  ngx_log_error(NGX_LOG_DEBUG, fetch->log_, 0,
+                "NgxFetch %p: Handle headers\n", fetch);
+
   if (n > size) {
     return false;
   } else if (fetch->parser_.headers_complete()) {
@@ -571,6 +898,9 @@ namespace net_instaweb {
         return false;
       } else {
         fetch->content_length_known_ = true;
+        if (fetch->content_length_ == 0) {
+          fetch->done_ = true;
+        }
       }
     }
 
@@ -581,45 +911,54 @@ namespace net_instaweb {
     }
 
     fetch->in_->pos += n;
-      fetch->set_response_handler(NgxFetchHandleBody);
+    fetch->set_response_handler(NgxFetch::HandleBody);
+    if ((fetch->in_->last - fetch->in_->pos) > 0) {
       return fetch->response_handler(c);
     }
-    return true;
+  } else {
+    fetch->in_->pos += n;
   }
+  return true;
+}
 
-  // Read the response body
+// Read the response body
-  bool NgxFetch::NgxFetchHandleBody(ngx_connection_t* c) {
+bool NgxFetch::HandleBody(ngx_connection_t* c) {
   NgxFetch* fetch = static_cast<NgxFetch*>(c->data);
   char* data = reinterpret_cast<char*>(fetch->in_->pos);
   size_t size = fetch->in_->last - fetch->in_->pos;
-    if (size == 0) {
-      return true;
-    }
 
   fetch->bytes_received_add(size);
 
-    if (fetch->async_fetch_->Write(StringPiece(data, size),
+  ngx_log_error(NGX_LOG_DEBUG, fetch->log_, 0,
-        fetch->message_handler())) {
+                "NgxFetch %p: Handle body (%d bytes)\n", fetch, size);
-      if (fetch->content_length_known_ &&
+
-          fetch->bytes_received_ == fetch->content_length_) {
+  if ( fetch->async_fetch_->Write(StringPiece(data, size),
+                                  fetch->message_handler()) ) {
+    if (fetch->bytes_received_ == fetch->content_length_) {
       fetch->done_ = true;
     }
-      return true;
+    fetch->in_->pos += size;
-    }
+  } else {
+    ngx_log_error(NGX_LOG_DEBUG, fetch->log_, 0,
+                  "NgxFetch %p: async fetch write failure\n", fetch);
     return false;
   }
+  return true;
+}
 
-  void NgxFetch::NgxFetchTimeout(ngx_event_t* tev) {
+void NgxFetch::TimeoutHandler(ngx_event_t* tev) {
   NgxFetch* fetch = static_cast<NgxFetch*>(tev->data);
+  ngx_log_error(NGX_LOG_DEBUG, fetch->log_, 0,
+                "NgxFetch %p: TimeoutHandler called\n", fetch);
   fetch->CallbackDone(false);
-  }
+}
 
-  void NgxFetch::FixUserAgent() {
+void NgxFetch::FixUserAgent() {
   GoogleString user_agent;
   ConstStringStarVector v;
   RequestHeaders* request_headers = async_fetch_->request_headers();
   if (request_headers->Lookup(HttpAttributes::kUserAgent, &v)) {
-      for (int i = 0, n = v.size(); i < n; i++) {
+    for (size_t i = 0, n = v.size(); i < n; i++) {
       if (i != 0) {
         user_agent += " ";
       }
@@ -640,5 +979,6 @@ namespace net_instaweb {
     user_agent += version;
   }
   request_headers->Add(HttpAttributes::kUserAgent, user_agent);
-  }
+}
+
 }  // namespace net_instaweb

src/ngx_fetch.h

@@ -51,12 +51,13 @@ namespace net_instaweb {
 typedef bool (*response_handler_pt)(ngx_connection_t* c);
 
 class NgxUrlAsyncFetcher;
+class NgxConnection;
+
 class NgxFetch : public PoolElement<NgxFetch> {
  public:
   NgxFetch(const GoogleString& url,
            AsyncFetch* async_fetch,
            MessageHandler* message_handler,
-           ngx_msec_t timeout_ms,
            ngx_log_t* log);
   ~NgxFetch();
 
@@ -112,19 +113,19 @@ class NgxFetch : public PoolElement<NgxFetch> {
     response_handler = handler;
   }
   // Only the Static functions could be used in callbacks.
-  static void NgxFetchResolveDone(ngx_resolver_ctx_t* ctx);
+  static void ResolveDoneHandler(ngx_resolver_ctx_t* ctx);
   // Write the request.
-  static void NgxFetchWrite(ngx_event_t* wev);
+  static void ConnectionWriteHandler(ngx_event_t* wev);
   // Wait for the response.
-  static void NgxFetchRead(ngx_event_t* rev);
+  static void ConnectionReadHandler(ngx_event_t* rev);
   // Read and parse the first status line.
-  static bool NgxFetchHandleStatusLine(ngx_connection_t* c);
+  static bool HandleStatusLine(ngx_connection_t* c);
   // Read and parse the HTTP headers.
-  static bool NgxFetchHandleHeader(ngx_connection_t* c);
+  static bool HandleHeader(ngx_connection_t* c);
   // Read the response body.
-  static bool NgxFetchHandleBody(ngx_connection_t* c);
+  static bool HandleBody(ngx_connection_t* c);
   // Cancel the fetch when it's timeout.
-  static void NgxFetchTimeout(ngx_event_t* tev);
+  static void TimeoutHandler(ngx_event_t* tev);
 
   // Add the pagespeed User-Agent.
   void FixUserAgent();
@@ -139,7 +140,6 @@ class NgxFetch : public PoolElement<NgxFetch> {
   int64 bytes_received_;
   int64 fetch_start_ms_;
   int64 fetch_end_ms_;
-  int64 timeout_ms_;
   bool done_;
   int64 content_length_;
   bool content_length_known_;
@@ -152,7 +152,7 @@ class NgxFetch : public PoolElement<NgxFetch> {
   ngx_http_request_t* r_;
   ngx_http_status_t* status_;
   ngx_event_t* timeout_event_;
-  ngx_connection_t* connection_;
+  NgxConnection* connection_;
   ngx_resolver_ctx_t* resolver_ctx_;
 
   DISALLOW_COPY_AND_ASSIGN(NgxFetch);

src/ngx_gzip_setter.cc

@@ -70,7 +70,7 @@ extern "C" {
   }
 }
 
-NgxGZipSetter::NgxGZipSetter() : enabled_(0) { }
+NgxGZipSetter::NgxGZipSetter() : enabled_(false), initialized_(false) { }
 NgxGZipSetter::~NgxGZipSetter() { }
 
 // Helper functions to determine signature.
@@ -95,7 +95,7 @@ bool IsNgxBitmaskCommand(ngx_command_t* command) {
           HasLocalConfig(command));
 }
 
-// Initialize the NgxGzipSetter.
+// Initialize the NgxGZipSetter.
 // Find the gzip, gzip_vary, gzip_http_version and gzip_types commands in the
 // gzip module. Enable if the signature of the zip command matches with what we
 // trust. Also sets up redirects for the configurations. These redirect handle
@@ -105,6 +105,16 @@ void NgxGZipSetter::Init(ngx_conf_t* cf) {
 #if (NGX_HTTP_GZIP)
   bool gzip_signature_mismatch = false;
   bool other_signature_mismatch = false;
+  // If we initialized already we don't have to scan again.
+  if (initialized_) {
+    // Config might have changed, so re-enable if we have gzip.
+    if (gzip_command_.command_ != NULL) {
+      enabled_ = true;
+    } else {
+      enabled_ = false;
+    }
+    return;
+  } 
   for (int m = 0; ngx_modules[m] != NULL; m++) {
     if (ngx_modules[m]->commands != NULL) {
       for (int c = 0; ngx_modules[m]->commands[c].name.len; c++) {
@@ -122,7 +132,7 @@ void NgxGZipSetter::Init(ngx_conf_t* cf) {
             current_command->set = ngx_gzip_redirect_conf_set_flag_slot;
             gzip_command_.command_ = current_command;
             gzip_command_.module_ = ngx_modules[m];
-            enabled_ = 1;
+            enabled_ = true;
           } else {
             ngx_conf_log_error(
                 NGX_LOG_WARN, cf, 0,
@@ -189,6 +199,7 @@ void NgxGZipSetter::Init(ngx_conf_t* cf) {
       }
     }
   }
+  initialized_ = true;
   if (gzip_signature_mismatch) {
     return;  // Already logged error.
   } else if (!enabled_) {
@@ -381,7 +392,7 @@ void NgxGZipSetter::AddGZipHTTPTypes(ngx_conf_t* cf) {
 }
 
 void NgxGZipSetter::RollBackAndDisable(ngx_conf_t* cf) {
-  ngx_conf_log_error(NGX_LOG_INFO, cf, 0,
+  ngx_conf_log_error(NGX_LOG_DEBUG, cf, 0,
                      "pagespeed: rollback gzip, explicit configuration");
   for (std::vector<ngx_flag_t*>::iterator i = ngx_flags_set_.begin();
        i != ngx_flags_set_.end(); ++i) {

src/ngx_gzip_setter.h

@@ -91,6 +91,7 @@ class NgxGZipSetter {
   ngx_command_ctx gzip_vary_command_;
   ngx_command_ctx gzip_http_version_command_;
   bool enabled_;
+  bool initialized_;
 
  public:
   NgxGZipSetter();

src/ngx_message_handler.h

@@ -18,6 +18,10 @@
 #define NGX_MESSAGE_HANDLER_H_
 
 extern "C" {
+  #include <ngx_auto_config.h>
+#if (NGX_THREADS)
+  #include <ngx_thread.h>
+#endif
   #include <ngx_core.h>
   #include <ngx_log.h>
 }

src/ngx_pagespeed.cc

@@ -89,6 +89,8 @@ extern ngx_module_t ngx_pagespeed;
 namespace net_instaweb {
 
 const char* kInternalEtagName = "@psol-etag";
+bool factory_init_called = false;
+
 // The process context takes care of proactively initialising
 // a few libraries for us, some of which are not thread-safe
 // when they are initialized lazily.
@@ -286,11 +288,6 @@ void copy_response_headers_from_ngx(const ngx_http_request_t* r,
 
   headers->set_status_code(r->headers_out.status);
 
-  if (r->headers_out.location != NULL) {
-    headers->Add(HttpAttributes::kLocation,
-                 str_to_string_piece(r->headers_out.location->value));
-  }
-
   // Manually copy over the content type because it's not included in
   // request_->headers_out.headers.
   headers->Add(HttpAttributes::kContentType,
@@ -298,7 +295,8 @@ void copy_response_headers_from_ngx(const ngx_http_request_t* r,
 
   // When we don't have a date header, set one with the current time.
   if (headers->Lookup1(HttpAttributes::kDate) == NULL) {
-    headers->SetDate(ngx_current_msec);
+    PosixTimer timer;
+    headers->SetDate(timer.NowMs());
   }
 
   // TODO(oschaaf): ComputeCaching should be called in setupforhtml()?
@@ -345,10 +343,17 @@ ngx_int_t copy_response_headers_to_ngx(
 
     ngx_str_t name, value;
 
+    // If the gzip module is not configured, we must not rename the header,
+    // because we will fail to inject the header filter that will rename the
+    // header back.
+    bool gzip_enabled = false;
+#if (NGX_HTTP_GZIP)
+    gzip_enabled = true;
+#endif
     // To prevent the gzip module from clearing weak etags, we output them
     // using a different name here. The etag header filter module runs behind
     // the gzip compressors header filter, and will rename it to 'ETag'
-    if (StringCaseEqual(name_gs, "etag")
+    if (gzip_enabled && StringCaseEqual(name_gs, "etag")
         && StringCaseStartsWith(value_gs, "W/")) {
       name.len = strlen(kInternalEtagName);
       name.data = reinterpret_cast<u_char*>(
@@ -357,6 +362,7 @@ ngx_int_t copy_response_headers_to_ngx(
       name.len = name_gs.length();
       name.data = reinterpret_cast<u_char*>(const_cast<char*>(name_gs.data()));
     }
+
     value.len = value_gs.length();
     value.data = reinterpret_cast<u_char*>(const_cast<char*>(value_gs.data()));
 
@@ -610,6 +616,16 @@ char* ps_configure(ngx_conf_t* cf,
                    NgxRewriteOptions** options,
                    MessageHandler* handler,
                    net_instaweb::RewriteOptions::OptionScope option_scope) {
+  ps_main_conf_t* cfg_m = static_cast<ps_main_conf_t*>(
+      ngx_http_conf_get_module_main_conf(cf, ngx_pagespeed));
+
+  if (!factory_init_called) {
+    // Init logging to nginx's default error_log.
+    cfg_m->driver_factory->LoggingInit(cf->cycle->log);
+    cfg_m->driver_factory->Init();
+    factory_init_called = true;
+  }
+
   // args[0] is always "pagespeed"; ignore it.
   ngx_uint_t n_args = cf->args->nelts - 1;
 
@@ -660,8 +676,6 @@ char* ps_configure(ngx_conf_t* cf,
     // directive yet.  That happens below in ParseAndSetOptions().
   }
 
-  ps_main_conf_t* cfg_m = static_cast<ps_main_conf_t*>(
-      ngx_http_cycle_get_module_main_conf(cf->cycle, ngx_pagespeed));
   if (*options == NULL) {
     *options = new NgxRewriteOptions(
         cfg_m->driver_factory->thread_system());
@@ -730,7 +744,16 @@ void ps_cleanup_srv_conf(void* data) {
   // from being executed
 
   if (!factory_deleted && cfg_s->server_context != NULL) {
-    delete cfg_s->server_context->factory();
+    NgxRewriteDriverFactory* factory = dynamic_cast<NgxRewriteDriverFactory*>(
+      cfg_s->server_context->factory());
+
+    if (!factory_init_called) {
+      factory->LoggingInit(ngx_cycle->log);
+      factory->Init();
+      factory_init_called = true;
+    }
+
+    delete factory;
     factory_deleted = true;
   }
   if (cfg_s->proxy_fetch_factory != NULL) {
@@ -800,7 +823,7 @@ void* ps_create_main_conf(ngx_conf_t* cf) {
       new SystemThreadSystem(),
       "" /* hostname, not used */,
       -1 /* port, not used */);
-  cfg_m->driver_factory->Init();
+  factory_init_called = false;
   ps_set_conf_cleanup_handler(cf, ps_cleanup_main_conf, cfg_m);
   return cfg_m;
 }
@@ -1178,6 +1201,12 @@ ngx_int_t ps_base_fetch_handler(ngx_http_request_t* r) {
                 STR_CASE_EQ_LITERAL(header->key, "Last-Modified") ||
                 STR_CASE_EQ_LITERAL(header->key, "Expires"))))) {
           header->hash = 0;
+          if (STR_CASE_EQ_LITERAL(header->key, "Location")) {
+            // There's a possible issue with the location header, where setting
+            // the hash to 0 is not enough. See:
+            // https://github.com/nginx/nginx/blob/master/src/http/ngx_http_header_filter_module.c#L314
+            r->headers_out.location = NULL;
+          }
         }
       }
     } else {
@@ -1280,6 +1309,15 @@ void ps_connection_read_handler(ngx_event_t* ev) {
     rc = read(c->fd, chr, 256);
   } while (rc > 0 || (rc == -1 && errno == EINTR));  // Retry on EINTR.
 
+  if (r->connection->error) {
+    ngx_log_error(NGX_LOG_DEBUG, ngx_cycle->log, 0,
+                  "pagespeed [%p] request already finalized", r);
+    ctx->pagespeed_connection = NULL;
+    ngx_close_connection(c);
+    ngx_http_finalize_request(r, NGX_ERROR);
+    return;
+  }
+
   if (rc == -1 && errno != EAGAIN && errno != EWOULDBLOCK) {
     ctx->pagespeed_connection = NULL;
     ngx_close_connection(c);
@@ -1601,7 +1639,8 @@ void ps_release_base_fetch(ps_request_ctx_t* ctx) {
 
 // TODO(chaizhenhua): merge into NgxBaseFetch ctor
 ngx_int_t ps_create_base_fetch(ps_request_ctx_t* ctx,
-                               RequestContextPtr request_context) {
+                               RequestContextPtr request_context,
+                               RequestHeaders* request_headers) {
   ngx_http_request_t* r = ctx->r;
   ps_srv_conf_t* cfg_s = ps_get_srv_config(r);
   int file_descriptors[2];
@@ -1644,6 +1683,7 @@ ngx_int_t ps_create_base_fetch(ps_request_ctx_t* ctx,
   ctx->base_fetch = new NgxBaseFetch(
       r, file_descriptors[1], cfg_s->server_context,
       request_context, ctx->preserve_caching_headers);
+  ctx->base_fetch->SetRequestHeadersTakingOwnership(request_headers);
 
   return NGX_OK;
 }
@@ -1779,7 +1819,10 @@ ngx_int_t ps_resource_handler(ngx_http_request_t* r,
   GoogleString url_string = ps_determine_url(r);
   GoogleUrl url(url_string);
 
-  CHECK(url.IsWebValid());
+  if (!url.IsWebValid()) {
+    ngx_log_error(NGX_LOG_ERR, r->connection->log, 0, "invalid url");
+    return NGX_DECLINED;
+  }
 
   scoped_ptr<RequestHeaders> request_headers(new RequestHeaders);
   scoped_ptr<ResponseHeaders> response_headers(new ResponseHeaders);
@@ -1844,12 +1887,19 @@ ngx_int_t ps_resource_handler(ngx_http_request_t* r,
     CHECK(ctx == NULL);
     ctx = new ps_request_ctx_t();
 
+    ctx->base_fetch = NULL;
+    ctx->pagespeed_connection = NULL;
     ctx->r = r;
-    ctx->write_pending = false;
     ctx->html_rewrite = false;
     ctx->in_place = false;
-    ctx->pagespeed_connection = NULL;
+    ctx->write_pending = false;
+    ctx->fetch_done = false;
     ctx->preserve_caching_headers = kDontPreserveHeaders;
+    ctx->proxy_fetch = NULL;
+    ctx->inflater_ = NULL;
+    ctx->driver = NULL;
+    ctx->recorder = NULL;
+    ctx->ipro_response_headers = NULL;
 
     // See build_context_for_request() in mod_instaweb.cc
     // TODO(jefftk): Is this the right place to be modifying caching headers for
@@ -1887,35 +1937,17 @@ ngx_int_t ps_resource_handler(ngx_http_request_t* r,
     ngx_http_set_ctx(r, ctx, ngx_pagespeed);
   }
 
-  if (ps_create_base_fetch(ctx, request_context) != NGX_OK) {
-    // Do not need to release request context 'ctx'.
-    // http_pool_cleanup will call ps_release_request_context
-    return NGX_ERROR;
-  }
-
-  ctx->base_fetch->SetRequestHeadersTakingOwnership(request_headers.release());
-
-  bool page_callback_added = false;
-  scoped_ptr<ProxyFetchPropertyCallbackCollector>
-      property_callback(
-          ProxyFetchFactory::InitiatePropertyCacheLookup(
-              !html_rewrite /* is_resource_fetch */,
-              url,
-              cfg_s->server_context,
-              options,
-              ctx->base_fetch,
-              false /* requires_blink_cohort (no longer unused) */,
-              &page_callback_added));
-
   if (pagespeed_resource) {
     // TODO(jefftk): Set using_spdy appropriately.  See
     // ProxyInterface::ProxyRequestCallback
+    ps_create_base_fetch(ctx, request_context, request_headers.release());
     ResourceFetch::Start(
         url,
         custom_options.release() /* null if there aren't custom options */,
         false /* using_spdy */, cfg_s->server_context, ctx->base_fetch);
     return ps_async_wait_response(r);
   } else if (is_an_admin_handler) {
+    ps_create_base_fetch(ctx, request_context, request_headers.release());
     QueryParams query_params;
     query_params.ParseFromUrl(url);
 
@@ -1959,6 +1991,7 @@ ngx_int_t ps_resource_handler(ngx_http_request_t* r,
   }
 
   if (html_rewrite) {
+    ps_create_base_fetch(ctx, request_context, request_headers.release());
     // Do not store driver in request_context, it's not safe.
     RewriteDriver* driver;
 
@@ -1985,12 +2018,22 @@ ngx_int_t ps_resource_handler(ngx_http_request_t* r,
     driver->set_pagespeed_option_cookies(pagespeed_option_cookies);
 
     // TODO(jefftk): FlushEarlyFlow would go here.
+    bool page_callback_added = false;
+    ProxyFetchPropertyCallbackCollector* property_callback =
+        ProxyFetchFactory::InitiatePropertyCacheLookup(
+            !html_rewrite /* is_resource_fetch */,
+            url,
+            cfg_s->server_context,
+            options,
+            ctx->base_fetch,
+            false /* requires_blink_cohort (no longer unused) */,
+            &page_callback_added);
 
     // Will call StartParse etc.  The rewrite driver will take care of deleting
     // itself if necessary.
     ctx->proxy_fetch = cfg_s->proxy_fetch_factory->CreateNewProxyFetch(
         url_string, ctx->base_fetch, driver,
-        property_callback.release(),
+        property_callback,
         NULL /* original_content_fetch */);
     return NGX_OK;
   }
@@ -1998,6 +2041,7 @@ ngx_int_t ps_resource_handler(ngx_http_request_t* r,
   if (options->in_place_rewriting_enabled() &&
       options->enabled() &&
       options->IsAllowed(url.Spec())) {
+    ps_create_base_fetch(ctx, request_context, request_headers.release());
     // Do not store driver in request_context, it's not safe.
     RewriteDriver* driver;
     if (custom_options.get() == NULL) {
@@ -2023,7 +2067,7 @@ ngx_int_t ps_resource_handler(ngx_http_request_t* r,
         url_string.c_str());
 
     ctx->in_place = true;
-    ctx->base_fetch->set_handle_error(false);
+    ctx->base_fetch->set_ipro_lookup(true);
     ctx->driver->FetchInPlaceResource(
         url, false /* proxy_mode */, ctx->base_fetch);
 
@@ -2037,8 +2081,7 @@ ngx_int_t ps_resource_handler(ngx_http_request_t* r,
                 "Passing on content handling for non-pagespeed resource '%s'",
                 url_string.c_str());
 
-  ctx->base_fetch->Done(false);
+  CHECK(ctx->base_fetch == NULL);
-  ps_release_base_fetch(ctx);
   // set html_rewrite flag.
   ctx->html_rewrite = true;
   return NGX_DECLINED;
@@ -2243,14 +2286,13 @@ ngx_int_t ps_html_rewrite_header_filter(ngx_http_request_t* r) {
     if (!ps_has_stacked_content_encoding(r)) {
       StringPiece content_encoding =
           str_to_string_piece(r->headers_out.content_encoding->value);
-      GzipInflater::InflateType inflate_type;
+      GzipInflater::InflateType inflate_type = GzipInflater::kGzip;
       bool is_encoded = false;
       if (StringCaseEqual(content_encoding, "deflate")) {
         is_encoded = true;
         inflate_type = GzipInflater::kDeflate;
       } else if (StringCaseEqual(content_encoding, "gzip")) {
         is_encoded = true;
-        inflate_type = GzipInflater::kGzip;
       }
 
       if (is_encoded) {
@@ -3074,10 +3116,15 @@ ngx_int_t ps_init_module(ngx_cycle_t* cycle) {
           "UseNativeFetcher is on, please configure a resolver.");
       return NGX_ERROR;
     }
-
+    // Update logging to the configured error_log in the http{} block.
     cfg_m->driver_factory->LoggingInit(cycle->log);
     cfg_m->driver_factory->RootInit();
   } else {
+    if (!factory_init_called) {
+      cfg_m->driver_factory->LoggingInit(cycle->log);
+      cfg_m->driver_factory->Init();
+      factory_init_called = true;
+    }
     delete cfg_m->driver_factory;
     cfg_m->driver_factory = NULL;
   }
@@ -3114,11 +3161,11 @@ ngx_int_t ps_init_child_process(ngx_cycle_t* cycle) {
     // Some server{} blocks may not have a ServerContext in that case we must
     // not instantiate a ProxyFetchFactory.
     if (cfg_s->server_context != NULL) {
-      cfg_s->proxy_fetch_factory = new ProxyFetchFactory(cfg_s->server_context);
       ngx_http_core_loc_conf_t* clcf = static_cast<ngx_http_core_loc_conf_t*>(
           cscfp[s]->ctx->loc_conf[ngx_http_core_module.ctx_index]);
       cfg_m->driver_factory->SetServerContextMessageHandler(
           cfg_s->server_context, clcf->error_log);
+      cfg_s->proxy_fetch_factory = new ProxyFetchFactory(cfg_s->server_context);
     }
   }
 

src/ngx_rewrite_driver_factory.cc

@@ -78,6 +78,8 @@ NgxRewriteDriverFactory::NgxRewriteDriverFactory(
       log_(NULL),
       resolver_timeout_(NGX_CONF_UNSET_MSEC),
       use_native_fetcher_(false),
+      // 100 Aligns to nginx's server-side default.
+      native_fetcher_max_keepalive_requests_(100),
       ngx_shared_circular_buffer_(NULL),
       hostname_(hostname.as_string()),
       port_(port),
@@ -112,6 +114,7 @@ UrlAsyncFetcher* NgxRewriteDriverFactory::AllocateFetcher(
         resolver_timeout_,
         config->blocking_fetch_timeout_ms(),
         resolver_,
+        native_fetcher_max_keepalive_requests_,
         thread_system(),
         message_handler());
     ngx_url_async_fetchers_.push_back(fetcher);

src/ngx_rewrite_driver_factory.h

@@ -20,6 +20,10 @@
 #define NGX_REWRITE_DRIVER_FACTORY_H_
 
 extern "C" {
+  #include <ngx_auto_config.h>
+#if (NGX_THREADS)
+  #include <ngx_thread.h>
+#endif
   #include <ngx_core.h>
   #include <ngx_http.h>
   #include <ngx_config.h>
@@ -105,6 +109,12 @@ class NgxRewriteDriverFactory : public SystemRewriteDriverFactory {
   void set_use_native_fetcher(bool x) {
     use_native_fetcher_ = x;
   }
+  int native_fetcher_max_keepalive_requests() {
+    return native_fetcher_max_keepalive_requests_;
+  }
+  void set_native_fetcher_max_keepalive_requests(int x) {
+    native_fetcher_max_keepalive_requests_ = x;
+  }
   bool process_script_variables() {
     return process_script_variables_;
   }
@@ -140,6 +150,8 @@ class NgxRewriteDriverFactory : public SystemRewriteDriverFactory {
   ngx_msec_t resolver_timeout_;
   ngx_resolver_t* resolver_;
   bool use_native_fetcher_;
+  int native_fetcher_max_keepalive_requests_;
+
   typedef std::set<NgxMessageHandler*> NgxMessageHandlerSet;
   NgxMessageHandlerSet server_context_message_handlers_;
 

src/ngx_rewrite_options.cc

@@ -72,12 +72,14 @@ const char* const server_only_options[] = {
   "LoadFromFileMatch",
   "LoadFromFileRule",
   "LoadFromFileRuleMatch",
-  "UseNativeFetcher"
+  "UseNativeFetcher",
+  "NativeFetcherMaxKeepaliveRequests"
 };
 
 // Options that can only be used in the main (http) option scope.
 const char* const main_only_options[] = {
-  "UseNativeFetcher"
+  "UseNativeFetcher",
+  "NativeFetcherMaxKeepaliveRequests"
 };
 
 }  // namespace
@@ -274,10 +276,15 @@ const char* NgxRewriteOptions::ParseAndSetOptions(
 
   ScriptLine* script_line;
   script_line = NULL;
-  // Only allow script variable support for LoadFromFile for now.
   // Note that LoadFromFile should not be scriptable on wildcard hosts,
   // as browsers might be able to manipulate its natural use-case: $http_host.
-  if (!StringCaseStartsWith(directive, "LoadFromFile")) {
+  if (!StringCaseStartsWith(directive, "LoadFromFile") &&
+      !StringCaseEqual(directive, "EnableFilters") &&
+      !StringCaseEqual(directive, "DisableFilters") &&
+      !StringCaseEqual(directive, "DownstreamCachePurgeLocationPrefix") &&
+      !StringCaseEqual(directive, "DownstreamCachePurgeMethod") &&
+      !StringCaseEqual(directive,
+                       "DownstreamCacheRewrittenPercentageThreshold")) {
     compile_scripts = false;
   }
 
@@ -343,6 +350,16 @@ const char* NgxRewriteOptions::ParseAndSetOptions(
       result = ParseAndSetOptionHelper<NgxRewriteDriverFactory>(
           arg, driver_factory,
           &NgxRewriteDriverFactory::set_use_native_fetcher);
+    } else if (IsDirective(directive, "NativeFetcherMaxKeepaliveRequests")) {
+      int max_keepalive_requests;
+      if (StringToInt(arg, &max_keepalive_requests) &&
+          max_keepalive_requests > 0) {
+        driver_factory->set_native_fetcher_max_keepalive_requests(
+            max_keepalive_requests);
+        result = RewriteOptions::kOptionOk;
+      } else {
+        result = RewriteOptions::kOptionValueInvalid;
+      }
     } else if (StringCaseEqual("ProcessScriptVariables", args[0])) {
       if (scope == RewriteOptions::kProcessScopeStrict) {
         if (StringCaseEqual(arg, "on")) {
@@ -498,6 +515,10 @@ NgxRewriteOptions* NgxRewriteOptions::Clone() const {
   return options;
 }
 
+void NgxRewriteOptions::Merge(const RewriteOptions& src) {
+  SystemRewriteOptions::Merge(src);
+}
+
 const NgxRewriteOptions* NgxRewriteOptions::DynamicCast(
     const RewriteOptions* instance) {
   return dynamic_cast<const NgxRewriteOptions*>(instance);

src/ngx_rewrite_options.h

@@ -140,6 +140,7 @@ class NgxRewriteOptions : public SystemRewriteOptions {
 
   // Make an identical copy of these options and return it.
   virtual NgxRewriteOptions* Clone() const;
+  virtual void Merge(const RewriteOptions& src);
 
   // Returns a suitably down cast version of 'instance' if it is an instance
   // of this class, NULL if not.

src/ngx_server_context.cc

@@ -48,7 +48,7 @@ SystemRequestContext* NgxServerContext::NewRequestContext(
     ngx_http_request_t* r) {
   // Based on ngx_http_variable_server_port.
   bool port_set = false;
-  int local_port;
+  int local_port = 0;
 #if (NGX_HAVE_INET6)
   if (r->connection->local_sockaddr->sa_family == AF_INET6) {
     local_port = ntohs(reinterpret_cast<struct sockaddr_in6*>(

src/ngx_url_async_fetcher.cc

@@ -55,6 +55,7 @@ namespace net_instaweb {
                                          ngx_msec_t resolver_timeout,
                                          ngx_msec_t fetch_timeout,
                                          ngx_resolver_t* resolver,
+                                         int max_keepalive_requests,
                                          ThreadSystem* thread_system,
                                          MessageHandler* handler)
     : fetchers_count_(0),
@@ -63,7 +64,8 @@ namespace net_instaweb {
       byte_count_(0),
       thread_system_(thread_system),
       message_handler_(handler),
-      mutex_(NULL) {
+      mutex_(NULL),
+      max_keepalive_requests_(max_keepalive_requests) {
     resolver_timeout_ = resolver_timeout;
     fetch_timeout_ = fetch_timeout;
     ngx_memzero(&proxy_, sizeof(proxy_));
@@ -223,7 +225,7 @@ namespace net_instaweb {
                                  AsyncFetch* async_fetch) {
     async_fetch = EnableInflation(async_fetch);
     NgxFetch* fetch = new NgxFetch(url, async_fetch,
-          message_handler, fetch_timeout_, log_);
+          message_handler, log_);
     ScopedMutex lock(mutex_);
     pending_fetches_.Add(fetch);
     SendCmd('F');

src/ngx_url_async_fetcher.h

@@ -53,12 +53,13 @@ class NgxUrlAsyncFetcher : public UrlAsyncFetcher {
   NgxUrlAsyncFetcher(
       const char* proxy, ngx_log_t* log, ngx_msec_t resolver_timeout,
       ngx_msec_t fetch_timeout, ngx_resolver_t* resolver,
-      ThreadSystem* thread_system, MessageHandler* handler);
+      int max_keepalive_requests, ThreadSystem* thread_system,
+      MessageHandler* handler);
 
   ~NgxUrlAsyncFetcher();
 
   // It should be called in the module init_process callback function. Do some
-  // intializations which can't be done in the master process
+  // initializations which can't be done in the master process
   bool Init();
 
   // shutdown all the fetches.
@@ -139,6 +140,7 @@ class NgxUrlAsyncFetcher : public UrlAsyncFetcher {
   ngx_connection_t* command_connection_;  // the command pipe
   int pipe_fd_;  // the write pipe end
   ngx_resolver_t* resolver_;
+  int max_keepalive_requests_;
   ngx_msec_t resolver_timeout_;
   ngx_msec_t fetch_timeout_;
 

test/pagespeed_test.conf.template

@@ -38,6 +38,10 @@ http {
   pagespeed StaticAssetPrefix /pagespeed_custom_static/;
 
   pagespeed MessageBufferSize 200000;
+  # Increase the default fetcher timeout to resolve sporadic flakeyness when
+  # the native fetcher uses 8.8.8.8 to resolve.
+  pagespeed FetcherTimeoutMs 10000;
+  pagespeed NativeFetcherMaxKeepaliveRequests 50;
 
   root "@@SERVER_ROOT@@";