[BlobAsync] Disk support for blob storage

content/child/blob_storage/blob_transport_controller.cc

 // Copyright 2015 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 "content/child/blob_storage/blob_transport_controller.h"
 
 #include <limits>
 #include <memory>
 #include <utility>
 #include <vector>
 
 #include "base/bind.h"
 #include "base/bind_helpers.h"
 #include "base/callback.h"
 #include "base/files/file.h"
 #include "base/lazy_instance.h"
 #include "base/location.h"
 #include "base/memory/ptr_util.h"
 #include "base/memory/scoped_vector.h"
 #include "base/memory/shared_memory.h"
 #include "base/metrics/histogram_macros.h"
 #include "base/numerics/safe_conversions.h"
-#include "base/optional.h"
 #include "base/single_thread_task_runner.h"
 #include "base/stl_util.h"
 #include "base/task_runner.h"
 #include "base/task_runner_util.h"
 #include "base/threading/thread_task_runner_handle.h"
 #include "base/time/time.h"
 #include "content/child/blob_storage/blob_consolidation.h"
 #include "content/child/child_process.h"
 #include "content/child/thread_safe_sender.h"
 #include "content/common/fileapi/webblob_messages.h"
 #include "ipc/ipc_message.h"
 #include "ipc/ipc_sender.h"
 #include "storage/common/blob_storage/blob_item_bytes_request.h"
 #include "storage/common/blob_storage/blob_item_bytes_response.h"
 #include "storage/common/data_element.h"
 #include "third_party/WebKit/public/platform/Platform.h"
 
 using base::File;
 using base::SharedMemory;
 using base::SharedMemoryHandle;
 using storage::BlobItemBytesRequest;
 using storage::BlobItemBytesResponse;
 using storage::IPCBlobItemRequestStrategy;
 using storage::DataElement;
-using storage::kBlobStorageIPCThresholdBytes;
 
 using storage::BlobItemBytesResponse;
 using storage::BlobItemBytesRequest;
-using storage::IPCBlobCreationCancelCode;
+using storage::BlobStatus;
 
 namespace content {
 using ConsolidatedItem = BlobConsolidation::ConsolidatedItem;
 using ReadStatus = BlobConsolidation::ReadStatus;
 
 namespace {
 static base::LazyInstance<BlobTransportController>::Leaky g_controller =
     LAZY_INSTANCE_INITIALIZER;
 
 // This keeps the process alive while blobs are being transferred.
@@ skipping 45 matching lines @@
   BlobConsolidation::ReadStatus status = consolidation->VisitMemory(
       request.renderer_item_index, request.renderer_item_offset, request.size,
       base::Bind(&WriteSingleChunk, file));
   if (status != BlobConsolidation::ReadStatus::OK)
     return base::nullopt;
   File::Info info;
   file->GetInfo(&info);
   return base::make_optional(info.last_modified);
 }
 
-// This returns either the responses, or if they're empty, an error code.
-std::pair<std::vector<storage::BlobItemBytesResponse>,
-          IPCBlobCreationCancelCode>
-WriteDiskRequests(
+base::Optional<std::vector<BlobItemBytesResponse>> WriteDiskRequests(
     scoped_refptr<BlobConsolidation> consolidation,
     std::unique_ptr<std::vector<BlobItemBytesRequest>> requests,
     const std::vector<IPC::PlatformFileForTransit>& file_handles) {
   std::vector<BlobItemBytesResponse> responses;
   std::vector<base::Time> last_modified_times;
   last_modified_times.resize(file_handles.size());
   // We grab ownership of the file handles here. When this vector is destroyed
   // it will close the files.
   std::vector<File> files;
   files.reserve(file_handles.size());
   for (const auto& file_handle : file_handles) {
     files.emplace_back(IPC::PlatformFileForTransitToFile(file_handle));
   }
   for (const auto& request : *requests) {
     base::Optional<base::Time> last_modified = WriteSingleRequestToDisk(
         consolidation.get(), request, &files[request.handle_index]);
     if (!last_modified) {
-      return std::make_pair(std::vector<storage::BlobItemBytesResponse>(),
-                            IPCBlobCreationCancelCode::FILE_WRITE_FAILED);
+      return base::nullopt;
     }
     last_modified_times[request.handle_index] = last_modified.value();
   }
   for (const auto& request : *requests) {
     responses.push_back(BlobItemBytesResponse(request.request_number));
     responses.back().time_file_modified =
         last_modified_times[request.handle_index];
   }
 
-  return std::make_pair(responses, IPCBlobCreationCancelCode::UNKNOWN);
+  return responses;
 }
 
 }  // namespace
 
 BlobTransportController* BlobTransportController::GetInstance() {
   return g_controller.Pointer();
 }
 
 // static
 void BlobTransportController::InitiateBlobTransfer(
     const std::string& uuid,
     const std::string& content_type,
     scoped_refptr<BlobConsolidation> consolidation,
     scoped_refptr<ThreadSafeSender> sender,
     base::SingleThreadTaskRunner* io_runner,
     scoped_refptr<base::SingleThreadTaskRunner> main_runner) {
   if (main_runner->BelongsToCurrentThread()) {
     IncChildProcessRefCount();
   } else {
     main_runner->PostTask(FROM_HERE, base::Bind(&IncChildProcessRefCount));
   }
 
+  storage::BlobStorageLimits quotas;
   std::vector<storage::DataElement> descriptions;
-  std::set<std::string> referenced_blobs = consolidation->referenced_blobs();
   BlobTransportController::GetDescriptions(
-      consolidation.get(), kBlobStorageIPCThresholdBytes, &descriptions);
+      consolidation.get(), quotas.max_ipc_memory_size, &descriptions);
   // I post the task first to make sure that we store our consolidation before
   // we get a request back from the browser.
   io_runner->PostTask(
       FROM_HERE,
       base::Bind(&BlobTransportController::StoreBlobDataForRequests,
                  base::Unretained(BlobTransportController::GetInstance()), uuid,
                  base::Passed(std::move(consolidation)),
                  base::Passed(std::move(main_runner))));
-  // TODO(dmurph): Merge register and start messages.
-  sender->Send(new BlobStorageMsg_RegisterBlobUUID(uuid, content_type, "",
-                                                   referenced_blobs));
-  sender->Send(new BlobStorageMsg_StartBuildingBlob(uuid, descriptions));
+  sender->Send(
+      new BlobStorageMsg_RegisterBlob(uuid, content_type, "", descriptions));
 }
 
 void BlobTransportController::OnMemoryRequest(
     const std::string& uuid,
     const std::vector<storage::BlobItemBytesRequest>& requests,
     std::vector<base::SharedMemoryHandle>* memory_handles,
     const std::vector<IPC::PlatformFileForTransit>& file_handles,
     base::TaskRunner* file_runner,
     IPC::Sender* sender) {
   std::vector<BlobItemBytesResponse> responses;
@@ skipping 98 matching lines @@
         base::Bind(&WriteDiskRequests, make_scoped_refptr(consolidation),
                    base::Passed(&file_requests), file_handles),
         base::Bind(&BlobTransportController::OnFileWriteComplete,
                    weak_factory_.GetWeakPtr(), sender, uuid));
   }
 
   if (!responses.empty())
     sender->Send(new BlobStorageMsg_MemoryItemResponse(uuid, responses));
 }
 
-void BlobTransportController::OnCancel(
-    const std::string& uuid,
-    storage::IPCBlobCreationCancelCode code) {
-  DVLOG(1) << "Received blob cancel for blob " << uuid
-           << " with code: " << static_cast<int>(code);
+void BlobTransportController::OnBlobFinalStatus(const std::string& uuid,
+                                                storage::BlobStatus code) {
+  DVLOG_IF(1, storage::BlobStatusIsError(code))
+      << "Received blob error for blob " << uuid
+      << " with code: " << static_cast<int>(code);
+  DCHECK(!BlobStatusIsPending(code));
   ReleaseBlobConsolidation(uuid);
 }
 
-void BlobTransportController::OnDone(const std::string& uuid) {
-  ReleaseBlobConsolidation(uuid);
-}
-
 void BlobTransportController::CancelAllBlobTransfers() {
   weak_factory_.InvalidateWeakPtrs();
   if (!blob_storage_.empty() && main_thread_runner_) {
     main_thread_runner_->PostTask(
         FROM_HERE,
         base::Bind(&DecChildProcessRefCountTimes, blob_storage_.size()));
   }
   main_thread_runner_ = nullptr;
   blob_storage_.clear();
 }
@@ skipping 52 matching lines @@
   }
 }
 
 BlobTransportController::BlobTransportController() : weak_factory_(this) {}
 
 BlobTransportController::~BlobTransportController() {}
 
 void BlobTransportController::OnFileWriteComplete(
     IPC::Sender* sender,
     const std::string& uuid,
-    const std::pair<std::vector<BlobItemBytesResponse>,
-                    IPCBlobCreationCancelCode>& result) {
+    const base::Optional<std::vector<storage::BlobItemBytesResponse>>& result) {
   if (blob_storage_.find(uuid) == blob_storage_.end())
     return;
-  if (!result.first.empty()) {
-    sender->Send(new BlobStorageMsg_MemoryItemResponse(uuid, result.first));
+  if (!result) {
+    sender->Send(new BlobStorageMsg_SendBlobStatus(
+        uuid, BlobStatus::ERR_FILE_WRITE_FAILED));
+    ReleaseBlobConsolidation(uuid);
     return;
   }
-  sender->Send(new BlobStorageMsg_CancelBuildingBlob(uuid, result.second));
-  ReleaseBlobConsolidation(uuid);
+  sender->Send(new BlobStorageMsg_MemoryItemResponse(uuid, result.value()));
 }
 
 void BlobTransportController::StoreBlobDataForRequests(
     const std::string& uuid,
     scoped_refptr<BlobConsolidation> consolidation,
     scoped_refptr<base::SingleThreadTaskRunner> main_runner) {
   if (!main_thread_runner_.get()) {
     main_thread_runner_ = std::move(main_runner);
   }
   blob_storage_[uuid] = std::move(consolidation);
 }
 
 void BlobTransportController::ReleaseBlobConsolidation(
     const std::string& uuid) {
   if (blob_storage_.erase(uuid)) {
     main_thread_runner_->PostTask(FROM_HERE,
                                   base::Bind(&DecChildProcessRefCount));
   }
 }
 
 }  // namespace content