Readd pthread_mutex.h include on ngx_fetch.cc.

Messed up the merge of 0290f52a88 originally, add the include back.

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 "     $ tar -xzvf 1.8.31.2.tar.gz # expands to psol/"
+    echo "     $ wget https://dl.google.com/dl/page-speed/psol/1.9.32.2.tar.gz"
+    echo "     $ tar -xzvf 1.9.32.2.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

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, supress 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>
 }
@@ -48,6 +56,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 +65,231 @@ 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;
+  }
+  DCHECK(false) << "NgxFetch: Unexpected write event" << n;
+}
+
+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.
-  bool NgxFetch::Start(NgxUrlAsyncFetcher* fetcher) {
+// This function is called by NgxUrlAsyncFetcher::StartFetch.
+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
-  // hook the DNS resolver if needed. Else we connect directly.
-  // When this returns false, our caller (NgxUrlAsyncFetcher::StartFetch)
-  // will call fetch->CallbackDone()
-  bool NgxFetch::Init() {
+// Create the pool, parse the url, add the timeout event and
+// hook the DNS resolver if needed. Else we connect directly.
+// When this returns false, our caller (NgxUrlAsyncFetcher::StartFetch)
+// will call fetch->CallbackDone()
+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_,
-                                           sizeof(ngx_http_request_t)));
+  r_ = static_cast<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_,
-                                              sizeof(ngx_http_status_t)));
+  status_ = static_cast<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 sucess 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,56 @@ 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
-  void NgxFetch::NgxFetchResolveDone(ngx_resolver_ctx_t* resolver_ctx) {
+// Issue a request after the resolver is done
+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++) {
+    if (reinterpret_cast<struct sockaddr_in*>(
+            resolver_ctx->addrs[i].sockaddr)->sin_family == AF_INET) {
+      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 +612,9 @@ namespace net_instaweb {
 
   char* ip_address = inet_ntoa(fetch->sin_.sin_addr);
 
-    fetch->message_handler()->Message(
-        kInfo, "NgxFetch: Resolved host [%.*s] to [%s]",
-        static_cast<int>(resolver_ctx->name.len), resolver_ctx->name.data,
-        ip_address);
+  ngx_log_error(NGX_LOG_DEBUG, fetch->log_, 0,
+                "NgxFetch %p: Resolved host [%V] to [%s]", fetch,
+                &resolver_ctx->name, ip_address);
 
   fetch->release_resolver();
 
@@ -327,10 +622,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.
-  int NgxFetch::InitRequest() {
+// Prepare the request data for this fetch, and hook the write event.
+int NgxFetch::InitRequest() {
   in_ = ngx_create_temp_buf(pool_, 4096);
   if (in_ == NULL) {
     return NGX_ERROR;
@@ -344,6 +639,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 +663,7 @@ namespace net_instaweb {
 
       // name: value\r\n
       size += request_headers->Name(i).length()
-           + request_headers->Value(i).length()
-           + 4;  // for ": \r\n"
+          + request_headers->Value(i).length() + 4;  // 4 for ": \r\n"
     }
 
     if (!have_host) {
@@ -402,20 +703,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;
   }
-
-    NgxFetchWrite(connection_->write);
+  CHECK(rc == NGX_OK);
+  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 +727,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) {
-      return rc;
+
+  connection_ = NgxConnection::Connect(&pc, message_handler(),
+                                       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_->write->handler = NgxFetchWrite;
-    connection_->read->handler = NgxFetchRead;
-    connection_->data = this;
+  connection_->c_->write->handler = NgxFetch::ConnectionWriteHandler;
+  connection_->c_->read->handler = NgxFetch::ConnectionReadHandler;
+  connection_->c_->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,
-  // and prepare to parse the response.
-  void NgxFetch::NgxFetchWrite(ngx_event_t* wev) {
+// When the fetch sends the request completely, it will hook the read event,
+// and prepare to parse the response.
+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 +780,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 +790,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 +839,14 @@ namespace net_instaweb {
   }
 
   // Timer set in Init() is still in effect.
-  }
+}
 
-  // Parse the status line: "HTTP/1.1 200 OK\r\n"
-  bool NgxFetch::NgxFetchHandleStatusLine(ngx_connection_t* c) {
+// Parse the status line: "HTTP/1.1 200 OK\r\n"
+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 +864,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
-  bool NgxFetch::NgxFetchHandleHeader(ngx_connection_t* c) {
+// Parse the HTTP headers
+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 +890,9 @@ namespace net_instaweb {
         return false;
       } else {
         fetch->content_length_known_ = true;
+        if (fetch->content_length_ == 0) {
+          fetch->done_ = true;
+        }
       }
     }
 
@@ -581,45 +903,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
-  bool NgxFetch::NgxFetchHandleBody(ngx_connection_t* c) {
+// Read the response body
+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),
-        fetch->message_handler())) {
-      if (fetch->content_length_known_ &&
-          fetch->bytes_received_ == fetch->content_length_) {
+  ngx_log_error(NGX_LOG_DEBUG, fetch->log_, 0,
+                "NgxFetch %p: Handle body (%d bytes)\n", fetch, size);
+
+  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 +971,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_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,
@@ -610,6 +607,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 +667,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 +735,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 +814,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 +1192,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 +1300,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);
@@ -1844,12 +1873,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
@@ -2023,7 +2059,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);
 
@@ -3074,10 +3110,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 +3155,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

@@ -105,6 +105,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 +146,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
@@ -343,6 +345,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 +510,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_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,7 +53,8 @@ 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();
 
@@ -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@@";