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_);
+
+  // 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.
   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) {
       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;
 }
 
+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