net: Make UploadDataStream::Read() asynchronous

net/base/upload_data_stream.cc

Index: net/base/upload_data_stream.cc
diff --git a/net/base/upload_data_stream.cc b/net/base/upload_data_stream.cc
index a62528d9bf816adacef194387cd06d21178f2fab..2a405600faca07adf01517251faf91bf921390f2 100644
--- a/net/base/upload_data_stream.cc
+++ b/net/base/upload_data_stream.cc
@@ -5,6 +5,7 @@
 #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"
@@ -62,7 +63,21 @@ int UploadDataStream::InitSync() {
   return OK;
 }
 
-int UploadDataStream::Read(IOBuffer* buf, int buf_len) {
+int UploadDataStream::Read(IOBuffer* buf,
+                           int buf_len,
+                           const CompletionCallback& callback) {
+  DCHECK(initialized_successfully_);
+
+  // Use fast path when all data is in memory, chunked data is also processed
+  // with ReadSync() since it requires special logic.

[willchan no longer on Chromium, 2012/09/24 23:37:08]

Are we going to be able to fix the chunked data to be async eventually? If so,
is there a bug open for that?

[hashimoto, 2012/09/25 10:47:56]

Filed http://crbug.com/152148

+  if (IsInMemory() || is_chunked())
+    return ReadSync(buf, buf_len);
+
+  ReadInternal(buf, buf_len, 0, callback, OK);
+  return ERR_IO_PENDING;
+}
+
+int UploadDataStream::ReadSync(IOBuffer* buf, int buf_len) {
   DCHECK(initialized_successfully_);
 
   // Initialize readers for newly appended chunks.
@@ -84,8 +99,10 @@ int UploadDataStream::Read(IOBuffer* buf, int buf_len) {
   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_;
 
@@ -179,4 +196,47 @@ void UploadDataStream::FinalizeInitialization() {
   initialized_successfully_ = true;
 }
 
+void UploadDataStream::ReadInternal(scoped_refptr<IOBuffer> buf,
+                                    int buf_len,
+                                    int bytes_copied,
+                                    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(&UploadDataStream::ReadInternal,
+                                               weak_ptr_factory_.GetWeakPtr(),
+                                               buf,
+                                               buf_len,
+                                               bytes_copied,
+                                               callback));
+    if (result == ERR_IO_PENDING)
+      return;
+
+    DCHECK_GE(bytes_copied, 0);
+    bytes_copied += result;
+  }
+  current_position_ += bytes_copied;
+
+  // Post task to the message loop to keep the execution order unchanged even if

[willchan no longer on Chromium, 2012/09/24 23:37:08]

What does this mean?

[hashimoto, 2012/09/25 10:47:56]

Sorry for my badly written comment.
Introduced more sophisticated way to determine if callback should be run or not.

Running the callback directly here may change the execution order of the
callback and the code placed after Read(), please consider the following
example.  What we want is always outputting "BeginEnd".

If there are functions below:
  void FuncA(const base::Closure& callback) {
    PostTask(FROM_HERE, callback);
  }
  void FuncB(const base::Closure& callback) {
    callback.Run();
  }
  void Begin() { std::cout << "Begin"; }
  void End() { std::cout << "End"; }

The code below outputs "BeginEnd":
  FuncA(base::Bind(&End));
  Begin();

But, the code below outputs "EndBegin":
  FuncB(base::Bind(&End));
  Begin();

+  // there were no asynchronous operation happening.
+  MessageLoop::current()->PostTask(FROM_HERE,
+                                   base::Bind(callback, bytes_copied));
+}
+
 }  // namespace net