[ServiceWorkerCache] Make the cache operations run serially.

content/browser/service_worker/service_worker_cache.cc

 // Copyright 2014 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_cache.h"
 
 #include <string>
 
 #include "base/files/file_path.h"
 #include "base/guid.h"
@@ skipping 410 matching lines @@
 ServiceWorkerCache::~ServiceWorkerCache() {
 }
 
 base::WeakPtr<ServiceWorkerCache> ServiceWorkerCache::AsWeakPtr() {
   return weak_ptr_factory_.GetWeakPtr();
 }
 
 void ServiceWorkerCache::Put(scoped_ptr<ServiceWorkerFetchRequest> request,
                              scoped_ptr<ServiceWorkerResponse> response,
                              const ResponseCallback& callback) {
-  IncPendingOps();
-  ResponseCallback pending_callback =
-      base::Bind(&ServiceWorkerCache::PendingResponseCallback,
-                 weak_ptr_factory_.GetWeakPtr(), callback);
   scoped_ptr<storage::BlobDataHandle> blob_data_handle;
 
   if (!response->blob_uuid.empty()) {
     if (!blob_storage_context_) {
-      pending_callback.Run(ErrorTypeStorage,
-                           scoped_ptr<ServiceWorkerResponse>(),
-                           scoped_ptr<storage::BlobDataHandle>());
+      callback.Run(ErrorTypeStorage, scoped_ptr<ServiceWorkerResponse>(),
+                   scoped_ptr<storage::BlobDataHandle>());
       return;
     }
     blob_data_handle =
         blob_storage_context_->GetBlobDataFromUUID(response->blob_uuid);
     if (!blob_data_handle) {
-      pending_callback.Run(ErrorTypeStorage,
-                           scoped_ptr<ServiceWorkerResponse>(),
-                           scoped_ptr<storage::BlobDataHandle>());
+      callback.Run(ErrorTypeStorage, scoped_ptr<ServiceWorkerResponse>(),
+                   scoped_ptr<storage::BlobDataHandle>());
       return;
     }
   }
 
+  ResponseCallback pending_callback =
+      base::Bind(&ServiceWorkerCache::PendingResponseCallback,
+                 weak_ptr_factory_.GetWeakPtr(), callback);
+
   scoped_ptr<PutContext> put_context(new PutContext(
       origin_, request.Pass(), response.Pass(), blob_data_handle.Pass(),
       pending_callback, request_context_, quota_manager_proxy_));
 
   if (put_context->blob_data_handle) {
     // Grab another handle to the blob for the callback response.
     put_context->out_blob_data_handle =
         blob_storage_context_->GetBlobDataFromUUID(
             put_context->response->blob_uuid);
   }
 
-  base::Closure continuation =
-      base::Bind(&ServiceWorkerCache::PutImpl, weak_ptr_factory_.GetWeakPtr(),
-                 base::Passed(put_context.Pass()));
+  pending_operations_.push_back(base::Bind(&ServiceWorkerCache::PutImpl,
+                                           weak_ptr_factory_.GetWeakPtr(),
+                                           base::Passed(put_context.Pass())));
 
   if (backend_state_ == BACKEND_UNINITIALIZED) {
-    InitBackend(continuation);
+    InitBackend(pending_operations_.back());
     return;
   }
 
-  continuation.Run();
+  RunOperationIfIdle();
 }
 
 void ServiceWorkerCache::Match(scoped_ptr<ServiceWorkerFetchRequest> request,
                                const ResponseCallback& callback) {
-  IncPendingOps();
   ResponseCallback pending_callback =
       base::Bind(&ServiceWorkerCache::PendingResponseCallback,
                  weak_ptr_factory_.GetWeakPtr(), callback);
+  pending_operations_.push_back(
+      base::Bind(&ServiceWorkerCache::MatchImpl, weak_ptr_factory_.GetWeakPtr(),
+                 base::Passed(request.Pass()), pending_callback));
 
   switch (backend_state_) {
     case BACKEND_UNINITIALIZED:
-      InitBackend(base::Bind(&ServiceWorkerCache::Match,
-                             weak_ptr_factory_.GetWeakPtr(),
-                             base::Passed(request.Pass()), pending_callback));
+      InitBackend(pending_operations_.back());
       return;
     case BACKEND_CLOSED:
       pending_callback.Run(ErrorTypeStorage,
                            scoped_ptr<ServiceWorkerResponse>(),
                            scoped_ptr<storage::BlobDataHandle>());
       return;
     case BACKEND_OPEN:
       DCHECK(backend_);
-      break;
+      RunOperationIfIdle();
+      return;
   }
-
-  scoped_ptr<MatchContext> match_context(new MatchContext(
-      request.Pass(), pending_callback, blob_storage_context_));
-
-  disk_cache::Entry** entry_ptr = &match_context->entry;
-  ServiceWorkerFetchRequest* request_ptr = match_context->request.get();
-
-  net::CompletionCallback open_entry_callback = base::Bind(
-      &ServiceWorkerCache::MatchDidOpenEntry, weak_ptr_factory_.GetWeakPtr(),
-      base::Passed(match_context.Pass()));
-
-  int rv = backend_->OpenEntry(
-      request_ptr->url.spec(), entry_ptr, open_entry_callback);
-  if (rv != net::ERR_IO_PENDING)
-    open_entry_callback.Run(rv);
+  NOTREACHED();
 }
 
 void ServiceWorkerCache::Delete(scoped_ptr<ServiceWorkerFetchRequest> request,
                                 const ErrorCallback& callback) {
-  IncPendingOps();
   ErrorCallback pending_callback =
       base::Bind(&ServiceWorkerCache::PendingErrorCallback,
                  weak_ptr_factory_.GetWeakPtr(), callback);
+  pending_operations_.push_back(base::Bind(
+      &ServiceWorkerCache::DeleteImpl, weak_ptr_factory_.GetWeakPtr(),
+      base::Passed(request.Pass()), pending_callback));
 
   switch (backend_state_) {
     case BACKEND_UNINITIALIZED:
-      InitBackend(base::Bind(&ServiceWorkerCache::Delete,
-                             weak_ptr_factory_.GetWeakPtr(),
-                             base::Passed(request.Pass()), pending_callback));
+      InitBackend(pending_operations_.back());
       return;
     case BACKEND_CLOSED:
       pending_callback.Run(ErrorTypeStorage);
       return;
     case BACKEND_OPEN:
       DCHECK(backend_);
-      break;
+      RunOperationIfIdle();
+      return;
   }
-
-  scoped_ptr<disk_cache::Entry*> entry(new disk_cache::Entry*);
-
-  disk_cache::Entry** entry_ptr = entry.get();
-
-  ServiceWorkerFetchRequest* request_ptr = request.get();
-
-  net::CompletionCallback open_entry_callback = base::Bind(
-      &ServiceWorkerCache::DeleteDidOpenEntry, weak_ptr_factory_.GetWeakPtr(),
-      origin_, base::Passed(request.Pass()), pending_callback,
-      base::Passed(entry.Pass()), quota_manager_proxy_);
-
-  int rv = backend_->OpenEntry(
-      request_ptr->url.spec(), entry_ptr, open_entry_callback);
-  if (rv != net::ERR_IO_PENDING)
-    open_entry_callback.Run(rv);
+  NOTREACHED();
 }
 
 void ServiceWorkerCache::Keys(const RequestsCallback& callback) {
-  IncPendingOps();
   RequestsCallback pending_callback =
       base::Bind(&ServiceWorkerCache::PendingRequestsCallback,
                  weak_ptr_factory_.GetWeakPtr(), callback);
+  pending_operations_.push_back(base::Bind(&ServiceWorkerCache::KeysImpl,
+                                           weak_ptr_factory_.GetWeakPtr(),
+                                           pending_callback));
 
   switch (backend_state_) {
     case BACKEND_UNINITIALIZED:
-      InitBackend(base::Bind(&ServiceWorkerCache::Keys,
-                             weak_ptr_factory_.GetWeakPtr(), pending_callback));
+      InitBackend(pending_operations_.back());
       return;
     case BACKEND_CLOSED:
       pending_callback.Run(ErrorTypeStorage, scoped_ptr<Requests>());
       return;
     case BACKEND_OPEN:
       DCHECK(backend_);
-      break;
+      RunOperationIfIdle();
+      return;
   }
-
-  // 1. Iterate through all of the entries, open them, and add them to a vector.
-  // 2. For each open entry:
-  //  2.1. Read the headers into a protobuf.
-  //  2.2. Copy the protobuf into a ServiceWorkerFetchRequest (a "key").
-  //  2.3. Push the response into a vector of requests to be returned.
-  // 3. Return the vector of requests (keys).
-
-  // The entries have to be loaded into a vector first because enumeration loops
-  // forever if you read data from a cache entry while enumerating.
-
-  scoped_ptr<KeysContext> keys_context(new KeysContext(pending_callback));
-
-  keys_context->backend_iterator = backend_->CreateIterator();
-  disk_cache::Backend::Iterator& iterator = *keys_context->backend_iterator;
-  disk_cache::Entry** enumerated_entry = &keys_context->enumerated_entry;
-
-  net::CompletionCallback open_entry_callback = base::Bind(
-      &ServiceWorkerCache::KeysDidOpenNextEntry, weak_ptr_factory_.GetWeakPtr(),
-      base::Passed(keys_context.Pass()));
-
-  int rv = iterator.OpenNextEntry(enumerated_entry, open_entry_callback);
-
-  if (rv != net::ERR_IO_PENDING)
-    open_entry_callback.Run(rv);
+  NOTREACHED();
 }
 
 void ServiceWorkerCache::Close(const base::Closure& callback) {
   DCHECK(backend_state_ != BACKEND_CLOSED)
       << "Don't call ServiceWorkerCache::Close() twice.";
 
   backend_state_ = BACKEND_CLOSED;
-  if (pending_ops_ > 0) {
-    DCHECK(ops_complete_callback_.is_null());
-    ops_complete_callback_ =
-        base::Bind(&ServiceWorkerCache::CloseImpl,
-                   weak_ptr_factory_.GetWeakPtr(), callback);
-    return;
-  }
+  base::Closure pending_callback =
+      base::Bind(&ServiceWorkerCache::PendingClosure,
+                 weak_ptr_factory_.GetWeakPtr(), callback);
 
-  CloseImpl(callback);
+  pending_operations_.push_back(base::Bind(&ServiceWorkerCache::CloseImpl,
+                                           weak_ptr_factory_.GetWeakPtr(),
+                                           pending_callback));
+  RunOperationIfIdle();
 }
 
 int64 ServiceWorkerCache::MemoryBackedSize() const {
   if (backend_state_ != BACKEND_OPEN || !memory_only_)
     return 0;
 
   scoped_ptr<disk_cache::Backend::Iterator> backend_iter =
       backend_->CreateIterator();
   disk_cache::Entry* entry = nullptr;
 
@@ skipping 22 matching lines @@
     const base::FilePath& path,
     net::URLRequestContext* request_context,
     const scoped_refptr<storage::QuotaManagerProxy>& quota_manager_proxy,
     base::WeakPtr<storage::BlobStorageContext> blob_context)
     : origin_(origin),
       path_(path),
       request_context_(request_context),
       quota_manager_proxy_(quota_manager_proxy),
       blob_storage_context_(blob_context),
       backend_state_(BACKEND_UNINITIALIZED),
+      initializing_(false),
       memory_only_(path.empty()),
-      pending_ops_(0),
       weak_ptr_factory_(this) {
 }
 
+void ServiceWorkerCache::MatchImpl(
+    scoped_ptr<ServiceWorkerFetchRequest> request,
+    const ResponseCallback& callback) {
+  DCHECK(backend_state_ != BACKEND_UNINITIALIZED);
+  if (backend_state_ != BACKEND_OPEN) {
+    callback.Run(ErrorTypeStorage, scoped_ptr<ServiceWorkerResponse>(),
+                 scoped_ptr<storage::BlobDataHandle>());
+    return;
+  }
+
+  scoped_ptr<MatchContext> match_context(
+      new MatchContext(request.Pass(), callback, blob_storage_context_));
+
+  disk_cache::Entry** entry_ptr = &match_context->entry;
+  ServiceWorkerFetchRequest* request_ptr = match_context->request.get();
+
+  net::CompletionCallback open_entry_callback = base::Bind(
+      &ServiceWorkerCache::MatchDidOpenEntry, weak_ptr_factory_.GetWeakPtr(),
+      base::Passed(match_context.Pass()));
+
+  int rv = backend_->OpenEntry(request_ptr->url.spec(), entry_ptr,
+                               open_entry_callback);
+  if (rv != net::ERR_IO_PENDING)
+    open_entry_callback.Run(rv);
+}
+
 void ServiceWorkerCache::MatchDidOpenEntry(
     scoped_ptr<MatchContext> match_context,
     int rv) {
   if (rv != net::OK) {
     match_context->original_callback.Run(ServiceWorkerCache::ErrorTypeNotFound,
                                          scoped_ptr<ServiceWorkerResponse>(),
                                          scoped_ptr<storage::BlobDataHandle>());
     return;
   }
 
@@ skipping 141 matching lines @@
   scoped_ptr<storage::BlobDataHandle> blob_data_handle(
       match_context->blob_storage_context->AddFinishedBlob(
           match_context->blob_data.get()));
 
   match_context->original_callback.Run(ServiceWorkerCache::ErrorTypeOK,
                                        match_context->response.Pass(),
                                        blob_data_handle.Pass());
 }
 
 void ServiceWorkerCache::PutImpl(scoped_ptr<PutContext> put_context) {
+  DCHECK(backend_state_ != BACKEND_UNINITIALIZED);
   if (backend_state_ != BACKEND_OPEN) {
     put_context->callback.Run(ErrorTypeStorage,
                               scoped_ptr<ServiceWorkerResponse>(),
                               scoped_ptr<storage::BlobDataHandle>());
     return;
   }
 
   scoped_ptr<ServiceWorkerFetchRequest> request_copy(
       new ServiceWorkerFetchRequest(*put_context->request));
 
-  Delete(request_copy.Pass(), base::Bind(&ServiceWorkerCache::PutDidDelete,
-                                         weak_ptr_factory_.GetWeakPtr(),
-                                         base::Passed(put_context.Pass())));
+  DeleteImpl(request_copy.Pass(), base::Bind(&ServiceWorkerCache::PutDidDelete,
+                                             weak_ptr_factory_.GetWeakPtr(),
+                                             base::Passed(put_context.Pass())));
 }
 
 void ServiceWorkerCache::PutDidDelete(scoped_ptr<PutContext> put_context,
                                       ErrorType delete_error) {
   if (backend_state_ != BACKEND_OPEN) {
     put_context->callback.Run(ErrorTypeStorage,
                               scoped_ptr<ServiceWorkerResponse>(),
                               scoped_ptr<storage::BlobDataHandle>());
     return;
   }
@@ skipping 152 matching lines @@
         storage::kStorageTypeTemporary,
         put_context->cache_entry->GetDataSize(INDEX_HEADERS) +
             put_context->cache_entry->GetDataSize(INDEX_RESPONSE_BODY));
   }
 
   put_context->callback.Run(ServiceWorkerCache::ErrorTypeOK,
                             put_context->response.Pass(),
                             put_context->out_blob_data_handle.Pass());
 }
 
+void ServiceWorkerCache::DeleteImpl(
+    scoped_ptr<ServiceWorkerFetchRequest> request,
+    const ErrorCallback& callback) {
+  DCHECK(backend_state_ != BACKEND_UNINITIALIZED);
+  if (backend_state_ != BACKEND_OPEN) {
+    callback.Run(ErrorTypeStorage);
+    return;
+  }
+  scoped_ptr<disk_cache::Entry*> entry(new disk_cache::Entry*);
+
+  disk_cache::Entry** entry_ptr = entry.get();
+
+  ServiceWorkerFetchRequest* request_ptr = request.get();
+
+  net::CompletionCallback open_entry_callback = base::Bind(
+      &ServiceWorkerCache::DeleteDidOpenEntry, weak_ptr_factory_.GetWeakPtr(),
+      origin_, base::Passed(request.Pass()), callback,
+      base::Passed(entry.Pass()), quota_manager_proxy_);
+
+  int rv = backend_->OpenEntry(request_ptr->url.spec(), entry_ptr,
+                               open_entry_callback);
+  if (rv != net::ERR_IO_PENDING)
+    open_entry_callback.Run(rv);
+}
+
 void ServiceWorkerCache::DeleteDidOpenEntry(
     const GURL& origin,
     scoped_ptr<ServiceWorkerFetchRequest> request,
     const ServiceWorkerCache::ErrorCallback& callback,
     scoped_ptr<disk_cache::Entry*> entry_ptr,
     const scoped_refptr<storage::QuotaManagerProxy>& quota_manager_proxy,
     int rv) {
   if (rv != net::OK) {
     callback.Run(ServiceWorkerCache::ErrorTypeNotFound);
     return;
   }
 
   DCHECK(entry_ptr);
   disk_cache::ScopedEntryPtr entry(*entry_ptr);
 
   if (quota_manager_proxy.get()) {
     quota_manager_proxy->NotifyStorageModified(
         storage::QuotaClient::kServiceWorkerCache, origin,
         storage::kStorageTypeTemporary,
         -1 * (entry->GetDataSize(INDEX_HEADERS) +
               entry->GetDataSize(INDEX_RESPONSE_BODY)));
   }
 
   entry->Doom();
   callback.Run(ServiceWorkerCache::ErrorTypeOK);
 }
 
+void ServiceWorkerCache::KeysImpl(const RequestsCallback& callback) {
+  DCHECK(backend_state_ != BACKEND_UNINITIALIZED);
+  if (backend_state_ != BACKEND_OPEN) {
+    callback.Run(ErrorTypeStorage, scoped_ptr<Requests>());
+    return;
+  }
+
+  // 1. Iterate through all of the entries, open them, and add them to a vector.
+  // 2. For each open entry:
+  //  2.1. Read the headers into a protobuf.
+  //  2.2. Copy the protobuf into a ServiceWorkerFetchRequest (a "key").
+  //  2.3. Push the response into a vector of requests to be returned.
+  // 3. Return the vector of requests (keys).
+
+  // The entries have to be loaded into a vector first because enumeration loops
+  // forever if you read data from a cache entry while enumerating.
+
+  scoped_ptr<KeysContext> keys_context(new KeysContext(callback));
+
+  keys_context->backend_iterator = backend_->CreateIterator();
+  disk_cache::Backend::Iterator& iterator = *keys_context->backend_iterator;
+  disk_cache::Entry** enumerated_entry = &keys_context->enumerated_entry;
+
+  net::CompletionCallback open_entry_callback = base::Bind(
+      &ServiceWorkerCache::KeysDidOpenNextEntry, weak_ptr_factory_.GetWeakPtr(),
+      base::Passed(keys_context.Pass()));
+
+  int rv = iterator.OpenNextEntry(enumerated_entry, open_entry_callback);
+
+  if (rv != net::ERR_IO_PENDING)
+    open_entry_callback.Run(rv);
+}
+
 void ServiceWorkerCache::KeysDidOpenNextEntry(
     scoped_ptr<KeysContext> keys_context,
     int rv) {
   if (rv == net::ERR_FAILED) {
     DCHECK(!keys_context->enumerated_entry);
     // Enumeration is complete, extract the requests from the entries.
     Entries::iterator iter = keys_context->entries.begin();
     KeysProcessNextEntry(keys_context.Pass(), iter);
     return;
   }
@@ skipping 65 matching lines @@
     }
   } else {
     entry->Doom();
   }
 
   KeysProcessNextEntry(keys_context.Pass(), iter + 1);
 }
 
 void ServiceWorkerCache::CloseImpl(const base::Closure& callback) {
   DCHECK(backend_state_ == BACKEND_CLOSED);
+  DCHECK(pending_operations_.size() == 1u);
   backend_.reset();
   callback.Run();
 }
 
 void ServiceWorkerCache::CreateBackend(const ErrorCallback& callback) {
   DCHECK(!backend_);
 
   // Use APP_CACHE as opposed to DISK_CACHE to prevent cache eviction.
   net::CacheType cache_type = memory_only_ ? net::MEMORY_CACHE : net::APP_CACHE;
 
@@ skipping 29 matching lines @@
     int rv) {
   if (rv != net::OK) {
     callback.Run(ServiceWorkerCache::ErrorTypeStorage);
     return;
   }
 
   backend_ = backend_ptr->Pass();
   callback.Run(ServiceWorkerCache::ErrorTypeOK);
 }
 
 void ServiceWorkerCache::InitBackend(const base::Closure& callback) {

[michaeln, 2015/01/23 00:55:25]

The contract is a little odd, the 'callback' provided by the first caller to
InitBackend runs, others are ignored. Now that you have an explicit Q of pending
ops, maybe you don't need the callback param anymore. Would it make sense to
have InitDone run the front most op?

[jkarlin, 2015/01/23 18:13:42]

Done.

   DCHECK(backend_state_ == BACKEND_UNINITIALIZED);
-  init_callbacks_.push_back(callback);
 
-  // If this isn't the first call to Init then return as the initialization
-  // has already started.
-  if (init_callbacks_.size() > 1u)
+  if (initializing_)
     return;
 
+  initializing_ = true;
   CreateBackend(base::Bind(&ServiceWorkerCache::InitDone,
-                           weak_ptr_factory_.GetWeakPtr()));
+                           weak_ptr_factory_.GetWeakPtr(), callback));
 }
 
-void ServiceWorkerCache::InitDone(ErrorType error) {
+void ServiceWorkerCache::InitDone(const base::Closure& callback,
+                                  ErrorType error) {
+  initializing_ = false;
   backend_state_ = (error == ErrorTypeOK && backend_ &&
                     backend_state_ == BACKEND_UNINITIALIZED)
                        ? BACKEND_OPEN
                        : BACKEND_CLOSED;
-  for (std::vector<base::Closure>::iterator it = init_callbacks_.begin();
-       it != init_callbacks_.end(); ++it) {
-    it->Run();
-  }
-  init_callbacks_.clear();
+  callback.Run();
 }
 
-void ServiceWorkerCache::DecPendingOps() {
-  DCHECK(pending_ops_ > 0);
-  pending_ops_--;
-  if (pending_ops_ == 0 && !ops_complete_callback_.is_null()) {
-    ops_complete_callback_.Run();
-    ops_complete_callback_.Reset();
-  }
+void ServiceWorkerCache::CompleteOperationAndRunNext() {
+  DCHECK(!pending_operations_.empty());
+
+  pending_operations_.pop_front();
+
+  // TODO(jkarlin): Run multiple operations in parallel where allowed (e.g., if
+  // they're for different keys then they won't interfere). See
+  // https://crbug.com/451174.
+  if (!pending_operations_.empty())
+    pending_operations_.front().Run();
+}
+
+void ServiceWorkerCache::RunOperationIfIdle() {

[michaeln, 2015/01/23 00:55:25]

nit: this method is a little dangerous, if called twice in a row it will have
two operations in the running, maybe consider using a flag know when an op is
running, you could call RunOperationIfIdle() from within
CompleteOperationAndRunNext() and InitDone().

[jkarlin, 2015/01/23 18:13:42]

Done.

+  if (pending_operations_.size() == 1u)
+    pending_operations_.front().Run();
+}
+
+void ServiceWorkerCache::PendingClosure(const base::Closure& callback) {
+  callback.Run();
+  CompleteOperationAndRunNext();
 }
 
 void ServiceWorkerCache::PendingErrorCallback(const ErrorCallback& callback,
                                               ErrorType error) {
   callback.Run(error);
-  DecPendingOps();
+  CompleteOperationAndRunNext();
 }
 
 void ServiceWorkerCache::PendingResponseCallback(
     const ResponseCallback& callback,
     ErrorType error,
     scoped_ptr<ServiceWorkerResponse> response,
     scoped_ptr<storage::BlobDataHandle> blob_data_handle) {
   callback.Run(error, response.Pass(), blob_data_handle.Pass());
-  DecPendingOps();
+  CompleteOperationAndRunNext();
 }
 
 void ServiceWorkerCache::PendingRequestsCallback(
     const RequestsCallback& callback,
     ErrorType error,
     scoped_ptr<Requests> requests) {
   callback.Run(error, requests.Pass());
-  DecPendingOps();
+  CompleteOperationAndRunNext();
 }
 
 }  // namespace content