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 757e274a830b96730950d0e119a69879fc84637b..e65b0c57df523f1e89ee0ef5dc2c055825d5e706 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 kDownloadByteStreamSize = 100 * 1024;
+
 void CallStartedCBOnUIThread(
     const DownloadResourceHandler::OnStartedCallback& started_cb,
     DownloadId id,
@@ -49,6 +52,33 @@ void CallStartedCBOnUIThread(
   started_cb.Run(id, error);
 }
 
+// Static function in order to prevent any accidental accesses to
+// DownloadResourceHandler members from the UI thread.
+static void StartOnUIThread(
+    scoped_ptr<DownloadCreateInfo> info,
+    scoped_ptr<content::ByteStreamReader> stream,
+    scoped_refptr<DownloadFileManager> download_file_manager,
+    const DownloadRequestHandle& handle,
+    const DownloadResourceHandler::OnStartedCallback& started_cb) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+
+  DownloadManager* download_manager = handle.GetDownloadManager();
+  if (!download_manager) {
+    // NULL in unittests or if the page closed right after starting the
+    // download.
+    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;
+
+  download_file_manager->StartDownload(info.Pass(), stream.Pass(), handle);
+
+  if (!started_cb.is_null())
+    started_cb.Run(download_id, net::OK);
+}
+
 }  // namespace
 
 DownloadResourceHandler::DownloadResourceHandler(
@@ -56,19 +86,17 @@ DownloadResourceHandler::DownloadResourceHandler(
     int render_view_id,
     int request_id,
     const GURL& url,
-    DownloadFileManager* download_file_manager,
+    scoped_refptr<DownloadFileManager> download_file_manager,
     net::URLRequest* request,
     const DownloadResourceHandler::OnStartedCallback& started_cb,
     const content::DownloadSaveInfo& save_info)
-    : download_id_(DownloadId::Invalid()),
-      global_id_(render_process_host_id, request_id),
+    : global_id_(render_process_host_id, request_id),
       render_view_id_(render_view_id),
       content_length_(0),
       download_file_manager_(download_file_manager),
       request_(request),
       started_cb_(started_cb),
       save_info_(save_info),
-      buffer_(new content::DownloadBuffer),
       last_buffer_size_(0),
       bytes_read_(0),
       pause_count_(0),
@@ -97,6 +125,7 @@ bool DownloadResourceHandler::OnResponseStarted(
     int request_id,
     content::ResourceResponse* response,
     bool* defer) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
   // There can be only one (call)
   DCHECK(!on_response_started_called_);
   on_response_started_called_ = true;
@@ -122,6 +151,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 for sending data to the download sink.
+  scoped_ptr<content::ByteStreamReader> stream_reader;
+  CreateByteStream(
+      base::MessageLoopProxy::current(),
+      BrowserThread::GetMessageLoopProxyForThread(BrowserThread::FILE),
+      kDownloadByteStreamSize, &stream_writer_, &stream_reader);
+  stream_writer_->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();
@@ -168,13 +208,23 @@ bool DownloadResourceHandler::OnResponseStarted(
 
   BrowserThread::PostTask(
       BrowserThread::UI, FROM_HERE,
-      base::Bind(&DownloadResourceHandler::StartOnUIThread, this,
-                 base::Passed(&info), request_handle));
+      base::Bind(&StartOnUIThread,
+                 base::Passed(info.Pass()),
+                 base::Passed(stream_reader.Pass()),
+                 download_file_manager_,
+                 request_handle,
+                 // Pass to StartOnUIThread so that variable
+                 // access is always on IO thread but function
+                 // is called on UI thread.
+                 started_cb_));
+  // Guaranteed to be called in StartOnUIThread
+  started_cb_.Reset();
 
   return true;
 }
 
 void DownloadResourceHandler::CallStartedCB(DownloadId id, net::Error error) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
   if (started_cb_.is_null())
     return;
   BrowserThread::PostTask(
@@ -193,6 +243,7 @@ bool DownloadResourceHandler::OnWillStart(int request_id,
 // writing and deletion.
 bool DownloadResourceHandler::OnWillRead(int request_id, net::IOBuffer** buf,
                                          int* buf_size, int min_size) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
   DCHECK(buf && buf_size);
   if (!read_buffer_) {
     *buf_size = min_size < 0 ? kReadBufSize : min_size;
@@ -206,6 +257,7 @@ 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,
                                               bool* defer) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
   if (!read_buffer_) {
     // Ignore spurious OnReadCompleted!  Deferring from OnReadCompleted tells
     // the ResourceDispatcherHost that we did not consume the data.
@@ -219,14 +271,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();
@@ -245,27 +289,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 stream.  If the stream is full, pause the
+  // request; the stream callback will resume it.
+  if (!stream_writer_->Write(read_buffer_, *bytes_read)) {
+    PauseRequest();
     *defer = was_deferred_ = true;
-    CheckWriteProgressLater();
+    last_stream_pause_time_ = now;
   }
 
+  read_buffer_ = NULL;  // Drop our reference.
+
   return true;
 }
 
@@ -273,37 +307,13 @@ bool DownloadResourceHandler::OnResponseCompleted(
     int request_id,
     const net::URLRequestStatus& status,
     const std::string& security_info) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
   VLOG(20) << __FUNCTION__ << "()" << DebugString()
            << " 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.
@@ -347,66 +357,38 @@ 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|.
-  read_buffer_ = NULL;
-}
-
-void DownloadResourceHandler::OnRequestClosed() {
-  UMA_HISTOGRAM_TIMES("SB2.DownloadDuration",
-                      base::TimeTicks::Now() - download_start_time_);
-}
+  // Send the info down the stream.  Conditional is in case we get
+  // OnResponseCompleted without OnResponseStarted.
+  if (stream_writer_.get())
+    stream_writer_->Close(reason);
 
-void DownloadResourceHandler::StartOnUIThread(
-    scoped_ptr<DownloadCreateInfo> info,
-    const DownloadRequestHandle& handle) {
-  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);
-    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);
+  stream_writer_.reset();  // We no longer need the stream.
+  read_buffer_ = NULL;
 
-  BrowserThread::PostTask(
-      BrowserThread::IO, FROM_HERE,
-      base::Bind(&DownloadResourceHandler::SetDownloadID, this,
-                 download_id));
+  // Stats
+  download_stats::RecordNetworkBandwidth(
+      bytes_read_, base::TimeTicks::Now() - download_start_time_,
+      total_pause_time_);
 
-  CallStartedCB(download_id, net::OK);
+  return true;
 }
 
-void DownloadResourceHandler::SetDownloadID(content::DownloadId id) {
+void DownloadResourceHandler::OnRequestClosed() {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
+  UMA_HISTOGRAM_TIMES("SB2.DownloadDuration",
+                      base::TimeTicks::Now() - download_start_time_);
 
-  download_id_ = id;
-  MaybeResumeRequest();
+  // Can't trust request_ being valid after this point.
+  request_ = NULL;
 }
 
 // If the content-length header is not present (or contains something other
 // than numbers), the incoming content_length is -1 (unknown size).
 // Set the content length to 0 to indicate unknown size to DownloadManager.
 void DownloadResourceHandler::SetContentLength(const int64& content_length) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
   content_length_ = 0;
   if (content_length > 0)
     content_length_ = content_length;
@@ -414,22 +396,10 @@ void DownloadResourceHandler::SetContentLength(const int64& content_length) {
 
 void DownloadResourceHandler::SetContentDisposition(
     const std::string& content_disposition) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
   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));
 
@@ -441,7 +411,20 @@ void DownloadResourceHandler::ResumeRequest() {
   DCHECK_LT(0, pause_count_);
 
   --pause_count_;
-  MaybeResumeRequest();
+
+  if (!was_deferred_)
+    return;
+  if (pause_count_ > 0)
+    return;
+
+  was_deferred_ = false;
+  if (!last_stream_pause_time_.is_null()) {
+    total_pause_time_ += (base::TimeTicks::Now() - last_stream_pause_time_);
+    last_stream_pause_time_ = base::TimeTicks();
+  }
+  ResourceDispatcherHostImpl::Get()->ResumeDeferredRequest(
+      global_id_.child_id,
+      global_id_.request_id);
 }
 
 void DownloadResourceHandler::CancelRequest() {
@@ -456,7 +439,6 @@ void DownloadResourceHandler::CancelRequest() {
 std::string DownloadResourceHandler::DebugString() const {
   return base::StringPrintf("{"
                             " url_ = " "\"%s\""
-                            " download_id_ = " "%d"
                             " global_id_ = {"
                             " child_id = " "%d"
                             " request_id = " "%d"
@@ -467,7 +449,6 @@ std::string DownloadResourceHandler::DebugString() const {
                             request_ ?
                                 request_->url().spec().c_str() :
                                 "<NULL request>",
-                            download_id_.local(),
                             global_id_.child_id,
                             global_id_.request_id,
                             render_view_id_,
@@ -475,35 +456,15 @@ std::string DownloadResourceHandler::DebugString() const {
 }
 
 DownloadResourceHandler::~DownloadResourceHandler() {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
+
   // 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);
-}
+  CallStartedCB(DownloadId(), 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 stream callback if a stream exists.
+  if (stream_writer_.get())
+    stream_writer_->RegisterCallback(base::Closure());
 }
 
-void DownloadResourceHandler::MaybeResumeRequest() {
-  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(
-      global_id_.child_id,
-      global_id_.request_id);
-}