NetworkService: Fix a pair of bugs in DataPipeElementReader.

When a Read() call finished asynchronously, it was clearing its buf_
pointer after calling into its consumer. If its consumer called Read()
again, that could have resulted in populating buf_ again, which would
then incorrectly be cleared by the higher level call.

Also, if Init() interrupted a previous call to Init(), the callback to
the first Init() call was not being cancelled.

Bug: 794330
Cq-Include-Trybots: master.tryserver.chromium.linux:linux_mojo
Change-Id: I71faee5bcde302acd1962e55b4f2bb75d8c31c8d
Reviewed-on: https://chromium-review.googlesource.com/825843
Commit-Queue: Matt Menke <mmenke@chromium.org>
Reviewed-by: Randy Smith <rdsmith@chromium.org>
Cr-Commit-Position: refs/heads/master@{#524448}

content/network/BUILD.gn

@@ -38,6 +38,8 @@ jumbo_source_set("network_sources") {
     "cache_url_loader.h",
     "cookie_manager.cc",
     "cookie_manager.h",
+    "data_pipe_element_reader.cc",
+    "data_pipe_element_reader.h",
     "http_server_properties_pref_delegate.cc",
     "http_server_properties_pref_delegate.h",
     "network_change_manager.cc",

content/network/data_pipe_element_reader.cc

+// Copyright 2017 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 "content/network/data_pipe_element_reader.h"
+
+#include "base/bind.h"
+#include "base/callback.h"
+#include "base/location.h"
+#include "base/logging.h"
+#include "mojo/public/c/system/types.h"
+#include "net/base/io_buffer.h"
+#include "net/base/net_errors.h"
+
+namespace content {
+
+DataPipeElementReader::DataPipeElementReader(
+    scoped_refptr<ResourceRequestBody> resource_request_body,
+    network::mojom::DataPipeGetterPtr data_pipe_getter_)
+    : resource_request_body_(std::move(resource_request_body)),
+      data_pipe_getter_(std::move(data_pipe_getter_)),
+      handle_watcher_(FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::MANUAL),
+      weak_factory_(this) {}
+
+DataPipeElementReader::~DataPipeElementReader() {}
+
+int DataPipeElementReader::Init(const net::CompletionCallback& callback) {
+  DCHECK(callback);
+
+  // Init rewinds the stream. Throw away current state.
+  read_callback_.Reset();
+  buf_ = nullptr;
+  buf_length_ = 0;
+  handle_watcher_.Cancel();
+  size_ = 0;
+  bytes_read_ = 0;
+  // Need to do this to prevent any previously pending ReadCallback() invocation
+  // from running.
+  weak_factory_.InvalidateWeakPtrs();
+
+  // Get a new data pipe and start.
+  mojo::DataPipe data_pipe;
+  data_pipe_getter_->Read(std::move(data_pipe.producer_handle),
+                          base::BindOnce(&DataPipeElementReader::ReadCallback,
+                                         weak_factory_.GetWeakPtr()));
+  data_pipe_ = std::move(data_pipe.consumer_handle);
+  handle_watcher_.Watch(
+      data_pipe_.get(), MOJO_HANDLE_SIGNAL_READABLE,
+      base::BindRepeating(&DataPipeElementReader::OnHandleReadable,
+                          base::Unretained(this)));
+
+  init_callback_ = std::move(callback);
+  return net::ERR_IO_PENDING;
+}
+
+uint64_t DataPipeElementReader::GetContentLength() const {
+  return size_;
+}
+
+uint64_t DataPipeElementReader::BytesRemaining() const {
+  return size_ - bytes_read_;
+}
+
+int DataPipeElementReader::Read(net::IOBuffer* buf,
+                                int buf_length,
+                                const net::CompletionCallback& callback) {
+  DCHECK(callback);
+  DCHECK(!read_callback_);
+  DCHECK(!init_callback_);
+  DCHECK(!buf_);
+
+  int result = ReadInternal(buf, buf_length);
+  if (result == net::ERR_IO_PENDING) {
+    buf_ = buf;
+    buf_length_ = buf_length;
+    read_callback_ = std::move(callback);
+  }
+  return result;
+}
+
+void DataPipeElementReader::ReadCallback(int32_t status, uint64_t size) {
+  if (status == net::OK)
+    size_ = size;
+  if (init_callback_)
+    std::move(init_callback_).Run(status);
+}
+
+void DataPipeElementReader::OnHandleReadable(MojoResult result) {
+  DCHECK(read_callback_);
+  DCHECK(buf_);
+
+  // Final result of the Read() call, to be passed to the consumer.
+  int read_result;
+  if (result == MOJO_RESULT_OK) {
+    read_result = ReadInternal(buf_.get(), buf_length_);
+  } else {
+    read_result = net::ERR_FAILED;
+  }
+
+  buf_ = nullptr;
+  buf_length_ = 0;
+
+  if (read_result != net::ERR_IO_PENDING)
+    std::move(read_callback_).Run(read_result);
+}
+
+int DataPipeElementReader::ReadInternal(net::IOBuffer* buf, int buf_length) {
+  DCHECK(buf);
+  DCHECK_GT(buf_length, 0);
+
+  if (BytesRemaining() == 0)
+    return net::OK;
+
+  uint32_t num_bytes = buf_length;
+  MojoResult rv =
+      data_pipe_->ReadData(buf->data(), &num_bytes, MOJO_READ_DATA_FLAG_NONE);
+  if (rv == MOJO_RESULT_OK) {
+    bytes_read_ += num_bytes;
+    return num_bytes;
+  }
+
+  if (rv == MOJO_RESULT_SHOULD_WAIT) {
+    handle_watcher_.ArmOrNotify();
+    return net::ERR_IO_PENDING;
+  }
+
+  return net::ERR_FAILED;
+}
+
+}  // namespace content

content/network/data_pipe_element_reader.h

+// Copyright 2017 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.
+
+#ifndef CONTENT_NETWORK_DATA_PIPE_ELEMENT_READER_H_
+#define CONTENT_NETWORK_DATA_PIPE_ELEMENT_READER_H_
+
+#include <stdint.h>
+
+#include <memory>
+
+#include "base/macros.h"
+#include "base/memory/ref_counted.h"
+#include "base/memory/weak_ptr.h"
+#include "content/common/content_export.h"
+#include "content/public/common/resource_request_body.h"
+#include "mojo/public/cpp/system/data_pipe.h"
+#include "mojo/public/cpp/system/simple_watcher.h"
+#include "net/base/completion_callback.h"
+#include "net/base/upload_element_reader.h"
+#include "services/network/public/interfaces/data_pipe_getter.mojom.h"
+
+namespace net {
+class IOBuffer;
+}
+
+namespace content {
+
+// A subclass of net::UploadElementReader to read data pipes.
+class CONTENT_EXPORT DataPipeElementReader : public net::UploadElementReader {
+ public:
+  // |resource_request_body| is just passed in to keep the object around for the
+  // life of the ElementReader.
+  DataPipeElementReader(
+      scoped_refptr<ResourceRequestBody> resource_request_body,
+      network::mojom::DataPipeGetterPtr data_pipe_getter_);
+
+  ~DataPipeElementReader() override;
+
+  // net::UploadElementReader implementation:
+  int Init(const net::CompletionCallback& callback) override;
+  uint64_t GetContentLength() const override;
+  uint64_t BytesRemaining() const override;
+  int Read(net::IOBuffer* buf,
+           int buf_length,
+           const net::CompletionCallback& callback) override;
+
+ private:
+  // Callback invoked by DataPipeGetter::Read.
+  void ReadCallback(int32_t status, uint64_t size);
+
+  // Called by |handle_watcher_| when data is available or the pipe was closed,
+  // and there's a pending Read() call.
+  void OnHandleReadable(MojoResult result);
+
+  // Attempts to read data from |data_pipe_| and write it to |buf|. On success,
+  // writes the amount of data written. On failure, returns a net error code. If
+  // no data was available yet, tells |handle_watcher_| to start watching the
+  // pipe for data to become available and returns ERR_IO_PENDING. It's up to
+  // the caller to update |buf_| and |buf_length_| if needed.
+  int ReadInternal(net::IOBuffer* buf, int buf_length);
+
+  scoped_refptr<ResourceRequestBody> resource_request_body_;
+  network::mojom::DataPipeGetterPtr data_pipe_getter_;
+  mojo::ScopedDataPipeConsumerHandle data_pipe_;
+  mojo::SimpleWatcher handle_watcher_;
+
+  // Write buffer and its length. Populated when Read() is called but returns
+  // ERR_IO_PENDING. Cleared once the read completes.
+  scoped_refptr<net::IOBuffer> buf_;
+  int buf_length_ = 0;
+
+  // Total size of input, as passed to ReadCallback().
+  uint64_t size_ = 0;
+
+  uint64_t bytes_read_ = 0;
+  net::CompletionCallback init_callback_;
+  net::CompletionCallback read_callback_;
+
+  base::WeakPtrFactory<DataPipeElementReader> weak_factory_;
+
+  DISALLOW_COPY_AND_ASSIGN(DataPipeElementReader);
+};
+
+}  // namespace content
+
+#endif  // CONTENT_NETWORK_DATA_PIPE_ELEMENT_READER_H_

content/network/data_pipe_element_reader_unittest.cc

+// Copyright 2017 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 "content/network/data_pipe_element_reader.h"
+
+#include <stdint.h>
+
+#include <limits>
+#include <memory>
+
+#include "base/macros.h"
+#include "base/memory/ref_counted.h"
+#include "base/message_loop/message_loop.h"
+#include "base/run_loop.h"
+#include "base/strings/string_number_conversions.h"
+#include "mojo/common/data_pipe_utils.h"
+#include "mojo/public/cpp/bindings/binding.h"
+#include "mojo/public/cpp/system/data_pipe.h"
+#include "net/base/completion_callback.h"
+#include "net/base/io_buffer.h"
+#include "net/base/test_completion_callback.h"
+#include "services/network/public/interfaces/data_pipe_getter.mojom.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+// Most tests of this class are at the URLLoader layer. These tests focus on
+// things too difficult to cover with integration tests.
+
+namespace content {
+
+namespace {
+
+class PassThroughDataPipeGetter : public network::mojom::DataPipeGetter {
+ public:
+  explicit PassThroughDataPipeGetter() : binding_(this) {}
+
+  network::mojom::DataPipeGetterPtr GetDataPipeGetterPtr() {
+    EXPECT_FALSE(binding_.is_bound());
+
+    network::mojom::DataPipeGetterPtr data_pipe_getter_ptr;
+    binding_.Bind(mojo::MakeRequest(&data_pipe_getter_ptr));
+    return data_pipe_getter_ptr;
+  }
+
+  void WaitForRead(mojo::ScopedDataPipeProducerHandle* write_pipe,
+                   ReadCallback* read_callback) {
+    DCHECK(!run_loop_);
+
+    if (!write_pipe_.is_valid()) {
+      run_loop_ = std::make_unique<base::RunLoop>();
+      run_loop_->Run();
+      run_loop_.reset();
+    }
+
+    EXPECT_TRUE(write_pipe_.is_valid());
+    EXPECT_TRUE(read_callback_);
+
+    *write_pipe = std::move(write_pipe_);
+    *read_callback = std::move(read_callback_);
+  }
+
+ private:
+  // network::mojom::DataPipeGetter implementation:
+  void Read(mojo::ScopedDataPipeProducerHandle pipe,
+            ReadCallback callback) override {
+    EXPECT_FALSE(write_pipe_.is_valid());
+    EXPECT_FALSE(read_callback_);
+
+    write_pipe_ = std::move(pipe);
+    read_callback_ = std::move(callback);
+
+    if (run_loop_)
+      run_loop_->Quit();
+  }
+
+  void Clone(network::mojom::DataPipeGetterRequest request) override {
+    NOTIMPLEMENTED();
+  }
+
+  std::unique_ptr<base::RunLoop> run_loop_;
+
+  mojo::Binding<network::mojom::DataPipeGetter> binding_;
+  mojo::ScopedDataPipeProducerHandle write_pipe_;
+  ReadCallback read_callback_;
+
+  DISALLOW_COPY_AND_ASSIGN(PassThroughDataPipeGetter);
+};
+
+class DataPipeElementReaderTest : public testing::Test {
+ public:
+  DataPipeElementReaderTest()
+      : element_reader_(nullptr, data_pipe_getter_.GetDataPipeGetterPtr()) {}
+
+ protected:
+  base::MessageLoopForIO message_loop_;
+  PassThroughDataPipeGetter data_pipe_getter_;
+  DataPipeElementReader element_reader_;
+};
+
+// Test the case where a second Init() call occurs when there's a pending Init()
+// call in progress. The first call should be dropped, in favor of the second
+// one.
+TEST_F(DataPipeElementReaderTest, InitInterruptsInit) {
+  // Value deliberately outside of the range of an uint32_t, to catch any
+  // accidental conversions to an int.
+  const uint64_t kResponseBodySize = std::numeric_limits<uint32_t>::max();
+
+  // The network stack calls Init.
+  net::TestCompletionCallback first_init_callback;
+  EXPECT_EQ(net::ERR_IO_PENDING,
+            element_reader_.Init(first_init_callback.callback()));
+
+  // Wait for DataPipeGetter::Read() to be called.
+  mojo::ScopedDataPipeProducerHandle first_write_pipe;
+  network::mojom::DataPipeGetter::ReadCallback first_read_pipe_callback;
+  data_pipe_getter_.WaitForRead(&first_write_pipe, &first_read_pipe_callback);
+
+  // The network stack calls Init again, interrupting the previous call.
+  net::TestCompletionCallback second_init_callback;
+  EXPECT_EQ(net::ERR_IO_PENDING,
+            element_reader_.Init(second_init_callback.callback()));
+
+  // Wait for DataPipeGetter::Read() to be called again.
+  mojo::ScopedDataPipeProducerHandle second_write_pipe;
+  network::mojom::DataPipeGetter::ReadCallback second_read_pipe_callback;
+  data_pipe_getter_.WaitForRead(&second_write_pipe, &second_read_pipe_callback);
+
+  // Sending data on the first read pipe should do nothing.
+  std::move(first_read_pipe_callback)
+      .Run(net::ERR_FAILED, kResponseBodySize - 1);
+  // Run any pending tasks, to make sure nothing unexpected is queued.
+  base::RunLoop().RunUntilIdle();
+  EXPECT_FALSE(first_init_callback.have_result());
+  EXPECT_FALSE(second_init_callback.have_result());
+
+  // Sending data on the second pipe should result in the second init callback
+  // being invoked.
+  std::move(second_read_pipe_callback).Run(net::OK, kResponseBodySize);
+  EXPECT_EQ(net::OK, second_init_callback.WaitForResult());
+  EXPECT_FALSE(first_init_callback.have_result());
+
+  EXPECT_EQ(kResponseBodySize, element_reader_.GetContentLength());
+  EXPECT_EQ(kResponseBodySize, element_reader_.BytesRemaining());
+  EXPECT_FALSE(element_reader_.IsInMemory());
+
+  // Try to read from the body.
+  scoped_refptr<net::IOBufferWithSize> io_buffer(new net::IOBufferWithSize(10));
+  net::TestCompletionCallback read_callback;
+  EXPECT_EQ(net::ERR_IO_PENDING,
+            element_reader_.Read(io_buffer.get(), io_buffer->size(),
+                                 read_callback.callback()));
+
+  // Writes to the first write pipe should either fail, or succeed but be
+  // ignored.
+  mojo::common::BlockingCopyFromString("foo", first_write_pipe);
+  base::RunLoop().RunUntilIdle();
+  EXPECT_FALSE(read_callback.have_result());
+}
+
+// Test the case where a second Init() call occurs when there's a pending Read()
+// call in progress. The old Read() should be dropped, in favor of the new
+// Init().
+TEST_F(DataPipeElementReaderTest, InitInterruptsRead) {
+  // Value deliberately outside of the range of an uint32_t, to catch any
+  // accidental conversions to an int.
+  const uint64_t kResponseBodySize = std::numeric_limits<uint32_t>::max();
+
+  // The network stack calls Init.
+  net::TestCompletionCallback first_init_callback;
+  EXPECT_EQ(net::ERR_IO_PENDING,
+            element_reader_.Init(first_init_callback.callback()));
+
+  // Wait for DataPipeGetter::Read() to be called.
+  mojo::ScopedDataPipeProducerHandle first_write_pipe;
+  network::mojom::DataPipeGetter::ReadCallback first_read_pipe_callback;
+  data_pipe_getter_.WaitForRead(&first_write_pipe, &first_read_pipe_callback);
+  std::move(first_read_pipe_callback).Run(net::OK, kResponseBodySize);
+
+  ASSERT_EQ(net::OK, first_init_callback.WaitForResult());
+
+  scoped_refptr<net::IOBufferWithSize> first_io_buffer(
+      new net::IOBufferWithSize(10));
+  net::TestCompletionCallback first_read_callback;
+  EXPECT_EQ(net::ERR_IO_PENDING,
+            element_reader_.Read(first_io_buffer.get(), first_io_buffer->size(),
+                                 first_read_callback.callback()));
+
+  // The network stack calls Init again, interrupting the previous read.
+  net::TestCompletionCallback second_init_callback;
+  EXPECT_EQ(net::ERR_IO_PENDING,
+            element_reader_.Init(second_init_callback.callback()));
+
+  // Wait for DataPipeGetter::Read() to be called again.
+  mojo::ScopedDataPipeProducerHandle second_write_pipe;
+  network::mojom::DataPipeGetter::ReadCallback second_read_pipe_callback;
+  data_pipe_getter_.WaitForRead(&second_write_pipe, &second_read_pipe_callback);
+
+  // Run any pending tasks, to make sure nothing unexpected is queued.
+  base::RunLoop().RunUntilIdle();
+  EXPECT_FALSE(first_read_callback.have_result());
+  EXPECT_FALSE(second_init_callback.have_result());
+
+  // Sending data on the second pipe should result in the second init callback
+  // being invoked.
+  std::move(second_read_pipe_callback).Run(net::OK, kResponseBodySize);
+  EXPECT_EQ(net::OK, second_init_callback.WaitForResult());
+
+  EXPECT_EQ(kResponseBodySize, element_reader_.GetContentLength());
+  EXPECT_EQ(kResponseBodySize, element_reader_.BytesRemaining());
+  EXPECT_FALSE(element_reader_.IsInMemory());
+
+  // Try to read from the body.
+  scoped_refptr<net::IOBufferWithSize> io_buffer(new net::IOBufferWithSize(10));
+  net::TestCompletionCallback second_read_callback;
+  EXPECT_EQ(net::ERR_IO_PENDING,
+            element_reader_.Read(io_buffer.get(), io_buffer->size(),
+                                 second_read_callback.callback()));
+
+  // Writes to the first write pipe should either fail, or succeed but be
+  // ignored.
+  mojo::common::BlockingCopyFromString("foo", first_write_pipe);
+  base::RunLoop().RunUntilIdle();
+  EXPECT_FALSE(second_read_callback.have_result());
+}
+
+}  // namespace
+
+}  // namespace content

content/network/url_loader.cc

@@ -13,6 +13,7 @@
 #include "base/threading/thread_task_runner_handle.h"
 #include "base/time/time.h"
 #include "content/common/loader_util.h"
+#include "content/network/data_pipe_element_reader.h"
 #include "content/network/network_context.h"
 #include "content/network/network_service_impl.h"
 #include "content/public/common/referrer.h"
@@ -144,109 +145,6 @@ class RawFileElementReader : public net::UploadFileElementReader {
   DISALLOW_COPY_AND_ASSIGN(RawFileElementReader);
 };
 
-// A subclass of net::UploadElementReader to read data pipes.
-class DataPipeElementReader : public net::UploadElementReader {
- public:
-  DataPipeElementReader(
-      scoped_refptr<ResourceRequestBody> resource_request_body,
-      network::mojom::DataPipeGetterPtr data_pipe_getter_)
-      : resource_request_body_(std::move(resource_request_body)),
-        data_pipe_getter_(std::move(data_pipe_getter_)),
-        handle_watcher_(FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::MANUAL),
-        weak_factory_(this) {}
-
-  ~DataPipeElementReader() override {}
-
- private:
-  void ReadCallback(int32_t status, uint64_t size) {
-    if (status == net::OK)
-      size_ = size;
-    if (!init_callback_.is_null())
-      std::move(init_callback_).Run(status);
-  }
-
-  void OnHandleReadable(MojoResult result) {
-    if (result == MOJO_RESULT_OK) {
-      int read = Read(buf_.get(), buf_length_, net::CompletionCallback());
-      DCHECK_GT(read, 0);
-      std::move(read_callback_).Run(read);
-    } else {
-      std::move(read_callback_).Run(net::ERR_FAILED);
-    }
-    buf_ = nullptr;
-    buf_length_ = 0;
-  }
-
-  // net::UploadElementReader implementation:
-  int Init(const net::CompletionCallback& callback) override {
-    // Init rewinds the stream. Throw away current state.
-    if (!read_callback_.is_null())
-      std::move(read_callback_).Run(net::ERR_FAILED);
-    buf_ = nullptr;
-    buf_length_ = 0;
-    handle_watcher_.Cancel();
-    size_ = 0;
-    bytes_read_ = 0;
-
-    // Get a new data pipe and start.
-    mojo::DataPipe data_pipe;
-    data_pipe_getter_->Read(std::move(data_pipe.producer_handle),
-                            base::BindOnce(&DataPipeElementReader::ReadCallback,
-                                           weak_factory_.GetWeakPtr()));
-    data_pipe_ = std::move(data_pipe.consumer_handle);
-    handle_watcher_.Watch(data_pipe_.get(), MOJO_HANDLE_SIGNAL_READABLE,
-                          base::Bind(&DataPipeElementReader::OnHandleReadable,
-                                     base::Unretained(this)));
-
-    init_callback_ = std::move(callback);
-    return net::ERR_IO_PENDING;
-  }
-
-  uint64_t GetContentLength() const override { return size_; }
-
-  uint64_t BytesRemaining() const override { return size_ - bytes_read_; }
-
-  int Read(net::IOBuffer* buf,
-           int buf_length,
-           const net::CompletionCallback& callback) override {
-    if (!BytesRemaining())
-      return net::OK;
-
-    uint32_t num_bytes = buf_length;
-    MojoResult rv =
-        data_pipe_->ReadData(buf->data(), &num_bytes, MOJO_READ_DATA_FLAG_NONE);
-    if (rv == MOJO_RESULT_OK) {
-      bytes_read_ += num_bytes;
-      return num_bytes;
-    }
-
-    if (rv == MOJO_RESULT_SHOULD_WAIT) {
-      buf_ = buf;
-      buf_length_ = buf_length;
-      handle_watcher_.ArmOrNotify();
-      read_callback_ = std::move(callback);
-      return net::ERR_IO_PENDING;
-    }
-
-    return net::ERR_FAILED;
-  }
-
-  scoped_refptr<ResourceRequestBody> resource_request_body_;
-  network::mojom::DataPipeGetterPtr data_pipe_getter_;
-  mojo::ScopedDataPipeConsumerHandle data_pipe_;
-  mojo::SimpleWatcher handle_watcher_;
-  scoped_refptr<net::IOBuffer> buf_;
-  int buf_length_ = 0;
-  uint64_t size_ = 0;
-  uint64_t bytes_read_ = 0;
-  net::CompletionCallback init_callback_;
-  net::CompletionCallback read_callback_;
-
-  base::WeakPtrFactory<DataPipeElementReader> weak_factory_;
-
-  DISALLOW_COPY_AND_ASSIGN(DataPipeElementReader);
-};
-
 // TODO: copied from content/browser/loader/upload_data_stream_builder.cc.
 std::unique_ptr<net::UploadDataStream> CreateUploadDataStream(
     ResourceRequestBody* body,

content/network/url_loader_unittest.cc

@@ -47,6 +47,7 @@
 #include "net/url_request/url_request_job.h"
 #include "net/url_request/url_request_status.h"
 #include "net/url_request/url_request_test_job.h"
+#include "services/network/public/interfaces/data_pipe_getter.mojom.h"
 #include "testing/gtest/include/gtest/gtest.h"
 #include "url/gurl.h"
 
@@ -162,21 +163,47 @@ class MultipleWritesInterceptor : public net::URLRequestInterceptor {
   DISALLOW_COPY_AND_ASSIGN(MultipleWritesInterceptor);
 };
 
-class SimpleDataPipeGetter : public network::mojom::DataPipeGetter {
+class TestDataPipeGetter : public network::mojom::DataPipeGetter {
  public:
-  SimpleDataPipeGetter(const std::string& str,
-                       network::mojom::DataPipeGetterRequest request)
-      : str_(str) {
+  TestDataPipeGetter(const std::string& upload_string,
+                     network::mojom::DataPipeGetterRequest request)
+      : upload_string_(upload_string) {
     bindings_.AddBinding(this, std::move(request));
   }
-  ~SimpleDataPipeGetter() override = default;
+
+  ~TestDataPipeGetter() override = default;
+
+  // If set to anything other than net::OK, won't bother to upload the body.
+  void set_start_error(net::Error start_error) { start_error_ = start_error; }
+
+  // If set to true, will advertise the body size as being 1 byte larger than
+  // upload string, so the data pipe will be closed when the URLLoader is still
+  // expecting more data.
+  void set_pipe_closed_early(bool pipe_closed_early) {
+    pipe_closed_early_ = pipe_closed_early;
+  }
 
   // network::mojom::DataPipeGetter implementation:
-  void Read(mojo::ScopedDataPipeProducerHandle handle,
+  void Read(mojo::ScopedDataPipeProducerHandle pipe,
             ReadCallback callback) override {
-    bool result = mojo::common::BlockingCopyFromString(str_, handle);
-    ASSERT_TRUE(result);
-    std::move(callback).Run(net::OK, str_.length());
+    uint64_t advertised_length = upload_string_.length();
+    if (pipe_closed_early_)
+      advertised_length += 1;
+    std::move(callback).Run(start_error_, advertised_length);
+    if (start_error_ != net::OK)
+      return;
+
+    write_position_ = 0;
+    upload_body_pipe_ = std::move(pipe);
+    handle_watcher_ = std::make_unique<mojo::SimpleWatcher>(
+        FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::MANUAL);
+    handle_watcher_->Watch(
+        upload_body_pipe_.get(),
+        // Don't bother watching for close - rely on read pipes for errors.
+        MOJO_HANDLE_SIGNAL_WRITABLE, MOJO_WATCH_CONDITION_SATISFIED,
+        base::BindRepeating(&TestDataPipeGetter::MojoReadyCallback,
+                            base::Unretained(this)));
+    WriteData();
   }
 
   void Clone(network::mojom::DataPipeGetterRequest request) override {
@@ -184,10 +211,60 @@ class SimpleDataPipeGetter : public network::mojom::DataPipeGetter {
   }
 
  private:
-  std::string str_;
+  void MojoReadyCallback(MojoResult result,
+                         const mojo::HandleSignalsState& state) {
+    WriteData();
+  }
+
+  void WriteData() {
+    DCHECK_LE(write_position_, upload_string_.length());
+
+    while (true) {
+      uint32_t write_size = static_cast<uint32_t>(
+          std::min(static_cast<size_t>(32 * 1024),
+                   upload_string_.length() - write_position_));
+      if (write_size == 0) {
+        // Upload is done. Close the upload body pipe and wait for another call
+        // to Read().
+        handle_watcher_.reset();
+        upload_body_pipe_.reset();
+        return;
+      }
+
+      int result =
+          upload_body_pipe_->WriteData(upload_string_.data() + write_position_,
+                                       &write_size, MOJO_WRITE_DATA_FLAG_NONE);
+      if (result == MOJO_RESULT_SHOULD_WAIT) {
+        handle_watcher_->ArmOrNotify();
+        return;
+      }
+
+      if (result != MOJO_RESULT_OK) {
+        // Ignore the pipe being closed - the upload may still be retried with
+        // another call to Read.
+        handle_watcher_.reset();
+        upload_body_pipe_.reset();
+        return;
+      }
+
+      write_position_ += write_size;
+      DCHECK_LE(write_position_, upload_string_.length());
+    }
+  }
+
+ private:
+  const std::string upload_string_;
+  net::Error start_error_ = net::OK;
+  bool pipe_closed_early_ = false;
+
   mojo::BindingSet<network::mojom::DataPipeGetter> bindings_;
 
-  DISALLOW_COPY_AND_ASSIGN(SimpleDataPipeGetter);
+  mojo::ScopedDataPipeProducerHandle upload_body_pipe_;
+  // Must be below |upload_body_pipe_|, so it's deleted first.
+  std::unique_ptr<mojo::SimpleWatcher> handle_watcher_;
+  size_t write_position_ = 0;
+
+  DISALLOW_COPY_AND_ASSIGN(TestDataPipeGetter);
 };
 
 class RequestInterceptor : public net::URLRequestInterceptor {
@@ -1130,12 +1207,12 @@ TEST_F(URLLoaderTest, UploadDataPipe) {
   const std::string kRequestBody = "Request Body";
 
   network::mojom::DataPipeGetterPtr data_pipe_getter_ptr;
-  auto data_pipe_getter = std::make_unique<SimpleDataPipeGetter>(
+  auto data_pipe_getter = std::make_unique<TestDataPipeGetter>(
       kRequestBody, mojo::MakeRequest(&data_pipe_getter_ptr));
 
-  auto request_body = base::MakeRefCounted<ResourceRequestBody>();
-  request_body->AppendDataPipe(std::move(data_pipe_getter_ptr));
-  set_request_body(std::move(request_body));
+  auto resource_request_body = base::MakeRefCounted<ResourceRequestBody>();
+  resource_request_body->AppendDataPipe(std::move(data_pipe_getter_ptr));
+  set_request_body(std::move(resource_request_body));
 
   std::string response_body;
   EXPECT_EQ(net::OK, Load(test_server()->GetURL("/echo"), &response_body));
@@ -1147,12 +1224,12 @@ TEST_F(URLLoaderTest, UploadDataPipe_Redirect307) {
   const std::string kRequestBody = "Request Body";
 
   network::mojom::DataPipeGetterPtr data_pipe_getter_ptr;
-  auto data_pipe_getter = std::make_unique<SimpleDataPipeGetter>(
+  auto data_pipe_getter = std::make_unique<TestDataPipeGetter>(
       kRequestBody, mojo::MakeRequest(&data_pipe_getter_ptr));
 
-  auto request_body = base::MakeRefCounted<ResourceRequestBody>();
-  request_body->AppendDataPipe(std::move(data_pipe_getter_ptr));
-  set_request_body(std::move(request_body));
+  auto resource_request_body = base::MakeRefCounted<ResourceRequestBody>();
+  resource_request_body->AppendDataPipe(std::move(data_pipe_getter_ptr));
+  set_request_body(std::move(resource_request_body));
   set_expect_redirect();
 
   std::string response_body;
@@ -1161,6 +1238,60 @@ TEST_F(URLLoaderTest, UploadDataPipe_Redirect307) {
   EXPECT_EQ(kRequestBody, response_body);
 }
 
+// Tests a large request body, which should result in multiple asynchronous
+// reads.
+TEST_F(URLLoaderTest, UploadDataPipeWithLotsOfData) {
+  std::string request_body;
+  request_body.reserve(5 * 1024 * 1024);
+  // Using a repeating patter with a length that's prime is more likely to spot
+  // out of order or repeated chunks of data.
+  while (request_body.size() < 5 * 1024 * 1024)
+    request_body.append("foppity");
+
+  network::mojom::DataPipeGetterPtr data_pipe_getter_ptr;
+  auto data_pipe_getter = std::make_unique<TestDataPipeGetter>(
+      request_body, mojo::MakeRequest(&data_pipe_getter_ptr));
+
+  auto resource_request_body = base::MakeRefCounted<ResourceRequestBody>();
+  resource_request_body->AppendDataPipe(std::move(data_pipe_getter_ptr));
+  set_request_body(std::move(resource_request_body));
+
+  std::string response_body;
+  EXPECT_EQ(net::OK, Load(test_server()->GetURL("/echo"), &response_body));
+  EXPECT_EQ(request_body, response_body);
+}
+
+TEST_F(URLLoaderTest, UploadDataPipeError) {
+  const std::string kRequestBody = "Request Body";
+
+  network::mojom::DataPipeGetterPtr data_pipe_getter_ptr;
+  auto data_pipe_getter = std::make_unique<TestDataPipeGetter>(
+      kRequestBody, mojo::MakeRequest(&data_pipe_getter_ptr));
+  data_pipe_getter->set_start_error(net::ERR_ACCESS_DENIED);
+
+  auto resource_request_body = base::MakeRefCounted<ResourceRequestBody>();
+  resource_request_body->AppendDataPipe(std::move(data_pipe_getter_ptr));
+  set_request_body(std::move(resource_request_body));
+
+  EXPECT_EQ(net::ERR_ACCESS_DENIED, Load(test_server()->GetURL("/echo")));
+}
+
+TEST_F(URLLoaderTest, UploadDataPipeClosedEarly) {
+  const std::string kRequestBody = "Request Body";
+
+  network::mojom::DataPipeGetterPtr data_pipe_getter_ptr;
+  auto data_pipe_getter = std::make_unique<TestDataPipeGetter>(
+      kRequestBody, mojo::MakeRequest(&data_pipe_getter_ptr));
+  data_pipe_getter->set_pipe_closed_early(true);
+
+  auto resource_request_body = base::MakeRefCounted<ResourceRequestBody>();
+  resource_request_body->AppendDataPipe(std::move(data_pipe_getter_ptr));
+  set_request_body(std::move(resource_request_body));
+
+  std::string response_body;
+  EXPECT_EQ(net::ERR_FAILED, Load(test_server()->GetURL("/echo")));
+}
+
 TEST_F(URLLoaderTest, UploadDoubleRawFile) {
   base::FilePath file_path = GetTestFilePath("simple_page.html");
 

content/test/BUILD.gn

@@ -1509,6 +1509,7 @@ test("content_unittests") {
     "../common/unique_name_helper_unittest.cc",
     "../common/webplugininfo_unittest.cc",
     "../network/cookie_manager_unittest.cc",
+    "../network/data_pipe_element_reader_unittest.cc",
     "../network/network_change_manager_unittest.cc",
     "../network/network_context_unittest.cc",
     "../network/network_service_unittest.cc",

net/base/upload_element_reader.h

@@ -33,8 +33,9 @@ class NET_EXPORT UploadElementReader {
 
   // This function must be called before calling any other method. It is not
   // valid to call any method (other than the destructor) if Init() fails.
-  // This method can be called multiple times. Calling this method after an
-  // Init() success results in resetting the state (i.e. the stream is rewound).
+  // This method can be called multiple times. Calling this results in resetting
+  // the state (i.e. the stream is rewound), and any previously pending Init()
+  // or Read() calls are aborted.
   //
   // Initializes the instance synchronously when possible, otherwise does
   // initialization aynschronously, returns ERR_IO_PENDING and runs callback.