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..4a17595e2e1afc8044e25ce967956bf08a4e401c 100644
--- a/net/base/upload_file_element_reader.cc
+++ b/net/base/upload_file_element_reader.cc
@@ -10,6 +10,7 @@
 #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,7 +21,7 @@ namespace {
 // UploadFileElementReader::GetContentLength() when set to non-zero.
 uint64 overriding_content_length = 0;
 
-// This method is used to implement Init().
+// This function is used to implement Init().
 void InitInternal(const FilePath& path,
                   uint64 range_offset,
                   uint64 range_length,
@@ -72,6 +73,35 @@ void InitInternal(const FilePath& path,
   *out_result = OK;
 }
 
+// This function is used to implement Read().
+void ReadInternal(scoped_refptr<IOBuffer> buf,
+                  int buf_length,
+                  uint64 bytes_remaining,
+                  FileStream* file_stream,
+                  int* result) {
+  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

[mmenke, 2012/10/15 19:54:45]

nit:  |file_stream|.

[hashimoto, 2012/10/16 11:52:20]

I don't think pipes are necessary here.

+    // 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);
+    }
+  }
+  *result = num_bytes_to_read;
+}
+
 }  // namespace
 
 UploadFileElementReader::UploadFileElementReader(
@@ -145,32 +175,42 @@ 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) {

[mmenke, 2012/10/15 19:54:45]

DCHECK(!callback.is_null());

[hashimoto, 2012/10/16 11:52:20]

Done.

+  if (BytesRemaining() == 0)
+    return 0;
+
+  // Save the value of file_stream_.get() before base::Passed() invalidates it.
+  FileStream* file_stream_ptr = file_stream_.get();
+  int* result = new int;
+  // 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::WorkerPool::PostTaskAndReply(
+      FROM_HERE,
+      base::Bind(&ReadInternal,
+                 scoped_refptr<IOBuffer>(buf),
+                 buf_length,
+                 BytesRemaining(),
+                 file_stream_ptr,
+                 result),
+      base::Bind(&UploadFileElementReader::OnReadCompleted,
+                 weak_ptr_factory_.GetWeakPtr(),
+                 base::Passed(&file_stream_),
+                 base::Owned(result),
+                 callback),
+      true /* task_is_slow */);
+  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;
+  int result = 0;
+  ReadInternal(buf, buf_length, BytesRemaining(), file_stream_.get(), &result);
+  OnReadCompleted(file_stream_.Pass(), &result, CompletionCallback());
+  return result;
 }
 
 void UploadFileElementReader::OnInitCompleted(
@@ -185,6 +225,17 @@ void UploadFileElementReader::OnInitCompleted(
     callback.Run(*result);
 }
 
+void UploadFileElementReader::OnReadCompleted(
+    scoped_ptr<FileStream> file_stream,
+    int* result,
+    const CompletionCallback& callback) {
+  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::
 ScopedOverridingContentLengthForTests(uint64 value) {
   overriding_content_length = value;