Create a TYPE_CHUNKED_DATA_PIPE DataElement.

And add support for it to services/network.

This CL introduces network::mojom::ChunkedDataPipeGetters which is
similar to DataPipeGettter, but is used for chunked uploads, which
allow consumers to start uploading a request body before its size
is known.

Cq-Include-Trybots: master.tryserver.chromium.linux:linux_mojo
Change-Id: I7d7dfad4b6e33329cf56f7c9d1666944a7b25475
Bug: 746977
Reviewed-on: https://chromium-review.googlesource.com/908828
Commit-Queue: Matt Menke <mmenke@chromium.org>
Reviewed-by: Daniel Cheng <dcheng@chromium.org>
Reviewed-by: Helen Li <xunjieli@chromium.org>
Reviewed-by: Kinuko Yasuda <kinuko@chromium.org>
Cr-Commit-Position: refs/heads/master@{#541594}

content/browser/loader/upload_data_stream_builder.cc

@@ -110,6 +110,7 @@ std::unique_ptr<net::UploadDataStream> UploadDataStreamBuilder::Build(
       }
       case network::DataElement::TYPE_RAW_FILE:
       case network::DataElement::TYPE_DATA_PIPE:
+      case network::DataElement::TYPE_CHUNKED_DATA_PIPE:
       case network::DataElement::TYPE_UNKNOWN:
         NOTREACHED();
         break;

content/common/page_state_serialization.cc

@@ -708,6 +708,7 @@ void WriteResourceRequestBody(const network::ResourceRequestBody& request_body,
         break;
       case network::DataElement::TYPE_RAW_FILE:
       case network::DataElement::TYPE_DATA_PIPE:
+      case network::DataElement::TYPE_CHUNKED_DATA_PIPE:
       case network::DataElement::TYPE_UNKNOWN:
         NOTREACHED();
         continue;

content/common/service_worker/service_worker_loader_helpers.cc

@@ -181,6 +181,10 @@ ServiceWorkerLoaderHelpers::CloneResourceRequestBody(
                                   element.offset(), element.length(),
                                   element.expected_modification_time());
         break;
+      case network::DataElement::TYPE_CHUNKED_DATA_PIPE:
+        NOTREACHED() << "There should be no chunked data pipes going through "
+                        "ServiceWorker";
+        break;
       case network::DataElement::TYPE_BLOB:
         NOTREACHED() << "There should be no blob elements in NetworkService";
         break;

content/renderer/loader/web_url_request_util.cc

@@ -424,6 +424,7 @@ WebHTTPBody GetWebHTTPBodyForRequestBody(
       }
       case network::DataElement::TYPE_UNKNOWN:
       case network::DataElement::TYPE_RAW_FILE:
+      case network::DataElement::TYPE_CHUNKED_DATA_PIPE:
         NOTREACHED();
         break;
     }

services/network/BUILD.gn

@@ -10,6 +10,8 @@ import("//services/service_manager/public/tools/test/service_test.gni")
 
 component("network_service") {
   sources = [
+    "chunked_data_pipe_upload_data_stream.cc",
+    "chunked_data_pipe_upload_data_stream.h",
     "conditional_cache_deletion_helper.cc",
     "conditional_cache_deletion_helper.h",
     "cookie_manager.cc",
@@ -119,6 +121,7 @@ source_set("tests") {
   testonly = true
 
   sources = [
+    "chunked_data_pipe_upload_data_stream_unittest.cc",
     "cookie_manager_unittest.cc",
     "cors/cors_url_loader_unittest.cc",
     "cors/preflight_controller_unittest.cc",
@@ -134,6 +137,8 @@ source_set("tests") {
     "resource_scheduler_unittest.cc",
     "restricted_cookie_manager_unittest.cc",
     "test/test_url_loader_factory_unittest.cc",
+    "test_chunked_data_pipe_getter.cc",
+    "test_chunked_data_pipe_getter.h",
     "throttling/throttling_controller_unittest.cc",
     "udp_socket_factory_unittest.cc",
     "udp_socket_unittest.cc",

services/network/chunked_data_pipe_upload_data_stream.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 "services/network/chunked_data_pipe_upload_data_stream.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"
+
+namespace network {
+
+ChunkedDataPipeUploadDataStream::ChunkedDataPipeUploadDataStream(
+    scoped_refptr<ResourceRequestBody> resource_request_body,
+    mojom::ChunkedDataPipeGetterPtr chunked_data_pipe_getter)
+    : net::UploadDataStream(true /* is_chunked */,
+                            resource_request_body->identifier()),
+      resource_request_body_(std::move(resource_request_body)),
+      chunked_data_pipe_getter_(std::move(chunked_data_pipe_getter)),
+      handle_watcher_(FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::MANUAL) {
+  chunked_data_pipe_getter_.set_connection_error_handler(
+      base::BindOnce(&ChunkedDataPipeUploadDataStream::OnDataPipeGetterClosed,
+                     base::Unretained(this)));
+  chunked_data_pipe_getter_->GetSize(
+      base::BindOnce(&ChunkedDataPipeUploadDataStream::OnSizeReceived,
+                     base::Unretained(this)));
+}
+
+ChunkedDataPipeUploadDataStream::~ChunkedDataPipeUploadDataStream() {}
+
+int ChunkedDataPipeUploadDataStream::InitInternal(
+    const net::NetLogWithSource& net_log) {
+  // If there was an error either passed to the ReadCallback or as a result of
+  // closing the DataPipeGetter pipe, fail the read.
+  if (status_ != net::OK)
+    return status_;
+
+  // If the data pipe was closed, just fail initialization.
+  if (chunked_data_pipe_getter_.encountered_error())
+    return net::ERR_FAILED;
+
+  // Get a new data pipe and start.
+  mojo::DataPipe data_pipe;
+  chunked_data_pipe_getter_->StartReading(std::move(data_pipe.producer_handle));
+  data_pipe_ = std::move(data_pipe.consumer_handle);
+  handle_watcher_.Watch(
+      data_pipe_.get(),
+      MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+      base::BindRepeating(&ChunkedDataPipeUploadDataStream::OnHandleReadable,
+                          base::Unretained(this)));
+
+  return net::OK;
+}
+
+int ChunkedDataPipeUploadDataStream::ReadInternal(net::IOBuffer* buf,
+                                                  int buf_len) {
+  DCHECK(!buf_);
+  DCHECK(buf);
+  DCHECK_GT(buf_len, 0);
+
+  // If there was an error either passed to the ReadCallback or as a result of
+  // closing the DataPipeGetter pipe, fail the read.
+  if (status_ != net::OK)
+    return status_;
+
+  // Nothing else to do, if the entire body was read.
+  if (size_ && bytes_read_ == *size_) {
+    // This shouldn't be called if the stream was already completed.
+    DCHECK(!IsEOF());
+
+    SetIsFinalChunk();
+    return net::OK;
+  }
+
+  uint32_t num_bytes = buf_len;
+  if (size_ && num_bytes > *size_ - bytes_read_)
+    num_bytes = *size_ - bytes_read_;
+  MojoResult rv =
+      data_pipe_->ReadData(buf->data(), &num_bytes, MOJO_READ_DATA_FLAG_NONE);
+  if (rv == MOJO_RESULT_OK) {
+    bytes_read_ += num_bytes;
+    // Not needed for correctness, but this allows the consumer to send the
+    // final chunk and the end of stream message together, for protocols that
+    // allow it.
+    if (size_ && *size_ == bytes_read_)
+      SetIsFinalChunk();
+    return num_bytes;
+  }
+
+  if (rv == MOJO_RESULT_SHOULD_WAIT) {
+    handle_watcher_.ArmOrNotify();
+    buf_ = buf;
+    buf_len_ = buf_len;
+    return net::ERR_IO_PENDING;
+  }
+
+  // The pipe was closed. If the size isn't known yet, could be a success or a
+  // failure.
+  if (!size_) {
+    buf_ = buf;
+    buf_len_ = buf_len;
+    data_pipe_.reset();
+    return net::ERR_IO_PENDING;
+  }
+
+  // |size_| was checked earlier, so if this point is reached, the pipe was
+  // closed before receiving all bytes.
+  DCHECK_LT(bytes_read_, *size_);
+
+  return net::ERR_FAILED;
+}
+
+void ChunkedDataPipeUploadDataStream::ResetInternal() {
+  // Init rewinds the stream. Throw away current state, other than |size_| and
+  // |status_|.
+  buf_ = nullptr;
+  buf_len_ = 0;
+  handle_watcher_.Cancel();
+  bytes_read_ = 0;
+  data_pipe_.reset();
+}
+
+void ChunkedDataPipeUploadDataStream::OnSizeReceived(int32_t status,
+                                                     uint64_t size) {
+  DCHECK(!size_);
+  DCHECK_EQ(net::OK, status_);
+
+  status_ = status;
+  if (status == net::OK) {
+    size_ = size;
+    if (size == bytes_read_) {
+      // Only set this as a final chunk if there's a read in progress. Setting
+      // it asynchronously could result in confusing consumers.
+      if (buf_)
+        SetIsFinalChunk();
+    } else if (size < bytes_read_ || !data_pipe_.is_valid()) {
+      // If more data was received than was expected, or the data pipe was
+      // closed without passing in as many bytes as expected, the upload can't
+      // continue.
+      status_ = net::ERR_FAILED;
+    }
+  }
+
+  // If this is done, and there's a pending read, complete the pending read.
+  // If there's not a pending read, either |status_| will be reported on the
+  // next read, the file will be marked as done, so ReadInternal() won't be
+  // called again.
+  if (buf_ && (IsEOF() || status_ != net::OK)) {
+    // |buf_| being non-null means the pipe is being watched, unless it was
+    // closed. Stop watching the pipe, to avoid another OnHandleReadable()
+    // notification.
+    handle_watcher_.Cancel();
+
+    OnReadCompleted(status_);
+
+    // |this| may have been deleted at this point.
+  }
+}
+
+void ChunkedDataPipeUploadDataStream::OnHandleReadable(MojoResult result) {
+  DCHECK(buf_);
+
+  // Final result of the Read() call, to be passed to the consumer.
+  // Swap out |buf_| and |buf_len_|
+  scoped_refptr<net::IOBuffer> buf(std::move(buf_));
+  int buf_len = buf_len_;
+  buf_len_ = 0;
+
+  int rv = ReadInternal(buf.get(), buf_len);
+
+  if (rv != net::ERR_IO_PENDING)
+    OnReadCompleted(rv);
+
+  // |this| may have been deleted at this point.
+}
+
+void ChunkedDataPipeUploadDataStream::OnDataPipeGetterClosed() {
+  // If the size hasn't been received yet, treat this as receiving an error.
+  // Otherwise, this will only be a problem if/when InitInternal() tries to
+  // start reading again, so do nothing.
+  if (!size_)
+    OnSizeReceived(net::ERR_FAILED, 0);
+}
+
+}  // namespace network

services/network/chunked_data_pipe_upload_data_stream.h

+// Copyright 2018 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 SERVICES_NETWORK_CHUNKED_DATA_PIPE_UPLOAD_DATA_STREAM_H_
+#define SERVICES_NETWORK_CHUNKED_DATA_PIPE_UPLOAD_DATA_STREAM_H_
+
+#include <stdint.h>
+
+#include <memory>
+
+#include "base/component_export.h"
+#include "base/macros.h"
+#include "base/memory/ref_counted.h"
+#include "base/optional.h"
+#include "mojo/public/cpp/system/data_pipe.h"
+#include "mojo/public/cpp/system/simple_watcher.h"
+#include "net/base/completion_once_callback.h"
+#include "net/base/net_errors.h"
+#include "net/base/upload_data_stream.h"
+#include "services/network/public/cpp/resource_request_body.h"
+#include "services/network/public/mojom/chunked_data_pipe_getter.mojom.h"
+
+namespace net {
+class IOBuffer;
+}
+
+namespace network {
+
+// A subclass of net::UploadDataStream to read data pipes and provide the result
+// as a chunked upload.
+class COMPONENT_EXPORT(NETWORK_SERVICE) ChunkedDataPipeUploadDataStream
+    : public net::UploadDataStream {
+ public:
+  // |resource_request_body| is just passed in to keep the object around for the
+  // life of the UploadDataStream.
+  ChunkedDataPipeUploadDataStream(
+      scoped_refptr<ResourceRequestBody> resource_request_body,
+      mojom::ChunkedDataPipeGetterPtr chunked_data_pipe_getter);
+
+  ~ChunkedDataPipeUploadDataStream() override;
+
+ private:
+  // net::UploadDataStream implementation.
+  int InitInternal(const net::NetLogWithSource& net_log) override;
+  int ReadInternal(net::IOBuffer* buf, int buf_len) override;
+  void ResetInternal() override;
+
+  // Callback invoked by ChunkedDataPipeGetter::GetSize. Can be called any time
+  // - could be called before any of the body is read, could be called after the
+  // entire body is read, could end up not being called, in the case the other
+  // end of the pipe is torn down before the entire upload body is received.
+  void OnSizeReceived(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);
+
+  void OnDataPipeGetterClosed();
+
+  scoped_refptr<ResourceRequestBody> resource_request_body_;
+  mojom::ChunkedDataPipeGetterPtr chunked_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, or ResetInternal() is
+  // invoked.
+  scoped_refptr<net::IOBuffer> buf_;
+  int buf_len_ = 0;
+
+  // Total size of input, as passed to ReadCallback(). nullptr until size is
+  // received.
+  base::Optional<uint64_t> size_;
+
+  uint64_t bytes_read_ = 0;
+
+  // Set to a net::Error other than net::OK if the DataPipeGetter returns an
+  // error.
+  int status_ = net::OK;
+
+  DISALLOW_COPY_AND_ASSIGN(ChunkedDataPipeUploadDataStream);
+};
+
+}  // namespace network
+
+#endif  // SERVICES_NETWORK_CHUNKED_DATA_PIPE_UPLOAD_DATA_STREAM_H_

services/network/data_pipe_element_reader.cc

@@ -16,9 +16,9 @@ namespace network {
 
 DataPipeElementReader::DataPipeElementReader(
     scoped_refptr<ResourceRequestBody> resource_request_body,
-    mojom::DataPipeGetterPtr data_pipe_getter_)
+    mojom::DataPipeGetterPtr data_pipe_getter)
     : resource_request_body_(std::move(resource_request_body)),
-      data_pipe_getter_(std::move(data_pipe_getter_)),
+      data_pipe_getter_(std::move(data_pipe_getter)),
       handle_watcher_(FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::MANUAL),
       weak_factory_(this) {}
 

services/network/data_pipe_element_reader.h

@@ -32,9 +32,12 @@ class COMPONENT_EXPORT(NETWORK_SERVICE) DataPipeElementReader
  public:
   // |resource_request_body| is just passed in to keep the object around for the
   // life of the ElementReader.
+  //
+  // TODO(mmenke): This class doesn't handle the case where the DataPipeGetter
+  // pipe is closed. That should be fixed.
   DataPipeElementReader(
       scoped_refptr<ResourceRequestBody> resource_request_body,
-      mojom::DataPipeGetterPtr data_pipe_getter_);
+      mojom::DataPipeGetterPtr data_pipe_getter);
 
   ~DataPipeElementReader() override;
 

services/network/public/cpp/BUILD.gn

@@ -79,7 +79,7 @@ component("cpp_base") {
     "url_request_mojom_traits.h",
   ]
   public_deps = [
-    "//services/network/public/mojom:data_pipe_interface",
+    "//services/network/public/mojom:data_pipe_interfaces",
     "//url/ipc:url_ipc",
     "//url/mojom:url_mojom_gurl",
   ]

services/network/public/cpp/data_element.cc

@@ -65,14 +65,26 @@ void DataElement::SetToDataPipe(mojom::DataPipeGetterPtr data_pipe_getter) {
   data_pipe_getter_ = std::move(data_pipe_getter);
 }
 
+void DataElement::SetToChunkedDataPipe(
+    mojom::ChunkedDataPipeGetterPtr chunked_data_pipe_getter) {
+  type_ = TYPE_CHUNKED_DATA_PIPE;
+  chunked_data_pipe_getter_ = std::move(chunked_data_pipe_getter);
+}
+
 base::File DataElement::ReleaseFile() {
   return std::move(file_);
 }
 
 mojom::DataPipeGetterPtr DataElement::ReleaseDataPipeGetter() {
+  DCHECK_EQ(TYPE_DATA_PIPE, type_);
   return std::move(data_pipe_getter_);
 }
 
+mojom::ChunkedDataPipeGetterPtr DataElement::ReleaseChunkedDataPipeGetter() {
+  DCHECK_EQ(TYPE_CHUNKED_DATA_PIPE, type_);
+  return std::move(chunked_data_pipe_getter_);
+}
+
 void PrintTo(const DataElement& x, std::ostream* os) {
   const uint64_t kMaxDataPrintLength = 40;
   *os << "<DataElement>{type: ";
@@ -101,6 +113,9 @@ void PrintTo(const DataElement& x, std::ostream* os) {
     case DataElement::TYPE_DATA_PIPE:
       *os << "TYPE_DATA_PIPE";
       break;
+    case DataElement::TYPE_CHUNKED_DATA_PIPE:
+      *os << "TYPE_CHUNKED_DATA_PIPE";
+      break;
     case DataElement::TYPE_UNKNOWN:
       *os << "TYPE_UNKNOWN";
       break;
@@ -125,6 +140,8 @@ bool operator==(const DataElement& a, const DataElement& b) {
       return a.blob_uuid() == b.blob_uuid();
     case DataElement::TYPE_DATA_PIPE:
       return false;
+    case DataElement::TYPE_CHUNKED_DATA_PIPE:
+      return false;
     case DataElement::TYPE_UNKNOWN:
       NOTREACHED();
       return false;

services/network/public/cpp/data_element.h

@@ -21,13 +21,14 @@
 #include "base/logging.h"
 #include "base/time/time.h"
 #include "mojo/public/cpp/system/data_pipe.h"
+#include "services/network/public/mojom/chunked_data_pipe_getter.mojom.h"
 #include "services/network/public/mojom/data_pipe_getter.mojom.h"
 #include "url/gurl.h"
 
 namespace network {
 
-// Represents a base Web data element. This could be either one of
-// bytes, file or blob data.
+// Represents part of an upload body. This could be either one of bytes, file or
+// blob data.
 class COMPONENT_EXPORT(NETWORK_CPP_BASE) DataElement {
  public:
   static const uint64_t kUnknownSize = std::numeric_limits<uint64_t>::max();
@@ -37,6 +38,7 @@ class COMPONENT_EXPORT(NETWORK_CPP_BASE) DataElement {
 
     // Only used for Upload with Network Service as of now:
     TYPE_DATA_PIPE,
+    TYPE_CHUNKED_DATA_PIPE,
     TYPE_RAW_FILE,
 
     // Used for Upload when Network Service is disabled:
@@ -151,9 +153,19 @@ class COMPONENT_EXPORT(NETWORK_CPP_BASE) DataElement {
                       uint64_t offset,
                       uint64_t length);
 
-  // Sets TYPE_DATA_PIPE data.
+  // Sets TYPE_DATA_PIPE data. The data pipe consumer can safely wait for the
+  // callback passed to Read() to be invoked before reading the request body.
   void SetToDataPipe(mojom::DataPipeGetterPtr data_pipe_getter);
 
+  // Sets TYPE_CHUNKED_DATA_PIPE data. The data pipe consumer must not wait
+  // for the callback passed to GetSize() to be invoked before reading the
+  // request body, as the length may not be known until the entire body has been
+  // sent. This method triggers a chunked upload, which not all servers may
+  // support, so SetToDataPipe should be used instead, unless talking with a
+  // server known to support chunked uploads.
+  void SetToChunkedDataPipe(
+      mojom::ChunkedDataPipeGetterPtr chunked_data_pipe_getter);
+
   // Takes ownership of the File, if this is of TYPE_RAW_FILE. The file is open
   // for reading (asynchronous reading on Windows).
   base::File ReleaseFile();
@@ -161,16 +173,28 @@ class COMPONENT_EXPORT(NETWORK_CPP_BASE) DataElement {
   // Takes ownership of the DataPipeGetter, if this is of TYPE_DATA_PIPE.
   mojom::DataPipeGetterPtr ReleaseDataPipeGetter();
 
+  // Takes ownership of the DataPipeGetter, if this is of
+  // TYPE_CHUNKED_DATA_PIPE.
+  mojom::ChunkedDataPipeGetterPtr ReleaseChunkedDataPipeGetter();
+
  private:
   FRIEND_TEST_ALL_PREFIXES(BlobAsyncTransportStrategyTest, TestInvalidParams);
   friend void PrintTo(const DataElement& x, ::std::ostream* os);
   Type type_;
-  std::vector<char> buf_;  // For TYPE_BYTES.
-  const char* bytes_;      // For TYPE_BYTES.
-  base::FilePath path_;    // For TYPE_FILE and TYPE_RAW_FILE.
-  base::File file_;        // For TYPE_RAW_FILE.
-  std::string blob_uuid_;  // For TYPE_BLOB.
-  mojom::DataPipeGetterPtr data_pipe_getter_;  // For TYPE_DATA_PIPE.
+  // For TYPE_BYTES.
+  std::vector<char> buf_;
+  // For TYPE_BYTES.
+  const char* bytes_;
+  // For TYPE_FILE and TYPE_RAW_FILE.
+  base::FilePath path_;
+  // For TYPE_RAW_FILE.
+  base::File file_;
+  // For TYPE_BLOB.
+  std::string blob_uuid_;
+  // For TYPE_DATA_PIPE.
+  mojom::DataPipeGetterPtr data_pipe_getter_;
+  // For TYPE_CHUNKED_DATA_PIPE.
+  mojom::ChunkedDataPipeGetterPtr chunked_data_pipe_getter_;
   uint64_t offset_;
   uint64_t length_;
   base::Time expected_modification_time_;

services/network/public/cpp/network_param_ipc_traits.cc

@@ -542,11 +542,19 @@ void ParamTraits<network::DataElement>::Write(base::Pickle* m,
       break;
     }
     case network::DataElement::TYPE_DATA_PIPE: {
-      WriteParam(m,
-                 const_cast<network::mojom::DataPipeGetterPtr&>(p.data_pipe())
-                     .PassInterface()
-                     .PassHandle()
-                     .release());
+      WriteParam(m, const_cast<network::DataElement&>(p)
+                        .ReleaseDataPipeGetter()
+                        .PassInterface()
+                        .PassHandle()
+                        .release());
+      break;
+    }
+    case network::DataElement::TYPE_CHUNKED_DATA_PIPE: {
+      WriteParam(m, const_cast<network::DataElement&>(p)
+                        .ReleaseChunkedDataPipeGetter()
+                        .PassInterface()
+                        .PassHandle()
+                        .release());
       break;
     }
     case network::DataElement::TYPE_UNKNOWN: {
@@ -630,6 +638,17 @@ bool ParamTraits<network::DataElement>::Read(const base::Pickle* m,
       r->SetToDataPipe(std::move(data_pipe_getter));
       return true;
     }
+    case network::DataElement::TYPE_CHUNKED_DATA_PIPE: {
+      network::mojom::ChunkedDataPipeGetterPtr chunked_data_pipe_getter;
+      mojo::MessagePipeHandle message_pipe;
+      if (!ReadParam(m, iter, &message_pipe))
+        return false;
+      chunked_data_pipe_getter.Bind(
+          network::mojom::ChunkedDataPipeGetterPtrInfo(
+              mojo::ScopedMessagePipeHandle(message_pipe), 0u));
+      r->SetToChunkedDataPipe(std::move(chunked_data_pipe_getter));
+      return true;
+    }
     case network::DataElement::TYPE_UNKNOWN: {
       NOTREACHED();
       return false;
@@ -666,6 +685,13 @@ bool ParamTraits<scoped_refptr<network::ResourceRequestBody>>::Read(
   std::vector<network::DataElement> elements;
   if (!ReadParam(m, iter, &elements))
     return false;
+  // A chunked element is only allowed if it's the only one element.
+  if (elements.size() > 1) {
+    for (const auto& element : elements) {
+      if (element.type() == network::DataElement::TYPE_CHUNKED_DATA_PIPE)
+        return false;
+    }
+  }
   int64_t identifier;
   if (!ReadParam(m, iter, &identifier))
     return false;

services/network/public/cpp/resource_request_body.cc

@@ -19,6 +19,9 @@ scoped_refptr<ResourceRequestBody> ResourceRequestBody::CreateFromBytes(
 }
 
 void ResourceRequestBody::AppendBytes(const char* bytes, int bytes_len) {
+  DCHECK(elements_.empty() ||
+         elements_.front().type() != DataElement::TYPE_CHUNKED_DATA_PIPE);
+
   if (bytes_len > 0) {
     elements_.push_back(DataElement());
     elements_.back().SetToBytes(bytes, bytes_len);
@@ -30,6 +33,9 @@ void ResourceRequestBody::AppendFileRange(
     uint64_t offset,
     uint64_t length,
     const base::Time& expected_modification_time) {
+  DCHECK(elements_.empty() ||
+         elements_.front().type() != DataElement::TYPE_CHUNKED_DATA_PIPE);
+
   elements_.push_back(DataElement());
   elements_.back().SetToFilePathRange(file_path, offset, length,
                                       expected_modification_time);
@@ -41,22 +47,39 @@ void ResourceRequestBody::AppendRawFileRange(
     uint64_t offset,
     uint64_t length,
     const base::Time& expected_modification_time) {
+  DCHECK(elements_.empty() ||
+         elements_.front().type() != DataElement::TYPE_CHUNKED_DATA_PIPE);
+
   elements_.push_back(DataElement());
   elements_.back().SetToFileRange(std::move(file), file_path, offset, length,
                                   expected_modification_time);
 }
 
 void ResourceRequestBody::AppendBlob(const std::string& uuid) {
+  DCHECK(elements_.empty() ||
+         elements_.front().type() != DataElement::TYPE_CHUNKED_DATA_PIPE);
+
   elements_.push_back(DataElement());
   elements_.back().SetToBlob(uuid);
 }
 
 void ResourceRequestBody::AppendDataPipe(
     mojom::DataPipeGetterPtr data_pipe_getter) {
+  DCHECK(elements_.empty() ||
+         elements_.front().type() != DataElement::TYPE_CHUNKED_DATA_PIPE);
+
   elements_.push_back(DataElement());
   elements_.back().SetToDataPipe(std::move(data_pipe_getter));
 }
 
+void ResourceRequestBody::SetToChunkedDataPipe(
+    mojom::ChunkedDataPipeGetterPtr chunked_data_pipe_getter) {
+  DCHECK(elements_.empty());
+
+  elements_.push_back(DataElement());
+  elements_.back().SetToChunkedDataPipe(std::move(chunked_data_pipe_getter));
+}
+
 std::vector<base::FilePath> ResourceRequestBody::GetReferencedFiles() const {
   std::vector<base::FilePath> result;
   for (const auto& element : *elements()) {

services/network/public/cpp/resource_request_body.h

@@ -46,6 +46,19 @@ class COMPONENT_EXPORT(NETWORK_CPP_BASE) ResourceRequestBody
   void AppendBlob(const std::string& uuid);
   void AppendDataPipe(mojom::DataPipeGetterPtr data_pipe_getter);
 
+  // |chunked_data_pipe_getter| will provide the upload body for a chunked
+  // upload. Unlike the other methods, which support concatenating data of
+  // various types, when this is called, |chunked_data_pipe_getter| will provide
+  // the entire response body, and other types of data may not added when
+  // sending chunked data.
+  //
+  // It's unclear how widespread support for chunked uploads is, since there are
+  // no web APIs that send uploads with unknown request body sizes, so this
+  // method should only be used when talking to servers that are are known to
+  // support chunked uploads.
+  void SetToChunkedDataPipe(
+      mojom::ChunkedDataPipeGetterPtr chunked_data_pipe_getter);
+
   const std::vector<DataElement>* elements() const { return &elements_; }
   std::vector<DataElement>* elements_mutable() { return &elements_; }
   void swap_elements(std::vector<DataElement>* elements) {

services/network/public/mojom/BUILD.gn

@@ -4,11 +4,12 @@
 
 import("//mojo/public/tools/bindings/mojom.gni")
 
-# This interface is put in its own target to avoid a circular dependency, which
-# comes from the fact that the typemap for url_loader.mojom
-# (ResourceRequestBody) uses the interface from data_pipe_getter.mojom.
-mojom("data_pipe_interface") {
+# These interfaces are put in their own target to avoid a circular dependency,
+# which comes from the fact that the typemap for url_loader.mojom
+# (ResourceRequestBody) uses these interfaces.
+mojom("data_pipe_interfaces") {
   sources = [
+    "chunked_data_pipe_getter.mojom",
     "data_pipe_getter.mojom",
   ]
 
@@ -73,7 +74,7 @@ mojom("mojom") {
   ]
 
   public_deps = [
-    ":data_pipe_interface",
+    ":data_pipe_interfaces",
     ":mutable_network_traffic_annotation_interface",
     ":udp_socket_interface",
     "//mojo/common:common_custom_types",

services/network/public/mojom/chunked_data_pipe_getter.mojom

+// Copyright 2018 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.
+
+module network.mojom;
+
+// Interface that wraps a data stream and allows it to be read via a data pipe.
+// Note that unlike DataPipeGetter, the size may not be known in advance.
+//
+// This is only used for chunked uploads, which send a final empty chunk to
+// indicate the end of the response, instead of using a Content-Length header.
+// For all web-exposed APIs, the upload body size is known in advance, so it's
+// unclear how many servers support chunked uploads. As a result, it's
+// recommended this class only be used with servers known to support chunked
+// uploads.
+//
+// Knowing the size in advance also allows servers to know if the upload is
+// bigger than they're willing to accept, and cancel the request early, which is
+// a more friendly behavior.
+interface ChunkedDataPipeGetter {
+  // Requests the total size of the data that will be provided by the data pipe.
+  // The Getter may invoke the callback any time, before, during, or after
+  // providing data through the pipe. On error, |status| should be populated
+  // with a net::Error value.
+  //
+  // Will be called only once, before StartReading() is invoked.
+  GetSize() => (int32 status, uint64 size);
+
+  // Starts reading from the beginning of the wrapped stream and writing it into
+  // the producer handle |pipe|. Will only be called after GetSize(), but unlike
+  // GetSize(), may be invoked multiple times. Data should only be written to the
+  // |pipe| that StartReading() was most recently called with; any previously
+  // passed pipes can be discarded and no longer need to be be written to.
+  StartReading(handle<data_pipe_producer> pipe);
+};

services/network/public/mojom/data_pipe_getter.mojom

@@ -8,9 +8,10 @@ module network.mojom;
 // as backed by content like a Blob that can be read multiple times.
 interface DataPipeGetter {
   // Reads all the content, writing into |pipe|. Calls the callback with |size|
-  // once the size of the content is known. If an error occurred before the
-  // size was known, the callback is instead called with the net::Error
-  // |status|.
+  // once the size of the content is known.  The callback must be invoked before
+  // writing to |pipe|, since the size needs to be known to generate the
+  // Content-Length request header. If an error occurred before the size was
+  // known, the callback is instead called with the net::Error |status|.
   Read(handle<data_pipe_producer> pipe) => (int32 status, uint64 size);
 
   // Makes a clone so there can be multiple handles to this interface. This

services/network/test_chunked_data_pipe_getter.cc

+// Copyright 2018 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 "services/network/test_chunked_data_pipe_getter.h"
+
+#include "base/logging.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace network {
+
+TestChunkedDataPipeGetter::TestChunkedDataPipeGetter() : binding_(this) {}
+
+TestChunkedDataPipeGetter::~TestChunkedDataPipeGetter() {}
+
+mojom::ChunkedDataPipeGetterPtr
+TestChunkedDataPipeGetter::GetDataPipeGetterPtr() {
+  EXPECT_FALSE(binding_.is_bound());
+
+  mojom::ChunkedDataPipeGetterPtr data_pipe_getter_ptr;
+  binding_.Bind(mojo::MakeRequest(&data_pipe_getter_ptr));
+  return data_pipe_getter_ptr;
+}
+
+void TestChunkedDataPipeGetter::ClosePipe() {
+  binding_.Close();
+}
+
+mojom::ChunkedDataPipeGetter::GetSizeCallback
+TestChunkedDataPipeGetter::WaitForGetSize() {
+  DCHECK(!get_size_run_loop_);
+
+  if (!get_size_callback_) {
+    get_size_run_loop_ = std::make_unique<base::RunLoop>();
+    get_size_run_loop_->Run();
+    get_size_run_loop_.reset();
+  }
+
+  EXPECT_TRUE(get_size_callback_);
+
+  return std::move(get_size_callback_);
+}
+
+mojo::ScopedDataPipeProducerHandle
+TestChunkedDataPipeGetter::WaitForStartReading() {
+  DCHECK(!start_reading_run_loop_);
+
+  if (!write_pipe_.is_valid()) {
+    start_reading_run_loop_ = std::make_unique<base::RunLoop>();
+    start_reading_run_loop_->Run();
+    start_reading_run_loop_.reset();
+  }
+
+  EXPECT_TRUE(write_pipe_.is_valid());
+
+  return std::move(write_pipe_);
+}
+
+void TestChunkedDataPipeGetter::GetSize(GetSizeCallback get_size_callback) {
+  EXPECT_FALSE(received_size_callback_);
+  EXPECT_FALSE(get_size_callback_);
+
+  received_size_callback_ = true;
+  get_size_callback_ = std::move(get_size_callback);
+
+  if (get_size_run_loop_)
+    get_size_run_loop_->Quit();
+}
+
+void TestChunkedDataPipeGetter::StartReading(
+    mojo::ScopedDataPipeProducerHandle pipe) {
+  EXPECT_FALSE(write_pipe_.is_valid());
+  EXPECT_TRUE(received_size_callback_);
+
+  write_pipe_ = std::move(pipe);
+
+  if (start_reading_run_loop_)
+    start_reading_run_loop_->Quit();
+}
+
+}  // namespace network

services/network/test_chunked_data_pipe_getter.h

+// Copyright 2018 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 SERVICES_NETWORK_TEST_CHUNKED_DATA_PIPE_GETTER_H_
+#define SERVICES_NETWORK_TEST_CHUNKED_DATA_PIPE_GETTER_H_
+
+#include <memory>
+
+#include "base/macros.h"
+#include "base/run_loop.h"
+#include "mojo/public/cpp/bindings/binding.h"
+#include "mojo/public/cpp/system/data_pipe.h"
+#include "services/network/public/mojom/chunked_data_pipe_getter.mojom.h"
+
+namespace network {
+
+// Test implementation of mojom::DataPipeGetter that lets tests wait for
+// the mojo::ScopedDataPipeProducerHandle and ReadCallback to be received
+// and then manage them both directly.
+class TestChunkedDataPipeGetter : public mojom::ChunkedDataPipeGetter {
+ public:
+  TestChunkedDataPipeGetter();
+  ~TestChunkedDataPipeGetter() override;
+
+  // Returns the mojo::ChunkedDataPipeGetterPtr corresponding to |this|. May
+  // only be called once.
+  mojom::ChunkedDataPipeGetterPtr GetDataPipeGetterPtr();
+
+  // Close the mojom::DataPipeGetter pipe.
+  void ClosePipe();
+
+  GetSizeCallback WaitForGetSize();
+  mojo::ScopedDataPipeProducerHandle WaitForStartReading();
+
+ private:
+  // mojom::ChunkedDataPipeGetter implementation:
+  void GetSize(GetSizeCallback get_size_callback) override;
+  void StartReading(mojo::ScopedDataPipeProducerHandle pipe) override;
+
+  std::unique_ptr<base::RunLoop> get_size_run_loop_;
+  std::unique_ptr<base::RunLoop> start_reading_run_loop_;
+
+  mojo::Binding<mojom::ChunkedDataPipeGetter> binding_;
+  mojo::ScopedDataPipeProducerHandle write_pipe_;
+  GetSizeCallback get_size_callback_;
+  bool received_size_callback_ = false;
+
+  DISALLOW_COPY_AND_ASSIGN(TestChunkedDataPipeGetter);
+};
+
+}  // namespace network
+
+#endif  // SERVICES_NETWORK_TEST_CHUNKED_DATA_PIPE_GETTER_H_

services/network/url_loader.cc

@@ -22,6 +22,7 @@
 #include "net/ssl/ssl_private_key.h"
 #include "net/url_request/url_request_context.h"
 #include "net/url_request/url_request_context_getter.h"
+#include "services/network/chunked_data_pipe_upload_data_stream.h"
 #include "services/network/data_pipe_element_reader.h"
 #include "services/network/loader_util.h"
 #include "services/network/public/cpp/features.h"
@@ -150,6 +151,15 @@ class RawFileElementReader : public net::UploadFileElementReader {
 std::unique_ptr<net::UploadDataStream> CreateUploadDataStream(
     ResourceRequestBody* body,
     base::SequencedTaskRunner* file_task_runner) {
+  // In the case of a chunked upload, there will just be one element.
+  if (body->elements()->size() == 1 &&
+      body->elements()->begin()->type() ==
+          DataElement::TYPE_CHUNKED_DATA_PIPE) {
+    return std::make_unique<network::ChunkedDataPipeUploadDataStream>(
+        body, const_cast<DataElement&>(body->elements()->front())
+                  .ReleaseChunkedDataPipeGetter());
+  }
+
   std::vector<std::unique_ptr<net::UploadElementReader>> element_readers;
   for (const auto& element : *body->elements()) {
     switch (element.type()) {
@@ -174,6 +184,12 @@ std::unique_ptr<net::UploadDataStream> CreateUploadDataStream(
             body, const_cast<DataElement*>(&element)->ReleaseDataPipeGetter()));
         break;
       }
+      case DataElement::TYPE_CHUNKED_DATA_PIPE: {
+        // This shouldn't happen, as the traits logic should ensure that if
+        // there's a chunked pipe, there's one and only one element.
+        NOTREACHED();
+        break;
+      }
       case DataElement::TYPE_UNKNOWN:
         NOTREACHED();
         break;

services/network/url_loader_unittest.cc

@@ -48,6 +48,7 @@
 #include "services/network/resource_scheduler_client.h"
 #include "services/network/test/test_data_pipe_getter.h"
 #include "services/network/test/test_url_loader_client.h"
+#include "services/network/test_chunked_data_pipe_getter.h"
 #include "services/network/url_loader.h"
 #include "services/network/url_request_context_owner.h"
 #include "testing/gtest/include/gtest/gtest.h"
@@ -1248,6 +1249,38 @@ TEST_F(URLLoaderTest, UploadDoubleRawFile) {
   EXPECT_EQ(expected_body + expected_body, response_body);
 }
 
+// Tests a request body with a chunked data pipe element.
+TEST_F(URLLoaderTest, UploadChunkedDataPipe) {
+  const std::string kRequestBody = "Request Body";
+
+  TestChunkedDataPipeGetter data_pipe_getter;
+
+  ResourceRequest request =
+      CreateResourceRequest("POST", test_server()->GetURL("/echo"));
+  request.request_body = base::MakeRefCounted<ResourceRequestBody>();
+  request.request_body->SetToChunkedDataPipe(
+      data_pipe_getter.GetDataPipeGetterPtr());
+
+  mojom::URLLoaderPtr loader;
+  // The loader is implicitly owned by the client and the NetworkContext.
+  new URLLoader(context(), nullptr /* network_service_client */,
+                mojo::MakeRequest(&loader), 0, request,
+                false /* report_raw_headers */, client()->CreateInterfacePtr(),
+                TRAFFIC_ANNOTATION_FOR_TESTS, 0 /* process_id */,
+                nullptr /* resource_scheduler_client */,
+                nullptr /* keepalive_statistics_reporter */);
+
+  mojom::ChunkedDataPipeGetter::GetSizeCallback get_size_callback =
+      data_pipe_getter.WaitForGetSize();
+  mojo::common::BlockingCopyFromString(kRequestBody,
+                                       data_pipe_getter.WaitForStartReading());
+  std::move(get_size_callback).Run(net::OK, kRequestBody.size());
+  client()->RunUntilComplete();
+
+  EXPECT_EQ(kRequestBody, ReadBody());
+  EXPECT_EQ(net::OK, client()->completion_status().error_code);
+}
+
 // Tests that SSLInfo is not attached to OnComplete messages when there is no
 // certificate error.
 TEST_F(URLLoaderTest, NoSSLInfoWithoutCertificateError) {

storage/browser/blob/blob_data_builder.cc

@@ -125,6 +125,7 @@ void BlobDataBuilder::AppendIPCDataElement(
     case network::DataElement::TYPE_UNKNOWN:
     // This type can't be sent by IPC.
     case network::DataElement::TYPE_DATA_PIPE:
+    case network::DataElement::TYPE_CHUNKED_DATA_PIPE:
       NOTREACHED();
       break;
   }