Use ByteStream in downloads system to decouple source and sink.

content/browser/download/download_resource_handler.cc

Index: content/browser/download/download_resource_handler.cc
diff --git a/content/browser/download/download_resource_handler.cc b/content/browser/download/download_resource_handler.cc
index bdcd1d7bfcaaea14eef61c04b2ff42051eb62c78..603d8d148616ec12b93429fa973b343fcfccbce8 100644
--- a/content/browser/download/download_resource_handler.cc
+++ b/content/browser/download/download_resource_handler.cc
@@ -8,15 +8,16 @@
 
 #include "base/bind.h"
 #include "base/logging.h"
+#include "base/message_loop_proxy.h"
 #include "base/metrics/histogram.h"
 #include "base/metrics/stats_counters.h"
 #include "base/stringprintf.h"
-#include "content/browser/download/download_buffer.h"
 #include "content/browser/download/download_create_info.h"
 #include "content/browser/download/download_file_manager.h"
 #include "content/browser/download/download_interrupt_reasons_impl.h"
 #include "content/browser/download/download_manager_impl.h"
 #include "content/browser/download/download_request_handle.h"
+#include "content/browser/download/byte_stream.h"
 #include "content/browser/download/download_stats.h"
 #include "content/browser/renderer_host/resource_dispatcher_host_impl.h"
 #include "content/browser/renderer_host/resource_request_info_impl.h"
@@ -39,6 +40,8 @@ using content::ResourceRequestInfoImpl;
 
 namespace {
 
+static const int kDownloadPipeSize = 100 * 1024;
+
 void CallStartedCBOnUIThread(
     const DownloadResourceHandler::OnStartedCallback& started_cb,
     DownloadId id,
@@ -68,8 +71,6 @@ DownloadResourceHandler::DownloadResourceHandler(
       request_(request),
       started_cb_(started_cb),
       save_info_(save_info),
-      buffer_(new content::DownloadBuffer),
-      is_paused_(false),
       last_buffer_size_(0),
       bytes_read_(0) {
   download_stats::RecordDownloadCount(download_stats::UNTHROTTLED_COUNT);
@@ -120,6 +121,17 @@ bool DownloadResourceHandler::OnResponseStarted(
       base::Time::Now(), 0, content_length_, DownloadItem::IN_PROGRESS,
       request_->net_log(), request_info->has_user_gesture(),
       request_info->transition_type()));
+
+  // Create the ByteStream pipe for sending data to the download sink.
+  scoped_ptr<content::ByteStreamOutput> pipe_output;
+  CreateByteStream(
+      base::MessageLoopProxy::current(),
+      BrowserThread::GetMessageLoopProxyForThread(BrowserThread::FILE),
+      kDownloadPipeSize, &pipe_input_, &pipe_output);
+  pipe_input_->RegisterCallback(
+      // We're refcounted, so this is a safe callback to pass around.
+      base::Bind(&DownloadResourceHandler::ResumeRequest, this));
+
   info->url_chain = request_->url_chain();
   info->referrer_url = GURL(request_->referrer());
   info->start_time = base::Time::Now();
@@ -164,18 +176,12 @@ bool DownloadResourceHandler::OnResponseStarted(
   info->referrer_charset = request_->context()->referrer_charset();
   info->save_info = save_info_;
 
-
   BrowserThread::PostTask(
       BrowserThread::UI, FROM_HERE,
       base::Bind(&DownloadResourceHandler::StartOnUIThread, this,
-                 base::Passed(&info), request_handle));
-
-  // We can't start saving the data before we create the file on disk and have a
-  // download id. The request will be un-paused in
-  // DownloadFileManager::CreateDownloadFile.
-  ResourceDispatcherHostImpl::Get()->PauseRequest(global_id_.child_id,
-                                                  global_id_.request_id,
-                                                  true);
+                 base::Passed(info.Pass()),
+                 base::Passed(pipe_output.Pass()),
+                 request_handle));
 
   return true;
 }
@@ -211,6 +217,19 @@ bool DownloadResourceHandler::OnWillRead(int request_id, net::IOBuffer** buf,
 
 // Pass the buffer to the download file writer.
 bool DownloadResourceHandler::OnReadCompleted(int request_id, int* bytes_read) {
+  if (!read_buffer_) {
+    // Ignore spurious OnReadCompleted!  PauseRequest(true) called from within
+    // OnReadCompleted tells the ResourceDispatcherHost that we did not consume
+    // the data.  PauseRequest(false) then repeats the last OnReadCompleted
+    // call.  We pause the request after we transmit the buffer onwards, so
+    // the ResourceDispatcherHost pause mechanism does not fit our use
+    // case very well.
+    // Note that this test makes redundant the DCHECK() below; it's being left
+    // in in hopes that the TODO will be executed soon.
+    // TODO(darin): Fix the ResourceDispatcherHost to avoid this hack!
+    return true;
+  }
+
   base::TimeTicks now(base::TimeTicks::Now());
   if (!last_read_time_.is_null()) {
     double seconds_since_last_read = (now - last_read_time_).InSecondsF();
@@ -229,24 +248,16 @@ bool DownloadResourceHandler::OnReadCompleted(int request_id, int* bytes_read) {
     return true;
   bytes_read_ += *bytes_read;
   DCHECK(read_buffer_);
-  // Swap the data.
-  net::IOBuffer* io_buffer = NULL;
-  read_buffer_.swap(&io_buffer);
-  size_t vector_size = buffer_->AddData(io_buffer, *bytes_read);
-  bool need_update = (vector_size == 1);  // Buffer was empty.
-
-  // We are passing ownership of this buffer to the download file manager.
-  if (need_update) {
-    BrowserThread::PostTask(
-        BrowserThread::FILE, FROM_HERE,
-        base::Bind(&DownloadFileManager::UpdateDownload,
-                   download_file_manager_, download_id_, buffer_));
+
+  // Take the data ship it down the pipe.  If the pipe is full, pause the
+  // request; the pipe callback will resume it.
+  if (!pipe_input_->Write(read_buffer_, *bytes_read)) {
+    ResourceDispatcherHostImpl::Get()->PauseRequest(global_id_.child_id,
+                                                    global_id_.request_id,
+                                                    true);
   }
 
-  // We schedule a pause outside of the read loop if there is too much file
-  // writing work to do.
-  if (vector_size > kLoadsToWrite)
-    StartPauseTimer();
+  read_buffer_ = NULL;  // Drop our reference.
 
   return true;
 }
@@ -332,14 +343,12 @@ void DownloadResourceHandler::OnResponseCompletedInternal(
   // If the callback was already run on the UI thread, this will be a noop.
   CallStartedCB(download_id_, error_code);
 
-  // We transfer ownership to |DownloadFileManager| to delete |buffer_|,
-  // so that any functions queued up on the FILE thread are executed
-  // before deletion.
-  BrowserThread::PostTask(
-      BrowserThread::FILE, FROM_HERE,
-      base::Bind(&DownloadFileManager::OnResponseCompleted,
-                 download_file_manager_, download_id_, reason, security_info));
-  buffer_ = NULL;  // The buffer is longer needed by |DownloadResourceHandler|.
+  // Send the info down the pipe.  Conditional is in case we get
+  // OnResponseCompleted without OnResponseStarted.
+  if (pipe_input_.get())
+    pipe_input_->Close(reason);
+
+  pipe_input_.reset();  // We no longer need the pipe.
   read_buffer_ = NULL;
 }
 
@@ -350,6 +359,7 @@ void DownloadResourceHandler::OnRequestClosed() {
 
 void DownloadResourceHandler::StartOnUIThread(
     scoped_ptr<DownloadCreateInfo> info,
+    scoped_ptr<content::ByteStreamOutput> pipe,
     const DownloadRequestHandle& handle) {
   DownloadManager* download_manager = handle.GetDownloadManager();
   if (!download_manager) {
@@ -368,7 +378,7 @@ void DownloadResourceHandler::StartOnUIThread(
   // confirmation that that download_id_ has been set on the IO thread, as any
   // messages generated by the UI thread that affect the IO thread will be
   // behind the message posted above.
-  download_file_manager_->StartDownload(info.release(), handle);
+  download_file_manager_->StartDownload(info.Pass(), pipe.Pass(), handle);
   CallStartedCB(download_id, net::OK);
 }
 
@@ -391,38 +401,21 @@ void DownloadResourceHandler::set_content_disposition(
   content_disposition_ = content_disposition;
 }
 
-void DownloadResourceHandler::CheckWriteProgress() {
-  if (!buffer_.get())
-    return;  // The download completed while we were waiting to run.
-
-  size_t contents_size = buffer_->size();
-
-  bool should_pause = contents_size > kLoadsToWrite;
-
-  // We'll come back later and see if it's okay to unpause the request.
-  if (should_pause)
-    StartPauseTimer();
-
-  if (is_paused_ != should_pause) {
-    ResourceDispatcherHostImpl::Get()->PauseRequest(global_id_.child_id,
-                                                    global_id_.request_id,
-                                                    should_pause);
-    is_paused_ = should_pause;
-  }
-}
-
 DownloadResourceHandler::~DownloadResourceHandler() {
   // This won't do anything if the callback was called before.
   // If it goes through, it will likely be because OnWillStart() returned
   // false somewhere in the chain of resource handlers.
   CallStartedCB(download_id_, net::ERR_ACCESS_DENIED);
+
+  // Remove output pipe callback if a pipe exists.
+  if (pipe_input_.get())
+    pipe_input_->RegisterCallback(base::Closure());
 }
 
-void DownloadResourceHandler::StartPauseTimer() {
-  if (!pause_timer_.IsRunning())
-    pause_timer_.Start(FROM_HERE,
-                       base::TimeDelta::FromMilliseconds(kThrottleTimeMs), this,
-                       &DownloadResourceHandler::CheckWriteProgress);
+void DownloadResourceHandler::ResumeRequest() {
+  ResourceDispatcherHostImpl::Get()->PauseRequest(global_id_.child_id,
+                                                  global_id_.request_id,
+                                                  false);
 }
 
 std::string DownloadResourceHandler::DebugString() const {