net: Make UploadDataStream::Read() asynchronous

net/base/upload_file_element_reader.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 "net/base/upload_file_element_reader.h"
 
 #include "base/bind.h"
 #include "base/file_util.h"
 #include "base/location.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 {
 
 namespace {
 
 // In tests, this value is used to override the return value of
 // 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,
                   const base::Time& expected_modification_time,
                   scoped_ptr<FileStream>* out_file_stream,
                   uint64* out_content_length,
                   int* out_result) {
   scoped_ptr<FileStream> file_stream(new FileStream(NULL));
   int64 rv = file_stream->OpenSync(
       path, base::PLATFORM_FILE_OPEN | base::PLATFORM_FILE_READ);
@@ skipping 31 matching lines @@
     if (file_util::GetFileInfo(path, &info) &&
         expected_modification_time.ToTimeT() != info.last_modified.ToTimeT()) {
       *out_result = ERR_UPLOAD_FILE_CHANGED;
       return;
     }
   }
 
   *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 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(
     const FilePath& path,
     uint64 range_offset,
     uint64 range_length,
     const base::Time& expected_modification_time)
     : path_(path),
       range_offset_(range_offset),
       range_length_(range_length),
       expected_modification_time_(expected_modification_time),
       content_length_(0),
       bytes_remaining_(0),
       weak_ptr_factory_(this) {
 }
 
 UploadFileElementReader::~UploadFileElementReader() {
   if (file_stream_.get()) {
     base::WorkerPool::PostTask(FROM_HERE,
                                base::Bind(&base::DeletePointer<FileStream>,
                                           file_stream_.release()),
                                true /* task_is_slow */);
   }
 }
 
 int UploadFileElementReader::Init(const CompletionCallback& callback) {

[mmenke, 2012/10/16 19:13:41]

For future consideration:  If we automatically called init on first read, rather
than having the parent call it directly, we'd know whether we were going to be
used synchronously or asynchronously.  Then we could open the file stream with
the necessary async flags in the async case, and use its own internal asyn
handling, which is nicer than the roll-your-own solution used here.

Alternatively, could just use where Init or InitSync was called to signal which
way we're going to be read from, and then construct the file with the
appropriate flag.

I really don't like passing stuff around between threads like this when we don't
have to.

[hashimoto, 2012/10/17 08:04:31]

I agree with you that ReadInternal() and InitInternal() look unusual and
unclean.
These functions are used to avoid code duplication between synchronous version
and asynchronous one.

Since the synchronous version methods are needed only for Chrome Frame, we
should consider fixing Chrome Frame rather than maintaining both
asynchronous/synchronous methods.

This class can be cleaned up if we fix Chrome Frame's UrlmonUploadDataStream to
support asynchronous initialization/reading and we get rid of InitSync() and
ReadSync() methods.

   scoped_ptr<FileStream>* file_stream = new scoped_ptr<FileStream>;
   uint64* content_length = new uint64;
   int* result = new int;
   const bool posted = base::WorkerPool::PostTaskAndReply(
       FROM_HERE,
       base::Bind(&InitInternal,
                  path_,
                  range_offset_,
                  range_length_,
                  expected_modification_time_,
@@ skipping 27 matching lines @@
 uint64 UploadFileElementReader::GetContentLength() const {
   if (overriding_content_length)
     return overriding_content_length;
   return content_length_;
 }
 
 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();
+  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(

[mmenke, 2012/10/16 19:13:41]

optional:  There's a PostTaskAndReplyWithResult that would remove the need for
passing result around.  Also makes ReadInternal's interface a little less
unusual.

[hashimoto, 2012/10/17 08:04:31]

Good to know, thanks.
Done.

+      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(
     scoped_ptr<FileStream>* file_stream,
     uint64* content_length,
     int* result,
     const CompletionCallback& callback) {
   file_stream_.swap(*file_stream);
   content_length_ = *content_length;
   bytes_remaining_ = GetContentLength();
   if (!callback.is_null())
     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;
 }
 
 UploadFileElementReader::ScopedOverridingContentLengthForTests::
 ~ScopedOverridingContentLengthForTests() {
   overriding_content_length = 0;
 }
 
 }  // namespace net