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..b10f396a4647bd73ea62a646e78af410294d8fb0 100644
--- a/net/http/http_cache.cc
+++ b/net/http/http_cache.cc
@@ -96,28 +96,29 @@ 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() &&
+         done_headers_queue.empty() && !headers_transaction;
+}
+
+bool HttpCache::ActiveEntry::HasNoActiveTransactions() {
+  return !writer && readers.empty() && !headers_transaction;
 }
 
 //-----------------------------------------------------------------------------
@@ -360,15 +361,17 @@ 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 queued transactions ,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_queued_transactions = false;
+    entry->add_to_entry_queue.clear();
     entry->readers.clear();
-    entry->writer = NULL;
+    entry->done_headers_queue.clear();
+    entry->headers_transaction = nullptr;
+    entry->writer = nullptr;
     DeactivateEntry(entry);
   }
 
@@ -628,7 +631,8 @@ 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->headers_transaction ||
+         entry_ptr->will_process_queued_transactions);
   return OK;
 }
 
@@ -684,7 +688,7 @@ 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;
 }
 
 HttpCache::ActiveEntry* HttpCache::ActivateEntry(
@@ -696,7 +700,7 @@ HttpCache::ActiveEntry* HttpCache::ActivateEntry(
 }
 
 void HttpCache::DeactivateEntry(ActiveEntry* entry) {
-  DCHECK(!entry->will_process_pending_queue);
+  DCHECK(!entry->will_process_queued_transactions);
   DCHECK(!entry->doomed);
   DCHECK(entry->disk_entry);
   DCHECK(entry->HasNoTransactions());
@@ -826,121 +830,276 @@ 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);
+  // Always add a new transaction to the queue to maintain FIFO order.
+  entry->add_to_entry_queue.push_back(transaction);
+  ProcessQueuedTransactions(entry);
+  return ERR_IO_PENDING;
+}
 
-  // 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).
+int HttpCache::DoneResponseHeaders(ActiveEntry* entry,
+                                   Transaction* transaction,
+                                   bool is_match) {
+  DCHECK_EQ(entry->headers_transaction, transaction);
+  entry->headers_transaction = nullptr;
 
-  if (entry->writer || entry->will_process_pending_queue) {
-    entry->pending_queue.push_back(trans);
-    return ERR_IO_PENDING;
+  if (!is_match) {
+    // Doom or destroy this entry based on whether it has active consumers.
+    ProcessEntryFailure(entry);
+    return OK;
+  }
+
+  entry->done_headers_queue.push_back(transaction);
+  ProcessQueuedTransactions(entry);
+  return ERR_IO_PENDING;
+}
+
+void HttpCache::RemoveAllQueuedTransactions(ActiveEntry* entry,
+                                            TransactionList* list) {
+  // Process done_headers_queue before add_to_entry_queue to maintain FIFO
+  // order.
+  for (auto* transaction : entry->done_headers_queue)
+    list->push_back(transaction);
+  entry->done_headers_queue.clear();
+
+  for (auto* transaction : entry->add_to_entry_queue)
+    list->push_back(transaction);
+  entry->add_to_entry_queue.clear();
+}
+
+void HttpCache::ProcessEntryFailure(ActiveEntry* entry) {
+  // Failure case is either writer failing to completely write the response to
+  // the cache or validating transaction received a non-304 response.
+  TransactionList list;
+  if (entry->HasNoActiveTransactions() &&
+      !entry->will_process_queued_transactions) {
+    entry->disk_entry->Doom();
+    RemoveAllQueuedTransactions(entry, &list);
+    DestroyEntry(entry);
+  } else {
+    DoomActiveEntry(entry->disk_entry->GetKey());
+    RemoveAllQueuedTransactions(entry, &list);
+  }
+  // ERR_CACHE_RACE causes the transaction to restart the whole process.
+  for (auto* transaction : list)
+    transaction->io_callback().Run(net::ERR_CACHE_RACE);
+}
+
+void HttpCache::ProcessQueuedTransactions(ActiveEntry* entry) {
+  // Multiple readers may finish with an entry at once, so we want to batch up
+  // calls to OnProcessQueuedTransactions. This flag also tells us that we
+  // should not delete the entry before OnProcessQueuedTransactions runs.
+  if (entry->will_process_queued_transactions)
+    return;
+
+  entry->will_process_queued_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::OnProcessQueuedTransactions, GetWeakPtr(), entry));
+}
+
+void HttpCache::OnProcessQueuedTransactions(ActiveEntry* entry) {
+  entry->will_process_queued_transactions = false;
+
+  // If no one is interested in this entry, then we can deactivate it.
+  if (entry->HasNoTransactions()) {
+    DestroyEntry(entry);
+    return;
   }
 
-  if (trans->mode() & Transaction::WRITE) {
-    // transaction needs exclusive access to the entry
-    if (entry->readers.empty()) {
-      entry->writer = trans;
+  if (entry->done_headers_queue.empty() && entry->add_to_entry_queue.empty())
+    return;
+
+  // To maintain FIFO order of transactions, done_headers_queue should be
+  // processed first, and then add_to_entry_queue.
+  // If another transaction is writing the response, let validated transactions
+  // wait till the response is complete. If the response is not yet started, the
+  // done_headers_queue transaction should start writing it.
+  if (!entry->writer && !entry->done_headers_queue.empty()) {
+    ProcessDoneHeadersTransaction(entry);
+    return;
+  }
+
+  if (!entry->add_to_entry_queue.empty())
+    ProcessAddToEntryTransaction(entry);
+}
+
+void HttpCache::ProcessAddToEntryTransaction(ActiveEntry* entry) {
+  if (entry->headers_transaction) {
+    // Note the entry may be new or may already have a response body written to
+    // it. In both cases, a transaction with write mode set, needs to wait
+    // because only one transaction is allowed to be validating at a time.
+    // If transaction is read-only and entry is not yet complete, then it
+    // needs to wait. If entry is complete, it needs to wait because its
+    // possible that the headers_transaction dooms the entry if its not a match.
+    return;
+  }
+
+  Transaction* transaction = entry->add_to_entry_queue.front();
+  if (entry->writer) {
+    // A transaction either starts headers phase or waits.
+    if (transaction->mode() & Transaction::WRITE) {
+      entry->headers_transaction = transaction;
     } else {
-      entry->pending_queue.push_back(trans);
-      return ERR_IO_PENDING;
+      return;
     }
   } else {
-    // transaction needs read access to the entry
-    entry->readers.insert(trans);
+    // This state may be reached either during the start of an entry or when the
+    // response is written.
+    if (transaction->mode() & Transaction::WRITE) {
+      entry->headers_transaction = transaction;
+    } else {
+      // A read-only transaction can start reading the response.
+      // If a read transaction is here, then it has to be an already created
+      // entry as the other case is already handled in the Transaction state
+      // machine.
+      entry->readers.insert(transaction);
+
+      // More readers can join too.
+      ProcessQueuedTransactions(entry);

[Randy Smith (Not in Mondays), 2017/03/30 19:18:05]

Is it reasonable to avoid another thread hop here and just loop?

[shivanisha, 2017/03/31 17:47:05]

Looping here will mean that the control flow goes in more than one transactions
io callbacks in sequence. Posting a task makes it clear that everytime it will
be just one transaction's io callback invoked.

[Randy Smith (Not in Mondays), 2017/03/31 19:26:12]

Why is that a useful invariant?

[shivanisha, 2017/04/03 20:16:48]

As discussed f2f, I have preferred to keep the post task to make one call stack
correspond to one transaction as it has been till now, but I do not see a
problem why we cannot have a loop here especially since entry should not get
deleted before the loop is entered a second time. Added a todo as well.

+    }
   }
 
-  // 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);
+  entry->add_to_entry_queue.erase(entry->add_to_entry_queue.begin());

[Randy Smith (Not in Mondays), 2017/03/30 19:18:05]

Would it be reasonable to rewrite the above so that this is more clearly
associated with the decision to actually move the transaction into a different
place in the active entry?  I'm thinking of something like:

Transaction* transaction = entry->add_to_entry_queue.front();
if (entry->writer && !(transaction->mode() & Transaction::WRITE))
  return
entry->add_to_entry_queue.erase(transaction);  // Or use begin() if you want, as
long as it's clear by proxy that you're talking about the same transaction.

if (transaction->mode() & Transaction::WRITE) 
  entry->headers_transaction = transaction;
else {
  entry->readers.insert(transaction); ProcessQueuedTransacations(); 
}

?

[shivanisha, 2017/03/31 17:47:05]

Done.

+  transaction->io_callback().Run(OK);
+}
 
-  return OK;
+void HttpCache::ProcessDoneHeadersTransaction(ActiveEntry* entry) {
+  DCHECK(!entry->writer && !entry->done_headers_queue.empty());
+  Transaction* transaction = entry->done_headers_queue.front();
+  // If this transaction is responsible for writing the response body.
+  if (transaction->mode() & Transaction::WRITE) {
+    DCHECK_EQ(entry->done_headers_queue.size(), (size_t)1);
+    entry->writer = transaction;
+  } else {
+    // Response body is already written, convert to a reader.

[Randy Smith (Not in Mondays), 2017/03/30 19:18:05]

Isn't the response body about the state of the ActiveEntry rather than the
transaction being a READ one?  READ transactions can occur based on
LOAD_ONLY_FROM_CACHE, I think.

[shivanisha, 2017/03/31 17:47:05]

This comment here signifies that since we are in ProcessDoneHeadersTransaction,
response body must have been complete. A transaction will be here with Read mode
not because of LOAD_ONLY_FROM_CACHE but because its validation led to a match. 
(For read-only transactions because of LOAD_ONLY_FROM_CACHE, their states are
from add_to_entry_queue to readers.)

+    auto return_val = entry->readers.insert(transaction);
+    DCHECK_EQ(return_val.second, true);
+  }
+  // Post another task to give a chance to more transactions to either join
+  // readers or another transaction to start parallel validation.
+  ProcessQueuedTransactions(entry);

[Randy Smith (Not in Mondays), 2017/03/30 19:18:05]

Given that we're already a PostTask away from the main thread, would it be
possible to loop instead?

[shivanisha, 2017/03/31 17:47:05]

Looping here will mean that the control flow goes in more than one transactions
io callbacks in sequence. Posting a task makes it clear that everytime it will
be just one transaction's io callback invoked.

+  entry->done_headers_queue.erase(entry->done_headers_queue.begin());
+  transaction->io_callback().Run(OK);
 }
 
-void HttpCache::DoneWithEntry(ActiveEntry* entry, Transaction* trans,
+bool HttpCache::IsCancelResponseBody(ActiveEntry* entry,

[jkarlin, 2017/03/30 14:58:14]

The name IsCancelResponseBody doesn't make sense to me.

Perhaps IsBackendEntryFinished?

[shivanisha, 2017/03/31 17:47:05]

Actually, what this function needs to return is this transaction's role in the
entry, how about:
IsTransactionCurrentOrFutureWriter()?

+                                     Transaction* transaction,
+                                     const std::string& method) {
+  bool cancel = (method != "HEAD");

[jkarlin, 2017/03/30 14:58:14]

The cancel variable is unnecessary and adds confusion. The following works just
as well:

if (method == "HEAD")
  return false;

if (transaction == entry->writer)
  return true;

...

[shivanisha, 2017/04/03 20:16:48]

done

+
+  if (!cancel)
+    return false;
+

[jkarlin, 2017/03/30 14:58:14]

It seems like this function shouldn't be necessary. Can the Transaction really
not tell:

1) If it has written bytes to the cache entry
2) If it isn't yet done writing bytes to the cache entry

[shivanisha, 2017/03/31 17:47:05]

Let's see... a transaction can tell above 1) and 2) but if 1) is not true, how
does it tell if it was the one responsible for writing the entry , that is done
by checking if the transaction's mode is WRITE and there is no writer currently
in the entry.

Simplified and renamed the function, PTAL.

+  if (transaction == entry->writer)
+    return cancel;
+
+  // Transaction's mode may have been set to NONE if StopCaching was invoked.
+  // Check if transaction was about to start writing to the cache.
+  cancel &= (transaction->mode() & Transaction::WRITE ||
+             transaction->mode() == Transaction::NONE);
+
+  if (!cancel)
+    return false;
+
+  if (transaction == entry->headers_transaction) {
+    cancel &= (!entry->writer);
+    return cancel;
+  }
+
+  if (!entry->done_headers_queue.empty()) {
+    auto i = std::find(entry->done_headers_queue.begin(),
+                       entry->done_headers_queue.end(), transaction);
+    if (i != entry->done_headers_queue.end()) {
+      cancel &= (!entry->writer);
+      return cancel;
+    }
+  }
+
+  return false;
+}
+
+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())
-    return;
+  // Transaction is waiting in the done_headers_queue.
+  if (!entry->done_headers_queue.empty()) {

[Randy Smith (Not in Mondays), 2017/03/30 19:18:05]

I think you can leave the outer conditional off without any performance cost and
with a slight decrease in code complexity?

[shivanisha, 2017/03/31 17:47:05]

Removed.

+    auto i = std::find(entry->done_headers_queue.begin(),
+                       entry->done_headers_queue.end(), transaction);
+    if (i != entry->done_headers_queue.end()) {
+      entry->done_headers_queue.erase(i);
+      if (cancel)
+        ProcessEntryFailure(entry);
+      return;
+    }
+  }
 
-  if (entry->writer) {
-    DCHECK(trans == entry->writer);
+  // Transaction is removed in the headers phase.
+  if (transaction == entry->headers_transaction) {
+    // If the response is not written (cancel is true), consider it a failure.
+    DoneWritingToEntry(entry, !cancel, transaction);
+    return;
+  }
 
+  // Transaction is removed in the writing phase.
+  if (transaction == 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();
+      success = transaction->AddTruncatedFlag();
       // The previous operation may have deleted the entry.
-      if (!trans->entry())
+      if (!transaction->entry())
         return;
     }
-    DoneWritingToEntry(entry, success);
-  } else {
-    DoneReadingFromEntry(entry, trans);
+    DoneWritingToEntry(entry, success, transaction);
+    return;
   }
-}
-
-void HttpCache::DoneWritingToEntry(ActiveEntry* entry, bool success) {
-  DCHECK(entry->readers.empty());
 
-  entry->writer = NULL;
-
-  if (success) {
-    ProcessPendingQueue(entry);
-  } else {
-    DCHECK(!entry->will_process_pending_queue);
+  // Transaction is reading from the entry.
+  DoneReadingFromEntry(entry, transaction);
+}
 
-    // We failed to create this entry.
-    TransactionList pending_queue;
-    pending_queue.swap(entry->pending_queue);
+void HttpCache::DoneWritingToEntry(ActiveEntry* entry,
+                                   bool success,
+                                   Transaction* transaction) {
+  DCHECK(transaction == entry->writer ||
+         transaction == entry->headers_transaction);
 
-    entry->disk_entry->Doom();
-    DestroyEntry(entry);
+  // Transaction is done writing to entry in the response body phase.
+  if (transaction == entry->writer)
+    entry->writer = nullptr;
+  else
+    entry->headers_transaction = nullptr;
 
-    // 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();
-    }
+  // If writer fails, all transactions should fail.

[Randy Smith (Not in Mondays), 2017/03/30 19:18:05]

All transactions?  This is just failing the headers_transaction; the ones in the
queues should not fail?

[shivanisha, 2017/03/31 17:47:05]

Changed the comment to : "... all transactions should be restarted"

[Randy Smith (Not in Mondays), 2017/03/31 19:26:12]

Still confused, though maybe it's just about placement.  The restart happens in
ProcessEntryFailure, correct?  So shouldn't this comment be above that call?  
What's happening in this conditional is about failing the
entry->headers_transaction, not the other transactions.

[shivanisha, 2017/04/03 20:16:48]

Ah I see, changed the comment to just mention headers_transaction.

+  if (!success && entry->headers_transaction) {
+    entry->headers_transaction->SetValidatingCannotProceed();
+    entry->headers_transaction = nullptr;
+    DCHECK(entry->HasNoActiveTransactions());
   }
+  if (!success)
+    ProcessEntryFailure(entry);
+  else
+    ProcessQueuedTransactions(entry);
 }
 
-void HttpCache::DoneReadingFromEntry(ActiveEntry* entry, Transaction* trans) {
+void HttpCache::DoneReadingFromEntry(ActiveEntry* entry,
+                                     Transaction* transaction) {
   DCHECK(!entry->writer);
-
-  auto it = entry->readers.find(trans);
+  auto it = entry->readers.find(transaction);
   DCHECK(it != entry->readers.end());
-
   entry->readers.erase(it);
 
-  ProcessPendingQueue(entry);
-}
-
-void HttpCache::ConvertWriterToReader(ActiveEntry* entry) {
-  DCHECK(entry->writer);
-  DCHECK(entry->writer->mode() == Transaction::READ_WRITE);
-  DCHECK(entry->readers.empty());
-
-  Transaction* trans = entry->writer;
-
-  entry->writer = NULL;
-  entry->readers.insert(trans);
-
-  ProcessPendingQueue(entry);
+  ProcessQueuedTransactions(entry);
 }
 
 LoadState HttpCache::GetLoadStateForPendingTransaction(
@@ -990,14 +1149,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,45 +1179,6 @@ bool HttpCache::RemovePendingTransactionFromPendingOp(PendingOp* pending_op,
   return false;
 }
 
-void HttpCache::ProcessPendingQueue(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)
-    return;
-  entry->will_process_pending_queue = true;
-
-  base::ThreadTaskRunnerHandle::Get()->PostTask(
-      FROM_HERE,
-      base::Bind(&HttpCache::OnProcessPendingQueue, GetWeakPtr(), entry));
-}
-
-void HttpCache::OnProcessPendingQueue(ActiveEntry* entry) {
-  entry->will_process_pending_queue = false;
-  DCHECK(!entry->writer);
-
-  // If no one is interested in this entry, then we can deactivate it.
-  if (entry->HasNoTransactions()) {
-    DestroyEntry(entry);
-    return;
-  }
-
-  if (entry->pending_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.
-
-  entry->pending_queue.erase(entry->pending_queue.begin());
-
-  int rv = AddTransactionToEntry(entry, next);
-  if (rv != ERR_IO_PENDING) {
-    next->io_callback().Run(rv);
-  }
-}
-
 void HttpCache::OnIOComplete(int result, PendingOp* pending_op) {
   WorkItemOperation op = pending_op->writer->operation();