Merge pull request #1137 from pagespeed/32_apache_buffering

backport apache fetch buffering to branch 32

Author: crowell

net/instaweb/apache/apache_slurp.cc

@@ -270,7 +270,7 @@ void SlurpUrl(ApacheServerContext* server_context, request_rec* r) {
     // We always disable downstream header filters when sending out
     // slurped resources, since we've captured them from the origin
     // in the fetch we did to write the slurp.
-    ApacheWriter apache_writer(r);
+    ApacheWriter apache_writer(r, server_context->thread_system());
     apache_writer.set_disable_downstream_header_filters(true);
     ChunkingWriter chunking_writer(
         &apache_writer, global_config->slurp_flush_limit());

net/instaweb/apache/apache_writer.cc

@@ -27,33 +27,35 @@
 
 namespace net_instaweb {
 
-ApacheWriter::ApacheWriter(request_rec* r)
+ApacheWriter::ApacheWriter(request_rec* r, ThreadSystem* thread_system)
     : request_(r),
       headers_out_(false),
       disable_downstream_header_filters_(false),
       strip_cookies_(false),
-      squelch_output_(false),
-      content_length_(AsyncFetch::kContentLengthUnknown) {
+      content_length_(AsyncFetch::kContentLengthUnknown),
+      thread_system_(thread_system) {
+  apache_request_thread_.reset(thread_system->GetThreadId());
 }
 
 ApacheWriter::~ApacheWriter() {
 }
 
 bool ApacheWriter::Write(const StringPiece& str, MessageHandler* handler) {
+  DCHECK(apache_request_thread_->IsCurrentThread());
   DCHECK(headers_out_);
-  if (!squelch_output_) {
-    ap_rwrite(str.data(), str.size(), request_);
-  }
+  ap_rwrite(str.data(), str.size(), request_);
   return true;
 }
 
 bool ApacheWriter::Flush(MessageHandler* handler) {
+  DCHECK(apache_request_thread_->IsCurrentThread());
   DCHECK(headers_out_);
   ap_rflush(request_);
   return true;
 }
 
 void ApacheWriter::OutputHeaders(ResponseHeaders* response_headers) {
+  DCHECK(apache_request_thread_->IsCurrentThread());
   DCHECK(!headers_out_);
   if (headers_out_) {
     return;

net/instaweb/apache/apache_writer.h

@@ -18,7 +18,9 @@
 #define NET_INSTAWEB_APACHE_APACHE_WRITER_H_
 
 #include "net/instaweb/util/public/basictypes.h"
+#include "net/instaweb/util/public/scoped_ptr.h"
 #include "net/instaweb/util/public/string_util.h"
+#include "net/instaweb/util/public/thread_system.h"
 #include "net/instaweb/util/public/writer.h"
 
 struct request_rec;
@@ -28,10 +30,12 @@ namespace net_instaweb {
 class MessageHandler;
 class ResponseHeaders;
 
-// Writer object that writes to an Apache Request stream.
+// Writer object that writes to an Apache Request stream.  Should only be used
+// from a single apache request thread, not from a rewrite thread or anything
+// else.
 class ApacheWriter : public Writer {
  public:
-  explicit ApacheWriter(request_rec* r);
+  ApacheWriter(request_rec* r, ThreadSystem* thread_system);
   virtual ~ApacheWriter();
 
   virtual bool Write(const StringPiece& str, MessageHandler* handler);
@@ -61,18 +65,14 @@ class ApacheWriter : public Writer {
     strip_cookies_ = x;
   }
 
-  // When proxying content we deem to be unsafe (e.g. lacking
-  // a Content-Type header) we must squelch the output.
-  void set_squelch_output(bool x) { squelch_output_ = true; }
-  bool squelch_output() const { return squelch_output_; }
-
  private:
   request_rec* request_;
   bool headers_out_;
   bool disable_downstream_header_filters_;
   bool strip_cookies_;
-  bool squelch_output_;
   int64 content_length_;
+  ThreadSystem* thread_system_;
+  scoped_ptr<ThreadSystem::ThreadId> apache_request_thread_;
 
   DISALLOW_COPY_AND_ASSIGN(ApacheWriter);
 };

net/instaweb/apache/instaweb_handler.cc

@@ -128,7 +128,7 @@ InstawebHandler::~InstawebHandler() {
   // If fetch_ is null we either never tried to fetch anything or it took
   // ownership of itself after timing out.
   if (fetch_ != NULL) {
-    if (WaitForFetch() == ApacheFetch::kWaitSuccess) {
+    if (WaitForFetch()) {
       delete fetch_;  // Fetch completed normally.
     } else {
       // Fetch took ownership of itself and will continue in the background.
@@ -137,19 +137,19 @@ InstawebHandler::~InstawebHandler() {
   }
 }
 
-ApacheFetch::WaitResult InstawebHandler::WaitForFetch() {
+bool InstawebHandler::WaitForFetch() {
   if (fetch_ == NULL) {
-    return ApacheFetch::kWaitSuccess;  // Nothing to wait for.
+    return true;  // Nothing to wait for.
   }
 
-  ApacheFetch::WaitResult wait_result = fetch_->Wait(rewrite_driver_);
-  if (wait_result != ApacheFetch::kWaitSuccess) {
+  bool ok = fetch_->Wait(rewrite_driver_);
+  if (!ok) {
     // Give up on waiting for the fetch so we stop tying up a thread.  Fetch has
     // taken ownership of itself, will discard any messages it receives if it's
     // abandoned, and will delete itself when Done() is called, if ever.
     fetch_ = NULL;
   }
-  return wait_result;
+  return ok;
 }
 
 void InstawebHandler::SetupSpdyConnectionIfNeeded() {
@@ -179,16 +179,19 @@ RewriteDriver* InstawebHandler::MakeDriver() {
 }
 
 ApacheFetch* InstawebHandler::MakeFetch(const GoogleString& url,
+                                        bool buffered,
                                         StringPiece debug_info) {
   DCHECK(fetch_ == NULL);
   // ApacheFetch takes ownership of request_headers.
   RequestHeaders* request_headers = new RequestHeaders();
   ApacheRequestToRequestHeaders(*request_, request_headers);
-  fetch_ = new ApacheFetch(url, debug_info, server_context_->thread_system(),
-                           server_context_->timer(),
-                           new ApacheWriter(request_),
-                           request_headers, request_context_, options_,
-                           server_context_->message_handler());
+  fetch_ = new ApacheFetch(
+      url, debug_info, server_context_->thread_system(),
+      server_context_->timer(),
+      new ApacheWriter(request_, server_context_->thread_system()),
+      request_headers, request_context_, options_,
+      server_context_->message_handler());
+  fetch_->set_buffered(buffered);
   return fetch_;
 }
 
@@ -441,13 +444,12 @@ bool InstawebHandler::HandleAsInPlace() {
        != NULL) || (request_->user != NULL));
 
   RewriteDriver* driver = MakeDriver();
-  MakeFetch(original_url_, "ipro");
+  MakeFetch(true /* buffered */, "ipro");
   fetch_->set_handle_error(false);
   driver->FetchInPlaceResource(stripped_gurl_, false /* proxy_mode */, fetch_);
-  ApacheFetch::WaitResult wait_result = WaitForFetch();
-  if (wait_result != ApacheFetch::kWaitSuccess) {
-    // Note: fetch_ has been released; no longer safe to look at;
-    handled = (wait_result == ApacheFetch::kAbandonedAndHandled);
+  bool ok = WaitForFetch();
+  if (!ok) {
+    // Nothing to do, fetch_ has been released, no longer safe to look at.
   } else if (fetch_->status_ok()) {
     server_context_->rewrite_stats()->ipro_served()->Add(1);
     handled = true;
@@ -504,12 +506,12 @@ bool InstawebHandler::HandleAsProxy() {
                                               &host_header, &is_proxy) &&
       is_proxy) {
     RewriteDriver* driver = MakeDriver();
-    MakeFetch(mapped_url, "proxy");
+    MakeFetch(mapped_url, true /* buffered */, "proxy");
     fetch_->set_is_proxy(true);
     driver->SetRequestHeaders(*fetch_->request_headers());
     server_context_->proxy_fetch_factory()->StartNewProxyFetch(
         mapped_url, fetch_, driver, NULL, NULL);
-    handled = WaitForFetch() != ApacheFetch::kAbandonedAndDeclined;
+    handled = WaitForFetch();
   }
 
   return handled;  // false == declined
@@ -867,39 +869,51 @@ apr_status_t InstawebHandler::instaweb_handler(request_rec* request) {
     server_context->StatisticsPage(is_global_statistics,
                                    instaweb_handler.query_params(),
                                    instaweb_handler.options(),
-                                   instaweb_handler.MakeFetch("statistics"));
+                                   instaweb_handler.MakeFetch(
+                                       false /* unbuffered */, "statistics"));
     return OK;
   } else if ((request_handler_str == kAdminHandler) ||
              (request_handler_str == kGlobalAdminHandler)) {
     InstawebHandler instaweb_handler(request);
+    // The fetch has to be buffered because if it's a cache lookup it could
+    // complete asynchrously via the rewrite thread.
     server_context->AdminPage((request_handler_str == kGlobalAdminHandler),
                               instaweb_handler.stripped_gurl(),
                               instaweb_handler.query_params(),
                               instaweb_handler.options(),
-                              instaweb_handler.MakeFetch("admin"));
+                              instaweb_handler.MakeFetch(
+                                  true /* buffered */, "admin"));
     ret = OK;
   } else if (global_config->enable_cache_purge() &&
              !global_config->purge_method().empty() &&
              (global_config->purge_method() == request->method)) {
     InstawebHandler instaweb_handler(request);
     AdminSite* admin_site = server_context->admin_site();
+    // I'm not convinced that the purge handler must complete synchronously.  It
+    // schedules work on the rewrite driver factory's scheduler, and while in my
+    // testing it processes everything on the calling thread I'm not sure this
+    // is part of the contract.  The response is just headers and a few bytes of
+    // body, so buffering is basically free.  To be on the safe side let's
+    // buffer this one too.
     admin_site->PurgeHandler(instaweb_handler.original_url_,
                              server_context->cache_path(),
-                             instaweb_handler.MakeFetch("purge"));
+                             instaweb_handler.MakeFetch(
+                                 true /* buffered */, "purge"));
     ret = OK;
   } else if (request_handler_str == kConsoleHandler) {
     InstawebHandler instaweb_handler(request);
     server_context->ConsoleHandler(*instaweb_handler.options(),
                                    AdminSite::kOther,
                                    instaweb_handler.query_params(),
-                                   instaweb_handler.MakeFetch("console"));
+                                   instaweb_handler.MakeFetch(
+                                       false /* unbuffered */, "console"));
     ret = OK;
   } else if (request_handler_str == kMessageHandler) {
     InstawebHandler instaweb_handler(request);
     server_context->MessageHistoryHandler(
         *instaweb_handler.options(),
         AdminSite::kOther,
-        instaweb_handler.MakeFetch("messages"));
+        instaweb_handler.MakeFetch(false /* unbuffered */, "messages"));
     ret = OK;
   } else if (request_handler_str == kLogRequestHeadersHandler) {
     // For testing CustomFetchHeader.

net/instaweb/apache/instaweb_handler.h

@@ -92,12 +92,16 @@ class InstawebHandler {
   // Allocates a Fetch object associated with the current request and
   // the specified URL.  Include in debug_info anything that's cheap to create
   // and would be informative if something went wrong with the fetch.
-  ApacheFetch* MakeFetch(const GoogleString& url, StringPiece debug_info);
-
-  // Allocates a Fetch object associated with the current request and its
-  // URL.
-  ApacheFetch* MakeFetch(StringPiece debug_info) {
-    return MakeFetch(original_url_, debug_info);
+  // If any uses will be from other threads you must set buffered=true to keep
+  // your other thread from getting blocked if our output is being read by a
+  // slow reader.
+  ApacheFetch* MakeFetch(
+      const GoogleString& url, bool buffered, StringPiece debug_info);
+
+  // Allocates a Fetch object associated with the current request and its URL.
+  // Please read the comment above before setting buffered=false.
+  ApacheFetch* MakeFetch(bool buffered, StringPiece debug_info) {
+    return MakeFetch(original_url_, buffered, debug_info);
   }
 
   // Attempts to handle this as a proxied resource (see
@@ -115,13 +119,11 @@ class InstawebHandler {
   void HandleAsPagespeedResource();
 
   // Waits for an outstanding fetch (obtained by MakeFetch) to complete or time
-  // out.  Returns kWaitSuccess on success, on timeout it needs to indicate
-  // whether the request should be declined or counted as handled.  If we didn't
-  // send out headers then it's safe to decline this request and another Apache
-  // content handler can look into it, so we'll return AbandonedAndDeclined.
-  // Returning AbandonedAndHandled means we got at least as far as sending out
-  // headers, and there's nothing else it's safe to do with this request.
-  ApacheFetch::WaitResult WaitForFetch();
+  // out.  Returns true if the fetch completes, false if we time out and abandon
+  // the request.  On failure we haven't sent out headers or any other content,
+  // so it's safe to decline this request and another Apache content handler can
+  // look into it.
+  bool WaitForFetch();
 
   RequestHeaders* ReleaseRequestHeaders() { return request_headers_.release(); }