Make SyncLoadContext work with mojo data pipe

Previously SyncLoadContext::OnStartLoadingResponseBody() is used for downloading
to a blob, but this CL makes it usable in both of downloading to a blob and
downloading to a SyncLoadResponse.

Bug: 911027
Change-Id: I9287f3b4e6908aaf55b5d49ab8d085792357cfde
Reviewed-on: https://chromium-review.googlesource.com/c/1377960
Commit-Queue: Makoto Shimazu <[email protected]>
Reviewed-by: Yutaka Hirano <[email protected]>
Cr-Commit-Position: refs/heads/master@{#617761}

content/renderer/loader/resource_dispatcher.h

@@ -134,7 +134,7 @@ class CONTENT_EXPORT ResourceDispatcher {
 
   // Removes a request from the |pending_requests_| list, returning true if the
   // request was found and removed.
-  bool RemovePendingRequest(
+  virtual bool RemovePendingRequest(
       int request_id,
       scoped_refptr<base::SingleThreadTaskRunner> task_runner);
 

content/renderer/loader/sync_load_context.cc

@@ -121,12 +121,16 @@ SyncLoadContext::SyncLoadContext(
     blink::mojom::BlobRegistryPtrInfo download_to_blob_registry,
     scoped_refptr<base::SingleThreadTaskRunner> task_runner)
     : response_(response),
+      body_watcher_(FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::MANUAL),
       download_to_blob_registry_(std::move(download_to_blob_registry)),
       task_runner_(std::move(task_runner)),
       signals_(std::make_unique<SignalHelper>(this,
                                               redirect_or_response_event,
                                               abort_event,
                                               timeout)) {
+  if (download_to_blob_registry_)
+    mode_ = Mode::kBlob;
+
   url_loader_factory_ =
       network::SharedURLLoaderFactory::Create(std::move(url_loader_factory));
 
@@ -182,20 +186,37 @@ void SyncLoadContext::OnReceivedResponse(
 
 void SyncLoadContext::OnStartLoadingResponseBody(
     mojo::ScopedDataPipeConsumerHandle body) {
-  DCHECK(download_to_blob_registry_);
-  DCHECK(!blob_response_started_);
-
-  blob_response_started_ = true;
-
-  download_to_blob_registry_->RegisterFromStream(
-      response_->info.mime_type, "",
-      std::max<int64_t>(0, response_->info.content_length), std::move(body),
-      nullptr,
-      base::BindOnce(&SyncLoadContext::OnFinishCreatingBlob,
-                     base::Unretained(this)));
+  if (mode_ == Mode::kBlob) {
+    DCHECK(download_to_blob_registry_);
+    DCHECK(!blob_response_started_);
+
+    blob_response_started_ = true;
+
+    download_to_blob_registry_->RegisterFromStream(
+        response_->info.mime_type, "",
+        std::max<int64_t>(0, response_->info.content_length), std::move(body),
+        nullptr,
+        base::BindOnce(&SyncLoadContext::OnFinishCreatingBlob,
+                       base::Unretained(this)));
+    return;
+  }
+  DCHECK_EQ(Mode::kInitial, mode_);
+  mode_ = Mode::kDataPipe;
+  // setup datapipe to read.
+  body_handle_ = std::move(body);
+  body_watcher_.Watch(
+      body_handle_.get(),
+      MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+      MOJO_TRIGGER_CONDITION_SIGNALS_SATISFIED,
+      base::BindRepeating(&SyncLoadContext::OnBodyReadable,
+                          base::Unretained(this)));
+  body_watcher_.ArmOrNotify();
 }
 
 void SyncLoadContext::OnReceivedData(std::unique_ptr<ReceivedData> data) {
+  if (mode_ == Mode::kInitial)
+    mode_ = Mode::kNonDataPipe;
+  DCHECK_EQ(Mode::kNonDataPipe, mode_);
   DCHECK(!Completed());
   response_->data.append(data->payload(), data->length());
 }
@@ -204,14 +225,20 @@ void SyncLoadContext::OnTransferSizeUpdated(int transfer_size_diff) {}
 
 void SyncLoadContext::OnCompletedRequest(
     const network::URLLoaderCompletionStatus& status) {
-  DCHECK(!Completed());
+  if (Completed()) {
+    // It means the response has been aborted due to an error before finishing
+    // the response.
+    return;
+  }
+  request_completed_ = true;
   response_->error_code = status.error_code;
   response_->extended_error_code = status.extended_error_code;
   response_->cors_error = status.cors_error_status;
   response_->info.encoded_data_length = status.encoded_data_length;
   response_->info.encoded_body_length = status.encoded_body_length;
-  if (blob_response_started_ && !blob_finished_) {
-    request_completed_ = true;
+  if ((blob_response_started_ && !blob_finished_) || body_handle_.is_valid()) {
+    // The body is still begin downloaded as a Blob, or being read through the
+    // handle. Wait until it's completed.
     return;
   }
   CompleteRequest();
@@ -230,6 +257,40 @@ void SyncLoadContext::OnFinishCreatingBlob(
     CompleteRequest();
 }
 
+void SyncLoadContext::OnBodyReadable(MojoResult,
+                                     const mojo::HandleSignalsState&) {
+  DCHECK_EQ(Mode::kDataPipe, mode_);
+  DCHECK(body_handle_.is_valid());
+  const void* buffer = nullptr;
+  uint32_t read_bytes = 0;
+  MojoResult result = body_handle_->BeginReadData(&buffer, &read_bytes,
+                                                  MOJO_READ_DATA_FLAG_NONE);
+  if (result == MOJO_RESULT_SHOULD_WAIT) {
+    body_watcher_.ArmOrNotify();
+    return;
+  }
+  if (result == MOJO_RESULT_FAILED_PRECONDITION) {
+    // Whole body has been read.
+    body_handle_.reset();
+    body_watcher_.Cancel();
+    if (request_completed_)
+      CompleteRequest();
+    return;
+  }
+  if (result != MOJO_RESULT_OK) {
+    // Something went wrong.
+    body_handle_.reset();
+    body_watcher_.Cancel();
+    response_->error_code = net::ERR_FAILED;
+    CompleteRequest();
+    return;
+  }
+
+  response_->data.append(static_cast<const char*>(buffer), read_bytes);
+  body_handle_->EndReadData(read_bytes);
+  body_watcher_.ArmOrNotify();
+}
+
 void SyncLoadContext::OnAbort(base::WaitableEvent* event) {
   DCHECK(!Completed());
   response_->error_code = net::ERR_ABORTED;
@@ -245,6 +306,8 @@ void SyncLoadContext::OnTimeout() {
 }
 
 void SyncLoadContext::CompleteRequest() {
+  DCHECK(blob_finished_ || (mode_ != Mode::kBlob));
+  DCHECK(!body_handle_.is_valid());
   signals_->SignalRedirectOrResponseComplete();
   signals_ = nullptr;
   response_ = nullptr;

content/renderer/loader/sync_load_context.h

@@ -10,8 +10,11 @@
 #include "base/single_thread_task_runner.h"
 #include "base/synchronization/waitable_event_watcher.h"
 #include "base/timer/timer.h"
+#include "content/common/content_export.h"
 #include "content/public/renderer/request_peer.h"
 #include "content/renderer/loader/resource_dispatcher.h"
+#include "mojo/public/cpp/system/data_pipe.h"
+#include "mojo/public/cpp/system/simple_watcher.h"
 #include "net/traffic_annotation/network_traffic_annotation.h"
 #include "services/network/public/cpp/shared_url_loader_factory.h"
 #include "third_party/blink/public/mojom/blob/blob_registry.mojom.h"
@@ -30,7 +33,15 @@ struct SyncLoadResponse;
 
 // This class owns the context necessary to perform an asynchronous request
 // while the main thread is blocked so that it appears to be synchronous.
-class SyncLoadContext : public RequestPeer {
+// There are a few mode to load a request:
+//   1) kNonDataPipe: body is received on OnReceivedData(), and the body is set
+//      to response_.data.
+//   2) kDataPipe; body is received on a data pipe passed on
+//      OnStartLoadingResponseBody(), and the body is set to response_.data.
+//   3) kBlob: body is received on a data pipe passed on
+//      OnStartLoadingResponseBody(), and wraps the data pipe with a
+//      SerializedBlobPtr.
+class CONTENT_EXPORT SyncLoadContext : public RequestPeer {
  public:
   // Begins a new asynchronous request on whatever sequence this method is
   // called on. |completed_event| will be signalled when the request is complete
@@ -59,6 +70,8 @@ class SyncLoadContext : public RequestPeer {
   void CancelRedirect();
 
  private:
+  friend class SyncLoadContextTest;
+
   SyncLoadContext(
       network::ResourceRequest* request,
       std::unique_ptr<network::SharedURLLoaderFactoryInfo> url_loader_factory,
@@ -83,6 +96,8 @@ class SyncLoadContext : public RequestPeer {
 
   void OnFinishCreatingBlob(blink::mojom::SerializedBlobPtr blob);
 
+  void OnBodyReadable(MojoResult, const mojo::HandleSignalsState&);
+
   void OnAbort(base::WaitableEvent* event);
   void OnTimeout();
 
@@ -94,6 +109,13 @@ class SyncLoadContext : public RequestPeer {
   // Set to null after CompleteRequest() is called.
   SyncLoadResponse* response_;
 
+  enum class Mode { kInitial, kNonDataPipe, kDataPipe, kBlob };
+  Mode mode_ = Mode::kInitial;
+
+  // Used when Mode::kDataPipe.
+  mojo::ScopedDataPipeConsumerHandle body_handle_;
+  mojo::SimpleWatcher body_watcher_;
+
   // State necessary to run a request on an independent thread.
   scoped_refptr<network::SharedURLLoaderFactory> url_loader_factory_;
   std::unique_ptr<ResourceDispatcher> resource_dispatcher_;