Use ByteStream in downloads system to decouple source and sink.

content/browser/download/download_file_impl.cc

Index: content/browser/download/download_file_impl.cc
diff --git a/content/browser/download/download_file_impl.cc b/content/browser/download/download_file_impl.cc
index ae8d4ec4395ddb88e5b5c1875f0c9b247055c5c5..c48197301f5457ae7b0aa9d946270cde386df75d 100644
--- a/content/browser/download/download_file_impl.cc
+++ b/content/browser/download/download_file_impl.cc
@@ -6,11 +6,18 @@
 
 #include <string>
 
+#include "base/bind.h"
 #include "base/file_util.h"
+#include "base/message_loop_proxy.h"
+#include "content/browser/download/byte_stream.h"
 #include "content/browser/download/download_create_info.h"
+#include "content/browser/download/download_interrupt_reasons_impl.h"
+#include "content/browser/download/download_net_log_parameters.h"
 #include "content/browser/power_save_blocker.h"
 #include "content/public/browser/browser_thread.h"
 #include "content/public/browser/download_manager.h"
+#include "content/browser/download/download_stats.h"
+#include "net/base/io_buffer.h"
 
 using content::BrowserThread;
 using content::DownloadId;
@@ -20,6 +27,7 @@ const int kUpdatePeriodMs = 500;
 
 DownloadFileImpl::DownloadFileImpl(
     const DownloadCreateInfo* info,
+    scoped_ptr<content::ByteStreamOutput> pipe,
     DownloadRequestHandleInterface* request_handle,
     DownloadManager* download_manager,
     bool calculate_hash,
@@ -33,21 +41,39 @@ DownloadFileImpl::DownloadFileImpl(
                 info->save_info.hash_state,
                 info->save_info.file_stream,
                 bound_net_log),
+          input_pipe_(pipe.Pass()),
           id_(info->download_id),
           request_handle_(request_handle),
           download_manager_(download_manager),
+          bound_net_log_(bound_net_log),
+          weak_factory_(ALLOW_THIS_IN_INITIALIZER_LIST(this)),
           power_save_blocker_(power_save_blocker.Pass()) {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::FILE));
 }
 
 DownloadFileImpl::~DownloadFileImpl() {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::FILE));
+  // Prevent any future callbacks from reaching us.
+  if (input_pipe_.get())
+    input_pipe_->RegisterCallback(base::Closure());
 }
 
 // BaseFile delegated functions.
 net::Error DownloadFileImpl::Initialize() {
   update_timer_.reset(new base::RepeatingTimer<DownloadFileImpl>());
-  return file_.Initialize();
+  net::Error result = file_.Initialize();
+  if (result != net::OK)
+    return result;
+
+  input_pipe_->RegisterCallback(
+      // Unretained is safe because the callback is nulled in the

[benjhayden, 2012/05/21 13:44:14]

Is there any particular reason not to use a weak pointer?

[Randy Smith (Not in Mondays), 2012/05/23 03:33:15]

Given that we already have the factory, no.  Done.

+      // destructor.
+      base::Bind(&DownloadFileImpl::PipeActive, base::Unretained(this)));
+
+  // Initial pull from the straw.
+  PipeActive();
+
+  return result;
 }
 
 net::Error DownloadFileImpl::AppendDataToFile(const char* data,
@@ -72,11 +98,6 @@ void DownloadFileImpl::Cancel() {
   file_.Cancel();
 }
 
-void DownloadFileImpl::Finish() {
-  file_.Finish();
-  update_timer_.reset();
-}
-
 void DownloadFileImpl::AnnotateWithSourceInformation() {
   file_.AnnotateWithSourceInformation();
 }
@@ -135,6 +156,99 @@ std::string DownloadFileImpl::DebugString() const {
                             file_.DebugString().c_str());
 }
 
+void DownloadFileImpl::PipeActive() {
+  base::Time start(base::Time::Now());
+  base::Time now;
+  scoped_refptr<net::IOBuffer> incoming_data;
+  size_t length = 0;
+  size_t total_length = 0;
+  size_t num_buffers = 0;
+  content::ByteStreamOutput::StreamState state(
+      content::ByteStreamOutput::STREAM_EMPTY);
+  content::DownloadInterruptReason reason =
+      content::DOWNLOAD_INTERRUPT_REASON_NONE;
+  base::TimeDelta delta(base::TimeDelta::FromMilliseconds(1000));

[benjhayden, 2012/05/21 13:44:14]

kConstant, please.

Also, why 1s?

[Randy Smith (Not in Mondays), 2012/05/23 03:33:15]

Done.
The flying monkeys delivered it (i.e. it's a completely made up number).  If
you've got preferences, I'm happy to adapt--I just wanted to put a limit on how
long we could block the file thread, having had the bad browser experience of
the file thread being blocked for the entire download.

+
+  // Take care of any file local activity required.
+  do {
+    state = input_pipe_->Read(&incoming_data, &length);

[benjhayden, 2012/05/21 13:44:14]

Would you mind changing |length| to |incoming_data_size| so that it isn't
confused with |total_length| or |num_buffers|?

[Randy Smith (Not in Mondays), 2012/05/23 03:33:15]

Done.

+
+    net::Error result = net::OK;
+    switch (state) {
+      case content::ByteStreamOutput::STREAM_EMPTY:
+        break;
+      case content::ByteStreamOutput::STREAM_HAS_DATA:
+        ++num_buffers;
+        result = AppendDataToFile(incoming_data.get()->data(), length);
+        total_length += length;
+        reason = content::ConvertNetErrorToInterruptReason(
+            result, content::DOWNLOAD_INTERRUPT_FROM_DISK);
+        break;
+      case content::ByteStreamOutput::STREAM_COMPLETE:
+        reason = input_pipe_->GetStatus();
+        SendUpdate();
+        file_.Finish();
+        update_timer_.reset();
+        break;
+      default:
+        NOTREACHED();
+        break;
+    }
+    now = base::Time::Now();
+  } while (state == content::ByteStreamOutput::STREAM_HAS_DATA &&
+           reason == content::DOWNLOAD_INTERRUPT_REASON_NONE &&
+           now - start <= delta);
+
+  // If we're stopping to yield the thread, post a task so we come back.

[benjhayden, 2012/05/21 13:44:14]

Why is it necessary to yield the thread?

[Randy Smith (Not in Mondays), 2012/05/23 03:33:15]

That's a complex question, to which the answer is "It's not, but ...".

If the ByteStream implementation is lock based (as it originally was before Will
talked me out of it), it's possible to lock the file thread for the entire
course of the download when the network bandwidth is >> than the file system
bandwidth (in my case it was a file: download from a local file system to an NFS
mounted file system).  That's not a good user experience.

Of course, the ByteStream implementation is no longer lock based.  But there's
nothing in the interface to guarantee that it's not lock based, and I could
imagine replacing it with a lock based implementation in the future (if it turns
out to be much better performing).  And I dislike having consumers dependent on
features that are not guaranteed in the interface.  So I left the yielding in. 
But I could be talked into taking it out and replacing it with a warning; I
don't like having excess code either.  I'm not certain what the right choice is
here--do you have an opinion?

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

I don't quite understand why the file thread doesn't lock without the ByteStream
lock.
My impression of the TaskRunner revolution was that it would allow us to spin up
a whole pool of worker threads to be shared by all of chrome's subsystems.
<slight hyperbole>Every single other long-running process at Google uses
threadpools, which is why I still chuckle and SMH when I think of Chrome's
4-thread architecture.
If TaskRunners ran on a threadpool, then we can run a risk of blocking one or
two worker threads, at least until metrics show a specific significant problem.

NodeJS has asynchronous file access APIs on all platforms. Maybe we could grab
its |fs| module.

For now, though, yielding after 1s sounds good.

+  if (state == content::ByteStreamOutput::STREAM_HAS_DATA &&
+      now - start > delta) {
+    BrowserThread::PostTask(
+        BrowserThread::FILE, FROM_HERE,
+        base::Bind(&DownloadFileImpl::PipeActive, weak_factory_.GetWeakPtr()));
+  }
+
+  if (total_length)
+    download_stats::RecordFileThreadReceiveBuffers(total_length);
+
+  download_stats::RecordContiguousWriteTime(now - start);
+
+  // Take care of communication with our controller.
+  if (reason != content::DOWNLOAD_INTERRUPT_REASON_NONE) {
+    // Error case for both upstream source and file write.
+    // Shut down processing and signal an error to our controller.
+    // Our controller will clean us up.
+    input_pipe_->RegisterCallback(base::Closure());
+    weak_factory_.InvalidateWeakPtrs();
+    if (download_manager_.get()) {
+      BrowserThread::PostTask(
+          BrowserThread::UI, FROM_HERE,
+          base::Bind(&DownloadManager::OnDownloadInterrupted,
+                     download_manager_, id_.local(),
+                     BytesSoFar(), GetHashState(), reason));
+    }
+  } else if (state == content::ByteStreamOutput::STREAM_COMPLETE) {
+    // Signal successful completion and shut down processing.
+    input_pipe_->RegisterCallback(base::Closure());
+    weak_factory_.InvalidateWeakPtrs();
+    if (download_manager_.get()) {
+      std::string hash;
+      if (!GetHash(&hash) || file_.IsEmptyHash(hash))
+        hash.clear();
+      BrowserThread::PostTask(
+          BrowserThread::UI, FROM_HERE,
+          base::Bind(&DownloadManager::OnResponseCompleted,
+                     download_manager_, id_.local(),
+                     BytesSoFar(), hash));
+    }
+  }
+  if (bound_net_log_.IsLoggingAllEvents()) {
+    bound_net_log_.AddEvent(
+        net::NetLog::TYPE_DOWNLOAD_PIPE_DRAINED,
+        make_scoped_refptr(new download_net_logs::FilePipeDrainedParameters(
+            total_length, num_buffers)));
+  }
+}
+
 void DownloadFileImpl::SendUpdate() {
   if (download_manager_.get()) {
     BrowserThread::PostTask(