ServiceWorker: Introduce ServiceWorkerDiskCacheMigrator

content/browser/service_worker/service_worker_disk_cache_migrator.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/browser/service_worker/service_worker_disk_cache_migrator.h"
+
+#include "base/memory/ref_counted.h"
+#include "base/strings/string_number_conversions.h"
+#include "content/common/service_worker/service_worker_types.h"
+#include "net/base/io_buffer.h"
+#include "net/base/net_errors.h"
+#include "net/disk_cache/disk_cache.h"
+
+namespace content {
+
+// A task to move a cached resource from the src DiskCache to the dest
+// DiskCache. This is owned by ServiceWorkerDiskCacheMigrator.
+class ServiceWorkerDiskCacheMigrator::Task
+    : public base::RefCounted<ServiceWorkerDiskCacheMigrator::Task> {

[kinuko, 2015/05/27 05:45:14]

nit: It's not obvious why this needs to be ref-counted?

[nhiroki, 2015/05/27 10:05:27]

I thought the owner of the disk_cache::Entry* should be ref-counted to ensure
that the entry ptr is alive until OnOpenNextEntry via both sync and async
routes, but it was misconception. I made this task non-refcounted.

+ public:
+  explicit Task(const base::WeakPtr<ServiceWorkerDiskCacheMigrator>& owner);
+
+  void Run(int task_id,
+           ServiceWorkerDiskCache* src,
+           ServiceWorkerDiskCache* dest);
+  void Abort();
+
+ private:
+  friend class base::RefCounted<ServiceWorkerDiskCacheMigrator::Task>;
+  friend class ServiceWorkerDiskCacheMigrator;

[kinuko, 2015/05/27 05:45:14]

nit: having this Migrator class friend seems to imply that 'private:'  basically
does nothing.

[nhiroki, 2015/05/27 10:05:27]

These 'friend' are no longer necessary. Removed.

+
+  ~Task();
+
+  void ReadResponseInfo();
+  void OnReadResponseInfo(
+      const scoped_refptr<HttpResponseInfoIOBuffer>& info_buffer,
+      int result);
+  void OnWriteResponseInfo(
+      const scoped_refptr<HttpResponseInfoIOBuffer>& info_buffer,
+      int result);
+  void WriteResponseMetadata(
+      const scoped_refptr<HttpResponseInfoIOBuffer>& info_buffer);
+  void OnWriteResponseMetadata(int result);
+  void ReadResponseData();
+  void OnReadResponseData(const scoped_refptr<net::IOBuffer>& buffer,
+                          int result);
+  void OnWriteResponseData(int result);
+  void DeleteResponse();
+  void OnDeleteResponse(int result);
+  void Finish(ServiceWorkerStatusCode status);
+
+  IDMap<Task>::KeyType task_id_;
+  int64 resource_id_ = kInvalidServiceWorkerResourceId;
+  bool is_aborted_ = false;
+
+  base::WeakPtr<ServiceWorkerDiskCacheMigrator> owner_;
+  ServiceWorkerDiskCache* src_ = nullptr;
+  disk_cache::Entry* entry_ = nullptr;
+
+  scoped_ptr<ServiceWorkerResponseReader> reader_;
+  scoped_ptr<ServiceWorkerResponseWriter> writer_;
+  scoped_ptr<ServiceWorkerResponseMetadataWriter> metadata_writer_;
+
+  DISALLOW_COPY_AND_ASSIGN(Task);
+};
+
+ServiceWorkerDiskCacheMigrator::Task::Task(
+    const base::WeakPtr<ServiceWorkerDiskCacheMigrator>& owner)
+    : owner_(owner) {
+}
+
+ServiceWorkerDiskCacheMigrator::Task::~Task() {
+  if (entry_)
+    entry_->Close();
+}
+
+void ServiceWorkerDiskCacheMigrator::Task::Run(IDMap<Task>::KeyType task_id,
+                                               ServiceWorkerDiskCache* src,
+                                               ServiceWorkerDiskCache* dest) {
+  task_id_ = task_id;
+  src_ = src;
+
+  DCHECK(entry_);
+  if (!base::StringToInt64(entry_->GetKey(), &resource_id_)) {
+    LOG(ERROR) << "Failed to read the resource id";
+    Finish(SERVICE_WORKER_ERROR_FAILED);
+    return;
+  }
+  DCHECK_NE(kInvalidServiceWorkerResourceId, resource_id_);
+
+  reader_.reset(new ServiceWorkerResponseReader(resource_id_, src));
+  writer_.reset(new ServiceWorkerResponseWriter(resource_id_, dest));
+  metadata_writer_.reset(
+      new ServiceWorkerResponseMetadataWriter(resource_id_, dest));
+
+  ReadResponseInfo();
+}
+
+void ServiceWorkerDiskCacheMigrator::Task::Abort() {
+  is_aborted_ = true;
+}
+
+void ServiceWorkerDiskCacheMigrator::Task::ReadResponseInfo() {
+  scoped_refptr<HttpResponseInfoIOBuffer> info_buffer(
+      new HttpResponseInfoIOBuffer);
+  reader_->ReadInfo(info_buffer.get(),
+                    base::Bind(&Task::OnReadResponseInfo, this, info_buffer));
+}
+
+void ServiceWorkerDiskCacheMigrator::Task::OnReadResponseInfo(
+    const scoped_refptr<HttpResponseInfoIOBuffer>& info_buffer,
+    int result) {
+  if (is_aborted_)
+    return;
+  if (result < 0) {
+    LOG(ERROR) << "Failed to read the response info";
+    Finish(SERVICE_WORKER_ERROR_FAILED);
+    return;
+  }
+  writer_->WriteInfo(info_buffer.get(),
+                     base::Bind(&Task::OnWriteResponseInfo, this, info_buffer));
+}
+
+void ServiceWorkerDiskCacheMigrator::Task::OnWriteResponseInfo(
+    const scoped_refptr<HttpResponseInfoIOBuffer>& buffer,
+    int result) {
+  if (is_aborted_)
+    return;
+  if (result < 0) {
+    LOG(ERROR) << "Failed to write the response info";
+    Finish(SERVICE_WORKER_ERROR_FAILED);
+    return;
+  }
+
+  const net::HttpResponseInfo* http_info = buffer->http_info.get();
+  if (http_info->metadata) {
+    WriteResponseMetadata(buffer);
+    return;
+  }
+  ReadResponseData();
+}
+
+void ServiceWorkerDiskCacheMigrator::Task::WriteResponseMetadata(
+    const scoped_refptr<HttpResponseInfoIOBuffer>& info_buffer) {
+  const net::HttpResponseInfo* http_info = info_buffer->http_info.get();
+  const int data_size = http_info->metadata->size();
+
+  scoped_refptr<net::IOBuffer> buffer = new net::IOBuffer(data_size);
+  if (data_size)
+    memmove(buffer->data(), http_info->metadata->data(), data_size);
+  metadata_writer_->WriteMetadata(
+      buffer.get(), data_size,
+      base::Bind(&Task::OnWriteResponseMetadata, this));
+}
+
+void ServiceWorkerDiskCacheMigrator::Task::OnWriteResponseMetadata(int result) {
+  if (is_aborted_)
+    return;
+  if (result < 0) {
+    LOG(ERROR) << "Failed to write the response metadata";
+    Finish(SERVICE_WORKER_ERROR_FAILED);
+    return;
+  }
+  ReadResponseData();
+}
+
+void ServiceWorkerDiskCacheMigrator::Task::ReadResponseData() {
+  scoped_refptr<net::IOBuffer> buffer =
+      new net::IOBuffer(entry_->GetDataSize(0));
+  reader_->ReadData(buffer.get(), entry_->GetDataSize(0),
+                    base::Bind(&Task::OnReadResponseData, this, buffer));
+}
+
+void ServiceWorkerDiskCacheMigrator::Task::OnReadResponseData(
+    const scoped_refptr<net::IOBuffer>& buffer,
+    int result) {
+  if (is_aborted_)
+    return;
+  if (result < 0) {
+    LOG(ERROR) << "Failed to read the response data";
+    Finish(SERVICE_WORKER_ERROR_FAILED);
+    return;
+  }
+  writer_->WriteData(buffer.get(), result,
+                     base::Bind(&Task::OnWriteResponseData, this));
+}
+
+void ServiceWorkerDiskCacheMigrator::Task::OnWriteResponseData(int result) {
+  if (is_aborted_)
+    return;
+  if (result < 0) {
+    LOG(ERROR) << "Failed to write the response data";
+    Finish(SERVICE_WORKER_ERROR_FAILED);
+    return;
+  }
+  DeleteResponse();
+}
+
+void ServiceWorkerDiskCacheMigrator::Task::DeleteResponse() {
+  // Delete the response from the src diskcache so that we can avoid migrating
+  // it twice when remaining tasks are cancelled due to the storage shutdown and
+  // a retry happens.

[kinuko, 2015/05/27 05:45:15]

Do we really want to delete entry one by one while migration is ongoing?
I think we could still writing the same resource twice if unexpected shutdown
happens?

Could we make this migration all or nothing model?  Or how are you planning to
call this migrator to make it 'resumable'?

[nhiroki, 2015/05/27 10:05:27]

Right, half-migrated resources or migrated-but-not-deleted resources could be
overwritten.
At first, I though the resumable model was more stable for (a few) powerful
users who had tons of cached resources. But, at this moment, the number of
stored resources would be small for most of users and the all-or-nothing model
would be sufficient. In addition, issuing a bunch of delete operations could
cause additional corruption...

so, I switched this CL to the all-or-nothing model.
On the PS5, Calling Start() on newly created migrator can resume migration. The
migrator just overwrites existing resources in the dest DiskCache (see
ServiceWorkerDiskCacheMigratorTest::MigrateResponses_RetryAfterAbort).

+  int result =
+      src_->DoomEntry(resource_id_, base::Bind(&Task::OnDeleteResponse, this));
+  if (result == net::ERR_IO_PENDING)
+    return;
+  OnDeleteResponse(result);
+}
+
+void ServiceWorkerDiskCacheMigrator::Task::OnDeleteResponse(int result) {
+  if (is_aborted_)
+    return;
+  if (result < 0) {
+    LOG(ERROR) << "Failed to delete the response";
+    // Ignore an error and try to continue migrating. A leftover response will
+    // be deleted with the diskcache directory when all tasks are completed.
+  }
+  Finish(SERVICE_WORKER_OK);
+}
+
+void ServiceWorkerDiskCacheMigrator::Task::Finish(
+    ServiceWorkerStatusCode status) {
+  if (owner_)
+    owner_->OnResourceMigrated(task_id_, status);
+}
+
+ServiceWorkerDiskCacheMigrator::ServiceWorkerDiskCacheMigrator(
+    ServiceWorkerDiskCache* src,
+    ServiceWorkerDiskCache* dest,
+    const StatusCallback& callback)
+    : src_(src), dest_(dest), callback_(callback), weak_factory_(this) {
+  DCHECK(!src_->is_disabled());
+  DCHECK(!dest_->is_disabled());
+}
+
+ServiceWorkerDiskCacheMigrator::~ServiceWorkerDiskCacheMigrator() {
+}
+
+void ServiceWorkerDiskCacheMigrator::Start() {
+  iterator_ = src_->disk_cache()->CreateIterator();
+  ContinueMigratingResources();
+}
+
+void ServiceWorkerDiskCacheMigrator::ContinueMigratingResources() {
+  scoped_refptr<Task> next_task(new Task(weak_factory_.GetWeakPtr()));

[kinuko, 2015/05/27 05:45:15]

Instead of doing this here and initializing necessary fields at multiple places,
is it possible/sensible to create a new task in OnOpenNextEntry, so that
Task::entry_, task_id_, src_, dest_ all initialized in the ctor?

[nhiroki, 2015/05/27 10:05:27]

Done.

+
+  int result = iterator_->OpenNextEntry(
+      &next_task->entry_,
+      base::Bind(&ServiceWorkerDiskCacheMigrator::OnOpenNextEntry,
+                 weak_factory_.GetWeakPtr(), next_task));
+  if (result == net::ERR_IO_PENDING)
+    return;
+  OnOpenNextEntry(next_task, result);
+}
+
+void ServiceWorkerDiskCacheMigrator::OnOpenNextEntry(
+    const scoped_refptr<Task>& next_task,
+    int result) {
+  if (result == net::ERR_FAILED) {
+    // ERR_FAILED means the iterator reaches the end of the enumeration.
+    if (inflight_tasks_.IsEmpty())
+      callback_.Run(SERVICE_WORKER_OK);
+    return;
+  }
+
+  if (result != net::OK) {
+    LOG(ERROR) << "Failed to open the next entry";
+    callback_.Run(SERVICE_WORKER_ERROR_FAILED);
+    return;
+  }
+
+  IDMap<Task>::KeyType task_id = inflight_tasks_.Add(next_task.get());
+  next_task->Run(task_id, src_, dest_);
+
+  ContinueMigratingResources();
+}
+
+void ServiceWorkerDiskCacheMigrator::OnResourceMigrated(
+    IDMap<Task>::KeyType task_id,
+    ServiceWorkerStatusCode status) {
+  DCHECK(inflight_tasks_.Lookup(task_id));
+  inflight_tasks_.Remove(task_id);
+
+  if (status != SERVICE_WORKER_OK) {
+    AbortAllTasks();
+    callback_.Run(status);
+    return;
+  }
+
+  if (inflight_tasks_.IsEmpty()) {
+    callback_.Run(SERVICE_WORKER_OK);
+    return;
+  }
+
+  // |callback_| will be invoked when all inflight tasks are completed.
+}
+
+void ServiceWorkerDiskCacheMigrator::AbortAllTasks() {
+  for (IDMap<Task>::iterator it(&inflight_tasks_); !it.IsAtEnd();
+       it.Advance()) {
+    Task* task = it.GetCurrentValue();
+    DCHECK(task);
+    task->Abort();
+  }
+  inflight_tasks_.Clear();
+}
+
+}  // namespace content