Make DownloadFileImpl handle multiple byte streams.

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 <utility>
 
 #include "base/bind.h"
 #include "base/files/file_util.h"
+#include "base/memory/ptr_util.h"
 #include "base/strings/stringprintf.h"
 #include "base/time/time.h"
 #include "base/values.h"
 #include "content/browser/byte_stream.h"
 #include "content/browser/download/download_create_info.h"
 #include "content/browser/download/download_destination_observer.h"
 #include "content/browser/download/download_interrupt_reasons_impl.h"
 #include "content/browser/download/download_net_log_parameters.h"
 #include "content/browser/download/download_stats.h"
 #include "content/public/browser/browser_thread.h"
@@ skipping 11 matching lines @@
 const int kMaxTimeBlockingFileThreadMs = 1000;
 
 // These constants control the default retry behavior for failing renames. Each
 // retry is performed after a delay that is twice the previous delay. The
 // initial delay is specified by kInitialRenameRetryDelayMs.
 const int kInitialRenameRetryDelayMs = 200;
 
 // Number of times a failing rename is retried before giving up.
 const int kMaxRenameRetries = 3;
 
+DownloadFileImpl::SourceStream::SourceStream(int64_t offset, int64_t length)
+    : offset_(offset), length_(length), bytes_written_(0), finished_(false) {}

[asanka, 2017/03/02 19:37:43]

Is this 'git cl format's doing?

[xingliu, 2017/03/02 23:06:14]

Yes, I think so.

+
+DownloadFileImpl::SourceStream::~SourceStream() = default;
+
+void DownloadFileImpl::SourceStream::SetByteStream(
+    std::unique_ptr<ByteStreamReader> stream_reader) {
+  stream_reader_ = std::move(stream_reader);
+}
+
+void DownloadFileImpl::SourceStream::OnWriteBytesToDisk(int64_t bytes_write) {
+  bytes_written_ += bytes_write;
+}
+
 DownloadFileImpl::DownloadFileImpl(
     std::unique_ptr<DownloadSaveInfo> save_info,
     const base::FilePath& default_download_directory,
-    std::unique_ptr<ByteStreamReader> stream,
+    std::unique_ptr<ByteStreamReader> stream_reader,
     const net::NetLogWithSource& download_item_net_log,
+    bool is_sparse_file,
     base::WeakPtr<DownloadDestinationObserver> observer)
     : net_log_(
           net::NetLogWithSource::Make(download_item_net_log.net_log(),
                                       net::NetLogSourceType::DOWNLOAD_FILE)),
       file_(net_log_),
       save_info_(std::move(save_info)),
       default_download_directory_(default_download_directory),
-      stream_reader_(std::move(stream)),
+      is_sparse_file_(is_sparse_file),
       bytes_seen_(0),
       observer_(observer),
       weak_factory_(this) {
+  source_streams_[save_info_->offset] =
+      base::MakeUnique<SourceStream>(save_info_->offset, 0);
+  DCHECK(source_streams_.size() == static_cast<size_t>(1));
+  source_streams_.begin()->second->SetByteStream(std::move(stream_reader));

[David Trainor- moved to gerrit, 2017/03/02 18:24:50]

source_streams_[save_info_->offset]->SetByteStream()?

[xingliu, 2017/03/02 19:21:20]

Done.

+
   download_item_net_log.AddEvent(
       net::NetLogEventType::DOWNLOAD_FILE_CREATED,
       net_log_.source().ToEventParametersCallback());
   net_log_.BeginEvent(
       net::NetLogEventType::DOWNLOAD_FILE_ACTIVE,
       download_item_net_log.source().ToEventParametersCallback());
 }
 
 DownloadFileImpl::~DownloadFileImpl() {
   DCHECK_CURRENTLY_ON(BrowserThread::FILE);
   net_log_.EndEvent(net::NetLogEventType::DOWNLOAD_FILE_ACTIVE);
 }
 
 void DownloadFileImpl::Initialize(const InitializeCallback& callback) {
   DCHECK_CURRENTLY_ON(BrowserThread::FILE);
 
   update_timer_.reset(new base::RepeatingTimer());
   DownloadInterruptReason result =
-      file_.Initialize(save_info_->file_path,
-                       default_download_directory_,
-                       std::move(save_info_->file),
-                       save_info_->offset,
+      file_.Initialize(save_info_->file_path, default_download_directory_,
+                       std::move(save_info_->file), save_info_->offset,
                        save_info_->hash_of_partial_file,
-                       std::move(save_info_->hash_state),
-                       false);
+                       std::move(save_info_->hash_state), is_sparse_file_);
   if (result != DOWNLOAD_INTERRUPT_REASON_NONE) {
     BrowserThread::PostTask(
         BrowserThread::UI, FROM_HERE, base::Bind(callback, result));
     return;
   }
 
-  stream_reader_->RegisterCallback(
-      base::Bind(&DownloadFileImpl::StreamActive, weak_factory_.GetWeakPtr()));
-
   download_start_ = base::TimeTicks::Now();
 
   // Primarily to make reset to zero in restart visible to owner.
   SendUpdate();
 
   // Initial pull from the straw.
-  StreamActive();
+  for (auto& source_stream : source_streams_)
+    RegisterAndActivateStream(source_stream.second.get());
 
   BrowserThread::PostTask(
       BrowserThread::UI, FROM_HERE, base::Bind(
           callback, DOWNLOAD_INTERRUPT_REASON_NONE));
 }
 
+void DownloadFileImpl::AddByteStream(
+    std::unique_ptr<ByteStreamReader> stream_reader,
+    int64_t offset) {
+  // |source_streams_| is not thread safe, must be modified on the same thread.

[asanka, 2017/03/02 19:37:43]

This is not safe.

The problem here is that the DownloadFileImpl might be getting destroyed or
probably already destroyed on the FILE thread. In that case you are indirecting
through a freed pointer.

The thread specificity of DownloadFileImpl is:
* It's created and initialized on the UI thread.
* Then it's tossed over to the FILE thread where it lives and dies.

If you intend AddByteStream to be called from a thread other than the FILE
thread, then:
* Make AddByteStream() be static. This way you won't accidentally touch an
unsafe member like this code is currently doing.

* Make the API contract state explicitly that AddByteStream can only be called
by an owner who can guarantee that the destruction of DownloadFileImpl hasn't
beed scheduled on the FILE thread. THis is currently only possible if the owner
is thread bound and controls the lifetime of the DownloadFileImpl. The latter
condition is true for DownloadItemImpl. So you should be safe.

* Have a single method on DownloadFileImpl that accepts an incoming
ByteStreamReader. Schedule that method using an unretained DownloadFileImpl
pointer.

Note that WeakPtrFactory isn't thread safe.

[xingliu, 2017/03/02 23:06:14]

Done,

Thanks for this suggestion. Yeah, good catch, DownloadFileImpl's
weak_factory_.GetWeakPtr() must be called on file thread here to bind to the
file thread task runner token.

Then in this way, object on UI thread must make sure download_file unique_ptr is
not null when call AddByteStream and use Unretained.

For the stream_reader, when we post this function from the UI thread, it's moved
and copied to the callback storage so it should also be ok.

For source_streams_(unordered_map), since we don't change its size but only pass
the parameters from callback storage it should be fine.


PTAL at the new code here.

+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  // The |source_streams_| must have an existing entry for the stream reader.
+  auto current_source_stream = source_streams_.find(offset);
+  DCHECK(current_source_stream != source_streams_.end());
+  SourceStream* stream = current_source_stream->second.get();
+  stream->SetByteStream(std::move(stream_reader));
+
+  // Start to pull data from the stream.
+  BrowserThread::PostTask(
+      BrowserThread::FILE, FROM_HERE,
+      base::Bind(&DownloadFileImpl::RegisterAndActivateStream,
+                 weak_factory_.GetWeakPtr(), stream));
+}
+
 DownloadInterruptReason DownloadFileImpl::AppendDataToFile(
     const char* data, size_t data_len) {
   DCHECK_CURRENTLY_ON(BrowserThread::FILE);
+  WillWriteToDisk(data_len);
+  return file_.AppendDataToFile(data, data_len);
+}
 
-  if (!update_timer_->IsRunning()) {
-    update_timer_->Start(FROM_HERE,
-                         base::TimeDelta::FromMilliseconds(kUpdatePeriodMs),
-                         this, &DownloadFileImpl::SendUpdate);
-  }
-  rate_estimator_.Increment(data_len);
-  return file_.AppendDataToFile(data, data_len);
+DownloadInterruptReason DownloadFileImpl::WriteDataToFile(int64_t offset,
+                                                          const char* data,
+                                                          size_t data_len) {
+  DCHECK_CURRENTLY_ON(BrowserThread::FILE);
+  WillWriteToDisk(data_len);
+  return file_.WriteDataToFile(offset, data, data_len);
 }
 
 void DownloadFileImpl::RenameAndUniquify(
     const base::FilePath& full_path,
     const RenameCompletionCallback& callback) {
   std::unique_ptr<RenameParameters> parameters(
       new RenameParameters(UNIQUIFY, full_path, callback));
   RenameWithRetryInternal(std::move(parameters));
 }
 
@@ skipping 79 matching lines @@
                                                  parameters->source_url,
                                                  parameters->referrer_url);
   }
 
   if (reason != DOWNLOAD_INTERRUPT_REASON_NONE) {
     // Make sure our information is updated, since we're about to
     // error out.
     SendUpdate();
 
     // Null out callback so that we don't do any more stream processing.
-    stream_reader_->RegisterCallback(base::Closure());
+    for (auto& stream : source_streams_) {
+      ByteStreamReader* stream_reader = stream.second->stream_reader();
+      if (stream_reader)
+        stream_reader->RegisterCallback(base::Closure());
+    }
 
     new_path.clear();
   }
 
   BrowserThread::PostTask(
       BrowserThread::UI,
       FROM_HERE,
       base::Bind(parameters->completion_callback, reason, new_path));
 }
 
 void DownloadFileImpl::Detach() {
   file_.Detach();
 }
 
 void DownloadFileImpl::Cancel() {
   file_.Cancel();
 }
 
 const base::FilePath& DownloadFileImpl::FullPath() const {
   return file_.full_path();
 }
 
 bool DownloadFileImpl::InProgress() const {
   return file_.in_progress();
 }
 
-void DownloadFileImpl::StreamActive() {
+void DownloadFileImpl::StreamActive(SourceStream* source_stream) {
+  DCHECK(source_stream->stream_reader());
   base::TimeTicks start(base::TimeTicks::Now());
   base::TimeTicks now;
   scoped_refptr<net::IOBuffer> incoming_data;
   size_t incoming_data_size = 0;
   size_t total_incoming_data_size = 0;
   size_t num_buffers = 0;
+  bool should_terminate = false;
   ByteStreamReader::StreamState state(ByteStreamReader::STREAM_EMPTY);
   DownloadInterruptReason reason = DOWNLOAD_INTERRUPT_REASON_NONE;
   base::TimeDelta delta(
       base::TimeDelta::FromMilliseconds(kMaxTimeBlockingFileThreadMs));
 
   // Take care of any file local activity required.
   do {
-    state = stream_reader_->Read(&incoming_data, &incoming_data_size);
+    state = source_stream->stream_reader()->Read(&incoming_data,
+                                                 &incoming_data_size);
 
     switch (state) {
       case ByteStreamReader::STREAM_EMPTY:
         break;
       case ByteStreamReader::STREAM_HAS_DATA:
         {
           ++num_buffers;
           base::TimeTicks write_start(base::TimeTicks::Now());
-          reason = AppendDataToFile(
-              incoming_data.get()->data(), incoming_data_size);
+          // Stop the stream if it writes more bytes than expected.
+          if (source_stream->length() > 0 &&
+              source_stream->bytes_written() +
+                      static_cast<int64_t>(incoming_data_size) >=
+                  source_stream->length()) {
+            should_terminate = true;
+            incoming_data_size =
+                source_stream->length() - source_stream->bytes_written();
+          }
+
+          if (is_sparse_file_) {
+            reason = WriteDataToFile(
+                source_stream->offset() + source_stream->bytes_written(),
+                incoming_data.get()->data(), incoming_data_size);
+          } else {
+            reason = AppendDataToFile(incoming_data.get()->data(),

[asanka, 2017/03/02 19:37:43]

I'm assuming that the plan is to get rid of this logic?

[xingliu, 2017/03/02 23:06:14]

Done.

+                                      incoming_data_size);
+          }
           disk_writes_time_ += (base::TimeTicks::Now() - write_start);
           bytes_seen_ += incoming_data_size;
           total_incoming_data_size += incoming_data_size;
+          if (reason == DOWNLOAD_INTERRUPT_REASON_NONE)
+            source_stream->OnWriteBytesToDisk(incoming_data_size);
         }
         break;
       case ByteStreamReader::STREAM_COMPLETE:
         {
           reason = static_cast<DownloadInterruptReason>(
-              stream_reader_->GetStatus());
+            source_stream->stream_reader()->GetStatus());
           SendUpdate();
-          base::TimeTicks close_start(base::TimeTicks::Now());
-          base::TimeTicks now(base::TimeTicks::Now());
-          disk_writes_time_ += (now - close_start);
-          RecordFileBandwidth(
-              bytes_seen_, disk_writes_time_, now - download_start_);
-          update_timer_.reset();
         }
         break;
       default:
         NOTREACHED();
         break;
     }
     now = base::TimeTicks::Now();
   } while (state == ByteStreamReader::STREAM_HAS_DATA &&
-           reason == DOWNLOAD_INTERRUPT_REASON_NONE &&
-           now - start <= delta);
+           reason == DOWNLOAD_INTERRUPT_REASON_NONE && now - start <= delta &&
+           !should_terminate);
 
   // If we're stopping to yield the thread, post a task so we come back.
-  if (state == ByteStreamReader::STREAM_HAS_DATA &&
-      now - start > delta) {
+  if (state == ByteStreamReader::STREAM_HAS_DATA && now - start > delta &&
+      !should_terminate) {
     BrowserThread::PostTask(
         BrowserThread::FILE, FROM_HERE,
-        base::Bind(&DownloadFileImpl::StreamActive,
-                   weak_factory_.GetWeakPtr()));
+        base::Bind(&DownloadFileImpl::StreamActive, weak_factory_.GetWeakPtr(),
+                   source_stream));
   }
 
   if (total_incoming_data_size)
     RecordFileThreadReceiveBuffers(num_buffers);
 
   RecordContiguousWriteTime(now - start);
 
   // Take care of communication with our observer.
   if (reason != DOWNLOAD_INTERRUPT_REASON_NONE) {
     // Error case for both upstream source and file write.
     // Shut down processing and signal an error to our observer.
     // Our observer will clean us up.
-    stream_reader_->RegisterCallback(base::Closure());
+    source_stream->stream_reader()->RegisterCallback(base::Closure());
     weak_factory_.InvalidateWeakPtrs();
     SendUpdate();  // Make info up to date before error.
     std::unique_ptr<crypto::SecureHash> hash_state = file_.Finish();
     BrowserThread::PostTask(
-        BrowserThread::UI,
-        FROM_HERE,
-        base::Bind(&DownloadDestinationObserver::DestinationError,
-                   observer_,
-                   reason,
-                   file_.bytes_so_far(),
-                   base::Passed(&hash_state)));
-  } else if (state == ByteStreamReader::STREAM_COMPLETE) {
-    // Signal successful completion and shut down processing.
-    stream_reader_->RegisterCallback(base::Closure());
-    weak_factory_.InvalidateWeakPtrs();
+        BrowserThread::UI, FROM_HERE,
+        base::Bind(&DownloadDestinationObserver::DestinationError, observer_,
+                   reason, TotalBytesReceived(), base::Passed(&hash_state)));
+  } else if (state == ByteStreamReader::STREAM_COMPLETE || should_terminate) {
+    // Signal successful completion or termination of the current stream.
+    source_stream->stream_reader()->RegisterCallback(base::Closure());
+    source_stream->set_finished(true);
+
+    // Inform observers.
     SendUpdate();
-    std::unique_ptr<crypto::SecureHash> hash_state = file_.Finish();
-    BrowserThread::PostTask(
-        BrowserThread::UI,
-        FROM_HERE,
-        base::Bind(&DownloadDestinationObserver::DestinationCompleted,
-                   observer_,
-                   file_.bytes_so_far(),
-                   base::Passed(&hash_state)));
+
+    bool all_stream_complete = true;
+    for (auto& stream : source_streams_) {
+      if (!stream.second->is_finished()) {
+        all_stream_complete = false;

[asanka, 2017/03/02 19:37:43]

This kind of logic tends to be quite fragile. Imagine what would happen if a new
ByteStreamReader addition has been scheduled to be added. This logic will cause
the DownloadFileImpl to prematurely declare completion before the new stream
reader is added.

(source_stream_ is also being accessed from both the UI and FILE threads due to
this logic here, but that's moot once the other thread safety issues are
addressed).

[xingliu, 2017/03/02 23:06:14]

Done, Added some comment in the header.

Already noticed this issue, SourceStreams is not supposed to dynamically add or
remove elements after the ctor(UI thread).

+        break;
+      }
+    }
+
+    // All the stream reader are completed, shut down file IO processing.
+    if (all_stream_complete) {
+      RecordFileBandwidth(bytes_seen_, disk_writes_time_,
+                          base::TimeTicks::Now() - download_start_);
+      weak_factory_.InvalidateWeakPtrs();
+      std::unique_ptr<crypto::SecureHash> hash_state = file_.Finish();
+      update_timer_.reset();
+      BrowserThread::PostTask(
+          BrowserThread::UI, FROM_HERE,
+          base::Bind(&DownloadDestinationObserver::DestinationCompleted,
+                     observer_, TotalBytesReceived(),
+                     base::Passed(&hash_state)));
+    }
   }
   if (net_log_.IsCapturing()) {
     net_log_.AddEvent(net::NetLogEventType::DOWNLOAD_STREAM_DRAINED,
                             base::Bind(&FileStreamDrainedNetLogCallback,
                                        total_incoming_data_size, num_buffers));
   }
 }
 
+void DownloadFileImpl::RegisterAndActivateStream(SourceStream* source_stream) {
+  DCHECK_CURRENTLY_ON(BrowserThread::FILE);
+  ByteStreamReader* stream_reader = source_stream->stream_reader();
+  if (stream_reader) {
+    stream_reader->RegisterCallback(
+        base::Bind(&DownloadFileImpl::StreamActive, weak_factory_.GetWeakPtr(),
+                   source_stream));
+    StreamActive(source_stream);
+  }
+}
+
+int64_t DownloadFileImpl::TotalBytesReceived() const {
+  // TODO(xingliu): Use slice info to figure out total bytes received.
+  return file_.bytes_so_far();
+}
+
 void DownloadFileImpl::SendUpdate() {
+  // TODO(xingliu): Update slice info to observer to update history db.
   BrowserThread::PostTask(
-      BrowserThread::UI,
-      FROM_HERE,
-      base::Bind(&DownloadDestinationObserver::DestinationUpdate,
-                 observer_,
-                 file_.bytes_so_far(),
-                 rate_estimator_.GetCountPerSecond()));
+      BrowserThread::UI, FROM_HERE,
+      base::Bind(&DownloadDestinationObserver::DestinationUpdate, observer_,
+                 TotalBytesReceived(), rate_estimator_.GetCountPerSecond()));
+}
+
+void DownloadFileImpl::WillWriteToDisk(size_t data_len) {
+  if (!update_timer_->IsRunning()) {
+    update_timer_->Start(FROM_HERE,
+                         base::TimeDelta::FromMilliseconds(kUpdatePeriodMs),
+                         this, &DownloadFileImpl::SendUpdate);
+  }
+  rate_estimator_.Increment(data_len);
 }
 
 DownloadFileImpl::RenameParameters::RenameParameters(
     RenameOption option,
     const base::FilePath& new_path,
     const RenameCompletionCallback& completion_callback)
     : option(option),
       new_path(new_path),
       retries_left(kMaxRenameRetries),
       completion_callback(completion_callback) {}
 
 DownloadFileImpl::RenameParameters::~RenameParameters() {}
 
 }  // namespace content