Use ByteStream in downloads system to decouple source and sink.

content/browser/download/download_file_impl.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_file_impl.h"
 
 #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;
 using content::DownloadManager;
 
 const int kUpdatePeriodMs = 500;
 
 DownloadFileImpl::DownloadFileImpl(
     const DownloadCreateInfo* info,
+    scoped_ptr<content::ByteStreamOutput> pipe,
     DownloadRequestHandleInterface* request_handle,
     DownloadManager* download_manager,
     bool calculate_hash,
     scoped_ptr<PowerSaveBlocker> power_save_blocker,
     const net::BoundNetLog& bound_net_log)
         : file_(info->save_info.file_path,
                 info->url(),
                 info->referrer_url,
                 info->received_bytes,
                 calculate_hash,
                 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,
                                               size_t data_len) {
   if (!update_timer_->IsRunning()) {
     update_timer_->Start(FROM_HERE,
                          base::TimeDelta::FromMilliseconds(kUpdatePeriodMs),
                          this, &DownloadFileImpl::SendUpdate);
   }
   return file_.AppendDataToFile(data, data_len);
 }
 
 net::Error DownloadFileImpl::Rename(const FilePath& full_path) {
   return file_.Rename(full_path);
 }
 
 void DownloadFileImpl::Detach() {
   file_.Detach();
 }
 
 void DownloadFileImpl::Cancel() {
   file_.Cancel();
 }
 
-void DownloadFileImpl::Finish() {
-  file_.Finish();
-  update_timer_.reset();
-}
-
 void DownloadFileImpl::AnnotateWithSourceInformation() {
   file_.AnnotateWithSourceInformation();
 }
 
 FilePath DownloadFileImpl::FullPath() const {
   return file_.full_path();
 }
 
 bool DownloadFileImpl::InProgress() const {
   return file_.in_progress();
@@ skipping 38 matching lines @@
   return base::StringPrintf("{"
                             " id_ = " "%d"
                             " request_handle = %s"
                             " Base File = %s"
                             " }",
                             id_.local(),
                             request_handle_->DebugString().c_str(),
                             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(
         BrowserThread::UI, FROM_HERE,
         base::Bind(&DownloadManager::UpdateDownload,
                    download_manager_, id_.local(),
                    BytesSoFar(), CurrentSpeed(), GetHashState()));
   }
 }