[BlobAsync] Disk support for blob storage

storage/browser/blob/blob_memory_controller.cc

+// Copyright 2016 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 "storage/browser/blob/blob_memory_controller.h"
+
+#include <algorithm>
+
+#include "base/callback.h"
+#include "base/callback_helpers.h"
+#include "base/files/file_util.h"
+#include "base/location.h"
+#include "base/memory/ptr_util.h"
+#include "base/metrics/histogram_macros.h"
+#include "base/numerics/safe_conversions.h"
+#include "base/numerics/safe_math.h"
+#include "base/single_thread_task_runner.h"
+#include "base/single_thread_task_runner.h"
+#include "base/strings/string_number_conversions.h"
+#include "base/task_runner.h"
+#include "base/task_runner_util.h"
+#include "base/time/time.h"
+#include "base/trace_event/trace_event.h"
+#include "base/tuple.h"
+#include "storage/browser/blob/blob_data_item.h"
+#include "storage/browser/blob/shareable_blob_data_item.h"
+#include "storage/browser/blob/shareable_file_reference.h"
+#include "storage/common/data_element.h"
+
+using base::File;
+using base::FilePath;
+using FileCreationInfo = storage::BlobMemoryController::FileCreationInfo;
+
+namespace storage {
+namespace {
+
+bool CalculateBlobMemorySize(const std::vector<DataElement>& elements,
+                             size_t* shortcut_bytes,
+                             uint64_t* total_bytes) {
+  DCHECK(shortcut_bytes);
+  DCHECK(total_bytes);
+
+  base::CheckedNumeric<uint64_t> total_size_checked = 0;
+  base::CheckedNumeric<size_t> shortcut_size_checked = 0;
+  for (const auto& e : elements) {
+    if (e.type() == DataElement::TYPE_BYTES) {
+      total_size_checked += e.length();
+      shortcut_size_checked += e.length();
+    } else if (e.type() == DataElement::TYPE_BYTES_DESCRIPTION) {
+      total_size_checked += e.length();
+    } else {
+      continue;
+    }
+    if (!total_size_checked.IsValid() || !shortcut_size_checked.IsValid())
+      return false;
+  }
+  *shortcut_bytes = shortcut_size_checked.ValueOrDie();
+  *total_bytes = total_size_checked.ValueOrDie();
+  return true;
+}
+
+// Creates a file in the given directory w/ the given filename and size.
+BlobMemoryController::FileCreationInfo CreateFile(
+    scoped_refptr<ShareableFileReference> file_reference,
+    size_t size_bytes) {
+  LOG(ERROR) << "creating file for renderer";
+  DCHECK_NE(0u, size_bytes);
+  BlobMemoryController::FileCreationInfo creation_info;
+
+  // Try to open our file.
+  File file(file_reference->path(),
+            File::FLAG_CREATE_ALWAYS | File::FLAG_WRITE);
+  creation_info.file_reference = std::move(file_reference);
+  creation_info.error = file.error_details();
+  UMA_HISTOGRAM_ENUMERATION("Storage.Blob.TransportFileCreate",
+                            -creation_info.error, -File::FILE_ERROR_MAX);
+  if (creation_info.error != File::FILE_OK)
+    return creation_info;
+
+  // Grab the file info to get the "last modified" time and store the file.
+  File::Info file_info;
+  bool success = file.GetInfo(&file_info);
+  UMA_HISTOGRAM_BOOLEAN("Storage.Blob.TransportFileInfoSuccess", success);
+  creation_info.error = success ? File::FILE_OK : File::FILE_ERROR_FAILED;
+  creation_info.last_modified = file_info.last_modified;
+  if (success)
+    creation_info.file = std::move(file);
+  return creation_info;
+}
+
+BlobMemoryController::FileCreationInfo WriteItemsToFile(
+    std::vector<scoped_refptr<ShareableBlobDataItem>>* items,
+    size_t total_size_bytes,
+    scoped_refptr<ShareableFileReference> file_reference) {
+  DCHECK_NE(0u, total_size_bytes);
+  LOG(ERROR) << "writing to file!";
+  UMA_HISTOGRAM_MEMORY_KB("Storage.Blob.PageFileSize", total_size_bytes / 1024);
+
+  // Create our file.
+  BlobMemoryController::FileCreationInfo creation_info;
+  File file(file_reference->path(),
+            File::FLAG_CREATE_ALWAYS | File::FLAG_WRITE);
+  creation_info.file_reference = std::move(file_reference);
+  creation_info.error = file.error_details();
+  UMA_HISTOGRAM_ENUMERATION("Storage.Blob.PageFileCreate", -creation_info.error,
+                            -File::FILE_ERROR_MAX);
+  if (creation_info.error != File::FILE_OK)
+    return creation_info;
+
+  // Write data.
+  file.SetLength(total_size_bytes);
+  int bytes_written = 0;
+  for (const auto& refptr : *items) {
+    const DataElement& element = refptr->item()->data_element();
+    DCHECK_EQ(DataElement::TYPE_BYTES, element.type());
+    size_t length = base::checked_cast<size_t>(element.length());
+    size_t bytes_left = length;
+    while (bytes_left > 0) {
+      bytes_written =
+          file.WriteAtCurrentPos(element.bytes() + (length - bytes_left),
+                                 base::saturated_cast<int>(bytes_left));
+      if (bytes_written < 0)
+        break;
+      DCHECK_LE(static_cast<size_t>(bytes_written), bytes_left);
+      bytes_left -= bytes_written;
+    }
+    if (bytes_written < 0)
+      break;
+  }
+  UMA_HISTOGRAM_BOOLEAN("Storage.Blob.PageFileWriteSuccess", bytes_written > 0);
+
+  // Grab our modification time and create our SharedFileReference to manage the
+  // lifetime of the file.
+  File::Info info;
+  bool success = file.GetInfo(&info);
+  UMA_HISTOGRAM_BOOLEAN("Storage.Blob.PageFileInfoSuccess", success);
+  creation_info.error =
+      bytes_written < 0 || !success ? File::FILE_ERROR_FAILED : File::FILE_OK;
+  creation_info.last_modified = info.last_modified;
+  return creation_info;
+}
+
+}  // namespace
+
+BlobMemoryController::FileCreationInfo::FileCreationInfo() {}
+
+BlobMemoryController::FileCreationInfo::~FileCreationInfo() {}
+
+FileCreationInfo::FileCreationInfo(FileCreationInfo&&) = default;
+FileCreationInfo& FileCreationInfo::operator=(FileCreationInfo&&) = default;
+
+BlobMemoryController::BlobMemoryController()
+    : recent_item_cache_(
+          base::MRUCache<uint64_t, ShareableBlobDataItem*>::NO_AUTO_EVICT),
+      ptr_factory_(this) {}
+
+BlobMemoryController::~BlobMemoryController() {}
+
+void BlobMemoryController::EnableDisk(
+    const base::FilePath& storage_directory,
+    scoped_refptr<base::TaskRunner> file_runner) {
+  LOG(ERROR) << "enbling disk";
+  file_runner_ = std::move(file_runner);
+  blob_storage_dir_ = storage_directory;
+  disk_enabled_ = true;
+}
+
+void BlobMemoryController::DisableDisk() {
+  disk_enabled_ = false;
+  blob_memory_used_ += in_flight_memory_used_;
+  in_flight_memory_used_ = 0;
+  for (const auto& size_callback_pair : blobs_waiting_for_paging_)
+    size_callback_pair.second.Run(false);
+  pending_pagings_ = 0;
+  blobs_waiting_for_paging_.clear();
+  blobs_waiting_for_paging_size_ = 0;
+  recent_item_cache_.Clear();
+  recent_item_cache_bytes_ = 0;
+  RecordTracingCounters();
+}
+
+bool BlobMemoryController::DecideBlobTransportationMemoryStrategy(
+    const std::vector<DataElement>& descriptions,
+    uint64_t* total_bytes,
+    BlobMemoryController::MemoryStrategyResult* result) const {
+  DCHECK(total_bytes);
+  DCHECK(result);
+
+  // Step 1: Get the sizes.
+  size_t shortcut_memory_size_bytes;
+  uint64_t total_memory_size_bytes;
+  if (!CalculateBlobMemorySize(descriptions, &shortcut_memory_size_bytes,
+                               &total_memory_size_bytes)) {
+    return false;
+  }
+  *total_bytes = total_memory_size_bytes;
+
+  // Step 2: Handle case where we have no memory to transport.
+  if (total_memory_size_bytes == 0) {
+    *result = MemoryStrategyResult::NONE_NEEDED;
+    return true;
+  }
+
+  // Step 3: Check if we have enough memory to store the blob.
+  if (total_memory_size_bytes >
+      GetAvailableMemoryForBlobs() + GetAvailableDiskSpaceForBlobs()) {
+    *result = MemoryStrategyResult::TOO_LARGE;
+    return true;
+  }
+
+  // From here on, we know we can fit the blob in memory or on disk.
+  // Step 4: Decide if we're using the shortcut method.
+  if (shortcut_memory_size_bytes == total_memory_size_bytes &&
+      blobs_waiting_for_paging_.empty() &&
+      shortcut_memory_size_bytes < GetAvailableMemoryForBlobs()) {
+    *result = MemoryStrategyResult::NONE_NEEDED;
+    return true;
+  }
+
+  // Step 5: Decide if we're going straight to disk.
+  if (disk_enabled_ && (total_memory_size_bytes > max_blob_in_memory_size_)) {
+    *result = MemoryStrategyResult::FILE;
+    return true;
+  }
+  // From here on, we know the blob's size is less than:
+  // * max_blob_in_memory_size_ if enable_disk_ is true
+  // * max_blob_memory_space_ if enable_disk_ is false
+  // So we know we're < max(size_t).
+  // Step 6: Decide if we're using shared memory.
+  if (total_memory_size_bytes > max_ipc_memory_size_) {
+    *result = MemoryStrategyResult::SHARED_MEMORY;
+    return true;
+  }
+  // Step 7: We can fit in IPC.
+  *result = MemoryStrategyResult::IPC;
+  return true;
+}
+
+void BlobMemoryController::CreateTemporaryFile(
+    uint64_t size_bytes,
+    const base::Callback<void(FileCreationInfo)>& file_callback) {
+  if (!disk_enabled_) {
+    BlobMemoryController::FileCreationInfo creation_info;
+    file_callback.Run(std::move(creation_info));
+    return;
+  }
+
+  disk_used_ += size_bytes;
+  std::string file_name = base::Uint64ToString(current_file_num_++);
+  scoped_refptr<ShareableFileReference> file_ref =
+      ShareableFileReference::GetOrCreate(
+          blob_storage_dir_.Append(file_name),
+          ShareableFileReference::DELETE_ON_FINAL_RELEASE, file_runner_.get());
+  base::PostTaskAndReplyWithResult(
+      file_runner_.get(), FROM_HERE,
+      base::Bind(&CreateFile, std::move(file_ref), size_bytes),
+      base::Bind(&BlobMemoryController::OnCreateFile, ptr_factory_.GetWeakPtr(),
+                 size_bytes, file_callback));
+
+  RecordTracingCounters();
+}
+
+void BlobMemoryController::FreeMemory(size_t memory_size_bytes) {
+  DCHECK_GE(blob_memory_used_, memory_size_bytes);
+  blob_memory_used_ -= memory_size_bytes;
+  if (memory_size_bytes != 0) {
+    LOG(ERROR) << "Freeing memory " << memory_size_bytes;
+    MaybeScheduleWaitingBlobs();
+  }
+}
+
+bool BlobMemoryController::CanFitInSystem(uint64_t size) const {
+  return size < GetAvailableMemoryForBlobs() + GetAvailableDiskSpaceForBlobs();
+}
+
+base::Optional<BlobMemoryController::PendingConstructionEntry>
+BlobMemoryController::NotifyWhenMemoryCanPopulated(
+    size_t memory_size,
+    const base::Callback<void(bool)>& can_request_callback) {
+  DCHECK(memory_size <=
+         GetAvailableMemoryForBlobs() + GetAvailableDiskSpaceForBlobs());
+
+  if (!disk_enabled_) {
+    LOG(ERROR) << "Yes, " << memory_size << " can fit in memory now.";
+    blob_memory_used_ += memory_size;
+    return base::nullopt;
+  }
+
+  // If we're currently waiting for blobs to page already, then we add
+  // ourselves to the end of the queue. Once paging is complete, we'll schedule
+  // more paging for any more pending blobs.
+  if (!blobs_waiting_for_paging_.empty()) {
+    LOG(ERROR) << "putting memory request " << memory_size << " in queue.";
+    blobs_waiting_for_paging_.push_back(
+        std::make_pair(memory_size, can_request_callback));
+    blobs_waiting_for_paging_size_ += memory_size;
+    base::Optional<BlobMemoryController::PendingConstructionEntry> entry =
+        --blobs_waiting_for_paging_.end();
+    return entry;
+  }
+
+  // Store right away if we can.
+  if (memory_size <= GetAvailableMemoryForBlobs()) {
+    // If we're past our blob memory limit, then schedule our paging.
+    LOG(ERROR) << "Yes, " << memory_size << " can fit in memory now.";
+    blob_memory_used_ += memory_size;
+    MaybeSchedulePagingUntilSystemHealthy();
+    return base::nullopt;
+  }
+
+  // This means we're too big for memory.
+  LOG(ERROR) << "waiting until " << memory_size << " can fit.";
+  DCHECK(blobs_waiting_for_paging_.empty());
+  DCHECK_EQ(0u, blobs_waiting_for_paging_size_);
+  blobs_waiting_for_paging_.push_back(
+      std::make_pair(memory_size, can_request_callback));
+  blobs_waiting_for_paging_size_ = memory_size;
+  base::Optional<BlobMemoryController::PendingConstructionEntry> entry =
+      --blobs_waiting_for_paging_.end();
+  LOG(ERROR) << "Scheduling paging on first item too big";
+  MaybeSchedulePagingUntilSystemHealthy();
+  return entry;
+}
+
+void BlobMemoryController::RemovePendingConstructionEntry(
+    const BlobMemoryController::PendingConstructionEntry& entry) {
+  if (entry == blobs_waiting_for_paging_.end())
+    return;
+  const std::pair<size_t, base::Callback<void(bool)>> pair = *entry;
+  blobs_waiting_for_paging_size_ -= pair.first;
+  blobs_waiting_for_paging_.erase(entry);
+}
+
+void BlobMemoryController::UpdateBlobItemInRecents(
+    ShareableBlobDataItem* item) {
+  auto iterator = recent_item_cache_.Get(item->item_id());
+  if (iterator == recent_item_cache_.end()) {
+    DCHECK_EQ(DataElement::TYPE_BYTES, item->item()->type());
+    recent_item_cache_bytes_ += static_cast<size_t>(item->item()->length());
+    recent_item_cache_.Put(item->item_id(), item);
+    MaybeSchedulePagingUntilSystemHealthy();
+  }
+}
+
+void BlobMemoryController::RemoveBlobItemInRecents(
+    const ShareableBlobDataItem& item) {
+  auto iterator = recent_item_cache_.Get(item.item_id());
+  if (iterator != recent_item_cache_.end()) {
+    size_t size = static_cast<size_t>(item.item()->length());
+    DCHECK_GE(recent_item_cache_bytes_, size);
+    recent_item_cache_bytes_ -= size;
+    recent_item_cache_.Erase(iterator);
+  }
+}
+
+void BlobMemoryController::OnCreateFile(
+    uint64_t file_size,
+    const base::Callback<void(FileCreationInfo)>& file_callback,
+    FileCreationInfo result) {
+  if (result.error == File::FILE_OK) {
+    result.file_reference->AddFinalReleaseCallback(
+        base::Bind(&BlobMemoryController::OnBlobFileDelete,
+                   ptr_factory_.GetWeakPtr(), file_size));
+  } else {
+    disk_used_ -= file_size;
+  }
+  LOG(ERROR) << "Created file!";
+  file_callback.Run(std::move(result));
+}
+
+void BlobMemoryController::MaybeScheduleWaitingBlobs() {
+  size_t space_available = max_blob_in_memory_size_ - blob_memory_used_;
+  while (!blobs_waiting_for_paging_.empty() &&
+         max_blob_in_memory_size_ - blob_memory_used_ >=
+             blobs_waiting_for_paging_.front().first) {
+    auto size_callback_pair = blobs_waiting_for_paging_.front();
+    blobs_waiting_for_paging_.pop_front();
+    space_available -= size_callback_pair.first;
+    blobs_waiting_for_paging_size_ -= size_callback_pair.first;
+    blob_memory_used_ += size_callback_pair.first;
+    size_callback_pair.second.Run(true);
+  }
+  RecordTracingCounters();
+}
+
+void BlobMemoryController::MaybeSchedulePagingUntilSystemHealthy() {
+  // Don't do paging when others are happening, as we don't change our
+  // blobs_waiting_for_paging_size_ value until after the paging files have
+  // been writen.
+  if (pending_pagings_ != 0 || !disk_enabled_)
+    return;
+
+  // We try to page items to disk until our current system size + requested
+  // memory is below our size limit.
+  while (blobs_waiting_for_paging_size_ + blob_memory_used_ >
+         max_blob_in_memory_size_) {
+    if (!HasEnoughMemoryToPage())
+      break;
+    DCHECK_LT(min_page_file_size_, static_cast<uint64_t>(blob_memory_used_));
+    size_t total_items_size = 0;
+    std::unique_ptr<std::vector<scoped_refptr<ShareableBlobDataItem>>>
+        items_for_disk(new std::vector<scoped_refptr<ShareableBlobDataItem>>());
+    // Collect our items.
+    while (total_items_size < min_page_file_size_ &&
+           !recent_item_cache_.empty()) {
+      auto iterator = --recent_item_cache_.end();
+      ShareableBlobDataItem* item = iterator->second;
+      DCHECK(item);
+      recent_item_cache_.Erase(iterator);
+      recent_item_cache_bytes_ -= static_cast<size_t>(item->item()->length());
+      size_t size = base::checked_cast<size_t>(item->item()->length());
+      total_items_size += size;
+      items_for_disk->push_back(make_scoped_refptr(item));
+    }
+    if (total_items_size == 0)
+      break;
+
+    // Update our bookkeeping.
+    pending_pagings_++;
+    disk_used_ += total_items_size;
+    DCHECK_GE(blob_memory_used_, total_items_size);
+    LOG(ERROR) << "saving " << total_items_size << " to disk.";
+    blob_memory_used_ -= total_items_size;
+    in_flight_memory_used_ += total_items_size;
+    std::string file_name = base::Uint64ToString(current_file_num_++);
+    // Create our file reference.
+    scoped_refptr<ShareableFileReference> file_ref =
+        ShareableFileReference::GetOrCreate(
+            blob_storage_dir_.Append(file_name),
+            ShareableFileReference::DELETE_ON_FINAL_RELEASE,
+            file_runner_.get());
+    // Add the release callback so we decrement our disk usage on file deletion.
+    file_ref->AddFinalReleaseCallback(
+        base::Bind(&BlobMemoryController::OnBlobFileDelete,
+                   ptr_factory_.GetWeakPtr(), total_items_size));
+    // Post the file writing task.
+    base::PostTaskAndReplyWithResult(
+        file_runner_.get(), FROM_HERE,
+        base::Bind(&WriteItemsToFile, items_for_disk.get(), total_items_size,
+                   std::move(file_ref)),
+        base::Bind(&BlobMemoryController::OnPagingComplete,
+                   ptr_factory_.GetWeakPtr(), base::Passed(&items_for_disk),
+                   total_items_size));
+  }
+  RecordTracingCounters();
+}
+
+void BlobMemoryController::OnPagingComplete(
+    std::unique_ptr<std::vector<scoped_refptr<ShareableBlobDataItem>>> items,
+    size_t total_items_size,
+    FileCreationInfo result) {
+  if (!disk_enabled_)
+    return;
+  if (result.error != File::FILE_OK) {
+    UMA_HISTOGRAM_ENUMERATION("Storage.Blob.PagingError", -result.error,
+                              -File::FILE_ERROR_MAX);
+    disk_used_ -= total_items_size;
+    DisableDisk();
+    return;
+  }
+  DCHECK_LT(0u, pending_pagings_);
+  pending_pagings_--;
+
+  // Switch from the data backing item to a new file backing item.
+  uint64_t offset = 0;
+  for (const scoped_refptr<ShareableBlobDataItem>& shareable_item :
+       *items.get()) {
+    scoped_refptr<BlobDataItem> new_item(new BlobDataItem(
+        base::WrapUnique(new DataElement()), result.file_reference));
+    new_item->data_element_ptr()->SetToFilePathRange(
+        result.file_reference->path(), offset, shareable_item->item()->length(),
+        result.last_modified);
+    shareable_item->item_ = new_item;
+    offset += shareable_item->item()->length();
+  }
+  in_flight_memory_used_ -= total_items_size;
+
+  // We want callback on blobs up to the amount we've freed.
+  MaybeScheduleWaitingBlobs();
+
+  // If we still have more blobs waiting and we're not waiting on more paging
+  // operations, schedule more.
+  MaybeSchedulePagingUntilSystemHealthy();
+
+  RecordTracingCounters();
+}
+
+void BlobMemoryController::RecordTracingCounters() {
+  TRACE_COUNTER2("Blob", "MemoryUsage", "RegularStorage", blob_memory_used_,
+                 "InFlightToDisk", in_flight_memory_used_);
+  TRACE_COUNTER1("Blob", "DiskUsage", disk_used_);
+  TRACE_COUNTER1("Blob", "TranfersPendingOnDisk",
+                 blobs_waiting_for_paging_.size());
+  TRACE_COUNTER1("Blob", "TranfersBytesPendingOnDisk",
+                 blobs_waiting_for_paging_size_);
+}
+
+size_t BlobMemoryController::GetAvailableMemoryForBlobs() const {
+  // If disk is enabled, then we include |in_flight_memory_used_|. Otherwise we
+  // combine the in memory and in flight quotas.
+  if (disk_enabled_) {
+    if (max_blob_in_memory_size_ + in_flight_space_ < memory_usage())
+      return 0;
+    return max_blob_in_memory_size_ + in_flight_space_ - memory_usage();
+  }
+  if (max_blob_in_memory_size_ + in_flight_space_ < memory_usage())
+    return 0;
+  return max_blob_in_memory_size_ + in_flight_space_ - memory_usage();
+}
+
+uint64_t BlobMemoryController::GetAvailableDiskSpaceForBlobs() const {
+  return disk_enabled_ ? max_blob_disk_space_ - disk_used_ : 0;
+}
+
+void BlobMemoryController::OnBlobFileDelete(uint64_t size,
+                                            const base::FilePath& path) {
+  DCHECK_LE(size, disk_used_);
+  LOG(ERROR) << "File deleted " << size;
+  disk_used_ -= size;
+}
+
+}  // namespace storage