Disk cache: Remove the queue for backend operations and

change the queue of entry operations so that the operation
that is posted is kept on the list.

BUG=54338
TEST=net_unittests

Review URL: http://codereview.chromium.org/3744007

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@65118 0039d316-1c4b-4281-b951-d872f2087c98

5 files changed, 143 insertions, 56 deletions

diff --git a/net/disk_cache/backend_impl.cc b/net/disk_cache/backend_impl.cc
index 2b17e48..5cada55 100644
--- a/net/disk_cache/backend_impl.cc
+++ b/net/disk_cache/backend_impl.cc
@@ -1315,6 +1315,13 @@ int BackendImpl::RunTaskForTest(Task* task, CompletionCallback* callback) {
   return net::ERR_IO_PENDING;
 }
 
+void BackendImpl::ThrottleRequestsForTest(bool throttle) {
+  if (throttle)
+    background_queue_.StartQueingOperations();
+  else
+    background_queue_.StopQueingOperations();
+}
+
 int BackendImpl::SelfCheck() {
   if (!init_) {
     LOG(ERROR) << "Init failed";
diff --git a/net/disk_cache/backend_impl.h b/net/disk_cache/backend_impl.h
index a8880d1..05e5016 100644
--- a/net/disk_cache/backend_impl.h
+++ b/net/disk_cache/backend_impl.h
@@ -249,6 +249,9 @@ class BackendImpl : public Backend {
   // deleted after it runs.
   int RunTaskForTest(Task* task, CompletionCallback* callback);
 
+  // Starts or stops throttling requests.
+  void ThrottleRequestsForTest(bool throttle);
+
   // Peforms a simple self-check, and returns the number of dirty items
   // or an error code (negative value).
   int SelfCheck();
diff --git a/net/disk_cache/entry_unittest.cc b/net/disk_cache/entry_unittest.cc
index 84568e3..bea940c 100644
--- a/net/disk_cache/entry_unittest.cc
+++ b/net/disk_cache/entry_unittest.cc
@@ -475,6 +475,75 @@ TEST_F(DiskCacheEntryTest, MemoryOnlyExternalAsyncIO) {
   ExternalAsyncIO();
 }
 
+TEST_F(DiskCacheEntryTest, RequestThrottling) {
+  SetDirectMode();
+  InitCache();
+  disk_cache::Entry* entry = NULL;
+  ASSERT_EQ(net::OK, CreateEntry("the first key", &entry));
+  ASSERT_TRUE(NULL != entry);
+
+  // Let's verify that each IO goes to the right callback object.
+  CallbackTest cb(true);
+
+  g_cache_tests_error = false;
+  g_cache_tests_received = 0;
+
+  MessageLoopHelper helper;
+
+  const int kSize = 200;
+  scoped_refptr<net::IOBuffer> buffer = new net::IOBuffer(kSize);
+  CacheTestFillBuffer(buffer->data(), kSize, false);
+
+  int expected = 0;
+  // Start with no throttling.
+  for (; expected < 10; expected++) {
+    int ret = entry->WriteData(0, 0, buffer, kSize, &cb, false);
+    EXPECT_EQ(net::ERR_IO_PENDING, ret);
+  }
+  EXPECT_TRUE(helper.WaitUntilCacheIoFinished(expected));
+
+  // And now with full throttling.
+  cache_impl_->ThrottleRequestsForTest(true);
+  for (; expected < 20; expected++) {
+    int ret = entry->WriteData(0, 0, buffer, kSize, &cb, false);
+    EXPECT_EQ(net::ERR_IO_PENDING, ret);
+  }
+  EXPECT_TRUE(helper.WaitUntilCacheIoFinished(expected));
+
+  for (; expected < 30; expected++) {
+    int ret = entry->WriteData(0, 0, buffer, kSize, &cb, false);
+    EXPECT_EQ(net::ERR_IO_PENDING, ret);
+  }
+  // We have 9 queued requests, lets dispatch them all at once.
+  cache_impl_->ThrottleRequestsForTest(false);
+  EXPECT_TRUE(helper.WaitUntilCacheIoFinished(expected));
+
+  cache_impl_->ThrottleRequestsForTest(true);
+  for (; expected < 40; expected++) {
+    int ret = entry->WriteData(0, 0, buffer, kSize, &cb, false);
+    EXPECT_EQ(net::ERR_IO_PENDING, ret);
+  }
+
+  // We can close the entry and keep receiving notifications.
+  entry->Close();
+  EXPECT_TRUE(helper.WaitUntilCacheIoFinished(expected));
+
+  ASSERT_EQ(net::OK, OpenEntry("the first key", &entry));
+  for (; expected < 50; expected++) {
+    int ret = entry->WriteData(0, 0, buffer, kSize, &cb, false);
+    EXPECT_EQ(net::ERR_IO_PENDING, ret);
+  }
+
+  // ... and even close the cache.
+  entry->Close();
+  delete cache_impl_;
+  cache_ = cache_impl_ = NULL;
+  EXPECT_TRUE(helper.WaitUntilCacheIoFinished(expected));
+
+  EXPECT_FALSE(g_cache_tests_error);
+  EXPECT_EQ(expected, g_cache_tests_received);
+}
+
 void DiskCacheEntryTest::StreamAccess() {
   disk_cache::Entry* entry = NULL;
   ASSERT_EQ(net::OK, CreateEntry("the first key", &entry));
@@ -755,11 +824,6 @@ void DiskCacheEntryTest::TruncateData() {
 TEST_F(DiskCacheEntryTest, TruncateData) {
   InitCache();
   TruncateData();
-
-  // We generate asynchronous IO that is not really tracked until completion
-  // so we just wait here before running the next test.
-  MessageLoopHelper helper;
-  helper.WaitUntilCacheIoFinished(1);
 }
 
 TEST_F(DiskCacheEntryTest, TruncateDataNoBuffer) {
@@ -767,11 +831,6 @@ TEST_F(DiskCacheEntryTest, TruncateDataNoBuffer) {
   InitCache();
   cache_impl_->SetFlags(disk_cache::kNoBuffering);
   TruncateData();
-
-  // We generate asynchronous IO that is not really tracked until completion
-  // so we just wait here before running the next test.
-  MessageLoopHelper helper;
-  helper.WaitUntilCacheIoFinished(1);
 }
 
 TEST_F(DiskCacheEntryTest, MemoryOnlyTruncateData) {
diff --git a/net/disk_cache/in_flight_backend_io.cc b/net/disk_cache/in_flight_backend_io.cc
index c4caed3..d83bd10 100644
--- a/net/disk_cache/in_flight_backend_io.cc
+++ b/net/disk_cache/in_flight_backend_io.cc
@@ -42,8 +42,9 @@ bool BackendIO::IsEntryOperation() {
   return operation_ > OP_MAX_BACKEND;
 }
 
-void BackendIO::ReleaseEntry() {
-  entry_ = NULL;
+// Runs on the background thread.
+void BackendIO::ReferenceEntry() {
+  entry_->AddRef();
 }
 
 base::TimeDelta BackendIO::ElapsedTime() const {
@@ -274,6 +275,8 @@ void BackendIO::ExecuteEntryOperation() {
       NOTREACHED() << "Invalid Operation";
       result_ = net::ERR_UNEXPECTED;
   }
+  // We added a reference to protect the queued operation.
+  entry_->Release();
   if (result_ != net::ERR_IO_PENDING)
     controller_->OnIOComplete(this);
 }
@@ -437,9 +440,6 @@ void InFlightBackendIO::ReadyForSparseIO(EntryImpl* entry,
 }
 
 void InFlightBackendIO::WaitForPendingIO() {
-  // We clear the list first so that we don't post more operations after this
-  // point.
-  pending_ops_.clear();
   InFlightIO::WaitForPendingIO();
 }
 
@@ -455,45 +455,51 @@ void InFlightBackendIO::OnOperationComplete(BackgroundIO* operation,
                                             bool cancel) {
   BackendIO* op = static_cast<BackendIO*>(operation);
 
-  if (!op->IsEntryOperation() && !pending_ops_.empty()) {
-    // Process the next request. Note that invoking the callback may result
-    // in the backend destruction (and with it this object), so we should deal
-    // with the next operation before invoking the callback.
-    PostQueuedOperation(&pending_ops_);
-  }
-
   if (op->IsEntryOperation()) {
     backend_->OnOperationCompleted(op->ElapsedTime());
-    if (!pending_entry_ops_.empty()) {
-      PostQueuedOperation(&pending_entry_ops_);
-
-      // If we are not throttling requests anymore, dispatch the whole queue.
-      if (!queue_entry_ops_) {
+    if (!pending_ops_.empty() && RemoveFirstQueuedOperation(op)) {
+      // Process the next request. Note that invoking the callback may result
+      // in the backend destruction (and with it this object), so we should deal
+      // with the next operation before invoking the callback.
+      if (queue_entry_ops_) {
+        PostQueuedOperation();
+      } else {
+        // If we are not throttling requests anymore, dispatch the whole queue.
         CACHE_UMA(COUNTS_10000, "FinalQueuedOperations", 0,
-                  pending_entry_ops_.size());
-        while (!pending_entry_ops_.empty())
-          PostQueuedOperation(&pending_entry_ops_);
+                  pending_ops_.size());
+        PostAllQueuedOperations();
       }
     }
   }
 
   if (op->callback() && (!cancel || op->IsEntryOperation()))
     op->callback()->Run(op->result());
-
-  if (cancel)
-    op->ReleaseEntry();
 }
 
 void InFlightBackendIO::QueueOperation(BackendIO* operation) {
   if (!operation->IsEntryOperation())
-    return QueueOperationToList(operation, &pending_ops_);
+    return PostOperation(operation);
 
-  if (!queue_entry_ops_)
+  // We have to protect the entry from deletion while it is on the queue.
+  // If the caller closes the entry right after writing to it, and the write is
+  // waiting on the queue, we could end up deleting the entry before the write
+  // operation is actually posted. Sending a task to reference the entry we make
+  // sure that there is an extra reference before the caller can post a task to
+  // release its reference.
+  background_thread_->PostTask(FROM_HERE,
+      NewRunnableMethod(operation, &BackendIO::ReferenceEntry));
+
+  bool empty_list = pending_ops_.empty();
+  if (!queue_entry_ops_ && empty_list)
     return PostOperation(operation);
 
-  CACHE_UMA(COUNTS_10000, "QueuedOperations", 0, pending_entry_ops_.size());
+  CACHE_UMA(COUNTS_10000, "QueuedOperations", 0, pending_ops_.size());
 
-  QueueOperationToList(operation, &pending_entry_ops_);
+  // We keep the operation that we are executing in the list so that we know
+  // when it completes.
+  pending_ops_.push_back(operation);
+  if (empty_list)
+    PostOperation(operation);
 }
 
 void InFlightBackendIO::PostOperation(BackendIO* operation) {
@@ -502,18 +508,31 @@ void InFlightBackendIO::PostOperation(BackendIO* operation) {
   OnOperationPosted(operation);
 }
 
-void InFlightBackendIO::PostQueuedOperation(OperationList* from_list) {
-  scoped_refptr<BackendIO> next_op = from_list->front();
-  from_list->pop_front();
+void InFlightBackendIO::PostQueuedOperation() {
+  if (pending_ops_.empty())
+    return;
+
+  // Keep it in the list until it's done.
+  scoped_refptr<BackendIO> next_op = pending_ops_.front();
   PostOperation(next_op);
 }
 
-void InFlightBackendIO::QueueOperationToList(BackendIO* operation,
-                                             OperationList* list) {
-  if (list->empty())
-    return PostOperation(operation);
+void InFlightBackendIO::PostAllQueuedOperations() {
+  for (OperationList::iterator it = pending_ops_.begin();
+       it != pending_ops_.end(); ++it) {
+    PostOperation(*it);
+  }
+  pending_ops_.clear();
+}
+
+bool InFlightBackendIO::RemoveFirstQueuedOperation(BackendIO* operation) {
+  DCHECK(!pending_ops_.empty());
+  scoped_refptr<BackendIO> next_op = pending_ops_.front();
+  if (operation != next_op)
+    return false;
 
-  list->push_back(make_scoped_refptr(operation));
+  pending_ops_.pop_front();
+  return true;
 }
 
 }  // namespace
diff --git a/net/disk_cache/in_flight_backend_io.h b/net/disk_cache/in_flight_backend_io.h
index 5eba131..ca239dd 100644
--- a/net/disk_cache/in_flight_backend_io.h
+++ b/net/disk_cache/in_flight_backend_io.h
@@ -37,7 +37,8 @@ class BackendIO : public BackgroundIO {
 
   net::CompletionCallback* callback() { return callback_; }
 
-  void ReleaseEntry();
+  // Grabs an extra reference of entry_.
+  void ReferenceEntry();
 
   // Returns the time that has passed since the operation was created.
   base::TimeDelta ElapsedTime() const;
@@ -72,12 +73,10 @@ class BackendIO : public BackgroundIO {
 
  private:
   // There are two types of operations to proxy: regular backend operations are
-  // queued so that we don't have more than one operation going on at the same
-  // time (for instance opening an entry and creating the same entry). On the
-  // other hand, operations targeted to a given entry can be long lived and
-  // support multiple simultaneous users (multiple reads or writes to the same
-  // entry), so they are not queued, just posted to the worker thread as they
-  // come.
+  // executed sequentially (queued by the message loop). On the other hand,
+  // operations targeted to a given entry can be long lived and support multiple
+  // simultaneous users (multiple reads or writes to the same entry), and they
+  // are subject to throttling, so we keep an explicit queue.
   enum Operation {
     OP_NONE = 0,
     OP_INIT,
@@ -200,13 +199,13 @@ class InFlightBackendIO : public InFlightIO {
   typedef std::list<scoped_refptr<BackendIO> > OperationList;
   void QueueOperation(BackendIO* operation);
   void PostOperation(BackendIO* operation);
-  void PostQueuedOperation(OperationList* from_list);
-  void QueueOperationToList(BackendIO* operation, OperationList* list);
+  void PostQueuedOperation();
+  void PostAllQueuedOperations();
+  bool RemoveFirstQueuedOperation(BackendIO* operation);
 
   BackendImpl* backend_;
   scoped_refptr<base::MessageLoopProxy> background_thread_;
-  OperationList pending_ops_;  // The list of operations to be posted.
-  OperationList pending_entry_ops_;  // Entry (async) operations to be posted.
+  OperationList pending_ops_;  // Entry (async) operations to be posted.
   bool queue_entry_ops_;  // True if we are queuing entry (async) operations.
 
   DISALLOW_COPY_AND_ASSIGN(InFlightBackendIO);