Tune the mojo::DataPipe capacity and read chunk size when reading blobs.

We have reports from some sites that data loaded from cache_storage via
a service worker is slower than when loaded from http cache.  One possible
explanation is that the mojo::DataPipes used for reading the BlobDataHandle
from cache_storage are not tuned like the DataPipes used by the network
loading path.  In the network loading path they use 512KB pipes and read in
32KB chunks.

This CL attempts to tune the various DataPipes used to read BlobDataHandle
objects.  It also includes some DataPipes in the service worker code that
are read in a similar way; for loading scripts, ReadableStreams, etc.

This CL uses 512KB pipes and 64KB read chunks as the default as those were
the best values in local testing.  It also adds features to control these
values in a future finch trial.  We can compare different values against
heartbeat metrics and site-specific metrics.

R=kinuko@chromium.org, mek@chromium.org, rkaplow@chromium.org

Bug: 881141
Change-Id: If8d956ad195c6cf85547f1c7172f4a35972513e4
Reviewed-on: https://chromium-review.googlesource.com/1211739Reviewed-by: Robert Kaplow <rkaplow@chromium.org>
Reviewed-by: Marijn Kruisselbrink <mek@chromium.org>
Reviewed-by: Kinuko Yasuda <kinuko@chromium.org>
Commit-Queue: Ben Kelly <wanderview@chromium.org>
Cr-Commit-Position: refs/heads/master@{#591135}

content/browser/cache_storage/cache_storage_blob_to_disk_cache.cc

@@ -13,6 +13,7 @@
 #include "storage/browser/blob/blob_data_handle.h"
 #include "storage/browser/blob/blob_url_request_job_factory.h"
 #include "storage/common/storage_histograms.h"
+#include "third_party/blink/public/common/blob/blob_utils.h"
 
 namespace content {
 
@@ -38,13 +39,14 @@ void CacheStorageBlobToDiskCache::StreamBlobToCache(
   DCHECK(!consumer_handle_.is_valid());
   DCHECK(!pending_read_);
 
-  mojo::ScopedDataPipeProducerHandle producer_handle;
-  MojoResult result =
-      CreateDataPipe(nullptr, &producer_handle, &consumer_handle_);
-  if (result != MOJO_RESULT_OK) {
+  mojo::DataPipe pipe(blink::BlobUtils::GetDataPipeCapacity());
+  if (!pipe.consumer_handle.is_valid()) {
     std::move(callback).Run(std::move(entry), false /* success */);
     return;
   }
+  DCHECK(pipe.producer_handle.is_valid());
+
+  consumer_handle_ = std::move(pipe.consumer_handle);
 
   disk_cache_body_index_ = disk_cache_body_index;
   entry_ = std::move(entry);
@@ -52,7 +54,7 @@ void CacheStorageBlobToDiskCache::StreamBlobToCache(
 
   blink::mojom::BlobReaderClientPtr client;
   client_binding_.Bind(MakeRequest(&client));
-  blob->ReadAll(std::move(producer_handle), std::move(client));
+  blob->ReadAll(std::move(pipe.producer_handle), std::move(client));
 
   handle_watcher_.Watch(
       consumer_handle_.get(), MOJO_HANDLE_SIGNAL_READABLE,

content/browser/service_worker/service_worker_installed_script_reader.cc

@@ -11,6 +11,7 @@
 #include "content/browser/service_worker/service_worker_metrics.h"
 #include "net/http/http_response_headers.h"
 #include "services/network/public/cpp/net_adapters.h"
+#include "third_party/blink/public/common/blob/blob_utils.h"
 
 namespace content {
 
@@ -114,25 +115,25 @@ void ServiceWorkerInstalledScriptReader::OnReadInfoComplete(
 
   DCHECK_GE(result, 0);
   mojo::ScopedDataPipeConsumerHandle meta_data_consumer;
-  mojo::ScopedDataPipeConsumerHandle body_consumer;
   DCHECK_GE(http_info->response_data_size, 0);
   uint64_t body_size = http_info->response_data_size;
   uint64_t meta_data_size = 0;
-  if (mojo::CreateDataPipe(nullptr, &body_handle_, &body_consumer) !=
-      MOJO_RESULT_OK) {
+  mojo::DataPipe body_pipe(blink::BlobUtils::GetDataPipeCapacity());
+  if (!body_pipe.producer_handle.is_valid()) {
     CompleteSendIfNeeded(FinishedReason::kCreateDataPipeError);
     return;
   }
+  body_handle_ = std::move(body_pipe.producer_handle);
   // Start sending meta data (V8 code cache data).
   if (http_info->http_info->metadata) {
-    mojo::ScopedDataPipeProducerHandle meta_data_producer;
-    if (mojo::CreateDataPipe(nullptr, &meta_data_producer,
-                             &meta_data_consumer) != MOJO_RESULT_OK) {
+    mojo::DataPipe meta_pipe(blink::BlobUtils::GetDataPipeCapacity());
+    if (!meta_pipe.producer_handle.is_valid()) {
       CompleteSendIfNeeded(FinishedReason::kCreateDataPipeError);
       return;
     }
+    meta_data_consumer = std::move(meta_pipe.consumer_handle);
     meta_data_sender_ = std::make_unique<MetaDataSender>(
-        http_info->http_info->metadata, std::move(meta_data_producer));
+        http_info->http_info->metadata, std::move(meta_pipe.producer_handle));
     meta_data_sender_->Start(base::BindOnce(
         &ServiceWorkerInstalledScriptReader::OnMetaDataSent, AsWeakPtr()));
     DCHECK_GE(http_info->http_info->metadata->size(), 0);
@@ -168,7 +169,7 @@ void ServiceWorkerInstalledScriptReader::OnReadInfoComplete(
   }
 
   client_->OnStarted(charset, std::move(header_strings),
-                     std::move(body_consumer), body_size,
+                     std::move(body_pipe.consumer_handle), body_size,
                      std::move(meta_data_consumer), meta_data_size);
   client_->OnHttpInfoRead(http_info);
 }
@@ -181,6 +182,9 @@ void ServiceWorkerInstalledScriptReader::OnWritableBody(MojoResult) {
   uint32_t num_bytes = 0;
   MojoResult rv = network::NetToMojoPendingBuffer::BeginWrite(
       &body_handle_, &body_pending_write_, &num_bytes);
+
+  num_bytes = std::min(num_bytes, blink::BlobUtils::GetDataPipeChunkSize());
+
   switch (rv) {
     case MOJO_RESULT_INVALID_ARGUMENT:
     case MOJO_RESULT_BUSY:

content/common/service_worker/service_worker_loader_helpers.cc

@@ -18,6 +18,7 @@
 #include "services/network/public/cpp/resource_request.h"
 #include "services/network/public/cpp/resource_request_body.h"
 #include "services/network/public/cpp/resource_response.h"
+#include "third_party/blink/public/common/blob/blob_utils.h"
 #include "ui/base/page_transition_types.h"
 
 namespace content {
@@ -193,7 +194,7 @@ int ServiceWorkerLoaderHelpers::ReadBlobResponseBody(
     return net::ERR_REQUEST_RANGE_NOT_SATISFIABLE;
   }
 
-  mojo::DataPipe data_pipe;
+  mojo::DataPipe data_pipe(blink::BlobUtils::GetDataPipeCapacity());
   blink::mojom::BlobReaderClientPtr blob_reader_client;
   mojo::MakeStrongBinding(
       std::make_unique<BlobCompleteCaller>(std::move(on_blob_read_complete)),

storage/DEPS

@@ -6,6 +6,7 @@ include_rules = [
   "+services/network/public/mojom",
   "+services/network/session_cleanup_cookie_store.h",
   "+sql",
+  "+third_party/blink/public/common",
   "+third_party/blink/public/mojom",
   "+third_party/blink/public/platform",
   "+third_party/leveldatabase",

storage/browser/blob/mojo_blob_reader.cc

@@ -8,6 +8,7 @@
 #include "net/base/io_buffer.h"
 #include "services/network/public/cpp/net_adapters.h"
 #include "storage/browser/blob/blob_data_handle.h"
+#include "third_party/blink/public/common/blob/blob_utils.h"
 
 namespace storage {
 
@@ -184,6 +185,8 @@ void MojoBlobReader::ReadMore() {
     return;
   }
 
+  num_bytes = std::min(num_bytes, blink::BlobUtils::GetDataPipeChunkSize());
+
   TRACE_EVENT_ASYNC_BEGIN0("Blob", "BlobReader::ReadMore", this);
   CHECK_GT(static_cast<uint32_t>(std::numeric_limits<int>::max()), num_bytes);
   DCHECK(pending_write_);

testing/variations/fieldtrial_testing_config.json

@@ -873,6 +873,29 @@
             ]
         }
     ],
+    "BlobDataPipeTuning": [
+        {
+            "platforms": [
+                "android",
+                "chromeos",
+                "linux",
+                "mac",
+                "windows"
+            ],
+            "experiments": [
+                {
+                    "name": "Capacity_512kb_Chunk_64kb",
+                    "params": {
+                        "capacity_bytes": "524288",
+                        "chunk_bytes": "65536"
+                    },
+                    "enable_features": [
+                        "BlobDataPipeTuning"
+                    ]
+                }
+            ]
+        }
+    ],
     "BlockTabUnders": [
         {
             "platforms": [

third_party/blink/common/blob/blob_utils.cc

@@ -4,7 +4,10 @@
 
 #include "third_party/blink/public/common/blob/blob_utils.h"
 
+#include <algorithm>
+
 #include "base/feature_list.h"
+#include "base/metrics/field_trial_params.h"
 #include "services/network/public/cpp/features.h"
 #include "third_party/blink/public/common/features.h"
 
@@ -16,4 +19,39 @@ bool BlobUtils::MojoBlobURLsEnabled() {
          base::FeatureList::IsEnabled(blink::features::kMojoBlobURLs);
 }
 
+namespace {
+
+const base::Feature kBlobDataPipeTuningFeature{
+    "BlobDataPipeTuning", base::FEATURE_DISABLED_BY_DEFAULT};
+
+constexpr int kBlobMinDataPipeCapacity = 1024;
+constexpr int kBlobDefaultDataPipeCapacity = 512 * 1024;
+
+constexpr base::FeatureParam<int> kBlobDataPipeCapacity{
+    &kBlobDataPipeTuningFeature, "capacity_bytes",
+    kBlobDefaultDataPipeCapacity};
+
+constexpr int kBlobMinDataPipeChunkSize = 1024;
+constexpr int kBlobDefaultDataPipeChunkSize = 64 * 1024;
+
+constexpr base::FeatureParam<int> kBlobDataPipeChunkSize{
+    &kBlobDataPipeTuningFeature, "chunk_bytes", kBlobDefaultDataPipeChunkSize};
+
+}  // namespace
+
+// static
+uint32_t BlobUtils::GetDataPipeCapacity() {
+  return std::max(kBlobDataPipeCapacity.Get(), kBlobMinDataPipeCapacity);
+}
+
+// static
+uint32_t BlobUtils::GetDataPipeChunkSize() {
+  // The mojo::DataPipe will allow up to 64KB to be written into it in
+  // a single chunk, but there may be some advantage to writing smaller
+  // chunks.  For example, the network stack uses 32KB chunks.  This could
+  // result in the faster delivery of the first byte of data when reading
+  // from a slow disk.
+  return std::max(kBlobDataPipeChunkSize.Get(), kBlobMinDataPipeChunkSize);
+}
+
 }  // namespace blink

third_party/blink/public/common/blob/blob_utils.h

@@ -6,6 +6,8 @@
 #ifndef THIRD_PARTY_BLINK_PUBLIC_COMMON_BLOB_BLOB_UTILS_H_
 #define THIRD_PARTY_BLINK_PUBLIC_COMMON_BLOB_BLOB_UTILS_H_
 
+#include <stdint.h>
+
 #include "third_party/blink/common/common_export.h"
 
 namespace blink {
@@ -15,6 +17,13 @@ class BlobUtils {
   // Whether the new Blob URL glue for NetworkService is enabled (i.e.,
   // the NetworkService or MojoBlobURLs feature is enabled).
   static bool BLINK_COMMON_EXPORT MojoBlobURLsEnabled();
+
+  // Get the preferred capacity a mojo::DataPipe being used to read a blob.
+  static uint32_t BLINK_COMMON_EXPORT GetDataPipeCapacity();
+
+  // Get the preferred chunk size to use when reading a blob to copy
+  // into a mojo::DataPipe.
+  static uint32_t BLINK_COMMON_EXPORT GetDataPipeChunkSize();
 };
 
 }  // namespace blink

third_party/blink/renderer/core/fetch/form_data_bytes_consumer.cc

@@ -4,6 +4,7 @@
 
 #include "third_party/blink/renderer/core/fetch/form_data_bytes_consumer.h"
 
+#include "third_party/blink/public/common/blob/blob_utils.h"
 #include "third_party/blink/public/platform/platform.h"
 #include "third_party/blink/renderer/core/execution_context/execution_context.h"
 #include "third_party/blink/renderer/core/fetch/blob_bytes_consumer.h"
@@ -185,7 +186,7 @@ class DataPipeAndDataBytesConsumer final : public BytesConsumer {
       if (!data_pipe_consumer_) {
         network::mojom::blink::DataPipeGetterPtr* data_pipe_getter =
             iter_->data_pipe_getter_->GetPtr();
-        mojo::DataPipe data_pipe;
+        mojo::DataPipe data_pipe(BlobUtils::GetDataPipeCapacity());
         (*data_pipe_getter)
             ->Read(
                 std::move(data_pipe.producer_handle),

third_party/blink/renderer/core/fileapi/file_reader_loader.cc

@@ -36,6 +36,7 @@
 
 #include "base/memory/scoped_refptr.h"
 #include "mojo/public/cpp/system/wait.h"
+#include "third_party/blink/public/common/blob/blob_utils.h"
 #include "third_party/blink/public/platform/web_url_request.h"
 #include "third_party/blink/renderer/core/execution_context/execution_context.h"
 #include "third_party/blink/renderer/core/fileapi/blob.h"
@@ -89,17 +90,16 @@ void FileReaderLoader::Start(scoped_refptr<BlobDataHandle> blob_data) {
   started_loading_ = true;
 #endif  // DCHECK_IS_ON()
 
-  mojo::ScopedDataPipeProducerHandle producer_handle;
-  MojoResult result =
-      CreateDataPipe(nullptr, &producer_handle, &consumer_handle_);
-  if (result != MOJO_RESULT_OK) {
+  mojo::DataPipe pipe(blink::BlobUtils::GetDataPipeCapacity());
+  if (!pipe.producer_handle.is_valid()) {
     Failed(FileError::kNotReadableErr, FailureType::kMojoPipeCreation);
     return;
   }
+  consumer_handle_ = std::move(pipe.consumer_handle);
 
   mojom::blink::BlobReaderClientPtr client_ptr;
   binding_.Bind(MakeRequest(&client_ptr));
-  blob_data->ReadAll(std::move(producer_handle), std::move(client_ptr));
+  blob_data->ReadAll(std::move(pipe.producer_handle), std::move(client_ptr));
 
   if (IsSyncLoad()) {
     // Wait for OnCalculatedSize, which will also synchronously drain the data

third_party/blink/renderer/modules/service_worker/fetch_respond_with_observer.cc

@@ -9,6 +9,7 @@
 
 #include "services/network/public/mojom/fetch_api.mojom-blink.h"
 #include "services/network/public/mojom/request_context_frame_type.mojom-blink.h"
+#include "third_party/blink/public/common/blob/blob_utils.h"
 #include "third_party/blink/public/platform/modules/service_worker/web_service_worker_response.h"
 #include "third_party/blink/public/platform/task_type.h"
 #include "third_party/blink/renderer/bindings/core/v8/script_value.h"
@@ -287,25 +288,23 @@ void FetchRespondWithObserver::OnResponseFulfilled(
       return;
     }
     // Handle the stream response body.
-    mojo::ScopedDataPipeProducerHandle producer;
-    mojo::ScopedDataPipeConsumerHandle consumer;
-    MojoResult result = mojo::CreateDataPipe(nullptr, &producer, &consumer);
-    if (result != MOJO_RESULT_OK) {
+    mojo::DataPipe pipe(BlobUtils::GetDataPipeCapacity());
+    if (!pipe.consumer_handle.is_valid()) {
       OnResponseRejected(ServiceWorkerResponseError::kDataPipeCreationFailed);
       return;
     }
-    DCHECK(producer.is_valid());
-    DCHECK(consumer.is_valid());
+    DCHECK(pipe.producer_handle.is_valid());
 
     std::unique_ptr<WebServiceWorkerStreamHandle> body_stream_handle =
-        std::make_unique<WebServiceWorkerStreamHandle>(std::move(consumer));
+        std::make_unique<WebServiceWorkerStreamHandle>(
+            std::move(pipe.consumer_handle));
     ServiceWorkerGlobalScopeClient::From(GetExecutionContext())
         ->RespondToFetchEventWithResponseStream(event_id_, web_response,
                                                 body_stream_handle.get(),
                                                 event_dispatch_time_);
 
     buffer->StartLoading(FetchDataLoader::CreateLoaderAsDataPipe(
-                             std::move(producer), task_runner_),
+                             std::move(pipe.producer_handle), task_runner_),
                          new FetchLoaderClient(std::move(body_stream_handle)),
                          exception_state);
     if (exception_state.HadException()) {