IPRO: Fix a race in processing writes() from NgxBaseFetch

For FetchInPlaceResource, NgxBaseFetch would send two bytes down its
pipe, one upon HeaderComplete() and one upon HandleDone(). We need
only one to resume processing on the nginx side.

There is a race between ps_connection_read_handler() and processing
of the byte send by NgxBaseFetch::HandleDone().
ps_connection_read_handler() clears the pipe when the request is
finalized, and also drains it on each event - so two writes could be
processed as one when lucky, masking the problem).

One concrete problem this solved for me was that SPDY + IPRO +
proxy_pass would segfault, hang, and/or pass on 5xx/404 responses
from IPRO lookup fetches to the browser, next to alerts about
r->count being zero in nginx's error.log

Might fix https://github.com/pagespeed/ngx_pagespeed/issues/788
Fixes https://github.com/pagespeed/ngx_pagespeed/issues/792

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);
     }
   }
 
-  RequestCollection();  // Headers available.
+  // 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_pagespeed.cc

@@ -2024,7 +2024,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);