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 84ab673c9cd59120e2757d50d579c76167f62092..f25fad1abf0f07d98eb820605f9bd763dfadff59 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,7 +71,6 @@ DownloadResourceHandler::DownloadResourceHandler(
       request_(request),
       started_cb_(started_cb),
       save_info_(save_info),
-      buffer_(new content::DownloadBuffer),
       last_buffer_size_(0),
       bytes_read_(0),
       pause_count_(0),
@@ -117,6 +119,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;

[darin (slow to review), 2012/05/24 22:01:36]

I am very concerned with this use of Input and Output terminology.  You are
using them in a grammatically correct fashion, but it is in conflict with the
way these terms are typically used in the context of streams.  Normally "Input"
is for the side with a "Read" function so that you can input data into a
program, and "Output" is for the side with a "Write" function so that you can
output data from a program.  Here you have inverted the sense, and I suspect
this will lead to confusion.  Why fight convention?

you might also consider using the terms "reader" and "writer", so you'd have
pipe_reader_->Read(...) and pipe_writer_->Write(...).  that would be fairly
conventional too.

[Randy Smith (Not in Mondays), 2012/05/26 02:36:54]

Done.

+  CreateByteStream(
+      base::MessageLoopProxy::current(),
+      BrowserThread::GetMessageLoopProxyForThread(BrowserThread::FILE),
+      kDownloadPipeSize, &pipe_input_, &pipe_output);

[darin (slow to review), 2012/05/24 22:01:36]

nit: you call the variables pipe_{input,output}, which suggests that the thing
that you are creating is a pipe, but the function is CreateByteStream.  Why did
you choose ByteStream instead of Pipe for the class name?

[Randy Smith (Not in Mondays), 2012/05/26 02:36:54]

I didn't put a lot of thought into the naming, actually.  I think I'd rather
bring Mohammed to the Mountain, so I'll change the variables I've named based
off of "pipe" to be named based off of "stream".  Let me know what you think.

+  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,12 +177,18 @@ bool DownloadResourceHandler::OnResponseStarted(
   BrowserThread::PostTask(
       BrowserThread::UI, FROM_HERE,
       base::Bind(&DownloadResourceHandler::StartOnUIThread, this,
-                 base::Passed(&info), request_handle));
+                 base::Passed(info.Pass()),
+                 base::Passed(pipe_output.Pass()),
+                 request_handle,
+                 started_cb_));

[benjhayden, 2012/05/24 15:35:57]

Can you comment why you're passing started_cb_ to StartOnUIThread and nullifying
it here instead of both running it and nullifying it there?
I think I see the race condition that you're fixing, but it still looks strange.

[Randy Smith (Not in Mondays), 2012/05/26 02:36:54]

Done.

+  // Guaranteed to be called in StartOnUIThread
+  started_cb_.Reset();
 
   return true;
 }
 
 void DownloadResourceHandler::CallStartedCB(DownloadId id, net::Error error) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));

[benjhayden, 2012/05/24 15:35:57]

Can you add this annotation to *all* the methods? It's super helpful.

[Randy Smith (Not in Mondays), 2012/05/26 02:36:54]

I did almost all the methods; I skipped the ones where I thought flexibility
might be useful (construction, though probably not necessary), and where the
routine didn't do anything but was just there for override purposes.

   if (started_cb_.is_null())
     return;
   BrowserThread::PostTask(
@@ -214,14 +233,6 @@ bool DownloadResourceHandler::OnReadCompleted(int request_id, int* bytes_read,
     return true;
   }
 
-  if (download_id_ == DownloadId::Invalid()) {
-    // 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.
-    *defer = was_deferred_ = true;
-    return true;
-  }
-
   base::TimeTicks now(base::TimeTicks::Now());
   if (!last_read_time_.is_null()) {
     double seconds_since_last_read = (now - last_read_time_).InSecondsF();
@@ -240,27 +251,17 @@ 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_));
-  }
 
-  // We schedule a pause outside of the read loop if there is too much file
-  // writing work to do.
-  if (vector_size > kLoadsToWrite) {
+  // 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)) {
     *defer = was_deferred_ = true;
-    CheckWriteProgressLater();
+    pause_count_++;

[darin (slow to review), 2012/05/24 22:01:36]

I'd call PauseRequest() here.

[Randy Smith (Not in Mondays), 2012/05/26 02:36:54]

Feels a little weird, since the *defer = true implementation is outside the
call, but it does collapse common code.  Done.

+    last_pause_time_ = now;
   }
 
+  read_buffer_ = NULL;  // Drop our reference.
+
   return true;
 }
 
@@ -272,33 +273,8 @@ bool DownloadResourceHandler::OnResponseCompleted(
            << " request_id = " << request_id
            << " status.status() = " << status.status()
            << " status.error() = " << status.error();
-  int response = status.is_success() ? request_->GetResponseCode() : 0;
-  if (download_id_.IsValid()) {
-    OnResponseCompletedInternal(request_id, status, security_info, response);
-  } else {
-    // We got cancelled before the task which sets the id ran on the IO thread.
-    // Wait for it.
-    BrowserThread::PostTaskAndReply(
-        BrowserThread::UI, FROM_HERE,
-        base::Bind(&base::DoNothing),
-        base::Bind(&DownloadResourceHandler::OnResponseCompletedInternal, this,
-                   request_id, status, security_info, response));
-  }
-  // Can't trust request_ being value after this point.
-  request_ = NULL;
-  return true;
-}
+  int response_code = status.is_success() ? request_->GetResponseCode() : 0;
 
-void DownloadResourceHandler::OnResponseCompletedInternal(
-    int request_id,
-    const net::URLRequestStatus& status,
-    const std::string& security_info,
-    int response_code) {
-  // NOTE: |request_| may be a dangling pointer at this point.
-  VLOG(20) << __FUNCTION__ << "()"
-           << " request_id = " << request_id
-           << " status.status() = " << status.status()
-           << " status.error() = " << status.error();
   net::Error error_code = net::OK;
   if (status.status() == net::URLRequestStatus::FAILED)
     error_code = static_cast<net::Error>(status.error());  // Normal case.
@@ -342,18 +318,25 @@ void DownloadResourceHandler::OnResponseCompletedInternal(
 
   download_stats::RecordAcceptsRanges(accept_ranges_, bytes_read_);
 
-  // If the callback was already run on the UI thread, this will be a noop.
-  CallStartedCB(download_id_, error_code);
+  CallStartedCB(DownloadId(), 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;
+
+  // Can't trust request_ being valid after this point.
+  request_ = NULL;

[darin (slow to review), 2012/05/24 22:01:36]

technically, the purpose of OnRequestClosed is to tell you when you must drop
any pointers to the URLRequest.  you can/should move this code to
OnRequestClosed instead.

[Randy Smith (Not in Mondays), 2012/05/26 02:36:54]

Done.  The code's a bit stale, anyway, as we no longer have any activity going
on after OnResponseCompleted() (because we no longer delay the implementation of
that call waiting for the download id).

+
+  // Stats
+  download_stats::RecordNetworkBandwidth(
+      bytes_read_, base::TimeTicks::Now() - download_start_time_,
+      total_pause_time_);
+
+  return true;
 }
 
 void DownloadResourceHandler::OnRequestClosed() {
@@ -363,32 +346,29 @@ void DownloadResourceHandler::OnRequestClosed() {
 
 void DownloadResourceHandler::StartOnUIThread(

[benjhayden, 2012/05/24 15:35:57]

If you want to avoid accessing member variables outside the IO thread, can you
make this method a static non-member function like CallStartedCBOnUIThread()?
And please comment that it's a static non-member function in order to prevent
member access because it runs off the IO thread. And annotate with the
DCHECK(BrowserThread...).

[Randy Smith (Not in Mondays), 2012/05/26 02:36:54]

Done.

     scoped_ptr<DownloadCreateInfo> info,
-    const DownloadRequestHandle& handle) {
+    scoped_ptr<content::ByteStreamOutput> pipe,
+    const DownloadRequestHandle& handle,
+    const OnStartedCallback& started_cb) {
   DownloadManager* download_manager = handle.GetDownloadManager();
   if (!download_manager) {
     // NULL in unittests or if the page closed right after starting the
     // download.
-    CallStartedCB(download_id_, net::ERR_ACCESS_DENIED);
+    if (!started_cb.is_null())
+      started_cb.Run(DownloadId(), net::ERR_ACCESS_DENIED);
     return;
   }
   DownloadId download_id = download_manager->delegate()->GetNextId();
   info->download_id = download_id;
 
-  // NOTE: StartDownload triggers creation of the download destination file
-  // that will hold the downloaded data.  SetDownloadID unblocks the
-  // DownloadResourceHandler to begin forwarding network data to the download
-  // destination file.  The sequence of these two steps is critical as creation
-  // of the downloaded destination file has to happen before we attempt to
-  // append data to it.  Both of those operations happen on the FILE thread.
-
-  download_file_manager_->StartDownload(info.release(), handle);
-
   BrowserThread::PostTask(
       BrowserThread::IO, FROM_HERE,
       base::Bind(&DownloadResourceHandler::SetDownloadID, this,
                  download_id));
 
-  CallStartedCB(download_id, net::OK);
+  download_file_manager_->StartDownload(info.Pass(), pipe.Pass(), handle);
+
+  if (!started_cb.is_null())
+    started_cb.Run(download_id, net::OK);
 }
 
 void DownloadResourceHandler::SetDownloadID(content::DownloadId id) {
@@ -412,19 +392,6 @@ void DownloadResourceHandler::SetContentDisposition(
   content_disposition_ = content_disposition;
 }
 
-void DownloadResourceHandler::CheckWriteProgress() {
-  if (!buffer_.get())
-    return;  // The download completed while we were waiting to run.
-
-  if (buffer_->size() > kLoadsToWrite) {
-    // We'll come back later and see if it's okay to unpause the request.
-    CheckWriteProgressLater();
-    return;
-  }
-
-  MaybeResumeRequest();
-}
-
 void DownloadResourceHandler::PauseRequest() {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
 
@@ -474,28 +441,17 @@ DownloadResourceHandler::~DownloadResourceHandler() {
   // 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);
-}
 
-void DownloadResourceHandler::CheckWriteProgressLater() {
-  if (!check_write_progress_timer_.IsRunning()) {
-    check_write_progress_timer_.Start(
-        FROM_HERE,
-        base::TimeDelta::FromMilliseconds(kThrottleTimeMs),
-        this,
-        &DownloadResourceHandler::CheckWriteProgress);
-  }
+  // Remove output pipe callback if a pipe exists.
+  if (pipe_input_.get())
+    pipe_input_->RegisterCallback(base::Closure());
 }
 
 void DownloadResourceHandler::MaybeResumeRequest() {

[darin (slow to review), 2012/05/24 22:01:36]

i'd probably just fold this code into ResumeRequest since that should now be the
only caller.

[Randy Smith (Not in Mondays), 2012/05/26 02:36:54]

Done.

   if (!was_deferred_)
     return;
-
   if (pause_count_ > 0)
     return;
-  if (download_id_ == DownloadId::Invalid())
-    return;
-  if (buffer_.get() && (buffer_->size() > kLoadsToWrite))
-    return;
 
   was_deferred_ = false;
   ResourceDispatcherHostImpl::Get()->ResumeDeferredRequest(