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

[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(
     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()),

[mmenke, 2012/10/15 21:34:01]

Are we leaking stuff here?  It looks like FileStreamWin, at least, only
automatically closes its Handle when created with the constructor that only
takes a NetLog as an argument, and FileStream does not use that constructor.

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

UploadFileElementReader is using the constructor that only takes a NetLog as an
argument.
Please see line 32. (new FileStream(NULL))

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

Ahhh...right.  For some reason, I was thinking both FileStream constructors were
using the 3-argument FileStreamWin constructor.

                                true /* task_is_slow */);
   }
 }
 
 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(
       FROM_HERE,
@@ skipping 32 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) {

[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(
     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