[BlobAsync] Disk support for blob storage

storage/browser/blob/blob_memory_controller.cc

Index: storage/browser/blob/blob_memory_controller.cc
diff --git a/storage/browser/blob/blob_memory_controller.cc b/storage/browser/blob/blob_memory_controller.cc
new file mode 100644
index 0000000000000000000000000000000000000000..834dc56076da638c5cf2a5cb9d3a41f7cba3e25b
--- /dev/null
+++ b/storage/browser/blob/blob_memory_controller.cc
@@ -0,0 +1,505 @@
+// 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/threading/sequenced_worker_pool.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"
+
+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 -= static_cast<size_t>(bytes_written);

[Marijn Kruisselbrink, 2016/07/12 21:33:06]

I don't think you need the static_cast here.

[dmurph, 2016/07/14 01:04:31]

Done.

+    }
+    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_(RecentItemsCache::NO_AUTO_EVICT), ptr_factory_(this) {}
+
+BlobMemoryController::~BlobMemoryController() {}
+
+void BlobMemoryController::EnableDisk(
+    const base::FilePath& storage_directory,
+    scoped_refptr<base::SequencedWorkerPool> file_worker_pool) {
+  LOG(ERROR) << "enbling disk";
+  file_worker_pool_ = std::move(file_worker_pool);
+  blob_storage_dir_ = storage_directory;
+  enable_disk_ = true;
+}
+
+void BlobMemoryController::DisableDisk() {
+  enable_disk_ = 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();
+  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() &&
+      total_memory_size_bytes > 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::SHORTCUT;
+    return true;
+  }
+
+  // Step 5: Decide if we're going straight to disk.
+  if (enable_disk_ && (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::CreateTemporaryFileForRenderer(
+    uint64_t size_bytes,
+    const base::Callback<void(FileCreationInfo)>& file_callback) {
+  if (!enable_disk_) {
+    BlobMemoryController::FileCreationInfo creation_info;
+    file_callback.Run(std::move(creation_info));
+    return;
+  }
+
+  disk_used_ += size_bytes;

[michaeln, 2016/07/14 01:44:43]

and this "allocates" memory on disk

[dmurph, 2016/07/15 02:45:27]

Added more documentation.

+  std::string file_name = base::Uint64ToString(current_file_num_++);
+  scoped_refptr<base::TaskRunner> file_runner =
+      file_worker_pool_->GetTaskRunnerWithShutdownBehavior(
+          base::SequencedWorkerPool::SKIP_ON_SHUTDOWN);
+  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));
+}
+
+void BlobMemoryController::FreeMemory(size_t memory_size_bytes) {
+  DCHECK_GE(blob_memory_used_, memory_size_bytes);
+  LOG(ERROR) << "Freeing memory " << memory_size_bytes;
+  blob_memory_used_ -= memory_size_bytes;
+  MaybeScheduleWaitingBlobs();
+}
+
+base::Optional<BlobMemoryController::PendingConstructionEntry>
+BlobMemoryController::NotifyWhenMemoryCanPopulated(
+    size_t memory_size,
+    const base::Callback<void(bool)>& can_request_callback) {
+  DCHECK(memory_size <=
+         GetAvailableMemoryForBlobs() + GetAvailableDiskSpaceForBlobs());
+
+  if (!enable_disk_) {
+    LOG(ERROR) << "Yes, " << memory_size << " can fit in memory now.";
+    blob_memory_used_ += memory_size;

[michaeln, 2016/07/14 01:44:43]

ok, this method "allocates" the memory

[dmurph, 2016/07/15 02:45:27]

Added more documentation.

+    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;
+    return base::make_optional(--blobs_waiting_for_paging_.end());
+  }
+
+  // Store right away if we can.
+  if (memory_size <= GetAvailableMemoryForBlobs()) {
+    // If we're past our blob memory limit, then schedule our paging.
+    if (ShouldSchedulePagingForSize(memory_size)) {
+      LOG(ERROR) << "Scheduling premptive paging";
+      ScheduleBlobPaging();
+    }
+    LOG(ERROR) << "Yes, " << memory_size << " can fit in memory now.";
+    blob_memory_used_ += memory_size;
+    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;
+  LOG(ERROR) << "Scheduling paging on first item too big";
+  SchedulePagingUntilWeCanFit(memory_size);
+  return base::make_optional(--blobs_waiting_for_paging_.end());
+}
+
+void BlobMemoryController::RemovePendingConstructionEntry(
+    const BlobMemoryController::PendingConstructionEntry& entry) {
+  const std::pair<size_t, base::Callback<void(bool)>> pair = *entry;

[Marijn Kruisselbrink, 2016/07/12 21:33:06]

I'm not sure I quite get why you're making of a copy of the thing entry points
to?

Also not sure what this method does, and where it is called from (if anywhere).

[dmurph, 2016/07/14 01:04:31]

I forgot to remove the entry from the list.

+  blobs_waiting_for_paging_size_ -= pair.first;
+}
+
+void BlobMemoryController::UpdateBlobItemInRecents(
+    ShareableBlobDataItem* item) {
+  recent_item_cache_.Put(item->item_id(), item);
+}
+
+void BlobMemoryController::RemoveBlobItemInRecents(
+    const ShareableBlobDataItem& item) {
+  auto iterator = recent_item_cache_.Get(item.item_id());
+  if (iterator != recent_item_cache_.end())
+    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() &&
+         space_available >= 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);
+  }
+}
+
+size_t BlobMemoryController::ScheduleBlobPaging() {
+  DCHECK(enable_disk_);
+  DCHECK_LT(min_page_file_size_, static_cast<uint64_t>(blob_memory_used_));
+  LOG(ERROR) << "scheduling paging";
+
+  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);
+    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)
+    return 0;
+
+  // 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_++);
+  scoped_refptr<base::TaskRunner> file_runner =
+      file_worker_pool_->GetTaskRunnerWithShutdownBehavior(
+          base::SequencedWorkerPool::SKIP_ON_SHUTDOWN);
+  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(&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));
+  return total_items_size;
+}
+
+void BlobMemoryController::SchedulePagingUntilWeCanFit(
+    size_t total_memory_needed) {
+  DCHECK_LT(total_memory_needed, max_blob_in_memory_size_);
+  while (total_memory_needed + blob_memory_used_ > max_blob_in_memory_size_) {
+    // schedule another page.
+    size_t memory_freeing = ScheduleBlobPaging();
+    DCHECK_GE(blob_memory_used_, memory_freeing);
+    if (memory_freeing == 0)
+      return;
+  }
+}
+
+void BlobMemoryController::OnPagingComplete(
+    std::unique_ptr<std::vector<scoped_refptr<ShareableBlobDataItem>>> items,
+    size_t total_items_size,
+    FileCreationInfo result) {
+  if (!enable_disk_)
+    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.
+  if (!blobs_waiting_for_paging_.empty() && pending_pagings_ == 0) {
+    SchedulePagingUntilWeCanFit(blobs_waiting_for_paging_size_);
+  }
+}
+
+void BlobMemoryController::RecordTracingCounters() {
+  TRACE_COUNTER2("Blob", "MemoryUsage", "RegularStorage", blob_memory_used_,
+                 "InFlightToDisk", in_flight_memory_used_);
+  TRACE_COUNTER1("Blob", "TranfersPendingOnDisk",
+                 blobs_waiting_for_paging_.size());
+  TRACE_COUNTER1("Blob", "TranfersBytesPendingOnDisk",
+                 blobs_waiting_for_paging_size_);
+}
+
+bool BlobMemoryController::ShouldSchedulePagingForSize(size_t size) const {
+  return blob_memory_used_ + size - in_flight_memory_used_ >
+         max_blob_in_memory_size_;
+}
+
+bool BlobMemoryController::CanFitInSystem(uint64_t size) const {
+  return size < GetAvailableMemoryForBlobs() + GetAvailableDiskSpaceForBlobs();
+}
+
+size_t BlobMemoryController::GetAvailableMemoryForBlobs() const {
+  if (blob_memory_used_ >= max_blob_in_memory_size_)
+    return 0;
+  if (enable_disk_) {
+    if (max_blob_in_memory_size_ + in_flight_space_ <
+        blob_memory_used_ + in_flight_memory_used_) {
+      return 0;
+    }
+    return max_blob_in_memory_size_ + in_flight_space_ - blob_memory_used_ -
+           in_flight_memory_used_;
+  }
+  return max_blob_memory_space_ - blob_memory_used_;
+}
+
+uint64_t BlobMemoryController::GetAvailableDiskSpaceForBlobs() const {
+  return enable_disk_ ? kBlobStorageMaxDiskSpace - disk_used_ : 0;
+}
+
+void BlobMemoryController::OnBlobFileDelete(uint64_t size,
+                                            const base::FilePath& path) {
+  DCHECK_LE(size, disk_used_);
+  LOG(ERROR) << "deleting file " << path.value();
+  disk_used_ -= size;
+}
+
+}  // namespace storage