net: Make UploadDataStream::Read() asynchronous

net/base/upload_data_stream.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_data_stream.h"
 
 #include "base/logging.h"
+#include "base/message_loop.h"
 #include "net/base/io_buffer.h"
 #include "net/base/net_errors.h"
 #include "net/base/upload_element_reader.h"
 
 namespace net {
 
 bool UploadDataStream::merge_chunks_ = true;
 
 // static
 void UploadDataStream::ResetMergeChunks() {
@@ skipping 37 matching lines @@
     if (result != OK) {
       element_readers_.clear();
       return result;
     }
   }
 
   FinalizeInitialization();
   return OK;
 }
 
-int UploadDataStream::Read(IOBuffer* buf, int buf_len) {
+int UploadDataStream::Read(IOBuffer* buf,
+                           int buf_len,
+                           const CompletionCallback& callback) {
+  DCHECK(initialized_successfully_);

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

DCHECK(!callback.is_null());

Probably also a good idea to DCHECK the |buf_len| > 0 both here and in ReadSync.

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

Done.

+
+  // Use fast path when all data is in memory, chunked data is also processed
+  // with ReadSync() since it requires special logic, and is always in memory.
+  if (IsInMemory() || is_chunked())
+    return ReadSync(buf, buf_len);
+
+  const bool invoked_asynchronously = false;
+  return ReadInternal(buf, buf_len, 0, invoked_asynchronously, callback, 0);
+}
+
+int UploadDataStream::ReadSync(IOBuffer* buf, int buf_len) {
   DCHECK(initialized_successfully_);
 
   // Initialize readers for newly appended chunks.
   if (is_chunked()) {
     const std::vector<UploadElement>& elements = *upload_data_->elements();
     DCHECK_LE(element_readers_.size(), elements.size());
 
     for (size_t i = element_readers_.size(); i < elements.size(); ++i) {
+      // We can initialize readeres synchrnously here because only bytes can be

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

nit:  readeres -> readers
synchrnously -> synchronously

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

Oops, thanks.
Done.

+      // appended for chunked data. We don't add each element's size to
+      // |toal_size_| because we treat the size of chunked data as zero.

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

nit:  |toal_size_| -> |total_size_|

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

"because we treat the size of chunked data as zero" isn't quite right.  Maybe
just remove that sentence, now that we've spelled out that it's zero in the
header.  Alternatively, "We leave |total_size_| at zero, since for chunked
uploads, we may not know the total size".

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

Done.

       const UploadElement& element = elements[i];
       DCHECK_EQ(UploadElement::TYPE_BYTES, element.type());
       UploadElementReader* reader = UploadElementReader::Create(element);
 
       const int rv = reader->InitSync();
       DCHECK_EQ(rv, OK);
       element_readers_.push_back(reader);
     }
   }
 
   int bytes_copied = 0;
   while (bytes_copied < buf_len && element_index_ < element_readers_.size()) {
     UploadElementReader* reader = element_readers_[element_index_];
-    bytes_copied += reader->ReadSync(buf->data() + bytes_copied,
-                                     buf_len - bytes_copied);
+    scoped_refptr<DrainableIOBuffer> sub_buffer =
+        new DrainableIOBuffer(buf, buf_len);
+    sub_buffer->SetOffset(bytes_copied);
+    bytes_copied += reader->ReadSync(sub_buffer, sub_buffer->BytesRemaining());
     if (reader->BytesRemaining() == 0)
       ++element_index_;
 
     if (is_chunked() && !merge_chunks_)
       break;
   }
 
   current_position_ += bytes_copied;
   if (is_chunked() && !IsEOF() && bytes_copied == 0)
     return ERR_IO_PENDING;
@@ skipping 73 matching lines @@
     uint64 total_size = 0;
     for (size_t i = 0; i < element_readers_.size(); ++i) {
       UploadElementReader* reader = element_readers_[i];
       total_size += reader->GetContentLength();
     }
     total_size_ = total_size;
   }
   initialized_successfully_ = true;
 }
 
+int UploadDataStream::ReadInternal(scoped_refptr<IOBuffer> buf,
+                                   int buf_len,
+                                   int bytes_copied,

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

I think this would be simpler if instead of passing around buf, buf_len, and
bytes_copied, we just created a DrainableIOBuffer (what you call sub_buffer)
before calling ReadInternal.  It already tracks buf_len (Its size() function)
and bytes_copies (Its BytesConsumed() function).

Also, I suggest making both the DrainableIOBuffer and the callback class
variables - this allows us to DCHECK that there's no other ReadInternal call
pending when Read and ReadSync are called, and is more consistent with other
net/ code.  You can also get rid of the invoked_asynchronously argument in that
case (Set callback after the call to ReadInternal, only if it returns
IO_PENDING).

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

Sounds good, done.
I think we should make UploadDataStream as stateless as possible and limit the
lifetime of variables even if it results in increasing the number of arguments.
If we really want to check that there is no pending tasks, we can have
DCHECK(!weak_ptr_factory_.HasWeakPtrs()).

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

I don't have particularly strong feelings about this, just a general dislike of
passing lots of variables around from callback to callback when we don't have
to, and hiding state.

What's the reasoning for keeping it as "stateless" as possible?  Making the
state owned by a callback rather than the parent class doesn't make the class
have less state, it just obfuscates the state.

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

IMHO the destination buffer and the callback are the states of a Read request,
not the UploadDataStream itself.
Since instance variables and method arguments are in different scopes and the
latter is narrower, I would like to avoid mixing the instance's states and the
Read request's by placing variables in the argument list when possible.

+                                   bool invoked_asynchronously,
+                                   const CompletionCallback& callback,
+                                   int previous_result) {
+  DCHECK(initialized_successfully_);
+  DCHECK_GE(previous_result, 0);
+
+  // Add the last result.
+  bytes_copied += previous_result;
+
+  while (bytes_copied < buf_len && element_index_ < element_readers_.size()) {
+    UploadElementReader* reader = element_readers_[element_index_];
+
+    if (reader->BytesRemaining() == 0) {
+      ++element_index_;
+      continue;
+    }
+    scoped_refptr<DrainableIOBuffer> sub_buffer =
+        new DrainableIOBuffer(buf, buf_len);
+    sub_buffer->SetOffset(bytes_copied);
+    const int result = reader->Read(
+        sub_buffer,
+        sub_buffer->BytesRemaining(),
+        base::Bind(base::IgnoreResult(&UploadDataStream::ReadInternal),
+                   weak_ptr_factory_.GetWeakPtr(),
+                   buf,
+                   buf_len,
+                   bytes_copied,
+                   true,  // invoked_asynchronously
+                   callback));
+    if (result == ERR_IO_PENDING)
+      return ERR_IO_PENDING;
+
+    DCHECK_GE(bytes_copied, 0);
+    bytes_copied += result;
+  }
+  current_position_ += bytes_copied;
+
+  // When invoked asynchronously, callback is the only way to return the result.
+  if (invoked_asynchronously)
+    callback.Run(bytes_copied);
+  return bytes_copied;
+}
+
 }  // namespace net