Introduce ResponseBodyLoader

ResponseBodyLoader is needed to provide a way to drain data pipe from
the resource loader, with the following restrictions:
 - We are going to provide a hook point on blink::Resource. Not every
   Resource loading body is associated with a ResourceLoader (e.g.,
   when MHTML is involved), so we need an abstraction layer.
 - Passing a broad interface is misleading and dangerous, so we want to
   pass an object with only "drain" function(s).

Bug: 894819
Change-Id: Ia3e849d1e100e2e6936fa88a5724a7adc3b07517
Reviewed-on: https://chromium-review.googlesource.com/c/1433662
Commit-Queue: Yutaka Hirano <yhirano@chromium.org>
Reviewed-by: Leszek Swirski <leszeks@chromium.org>
Reviewed-by: Makoto Shimazu <shimazu@chromium.org>
Cr-Commit-Position: refs/heads/master@{#626991}

third_party/blink/renderer/platform/loader/BUILD.gn

@@ -83,6 +83,9 @@ blink_platform_sources("loader") {
     "fetch/resource_status.h",
     "fetch/resource_timing_info.cc",
     "fetch/resource_timing_info.h",
+    "fetch/response_body_loader.cc",
+    "fetch/response_body_loader.h",
+    "fetch/response_body_loader_client.h",
     "fetch/script_cached_metadata_handler.cc",
     "fetch/script_cached_metadata_handler.h",
     "fetch/script_fetch_options.cc",
@@ -140,6 +143,7 @@ jumbo_source_set("unit_tests") {
     "fetch/resource_request_test.cc",
     "fetch/resource_response_test.cc",
     "fetch/resource_test.cc",
+    "fetch/response_body_loader_test.cc",
     "fetch/source_keyed_cached_metadata_handler_test.cc",
     "ftp_directory_listing_test.cc",
     "link_header_test.cc",

third_party/blink/renderer/platform/loader/fetch/response_body_loader.cc

+// Copyright 2019 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 "third_party/blink/renderer/platform/loader/fetch/response_body_loader.h"
+
+#include <algorithm>
+#include <utility>
+#include "third_party/blink/renderer/platform/loader/fetch/fetch_context.h"
+#include "third_party/blink/renderer/platform/loader/fetch/resource.h"
+#include "third_party/blink/renderer/platform/loader/fetch/resource_loader.h"
+#include "third_party/blink/renderer/platform/shared_buffer.h"
+
+namespace blink {
+
+constexpr size_t ResponseBodyLoader::kMaxNumConsumedBytesInTask;
+
+ResponseBodyLoader::ResponseBodyLoader(
+    BytesConsumer& bytes_consumer,
+    ResponseBodyLoaderClient& client,
+    scoped_refptr<base::SingleThreadTaskRunner> task_runner)
+    : bytes_consumer_(bytes_consumer),
+      client_(client),
+      task_runner_(std::move(task_runner)) {
+  bytes_consumer_->SetClient(this);
+}
+
+mojo::ScopedDataPipeConsumerHandle ResponseBodyLoader::DrainAsDataPipe(
+    ResponseBodyLoaderClient** client) {
+  DCHECK(!started_);
+
+  *client = nullptr;
+  if (drained_ || aborted_) {
+    return {};
+  }
+
+  DCHECK(bytes_consumer_);
+  auto data_pipe = bytes_consumer_->DrainAsDataPipe();
+
+  if (data_pipe) {
+    drained_ = true;
+    bytes_consumer_ = nullptr;
+    *client = this;
+  }
+
+  return data_pipe;
+}
+
+void ResponseBodyLoader::DidReceiveData(base::span<const char> data) {
+  if (aborted_)
+    return;
+
+  client_->DidReceiveData(data);
+}
+
+void ResponseBodyLoader::DidFinishLoadingBody() {
+  if (aborted_)
+    return;
+
+  if (suspended_) {
+    finish_signal_is_pending_ = true;
+    return;
+  }
+
+  finish_signal_is_pending_ = false;
+  client_->DidFinishLoadingBody();
+}
+
+void ResponseBodyLoader::DidFailLoadingBody() {
+  if (aborted_)
+    return;
+
+  if (suspended_) {
+    fail_signal_is_pending_ = true;
+    return;
+  }
+
+  fail_signal_is_pending_ = false;
+  client_->DidFailLoadingBody();
+}
+
+void ResponseBodyLoader::Start() {
+  DCHECK(!started_);
+  DCHECK(!drained_);
+
+  started_ = true;
+  OnStateChange();
+}
+
+void ResponseBodyLoader::Abort() {
+  if (aborted_)
+    return;
+
+  aborted_ = true;
+
+  if (bytes_consumer_ && !in_two_phase_read_)
+    bytes_consumer_->Cancel();
+}
+
+void ResponseBodyLoader::Suspend() {
+  if (aborted_)
+    return;
+
+  DCHECK(!suspended_);
+  suspended_ = true;
+}
+
+void ResponseBodyLoader::Resume() {
+  if (aborted_)
+    return;
+
+  DCHECK(suspended_);
+  suspended_ = false;
+
+  if (finish_signal_is_pending_) {
+    task_runner_->PostTask(
+        FROM_HERE, base::BindOnce(&ResponseBodyLoader::DidFinishLoadingBody,
+                                  WrapPersistent(this)));
+  } else if (fail_signal_is_pending_) {
+    task_runner_->PostTask(
+        FROM_HERE, base::BindOnce(&ResponseBodyLoader::DidFailLoadingBody,
+                                  WrapPersistent(this)));
+  } else {
+    task_runner_->PostTask(FROM_HERE,
+                           base::BindOnce(&ResponseBodyLoader::OnStateChange,
+                                          WrapPersistent(this)));
+  }
+}
+
+void ResponseBodyLoader::OnStateChange() {
+  if (!started_)
+    return;
+
+  size_t num_bytes_consumed = 0;
+
+  while (!aborted_ && !suspended_) {
+    if (kMaxNumConsumedBytesInTask == num_bytes_consumed) {
+      // We've already consumed many bytes in this task. Defer the remaining
+      // to the next task.
+      task_runner_->PostTask(FROM_HERE,
+                             base::BindOnce(&ResponseBodyLoader::OnStateChange,
+                                            WrapPersistent(this)));
+      return;
+    }
+
+    const char* buffer = nullptr;
+    size_t available = 0;
+    auto result = bytes_consumer_->BeginRead(&buffer, &available);
+    if (result == BytesConsumer::Result::kShouldWait)
+      return;
+    if (result == BytesConsumer::Result::kOk) {
+      in_two_phase_read_ = true;
+
+      available =
+          std::min(available, kMaxNumConsumedBytesInTask - num_bytes_consumed);
+      DidReceiveData(base::make_span(buffer, available));
+      result = bytes_consumer_->EndRead(available);
+      in_two_phase_read_ = false;
+      num_bytes_consumed += available;
+
+      if (aborted_) {
+        // As we cannot call Cancel in two-phase read, we need to call it here.
+        bytes_consumer_->Cancel();
+      }
+    }
+    DCHECK_NE(result, BytesConsumer::Result::kShouldWait);
+    if (result == BytesConsumer::Result::kDone) {
+      DidFinishLoadingBody();
+      return;
+    }
+    if (result != BytesConsumer::Result::kOk) {
+      DidFailLoadingBody();
+      Abort();
+      return;
+    }
+  }
+}
+
+void ResponseBodyLoader::Trace(Visitor* visitor) {
+  visitor->Trace(bytes_consumer_);
+  visitor->Trace(client_);
+  ResponseBodyLoaderDrainableInterface::Trace(visitor);
+  ResponseBodyLoaderClient::Trace(visitor);
+  BytesConsumer::Client::Trace(visitor);
+}
+
+}  // namespace blink

third_party/blink/renderer/platform/loader/fetch/response_body_loader.h

+// Copyright 2019 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 THIRD_PARTY_BLINK_RENDERER_PLATFORM_LOADER_FETCH_RESPONSE_BODY_LOADER_H_
+#define THIRD_PARTY_BLINK_RENDERER_PLATFORM_LOADER_FETCH_RESPONSE_BODY_LOADER_H_
+
+#include "base/containers/span.h"
+#include "base/memory/scoped_refptr.h"
+#include "mojo/public/cpp/system/data_pipe.h"
+#include "third_party/blink/renderer/platform/heap/garbage_collected.h"
+#include "third_party/blink/renderer/platform/heap/member.h"
+#include "third_party/blink/renderer/platform/loader/fetch/bytes_consumer.h"
+#include "third_party/blink/renderer/platform/loader/fetch/response_body_loader_client.h"
+#include "third_party/blink/renderer/platform/platform_export.h"
+#include "third_party/blink/renderer/platform/wtf/text/wtf_string.h"
+
+namespace base {
+class SingleThreadTaskRunner;
+}  // namespace base
+
+namespace blink {
+
+class ResponseBodyLoader;
+
+// See ResponseBodyLoader for details. This is a virtual interface to expose
+// only DrainAsDataPipe function.
+class PLATFORM_EXPORT ResponseBodyLoaderDrainableInterface
+    : public GarbageCollectedFinalized<ResponseBodyLoaderDrainableInterface> {
+ public:
+  virtual ~ResponseBodyLoaderDrainableInterface() = default;
+
+  // Drains the response body and returns it. This function must be called
+  // before calling Start(). This function may return an invalid handle when
+  // it is unable to convert the body to a data pipe, even when the body itself
+  // is valid. In that case, this function is no-op.
+  // If this function returns a valid handle, the caller is responsible for
+  // reading the body and providing the information to the client this
+  // function provides.
+  virtual mojo::ScopedDataPipeConsumerHandle DrainAsDataPipe(
+      ResponseBodyLoaderClient** client) = 0;
+
+  virtual void Trace(Visitor*) {}
+};
+
+// ResponseBodyLoader reads the response body and reports the contents to the
+// associated client. There are two ways:
+//  - By calling Start(), ResponseBodyLoader reads the response body. and
+//    reports the contents to the client. Abort() aborts reading.
+//  - By calling DrainAsDataPipe, a user can "drain" the contents from
+//    ResponseBodyLoader. The caller is responsible for reading the body and
+//    providing the information to the client this function provides.
+// A ResponseBodyLoader is bound to the thread on which it is created.
+class PLATFORM_EXPORT ResponseBodyLoader final
+    : public ResponseBodyLoaderDrainableInterface,
+      private ResponseBodyLoaderClient,
+      private BytesConsumer::Client {
+  USING_GARBAGE_COLLECTED_MIXIN(ResponseBodyLoader);
+
+ public:
+  ResponseBodyLoader(BytesConsumer&,
+                     ResponseBodyLoaderClient&,
+                     scoped_refptr<base::SingleThreadTaskRunner>);
+
+  // ResponseBodyLoaderDrainableInterface implementation.
+  mojo::ScopedDataPipeConsumerHandle DrainAsDataPipe(
+      ResponseBodyLoaderClient**) override;
+
+  // Starts loading.
+  void Start();
+
+  // Aborts loading. This is expected to be called from the client's side, and
+  // does not report the failure to the client. This doesn't affect a
+  // drained data pipe.
+  void Abort();
+
+  // Suspendes loading.
+  void Suspend();
+
+  // Resumes loading.
+  void Resume();
+
+  bool IsAborted() const { return aborted_; }
+  bool IsSuspended() const { return suspended_; }
+  bool IsDrained() const { return drained_; }
+
+  void Trace(Visitor*) override;
+
+  // The maximal number of bytes consumed in a task. When there are more bytes
+  // in the data pipe, they will be consumed in following tasks. Setting a too
+  // small number will generate ton of tasks but setting a too large number will
+  // lead to thread janks. Also, some clients cannot handle too large chunks
+  // (512k for example).
+  static constexpr size_t kMaxNumConsumedBytesInTask = 64 * 1024;
+
+ private:
+  // ResponseBodyLoaderClient implementation.
+  void DidReceiveData(base::span<const char> data) override;
+  void DidFinishLoadingBody() override;
+  void DidFailLoadingBody() override;
+
+  // BytesConsumer::Client implementation.
+  void OnStateChange() override;
+  String DebugName() const override { return "ResponseBodyLoader"; }
+
+  Member<BytesConsumer> bytes_consumer_;
+  const Member<ResponseBodyLoaderClient> client_;
+  const scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
+  bool started_ = false;
+  bool aborted_ = false;
+  bool suspended_ = false;
+  bool drained_ = false;
+  bool finish_signal_is_pending_ = false;
+  bool fail_signal_is_pending_ = false;
+  bool in_two_phase_read_ = false;
+};
+
+}  // namespace blink
+
+#endif  // THIRD_PARTY_BLINK_RENDERER_PLATFORM_LOADER_FETCH_RESPONSE_BODY_LOADER_H_

third_party/blink/renderer/platform/loader/fetch/response_body_loader_client.h

+// Copyright 2019 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 THIRD_PARTY_BLINK_RENDERER_PLATFORM_LOADER_FETCH_RESPONSE_BODY_LOADER_CLIENT_H_
+#define THIRD_PARTY_BLINK_RENDERER_PLATFORM_LOADER_FETCH_RESPONSE_BODY_LOADER_CLIENT_H_
+
+namespace blink {
+
+// A ResponseBodyLoaderClient receives signals for loading a response body.
+class ResponseBodyLoaderClient : public GarbageCollectedMixin {
+ public:
+  ~ResponseBodyLoaderClient() = default;
+
+  // Called when reading a chunk, with the chunk.
+  virtual void DidReceiveData(base::span<const char> data) = 0;
+
+  // Called when finishing reading the entire body. This must be the last
+  // signal.
+  virtual void DidFinishLoadingBody() = 0;
+
+  // Called when seeing an error while reading the body. This must be the last
+  // signal.
+  virtual void DidFailLoadingBody() = 0;
+};
+
+}  // namespace blink
+
+#endif  // THIRD_PARTY_BLINK_RENDERER_PLATFORM_LOADER_FETCH_RESPONSE_BODY_LOADER_CLIENT_H_