HttpCache::Transaction layer allowing parallel validation

net/http/http_cache.cc

Index: net/http/http_cache.cc
diff --git a/net/http/http_cache.cc b/net/http/http_cache.cc
index 1dc390a474c3abcc21e95539ed56ef75fb9dda16..d49d778597a221a2adf6412b7567465fd0bafb0f 100644
--- a/net/http/http_cache.cc
+++ b/net/http/http_cache.cc
@@ -96,28 +96,33 @@ int HttpCache::DefaultBackend::CreateBackend(
 //-----------------------------------------------------------------------------
 
 HttpCache::ActiveEntry::ActiveEntry(disk_cache::Entry* entry)
-    : disk_entry(entry),
-      writer(NULL),
-      will_process_pending_queue(false),
-      doomed(false) {
-}
+    : disk_entry(entry) {}
 
 HttpCache::ActiveEntry::~ActiveEntry() {
   if (disk_entry) {
     disk_entry->Close();
-    disk_entry = NULL;
+    disk_entry = nullptr;
   }
 }
 
 size_t HttpCache::ActiveEntry::EstimateMemoryUsage() const {
   // Skip |disk_entry| which is tracked in simple_backend_impl; Skip |readers|
-  // and |pending_queue| because the Transactions are owned by their respective
-  // URLRequestHttpJobs.
+  // and |add_to_entry_queue| because the Transactions are owned by their
+  // respective URLRequestHttpJobs.
   return 0;
 }
 
 bool HttpCache::ActiveEntry::HasNoTransactions() {
-  return !writer && readers.empty() && pending_queue.empty();
+  return !writer && readers.empty() && add_to_entry_queue.empty() &&
+         validated_queue.empty() && !validating_transaction;
+}
+
+bool HttpCache::ActiveEntry::HasNoActiveTransactions() {
+  return !writer && readers.empty() && !validating_transaction;
+}
+
+bool HttpCache::ActiveEntry::IsReader(Transaction* transaction) {
+  return readers.count(transaction);
 }
 
 //-----------------------------------------------------------------------------
@@ -360,15 +365,18 @@ HttpCache::~HttpCache() {
   weak_factory_.InvalidateWeakPtrs();
 
   // If we have any active entries remaining, then we need to deactivate them.
-  // We may have some pending calls to OnProcessPendingQueue, but since those
-  // won't run (due to our destruction), we can simply ignore the corresponding
-  // will_process_pending_queue flag.
+  // We may have some pending tasks to process pending queue for reading or
+  // parallel validation, but since those won't run (due to our destruction),
+  // we can simply ignore the corresponding flags.
   while (!active_entries_.empty()) {
     ActiveEntry* entry = active_entries_.begin()->second.get();
-    entry->will_process_pending_queue = false;
-    entry->pending_queue.clear();
+    entry->will_process_waiting_transactions = false;
+    entry->will_process_parallel_validation = false;
+    entry->add_to_entry_queue.clear();
     entry->readers.clear();
-    entry->writer = NULL;
+    entry->validated_queue.clear();
+    entry->validating_transaction = nullptr;
+    entry->writer = nullptr;
     DeactivateEntry(entry);
   }
 
@@ -628,7 +636,10 @@ int HttpCache::DoomEntry(const std::string& key, Transaction* trans) {
   entry_ptr->doomed = true;
 
   DCHECK(entry_ptr->writer || !entry_ptr->readers.empty() ||
-         entry_ptr->will_process_pending_queue);
+         !entry_ptr->validated_queue.empty() ||
+         entry_ptr->validating_transaction ||
+         entry_ptr->will_process_waiting_transactions ||
+         entry_ptr->will_process_parallel_validation);
   return OK;
 }
 
@@ -684,7 +695,14 @@ void HttpCache::FinalizeDoomedEntry(ActiveEntry* entry) {
 
 HttpCache::ActiveEntry* HttpCache::FindActiveEntry(const std::string& key) {
   auto it = active_entries_.find(key);
-  return it != active_entries_.end() ? it->second.get() : NULL;
+  return it != active_entries_.end() ? it->second.get() : nullptr;
+}
+
+bool HttpCache::IsEntryAlive(const std::string& key, ActiveEntry* entry) {
+  ActiveEntry* active_entry = FindActiveEntry(key);
+  if (active_entry && active_entry == entry)
+    return true;
+  return doomed_entries_.count(entry);
 }
 
 HttpCache::ActiveEntry* HttpCache::ActivateEntry(
@@ -696,7 +714,6 @@ HttpCache::ActiveEntry* HttpCache::ActivateEntry(
 }
 
 void HttpCache::DeactivateEntry(ActiveEntry* entry) {
-  DCHECK(!entry->will_process_pending_queue);
   DCHECK(!entry->doomed);
   DCHECK(entry->disk_entry);
   DCHECK(entry->HasNoTransactions());
@@ -826,97 +843,184 @@ void HttpCache::DestroyEntry(ActiveEntry* entry) {
   }
 }
 
-int HttpCache::AddTransactionToEntry(ActiveEntry* entry, Transaction* trans) {
+int HttpCache::AddTransactionToEntry(ActiveEntry* entry,
+                                     Transaction* transaction) {
   DCHECK(entry);
   DCHECK(entry->disk_entry);
 
-  // We implement a basic reader/writer lock for the disk cache entry.  If
-  // there is already a writer, then everyone has to wait for the writer to
-  // finish before they can access the cache entry.  There can be multiple
-  // readers.
-  //
-  // NOTE: If the transaction can only write, then the entry should not be in
-  // use (since any existing entry should have already been doomed).
-
-  if (entry->writer || entry->will_process_pending_queue) {
-    entry->pending_queue.push_back(trans);
+  // If a pending queue processing task is posted, then this transaction
+  // should be added after the queue task is processed, to maintain FIFO
+  // ordering.
+  if (entry->will_process_waiting_transactions ||
+      entry->will_process_parallel_validation) {
+    entry->add_to_entry_queue.push_back(transaction);
     return ERR_IO_PENDING;
   }
 
-  if (trans->mode() & Transaction::WRITE) {
-    // transaction needs exclusive access to the entry
-    if (entry->readers.empty()) {
-      entry->writer = trans;
-    } else {
-      entry->pending_queue.push_back(trans);
-      return ERR_IO_PENDING;
-    }
-  } else {
-    // transaction needs read access to the entry
-    entry->readers.insert(trans);
+  int rv = AddTransactionToEntryImpl(entry, transaction);
+  if (rv == ERR_IO_PENDING)
+    entry->add_to_entry_queue.push_back(transaction);
+  return rv;
+}
+
+// We implement a basic reader/writer lock for the disk cache entry. If there
+// is a writer, then all read-only transactions must wait. Non read-only
+// transactions can proceed to their validation phase. Validation is allowed
+// for one transaction at a time so that we do not end up with wasted network
+// requests.
+int HttpCache::AddTransactionToEntryImpl(ActiveEntry* entry,
+                                         Transaction* transaction) {
+  if (transaction->mode() & Transaction::WRITE)
+    return AddTransactionToEntryForWrite(entry, transaction);
+  return AddTransactionToEntryForRead(entry, transaction);
+}
+
+int HttpCache::AddTransactionToEntryForRead(ActiveEntry* entry,
+                                            Transaction* transaction) {
+  if (entry->writer) {
+    return ERR_IO_PENDING;
   }
+  entry->readers.insert(transaction);
+  if (!entry->add_to_entry_queue.empty())
+    ProcessWaitingTransactions(entry);
+  return OK;
+}
 
-  // We do this before calling EntryAvailable to force any further calls to
-  // AddTransactionToEntry to add their transaction to the pending queue, which
-  // ensures FIFO ordering.
-  if (!entry->writer && !entry->pending_queue.empty())
-    ProcessPendingQueue(entry);
+int HttpCache::AddTransactionToEntryForWrite(ActiveEntry* entry,
+                                             Transaction* transaction) {
+  // A transaction that has the write mode can become the next validating
+  // transaction. It can also become the writer if there is no active writer at
+  // this time.
+  if (entry->validating_transaction) {
+    return ERR_IO_PENDING;
+  }
+  entry->validating_transaction = transaction;
+  if (!entry->writer && entry->readers.empty()) {
+    entry->writer = transaction;
+  }
 
+  // If there is no writer, we can add more readers.
+  if (!entry->writer && !entry->add_to_entry_queue.empty())
+    ProcessWaitingTransactions(entry);
   return OK;
 }
 
-void HttpCache::DoneWithEntry(ActiveEntry* entry, Transaction* trans,
+int HttpCache::DoneValidationWithEntry(ActiveEntry* entry,
+                                       Transaction* transaction,
+                                       bool is_match) {
+  DCHECK(entry->writer == transaction ||
+         entry->validating_transaction == transaction);
+  entry->validating_transaction = nullptr;
+
+  if (!is_match) {
+    if (entry->writer == transaction)
+      entry->writer = nullptr;
+    if (entry->HasNoActiveTransactions())
+      DestroyEntryRestartPendingTransactions(entry);
+    else
+      DoomEntryRestartPendingTransactions(entry);
+    return OK;
+  }
+
+  int rv = OK;
+  // If another transaction is writing the response, let this transaction wait
+  // till the response is complete.
+  if (entry->writer != transaction) {
+    entry->validated_queue.push_back(transaction);
+    rv = ERR_IO_PENDING;
+  }
+  // Else the transaction should either be the writer if its responsible for
+  // writing the response or should have been converted to a reader.
+
+  // In all cases another transaction can start its validation phase.
+  ProcessParallelValidation(entry);
+  return rv;
+}
+
+void HttpCache::DoneValidationWriteEntry(ActiveEntry* entry,
+                                         Transaction* transaction) {
+  entry->validating_transaction = nullptr;
+  ProcessParallelValidation(entry);
+}
+
+void HttpCache::DoneWithEntry(ActiveEntry* entry,
+                              Transaction* transaction,
                               bool cancel) {
   // If we already posted a task to move on to the next transaction and this was
   // the writer, there is nothing to cancel.
-  if (entry->will_process_pending_queue && entry->readers.empty())
+  if (entry->will_process_waiting_transactions && entry->writer == transaction)
     return;
 
-  if (entry->writer) {
-    DCHECK(trans == entry->writer);
-
-    // Assume there was a failure.
-    bool success = false;
-    if (cancel) {
-      DCHECK(entry->disk_entry);
-      // This is a successful operation in the sense that we want to keep the
-      // entry.
-      success = trans->AddTruncatedFlag();
-      // The previous operation may have deleted the entry.
-      if (!trans->entry())
-        return;
-    }
-    DoneWritingToEntry(entry, success);
-  } else {
-    DoneReadingFromEntry(entry, trans);
+  // Transaction is waiting in the validated_queue.
+  auto i = std::find(entry->validated_queue.begin(),
+                     entry->validated_queue.end(), transaction);
+  if (i != entry->validated_queue.end()) {
+    entry->validated_queue.erase(i);
+    return;
+  }
+
+  // Assume there was a failure.
+  bool success = false;
+  if (transaction == entry->writer && cancel) {
+    DCHECK(entry->disk_entry);
+    // This is a successful operation in the sense that we want to keep the
+    // entry.
+    success = transaction->AddTruncatedFlag();
+    // The previous operation may have deleted the entry.
+    if (!transaction->entry())
+      return;
   }
+
+  // Transaction is writing to the entry, either in the validation phase or
+  // response-body phase.
+  if (transaction == entry->writer ||
+      transaction == entry->validating_transaction) {
+    DoneWritingToEntry(entry, success, transaction);
+    return;
+  }
+
+  // Transaction is reading from the entry.
+  DoneReadingFromEntry(entry, transaction);
 }
 
-void HttpCache::DoneWritingToEntry(ActiveEntry* entry, bool success) {
-  DCHECK(entry->readers.empty());
+void HttpCache::DoneWritingToEntry(ActiveEntry* entry,
+                                   bool success,
+                                   Transaction* transaction) {
+  DCHECK(transaction == entry->writer ||
+         transaction == entry->validating_transaction);
 
-  entry->writer = NULL;
+  if (transaction == entry->writer)
+    entry->writer = nullptr;
 
-  if (success) {
-    ProcessPendingQueue(entry);
-  } else {
-    DCHECK(!entry->will_process_pending_queue);
+  // Transaction is done writing to entry in the validation phase.
+  if (transaction == entry->validating_transaction) {
+    entry->validating_transaction = nullptr;
 
-    // We failed to create this entry.
-    TransactionList pending_queue;
-    pending_queue.swap(entry->pending_queue);
+    if (!entry->writer)
+      DoneWritingToEntryProcessOtherTransactions(entry, success);
+    return;
+  }
 
-    entry->disk_entry->Doom();
-    DestroyEntry(entry);
+  // Transaction is done writing to entry in the response body phase.
+  // Whether its a failure or success, validating transaction can be the new
+  // writer.
+  if (entry->validating_transaction) {
+    entry->writer = entry->validating_transaction;
+    return;
+  }
 
-    // We need to do something about these pending entries, which now need to
-    // be added to a new entry.
-    while (!pending_queue.empty()) {
-      // ERR_CACHE_RACE causes the transaction to restart the whole process.
-      pending_queue.front()->io_callback().Run(ERR_CACHE_RACE);
-      pending_queue.pop_front();
-    }
+  DoneWritingToEntryProcessOtherTransactions(entry, success);
+}
+
+void HttpCache::DoneWritingToEntryProcessOtherTransactions(ActiveEntry* entry,
+                                                           bool success) {
+  if (success) {
+    DCHECK(entry->readers.empty());
+    ProcessWaitingTransactions(entry);
+    return;
   }
+
+  DestroyEntryRestartPendingTransactions(entry);
 }
 
 void HttpCache::DoneReadingFromEntry(ActiveEntry* entry, Transaction* trans) {
@@ -927,20 +1031,29 @@ void HttpCache::DoneReadingFromEntry(ActiveEntry* entry, Transaction* trans) {
 
   entry->readers.erase(it);
 
-  ProcessPendingQueue(entry);
+  ProcessWaitingTransactions(entry);
 }
 
-void HttpCache::ConvertWriterToReader(ActiveEntry* entry) {
-  DCHECK(entry->writer);
-  DCHECK(entry->writer->mode() == Transaction::READ_WRITE);
-  DCHECK(entry->readers.empty());
+bool HttpCache::ConvertWriterToReader(ActiveEntry* entry,
+                                      Transaction* transaction) {
+  DCHECK_EQ(entry->validating_transaction, transaction);
+  DCHECK_EQ(transaction->mode(), Transaction::READ_WRITE);
 
-  Transaction* trans = entry->writer;
+  // We need to differentiate between the scenarios when the entry is already
+  // written to the cache vs. when the entry is not completely written and this
+  // is a validating transaction. In the latter case, it cannot become the
+  // reader and will be added to the validated_queue when
+  // DoneValidationWithEntry will be called for this transaction.
 
-  entry->writer = NULL;
-  entry->readers.insert(trans);
+  // Entry is not completely written yet.
+  if (entry->writer && entry->writer != transaction)
+    return false;
 
-  ProcessPendingQueue(entry);
+  entry->validating_transaction = nullptr;
+  entry->writer = nullptr;
+  entry->readers.insert(transaction);
+  ProcessWaitingTransactions(entry);
+  return true;
 }
 
 LoadState HttpCache::GetLoadStateForPendingTransaction(
@@ -990,14 +1103,15 @@ void HttpCache::RemovePendingTransaction(Transaction* trans) {
 }
 
 bool HttpCache::RemovePendingTransactionFromEntry(ActiveEntry* entry,
-                                                  Transaction* trans) {
-  TransactionList& pending_queue = entry->pending_queue;
+                                                  Transaction* transaction) {
+  TransactionList& add_to_entry_queue = entry->add_to_entry_queue;
 
-  auto j = find(pending_queue.begin(), pending_queue.end(), trans);
-  if (j == pending_queue.end())
+  auto j =
+      find(add_to_entry_queue.begin(), add_to_entry_queue.end(), transaction);
+  if (j == add_to_entry_queue.end())
     return false;
 
-  pending_queue.erase(j);
+  add_to_entry_queue.erase(j);
   return true;
 }
 
@@ -1019,22 +1133,89 @@ bool HttpCache::RemovePendingTransactionFromPendingOp(PendingOp* pending_op,
   return false;
 }
 
-void HttpCache::ProcessPendingQueue(ActiveEntry* entry) {
+void HttpCache::ProcessWaitingTransactions(ActiveEntry* entry) {
   // Multiple readers may finish with an entry at once, so we want to batch up
-  // calls to OnProcessPendingQueue.  This flag also tells us that we should
-  // not delete the entry before OnProcessPendingQueue runs.
-  if (entry->will_process_pending_queue)
+  // calls to OnProcessWaitingTransactions.  This flag also tells us that we
+  // should not delete the entry before OnProcessWaitingTransactions runs.
+  if (entry->will_process_waiting_transactions)
     return;
-  entry->will_process_pending_queue = true;
 
+  entry->will_process_waiting_transactions = true;
+
+  // Post a task instead of invoking the io callback of another transaction here
+  // to avoid re-entrancy.
   base::ThreadTaskRunnerHandle::Get()->PostTask(
-      FROM_HERE,
-      base::Bind(&HttpCache::OnProcessPendingQueue, GetWeakPtr(), entry));
+      FROM_HERE, base::Bind(&HttpCache::OnProcessWaitingTransactions,
+                            GetWeakPtr(), entry->disk_entry->GetKey(), entry));
 }
 
-void HttpCache::OnProcessPendingQueue(ActiveEntry* entry) {
-  entry->will_process_pending_queue = false;
-  DCHECK(!entry->writer);
+void HttpCache::OnProcessWaitingTransactions(const std::string& key,
+                                             ActiveEntry* entry) {
+  // Check if the entry is still alive either as an active entry or doomed
+  // entry.
+  if (!IsEntryAlive(key, entry))
+    return;
+
+  entry->will_process_waiting_transactions = false;
+
+  // If no one is interested in this entry, then we can deactivate it.
+  if (entry->HasNoTransactions()) {
+    DestroyEntry(entry);
+    return;
+  }
+
+  if (entry->validated_queue.empty() && entry->add_to_entry_queue.empty())
+    return;
+
+  Transaction* next = nullptr;
+  TransactionList* list = nullptr;
+  int rv = OK;
+
+  // To maintain FIFO order of transactions, validated_queue should be processed
+  // first and then add_to_entry_queue.
+  if (!entry->validated_queue.empty()) {
+    next = entry->validated_queue.front();
+    list = &entry->validated_queue;
+    rv = AddTransactionToEntryForRead(entry, next);
+  } else {
+    // Promote next transaction from the pending queue.
+    next = entry->add_to_entry_queue.front();
+    list = &entry->add_to_entry_queue;
+    rv = AddTransactionToEntryImpl(entry, next);
+  }
+
+  if (rv != ERR_IO_PENDING) {
+    list->erase(list->begin());
+    next->io_callback().Run(rv);
+  }
+}
+
+void HttpCache::ProcessParallelValidation(ActiveEntry* entry) {
+  if (entry->will_process_parallel_validation)
+    return;
+
+  entry->will_process_parallel_validation = true;
+
+  // Post a task instead of invoking the io callback of another transaction here
+  // to avoid re-entrancy.
+  base::ThreadTaskRunnerHandle::Get()->PostTask(
+      FROM_HERE, base::Bind(&HttpCache::OnProcessParallelValidation,
+                            GetWeakPtr(), entry->disk_entry->GetKey(), entry));
+}
+
+void HttpCache::OnProcessParallelValidation(const std::string& key,
+                                            ActiveEntry* entry) {
+  // Check if the entry is still alive either as an active entry or doomed
+  // entry.
+  if (!IsEntryAlive(key, entry))
+    return;
+
+  entry->will_process_parallel_validation = false;
+
+  // A ProcessWaitingTransactions task before this may add a
+  // validating_transaction. In that case, do nothing.
+  if (entry->validating_transaction)
+    return;
 
   // If no one is interested in this entry, then we can deactivate it.
   if (entry->HasNoTransactions()) {
@@ -1042,22 +1223,56 @@ void HttpCache::OnProcessPendingQueue(ActiveEntry* entry) {
     return;
   }
 
-  if (entry->pending_queue.empty())
+  if (entry->add_to_entry_queue.empty())
     return;
 
-  // Promote next transaction from the pending queue.
-  Transaction* next = entry->pending_queue.front();
-  if ((next->mode() & Transaction::WRITE) && !entry->readers.empty())
-    return;  // Have to wait.
+  Transaction* next = entry->add_to_entry_queue.front();
+
+  int rv = AddTransactionToEntryImpl(entry, next);
 
-  entry->pending_queue.erase(entry->pending_queue.begin());
+  // Read only transaction cannot validate the entry.
+  if (!(next->mode() & Transaction::WRITE))
+    return;
 
-  int rv = AddTransactionToEntry(entry, next);
   if (rv != ERR_IO_PENDING) {
+    entry->add_to_entry_queue.erase(entry->add_to_entry_queue.begin());
     next->io_callback().Run(rv);
   }
 }
 
+void HttpCache::DoomEntryRestartPendingTransactions(ActiveEntry* entry) {
+  DoomActiveEntry(entry->disk_entry->GetKey());
+  TransactionList list = GetAllPendingTransactions(entry);
+  RestartPendingTransactions(list);
+}
+
+void HttpCache::DestroyEntryRestartPendingTransactions(ActiveEntry* entry) {
+  entry->disk_entry->Doom();
+  TransactionList list = GetAllPendingTransactions(entry);
+  DestroyEntry(entry);
+  RestartPendingTransactions(list);
+}
+
+HttpCache::TransactionList HttpCache::GetAllPendingTransactions(
+    ActiveEntry* entry) {
+  TransactionList list;
+  for (auto* transaction : entry->validated_queue)
+    list.push_back(transaction);
+  entry->validated_queue.clear();
+
+  for (auto* transaction : entry->add_to_entry_queue)
+    list.push_back(transaction);
+  entry->add_to_entry_queue.clear();
+
+  return list;
+}
+
+void HttpCache::RestartPendingTransactions(const TransactionList& list) {
+  // ERR_CACHE_RACE causes the transaction to restart the whole process.
+  for (auto* transaction : list)
+    transaction->io_callback().Run(ERR_CACHE_RACE);
+}
+
 void HttpCache::OnIOComplete(int result, PendingOp* pending_op) {
   WorkItemOperation op = pending_op->writer->operation();