[fetch] Cache request stream body up to 64k bytes.

This CL introduces caching in ChunkedDataPipeUploadDataStream (CDPUDS
here in after) so that fetch upload streams manages inevitable replay
required situations like socket time out.

If the caching is enabled,
CDPUDS caches streams content from the data pipe while uploading.
1 If we hit some error requiring CDPUDS restart, InitInternal() doesn't
reset the data pipe and CDPUDS replays the cached content and continues
data pipe withdrawing after that.
2 If we hit the error when the cache is over the max size, we will fail
the request because we cannot replay the stream any longer.
3 If no error happens, the cache will never be used.

We don't clear the cache because it is small and will be discarded after
the transaction.

CDPUDS caches chunks from the date pipe until whole size gets over the
max size. For example, if the date pipe sends [60k, 60k, 60k] chunks,
CDPUDS caches the first 2 60ks. If it is [120k, 1k,], CDPUDS caches the
120k chunk or any size if it is the first chunk.

Since this change is neutral to the HTTP protocol, this works over
H1/SPDY and QUIC.

Bug: 1077174
Change-Id: Ib87b9d7ff97577f25e7355aa4b0764e03157d7bd
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2269622Reviewed-by: Yutaka Hirano <yhirano@chromium.org>
Reviewed-by: Matt Menke <mmenke@chromium.org>
Commit-Queue: Yoichi Osato <yoichio@chromium.org>
Cr-Commit-Position: refs/heads/master@{#801213}

services/network/chunked_data_pipe_upload_data_stream.cc

@@ -48,6 +48,18 @@ int ChunkedDataPipeUploadDataStream::InitInternal(
   if (!chunked_data_pipe_getter_.is_connected())
     return net::ERR_FAILED;
 
+  switch (cache_state_) {
+    case CacheState::kActive:
+      if (data_pipe_.is_valid())
+        return net::OK;
+      else
+        break;
+    case CacheState::kExhausted:
+      return net::ERR_FAILED;
+    case CacheState::kDisabled:
+      break;
+  }
+
   // Get a new data pipe and start.
   mojo::ScopedDataPipeProducerHandle data_pipe_producer;
   mojo::ScopedDataPipeConsumerHandle data_pipe_consumer;
@@ -81,6 +93,10 @@ int ChunkedDataPipeUploadDataStream::ReadInternal(net::IOBuffer* buf,
     return net::OK;
   }
 
+  int cache_read_len = ReadFromCacheIfNeeded(buf, buf_len);
+  if (cache_read_len > 0)
+    return cache_read_len;
+
   // Only start watching once a read starts. This is because OnHandleReadable()
   // uses |buf_| implicitly assuming that this method has already been called.
   if (!handle_watcher_.IsWatching()) {
@@ -103,6 +119,7 @@ int ChunkedDataPipeUploadDataStream::ReadInternal(net::IOBuffer* buf,
     // allow it.
     if (size_ && *size_ == bytes_read_)
       SetIsFinalChunk();
+    WriteToCacheIfNeeded(buf, num_bytes);
     return num_bytes;
   }
 
@@ -134,12 +151,14 @@ int ChunkedDataPipeUploadDataStream::ReadInternal(net::IOBuffer* buf,
 }
 
 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;
+  if (cache_state_ != CacheState::kDisabled)
+    return;
+  // Init rewinds the stream. Throw away current state, other than |size_| and
+  // |status_|.
   data_pipe_.reset();
 }
 
@@ -211,4 +230,49 @@ void ChunkedDataPipeUploadDataStream::OnDataPipeGetterClosed() {
     OnSizeReceived(net::ERR_FAILED, 0);
 }
 
+void ChunkedDataPipeUploadDataStream::EnableCache(size_t dst_window_size) {
+  DCHECK_EQ(bytes_read_, 0u);
+  DCHECK_EQ(cache_state_, CacheState::kDisabled);
+  DCHECK(cache_.empty());
+  cache_state_ = CacheState::kActive;
+  dst_window_size_ = dst_window_size;
+}
+
+void ChunkedDataPipeUploadDataStream::WriteToCacheIfNeeded(net::IOBuffer* buf,
+                                                           uint32_t num_bytes) {
+  if (cache_state_ != CacheState::kActive)
+    return;
+
+  // |cache_state_ == CacheState::kActive| and |cache_.size() >= bytes_read_|
+  // means we're reading from the cache.
+  if (cache_.size() >= bytes_read_)
+    return;
+
+  if (cache_.size() >= dst_window_size_) {
+    // Attempted to write over the max size. Replay must be failed.
+    // Notes: CDPUDS caches chunks from the date pipe until whole size gets over
+    // the max size. For example, if the date pipe sends [60k, 60k, 60k] chunks,
+    // CDPUDS caches the first 2 60ks. If it is [120k, 1k,], CDPUDS caches the
+    // 120k chunk or any size if it is the first chunk.
+    cache_state_ = CacheState::kExhausted;
+    return;
+  }
+  cache_.insert(cache_.end(), buf->data(), buf->data() + num_bytes);
+}
+
+int ChunkedDataPipeUploadDataStream::ReadFromCacheIfNeeded(net::IOBuffer* buf,
+                                                           int buf_len) {
+  if (cache_state_ != CacheState::kActive)
+    return 0;
+  if (cache_.size() <= bytes_read_)
+    return 0;
+
+  int read_size =
+      std::min(static_cast<int>(cache_.size() - bytes_read_), buf_len);
+  DCHECK_GT(read_size, 0);
+  memcpy(buf->data(), &cache_[bytes_read_], read_size);
+  bytes_read_ += read_size;
+  return read_size;
+}
+
 }  // namespace network

services/network/chunked_data_pipe_upload_data_stream.h

@@ -8,6 +8,7 @@
 #include <stdint.h>
 
 #include <memory>
+#include <vector>
 
 #include "base/component_export.h"
 #include "base/macros.h"
@@ -45,7 +46,21 @@ class COMPONENT_EXPORT(NETWORK_SERVICE) ChunkedDataPipeUploadDataStream
 
   bool AllowHTTP1() const override;
 
+  static const size_t kDefaultDestinationWindowSize = 65535;
+  // This has ChunkedDataPipeUploadDataStream cache each chunk from the datapipe
+  // up to the size that just exceeds |dst_window_size| so that the destination
+  // resends the data.
+  // InitInternal() doesn't reset the data pipe and this instance replays the
+  // all cached chunks and continues datapipe withdrawing after that.
+  void EnableCache(size_t dst_window_size = kDefaultDestinationWindowSize);
+
  private:
+  enum class CacheState {
+    kDisabled,
+    kActive,
+    kExhausted,
+  };
+
   // net::UploadDataStream implementation.
   int InitInternal(const net::NetLogWithSource& net_log) override;
   int ReadInternal(net::IOBuffer* buf, int buf_len) override;
@@ -63,6 +78,9 @@ class COMPONENT_EXPORT(NETWORK_SERVICE) ChunkedDataPipeUploadDataStream
 
   void OnDataPipeGetterClosed();
 
+  void WriteToCacheIfNeeded(net::IOBuffer* buf, uint32_t num_bytes);
+  int ReadFromCacheIfNeeded(net::IOBuffer* buf, int buf_len);
+
   scoped_refptr<ResourceRequestBody> resource_request_body_;
   mojo::Remote<mojom::ChunkedDataPipeGetter> chunked_data_pipe_getter_;
   mojo::ScopedDataPipeConsumerHandle data_pipe_;
@@ -85,6 +103,11 @@ class COMPONENT_EXPORT(NETWORK_SERVICE) ChunkedDataPipeUploadDataStream
   // error.
   int status_ = net::OK;
 
+  // Variables used for EnableCache().
+  CacheState cache_state_ = CacheState::kDisabled;
+  size_t dst_window_size_ = kDefaultDestinationWindowSize;
+  std::vector<char> cache_;
+
   DISALLOW_COPY_AND_ASSIGN(ChunkedDataPipeUploadDataStream);
 };
 

services/network/url_loader.cc

@@ -216,9 +216,14 @@ std::unique_ptr<net::UploadDataStream> CreateUploadDataStream(
     network::mojom::DataElementType type = body->elements()->begin()->type();
     if (type == network::mojom::DataElementType::kChunkedDataPipe ||
         type == network::mojom::DataElementType::kReadOnceStream) {
-      return std::make_unique<ChunkedDataPipeUploadDataStream>(
-          body,
-          body->elements_mutable()->begin()->ReleaseChunkedDataPipeGetter());
+      auto upload_data_stream =
+          std::make_unique<ChunkedDataPipeUploadDataStream>(
+              body, body->elements_mutable()
+                        ->begin()
+                        ->ReleaseChunkedDataPipeGetter());
+      if (type == network::mojom::DataElementType::kReadOnceStream)
+        upload_data_stream->EnableCache();
+      return upload_data_stream;
     }
   }