net: Make UploadDataStream::Read() asynchronous

net/base/upload_file_element_reader.cc

Index: net/base/upload_file_element_reader.cc
diff --git a/net/base/upload_file_element_reader.cc b/net/base/upload_file_element_reader.cc
index 5cb143984c9e39f6bff1dcbcaf4185ed65fb4ab1..0c1d08bf4e11ed710205d48b71949ad611ee9732 100644
--- a/net/base/upload_file_element_reader.cc
+++ b/net/base/upload_file_element_reader.cc
@@ -7,9 +7,11 @@
 #include "base/bind.h"
 #include "base/file_util.h"
 #include "base/location.h"
+#include "base/task_runner_util.h"
 #include "base/threading/thread_restrictions.h"
 #include "base/threading/worker_pool.h"
 #include "net/base/file_stream.h"
+#include "net/base/io_buffer.h"
 #include "net/base/net_errors.h"
 
 namespace net {
@@ -20,14 +22,13 @@ namespace {
 // UploadFileElementReader::GetContentLength() when set to non-zero.
 uint64 overriding_content_length = 0;
 
-// This method is used to implement Init().
-void InitInternal(const FilePath& path,
-                  uint64 range_offset,
-                  uint64 range_length,
-                  const base::Time& expected_modification_time,
-                  scoped_ptr<FileStream>* out_file_stream,
-                  uint64* out_content_length,
-                  int* out_result) {
+// This function is used to implement Init().
+int InitInternal(const FilePath& path,
+                 uint64 range_offset,
+                 uint64 range_length,
+                 const base::Time& expected_modification_time,
+                 scoped_ptr<FileStream>* out_file_stream,
+                 uint64* out_content_length) {
   scoped_ptr<FileStream> file_stream(new FileStream(NULL));
   int64 rv = file_stream->OpenSync(
       path, base::PLATFORM_FILE_OPEN | base::PLATFORM_FILE_READ);
@@ -64,12 +65,38 @@ void InitInternal(const FilePath& path,
     base::PlatformFileInfo info;
     if (file_util::GetFileInfo(path, &info) &&
         expected_modification_time.ToTimeT() != info.last_modified.ToTimeT()) {
-      *out_result = ERR_UPLOAD_FILE_CHANGED;
-      return;
+      return ERR_UPLOAD_FILE_CHANGED;
     }
   }
 
-  *out_result = OK;
+  return OK;
+}
+
+// This function is used to implement Read().
+int ReadInternal(scoped_refptr<IOBuffer> buf,
+                 int buf_length,
+                 uint64 bytes_remaining,
+                 FileStream* file_stream) {
+  DCHECK_LT(0, buf_length);
+
+  const uint64 num_bytes_to_read =
+      std::min(bytes_remaining, static_cast<uint64>(buf_length));
+
+  if (num_bytes_to_read > 0) {
+    int num_bytes_consumed = 0;
+    // file_stream is NULL if the target file is missing or not readable.
+    if (file_stream) {
+      num_bytes_consumed = file_stream->ReadSync(buf->data(),
+                                                 num_bytes_to_read);
+    }
+    if (num_bytes_consumed <= 0) {
+      // If there's less data to read than we initially observed, then
+      // pad with zero.  Otherwise the server will hang waiting for the
+      // rest of the data.
+      memset(buf->data(), 0, num_bytes_to_read);
+    }
+  }
+  return num_bytes_to_read;
 }
 
 }  // namespace
@@ -100,8 +127,8 @@ UploadFileElementReader::~UploadFileElementReader() {
 int UploadFileElementReader::Init(const CompletionCallback& callback) {
   scoped_ptr<FileStream>* file_stream = new scoped_ptr<FileStream>;
   uint64* content_length = new uint64;
-  int* result = new int;
-  const bool posted = base::WorkerPool::PostTaskAndReply(
+  const bool posted = base::PostTaskAndReplyWithResult(
+      base::WorkerPool::GetTaskRunner(true /* task_is_slow */),
       FROM_HERE,
       base::Bind(&InitInternal,
                  path_,
@@ -109,15 +136,12 @@ int UploadFileElementReader::Init(const CompletionCallback& callback) {
                  range_length_,
                  expected_modification_time_,
                  file_stream,
-                 content_length,
-                 result),
+                 content_length),
       base::Bind(&UploadFileElementReader::OnInitCompleted,
                  weak_ptr_factory_.GetWeakPtr(),
                  base::Owned(file_stream),
                  base::Owned(content_length),
-                 base::Owned(result),
-                 callback),
-      true /* task_is_slow */);
+                 callback));
   DCHECK(posted);
   return ERR_IO_PENDING;
 }
@@ -128,10 +152,10 @@ int UploadFileElementReader::InitSync() {
 
   scoped_ptr<FileStream> file_stream;
   uint64 content_length = 0;
-  int result = OK;
-  InitInternal(path_, range_offset_, range_length_, expected_modification_time_,
-               &file_stream, &content_length, &result);
-  OnInitCompleted(&file_stream, &content_length, &result, CompletionCallback());
+  const int result = InitInternal(path_, range_offset_, range_length_,
+                                  expected_modification_time_,
+                                  &file_stream, &content_length);
+  OnInitCompleted(&file_stream, &content_length, CompletionCallback(), result);
   return result;
 }
 
@@ -145,44 +169,64 @@ uint64 UploadFileElementReader::BytesRemaining() const {
   return bytes_remaining_;
 }
 
-int UploadFileElementReader::ReadSync(char* buf, int buf_length) {
+int UploadFileElementReader::Read(IOBuffer* buf,
+                                  int buf_length,
+                                  const CompletionCallback& callback) {
+  DCHECK(!callback.is_null());
+
+  if (BytesRemaining() == 0)
+    return 0;
+
+  // Save the value of file_stream_.get() before base::Passed() invalidates it.
+  FileStream* file_stream_ptr = file_stream_.get();
+  // Pass the ownership of file_stream_ to the worker pool to safely perform
+  // operation even when |this| is destructed before the read completes.
+  const bool posted = base::PostTaskAndReplyWithResult(
+      base::WorkerPool::GetTaskRunner(true /* task_is_slow */),
+      FROM_HERE,
+      base::Bind(&ReadInternal,
+                 scoped_refptr<IOBuffer>(buf),
+                 buf_length,
+                 BytesRemaining(),
+                 file_stream_ptr),
+      base::Bind(&UploadFileElementReader::OnReadCompleted,
+                 weak_ptr_factory_.GetWeakPtr(),
+                 base::Passed(&file_stream_),
+                 callback));
+  DCHECK(posted);
+  return ERR_IO_PENDING;
+}
+
+int UploadFileElementReader::ReadSync(IOBuffer* buf, int buf_length) {
   // Temporarily allow until fix: http://crbug.com/72001.
   base::ThreadRestrictions::ScopedAllowIO allow_io;
-  DCHECK_LT(0, buf_length);
-
-  const uint64 num_bytes_to_read =
-      static_cast<int>(std::min(BytesRemaining(),
-                                static_cast<uint64>(buf_length)));
-  if (num_bytes_to_read > 0) {
-    int num_bytes_consumed = 0;
-    // file_stream_ is NULL if the target file is
-    // missing or not readable.
-    if (file_stream_.get()) {
-      num_bytes_consumed =
-          file_stream_->ReadSync(buf, num_bytes_to_read);
-    }
-    if (num_bytes_consumed <= 0) {
-      // If there's less data to read than we initially observed, then
-      // pad with zero.  Otherwise the server will hang waiting for the
-      // rest of the data.
-      memset(buf, 0, num_bytes_to_read);
-    }
-  }
-  DCHECK_GE(bytes_remaining_, num_bytes_to_read);
-  bytes_remaining_ -= num_bytes_to_read;
-  return num_bytes_to_read;
+  const int result = ReadInternal(buf, buf_length, BytesRemaining(),
+                                  file_stream_.get());
+  OnReadCompleted(file_stream_.Pass(), CompletionCallback(), result);
+  return result;
 }
 
 void UploadFileElementReader::OnInitCompleted(
     scoped_ptr<FileStream>* file_stream,
     uint64* content_length,
-    int* result,
-    const CompletionCallback& callback) {
+    const CompletionCallback& callback,
+    int result) {
   file_stream_.swap(*file_stream);
   content_length_ = *content_length;
   bytes_remaining_ = GetContentLength();
   if (!callback.is_null())
-    callback.Run(*result);
+    callback.Run(result);
+}
+
+void UploadFileElementReader::OnReadCompleted(
+    scoped_ptr<FileStream> file_stream,
+    const CompletionCallback& callback,
+    int result) {
+  file_stream_.swap(file_stream);
+  DCHECK_GE(static_cast<int>(bytes_remaining_), result);
+  bytes_remaining_ -= result;
+  if (!callback.is_null())
+    callback.Run(result);
 }
 
 UploadFileElementReader::ScopedOverridingContentLengthForTests::