Initial implementation of the ByteStream refactor.

content/browser/download/download_resource_handler.cc

 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "content/browser/download/download_resource_handler.h"
 
 #include <string>
 
 #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"
 #include "content/public/browser/browser_thread.h"
 #include "content/public/browser/download_interrupt_reasons.h"
 #include "content/public/browser/download_item.h"
 #include "content/public/browser/download_manager_delegate.h"
 #include "content/public/common/resource_response.h"
 #include "net/base/io_buffer.h"
 #include "net/base/net_errors.h"
@@ skipping 31 matching lines @@
     const DownloadResourceHandler::OnStartedCallback& started_cb,
     const content::DownloadSaveInfo& save_info)
     : download_id_(DownloadId::Invalid()),
       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),
       is_paused_(false),
       last_buffer_size_(0),
-      bytes_read_(0) {
+      bytes_read_(0),
+      ALLOW_THIS_IN_INITIALIZER_LIST(weak_factory_(this)) {
   download_stats::RecordDownloadCount(download_stats::UNTHROTTLED_COUNT);
 }
 
 bool DownloadResourceHandler::OnUploadProgress(int request_id,
                                                uint64 position,
                                                uint64 size) {
   return true;
 }
 
 // Not needed, as this event handler ought to be the final resource.
@@ skipping 23 matching lines @@
 
   std::string content_disposition;
   request_->GetResponseHeaderByName("content-disposition",
                                     &content_disposition);
   set_content_disposition(content_disposition);
   set_content_length(response->content_length);
 
   const ResourceRequestInfoImpl* request_info =
       ResourceRequestInfoImpl::ForRequest(request_);
 
+  // Create the ByteStream pipe for sending data to the download sink.
+  output_pipe_ = new content::ByteStream();
+  output_pipe_->RegisterSourceCallback(
+      base::MessageLoopProxy::current(),
+      base::Bind(&DownloadResourceHandler::ThrottleRequest,
+                 weak_factory_.GetWeakPtr()),
+      33);
+
   // Deleted in DownloadManager.
   scoped_ptr<DownloadCreateInfo> info(new DownloadCreateInfo(
       base::Time::Now(), 0, content_length_, DownloadItem::IN_PROGRESS,
       request_->net_log(), request_info->has_user_gesture(),
       request_info->transition_type()));
   info->url_chain = request_->url_chain();
   info->referrer_url = GURL(request_->referrer());
   info->start_time = base::Time::Now();
   info->received_bytes = save_info_.offset;
   info->total_bytes = content_length_;
@@ skipping 28 matching lines @@
       !response->headers->EnumerateHeader(NULL,
                                           "Accept-Ranges",
                                           &accept_ranges_)) {
     accept_ranges_ = "";
   }
 
   info->prompt_user_for_save_location =
       save_info_.prompt_for_save_location && save_info_.file_path.empty();
   info->referrer_charset = request_->context()->referrer_charset();
   info->save_info = save_info_;
-
+  info->pipe = output_pipe_;
 
   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);
-
   return true;
 }
 
 void DownloadResourceHandler::CallStartedCB(DownloadId id, net::Error error) {
   if (started_cb_.is_null())
     return;
   BrowserThread::PostTask(
       BrowserThread::UI, FROM_HERE,
       base::Bind(&CallStartedCBOnUIThread, started_cb_, id, error));
   started_cb_.Reset();
@@ skipping 32 matching lines @@
     double actual_bandwidth = (*bytes_read)/seconds_since_last_read;
     double potential_bandwidth = last_buffer_size_/seconds_since_last_read;
     download_stats::RecordBandwidth(actual_bandwidth, potential_bandwidth);
   }
   last_read_time_ = now;
 
   if (!*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) {
+  // Take the data from the DataLoanIOBuffer and ship it down the pipe.
+  // If the pipe is full, pause the request; the pipe callback will resume it.
+  if (!output_pipe_->AddData(read_buffer_, *bytes_read)) {
+    // Send a message to ourselves to pause the request.
+    // Posting to get around fragile/broken rentrancy semantics in
+    // ResourceDispatcherHost.
+    // TODO(rdsmith): Remove PostTask when RDH is fixed.
     BrowserThread::PostTask(
-        BrowserThread::FILE, FROM_HERE,
-        base::Bind(&DownloadFileManager::UpdateDownload,
-                   download_file_manager_, download_id_, buffer_));
+        BrowserThread::IO, FROM_HERE,
+        base::Bind(
+            &DownloadResourceHandler::ThrottleRequest,
+            weak_factory_.GetWeakPtr()));
   }
 
-  // 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;
 }
 
 bool DownloadResourceHandler::OnResponseCompleted(
     int request_id,
     const net::URLRequestStatus& status,
     const std::string& security_info) {
   VLOG(20) << __FUNCTION__ << "()" << DebugString()
            << " request_id = " << request_id
@@ skipping 65 matching lines @@
           break;
       }
     }
   }
 
   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);
 
-  // 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.
+  if (output_pipe_.get())
+    output_pipe_->SourceComplete(reason);
+
+  output_pipe_ = NULL;                  // We no longer need the pipe.
   read_buffer_ = NULL;
 }
 
 void DownloadResourceHandler::OnRequestClosed() {
   UMA_HISTOGRAM_TIMES("SB2.DownloadDuration",
                       base::TimeTicks::Now() - download_start_time_);
 }
 
 void DownloadResourceHandler::StartOnUIThread(
     scoped_ptr<DownloadCreateInfo> info,
@@ skipping 31 matching lines @@
   content_length_ = 0;
   if (content_length > 0)
     content_length_ = content_length;
 }
 
 void DownloadResourceHandler::set_content_disposition(
     const std::string& 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 (output_pipe_.get())
+    output_pipe_->RegisterSourceCallback(scoped_refptr<base::TaskRunner>(),
+                                         base::Closure(), 0);
 }
 
-void DownloadResourceHandler::StartPauseTimer() {
-  if (!pause_timer_.IsRunning())
-    pause_timer_.Start(FROM_HERE,
-                       base::TimeDelta::FromMilliseconds(kThrottleTimeMs), this,
-                       &DownloadResourceHandler::CheckWriteProgress);
+// Note subtlety here!!  We need to post a task to ourselves to throttle the
+// request because ResourceDispatcherHost isn't reentrant around pausing
+// requests from ResourceHandlers.  But that opens up a window during which
+// notification of having emptied the pipe might arrive from the destination,
+// and be dropped because we weren't paused at that point.  So whenever we
+// get a throttle request, we figure out for ourselves if we should be
+// paused or not, and adjust state appropriately.  This will only fail if
+// we don't get a notification of state transition, and that shouldn't
+// happen; pipe full events are detected in OnReadCompleted and posted here,
+// and pipe unfull events are detected on destination read and posted here.
+void DownloadResourceHandler::ThrottleRequest() {
+  bool pause_goal = output_pipe_->IsFull();
+
+  if (pause_goal == is_paused_)
+    // No need to do anything.
+    return;
+
+  ResourceDispatcherHostImpl::Get()->PauseRequest(global_id_.child_id,
+                                                  global_id_.request_id,
+                                                  pause_goal);
+  is_paused_ = pause_goal;
 }
 
+
 std::string DownloadResourceHandler::DebugString() const {
   return base::StringPrintf("{"
                             " url_ = " "\"%s\""
                             " download_id_ = " "%d"
                             " global_id_ = {"
                             " child_id = " "%d"
                             " request_id = " "%d"
                             " }"
                             " render_view_id_ = " "%d"
                             " save_info_.file_path = \"%" PRFilePath "\""
                             " }",
                             request_ ?
                                 request_->url().spec().c_str() :
                                 "<NULL request>",
                             download_id_.local(),
                             global_id_.child_id,
                             global_id_.request_id,
                             render_view_id_,
                             save_info_.file_path.value().c_str());
 }