[ServiceWorkerCache] Make the cache operations run serially.

content/browser/service_worker/service_worker_cache.cc

Index: content/browser/service_worker/service_worker_cache.cc
diff --git a/content/browser/service_worker/service_worker_cache.cc b/content/browser/service_worker/service_worker_cache.cc
index 3011356f4f17f3b81f8ae0867ae561e265da2fd2..0fe4c9e656aa8db9841708ce76e0b647410cbbaf 100644
--- a/content/browser/service_worker/service_worker_cache.cc
+++ b/content/browser/service_worker/service_worker_cache.cc
@@ -428,29 +428,27 @@ base::WeakPtr<ServiceWorkerCache> ServiceWorkerCache::AsWeakPtr() {
 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_));
@@ -462,30 +460,30 @@ void ServiceWorkerCache::Put(scoped_ptr<ServiceWorkerFetchRequest> request,
             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,
@@ -494,106 +492,57 @@ void ServiceWorkerCache::Match(scoped_ptr<ServiceWorkerFetchRequest> request,
       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) {
@@ -601,15 +550,14 @@ void ServiceWorkerCache::Close(const base::Closure& callback) {
       << "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 {
@@ -652,11 +600,37 @@ ServiceWorkerCache::ServiceWorkerCache(
       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) {
@@ -818,6 +792,7 @@ void ServiceWorkerCache::MatchDoneWithBody(
 }
 
 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>(),
@@ -828,9 +803,9 @@ void ServiceWorkerCache::PutImpl(scoped_ptr<PutContext> put_context) {
   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,
@@ -1003,6 +978,31 @@ void ServiceWorkerCache::PutDidWriteBlobToCache(
                             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,
@@ -1030,6 +1030,39 @@ void ServiceWorkerCache::DeleteDidOpenEntry(
   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) {
@@ -1115,6 +1148,7 @@ void ServiceWorkerCache::KeysDidReadMetadata(
 
 void ServiceWorkerCache::CloseImpl(const base::Closure& callback) {
   DCHECK(backend_state_ == BACKEND_CLOSED);
+  DCHECK(pending_operations_.size() == 1u);
   backend_.reset();
   callback.Run();
 }
@@ -1166,42 +1200,51 @@ void ServiceWorkerCache::CreateBackendDidCreate(
 
 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(
@@ -1210,7 +1253,7 @@ void ServiceWorkerCache::PendingResponseCallback(
     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(
@@ -1218,7 +1261,7 @@ void ServiceWorkerCache::PendingRequestsCallback(
     ErrorType error,
     scoped_ptr<Requests> requests) {
   callback.Run(error, requests.Pass());
-  DecPendingOps();
+  CompleteOperationAndRunNext();
 }
 
 }  // namespace content