Switch to using SequencedWorkerPool in Simple Cache backend.

At present, no sequencing behaviour is used, only the thread maximum,
which still can provide performance benefits.

R=pasko
BUG=248570

Review URL: https://chromiumcodereview.appspot.com/15817015

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

7 files changed, 83 insertions, 31 deletions

diff --git a/net/disk_cache/simple/simple_backend_impl.cc b/net/disk_cache/simple/simple_backend_impl.cc
index 4e74106..2fd425a 100644
--- a/net/disk_cache/simple/simple_backend_impl.cc
+++ b/net/disk_cache/simple/simple_backend_impl.cc
@@ -4,14 +4,19 @@
 
 #include "net/disk_cache/simple/simple_backend_impl.h"
 
+#include <algorithm>
+#include <cstdlib>
+
 #include "base/bind.h"
 #include "base/callback.h"
 #include "base/file_util.h"
 #include "base/location.h"
 #include "base/message_loop/message_loop_proxy.h"
+#include "base/metrics/field_trial.h"
 #include "base/metrics/histogram.h"
+#include "base/single_thread_task_runner.h"
 #include "base/sys_info.h"
-#include "base/threading/worker_pool.h"
+#include "base/threading/sequenced_worker_pool.h"
 #include "base/time.h"
 #include "net/base/net_errors.h"
 #include "net/disk_cache/backend_impl.h"
@@ -25,20 +30,46 @@
 using base::Closure;
 using base::FilePath;
 using base::MessageLoopProxy;
+using base::SequencedWorkerPool;
 using base::SingleThreadTaskRunner;
 using base::Time;
-using base::WorkerPool;
 using file_util::DirectoryExists;
 using file_util::CreateDirectory;
 
 namespace {
 
+// Maximum number of concurrent worker pool threads, which also is the limit
+// on concurrent IO (as we use one thread per IO request).
+const int kDefaultMaxWorkerThreads = 50;
+
+const char kThreadNamePrefix[] = "SimpleCacheWorker";
+
 // Cache size when all other size heuristics failed.
 const uint64 kDefaultCacheSize = 80 * 1024 * 1024;
 
 // Maximum fraction of the cache that one entry can consume.
 const int kMaxFileRatio = 8;
 
+// A global sequenced worker pool to use for launching all tasks.
+SequencedWorkerPool* g_sequenced_worker_pool = NULL;
+
+void MaybeCreateSequencedWorkerPool() {
+  if (!g_sequenced_worker_pool) {
+    int max_worker_threads = kDefaultMaxWorkerThreads;
+
+    const std::string thread_count_field_trial =
+        base::FieldTrialList::FindFullName("SimpleCacheMaxThreads");
+    if (!thread_count_field_trial.empty()) {
+      max_worker_threads =
+          std::max(1, std::atoi(thread_count_field_trial.c_str()));
+    }
+
+    g_sequenced_worker_pool = new SequencedWorkerPool(max_worker_threads,
+                                                      kThreadNamePrefix);
+    g_sequenced_worker_pool->AddRef();  // Leak it.
+  }
+}
+
 // Must run on IO Thread.
 void DeleteBackendImpl(disk_cache::Backend** backend,
                        const net::CompletionCallback& callback,
@@ -139,14 +170,8 @@ SimpleBackendImpl::SimpleBackendImpl(const FilePath& path,
                                      base::SingleThreadTaskRunner* cache_thread,
                                      net::NetLog* net_log)
     : path_(path),
-      index_(new SimpleIndex(MessageLoopProxy::current(),  // io_thread
-                             path,
-                             scoped_ptr<SimpleIndexFile>(
-                                 new SimpleIndexFile(cache_thread, path)))),
       cache_thread_(cache_thread),
       orig_max_size_(max_bytes) {
-  index_->ExecuteWhenReady(base::Bind(&RecordIndexLoad,
-                                      base::TimeTicks::Now()));
 }
 
 SimpleBackendImpl::~SimpleBackendImpl() {
@@ -154,6 +179,18 @@ SimpleBackendImpl::~SimpleBackendImpl() {
 }
 
 int SimpleBackendImpl::Init(const CompletionCallback& completion_callback) {
+  MaybeCreateSequencedWorkerPool();
+
+  worker_pool_ = g_sequenced_worker_pool->GetTaskRunnerWithShutdownBehavior(
+      SequencedWorkerPool::CONTINUE_ON_SHUTDOWN);
+
+  index_.reset(new SimpleIndex(MessageLoopProxy::current(), path_,
+                               make_scoped_ptr(new SimpleIndexFile(
+                                   cache_thread_.get(), worker_pool_.get(),
+                                   path_))));
+  index_->ExecuteWhenReady(base::Bind(&RecordIndexLoad,
+                                      base::TimeTicks::Now()));
+
   InitializeIndexCallback initialize_index_callback =
       base::Bind(&SimpleBackendImpl::InitializeIndex,
                  base::Unretained(this),
@@ -247,7 +284,7 @@ void SimpleBackendImpl::IndexReadyForDoom(Time initial_time,
                             new_result.get());
   Closure reply = base::Bind(&CallCompletionCallback,
                              callback, base::Passed(&new_result));
-  WorkerPool::PostTaskAndReply(FROM_HERE, task, reply, true);
+  worker_pool_->PostTaskAndReply(FROM_HERE, task, reply);
 }
 
 int SimpleBackendImpl::DoomEntriesBetween(
diff --git a/net/disk_cache/simple/simple_backend_impl.h b/net/disk_cache/simple/simple_backend_impl.h
index ce98ebc..0abe54c 100644
--- a/net/disk_cache/simple/simple_backend_impl.h
+++ b/net/disk_cache/simple/simple_backend_impl.h
@@ -21,6 +21,7 @@
 
 namespace base {
 class SingleThreadTaskRunner;
+class TaskRunner;
 }
 
 namespace disk_cache {
@@ -48,6 +49,8 @@ class NET_EXPORT_PRIVATE SimpleBackendImpl : public Backend,
 
   SimpleIndex* index() { return index_.get(); }
 
+  base::TaskRunner* worker_pool() { return worker_pool_.get(); }
+
   // Must run on IO Thread.
   int Init(const CompletionCallback& completion_callback);
 
@@ -117,6 +120,7 @@ class NET_EXPORT_PRIVATE SimpleBackendImpl : public Backend,
   const base::FilePath path_;
   scoped_ptr<SimpleIndex> index_;
   const scoped_refptr<base::SingleThreadTaskRunner> cache_thread_;
+  scoped_refptr<base::TaskRunner> worker_pool_;
 
   int orig_max_size_;
 
diff --git a/net/disk_cache/simple/simple_entry_impl.cc b/net/disk_cache/simple/simple_entry_impl.cc
index 0c104d1..fe92f29 100644
--- a/net/disk_cache/simple/simple_entry_impl.cc
+++ b/net/disk_cache/simple/simple_entry_impl.cc
@@ -15,7 +15,7 @@
 #include "base/logging.h"
 #include "base/message_loop/message_loop_proxy.h"
 #include "base/metrics/histogram.h"
-#include "base/threading/worker_pool.h"
+#include "base/task_runner.h"
 #include "base/time.h"
 #include "net/base/io_buffer.h"
 #include "net/base/net_errors.h"
@@ -75,7 +75,7 @@ using base::Closure;
 using base::FilePath;
 using base::MessageLoopProxy;
 using base::Time;
-using base::WorkerPool;
+using base::TaskRunner;
 
 // A helper class to insure that RunNextOperationIfNeeded() is called when
 // exiting the current stack frame.
@@ -97,6 +97,7 @@ SimpleEntryImpl::SimpleEntryImpl(SimpleBackendImpl* backend,
                                  const std::string& key,
                                  const uint64 entry_hash)
     : backend_(backend->AsWeakPtr()),
+      worker_pool_(backend->worker_pool()),
       path_(path),
       key_(key),
       entry_hash_(entry_hash),
@@ -187,7 +188,7 @@ int SimpleEntryImpl::DoomEntry(const CompletionCallback& callback) {
                             entry_hash_, result.get());
   Closure reply = base::Bind(&CallCompletionCallback,
                              callback, base::Passed(&result));
-  WorkerPool::PostTaskAndReply(FROM_HERE, task, reply, true);
+  worker_pool_->PostTaskAndReply(FROM_HERE, task, reply);
   return net::ERR_IO_PENDING;
 }
 
@@ -366,7 +367,7 @@ int SimpleEntryImpl::ReadyForSparseIO(const CompletionCallback& callback) {
 SimpleEntryImpl::~SimpleEntryImpl() {
   DCHECK(io_thread_checker_.CalledOnValidThread());
   DCHECK_EQ(0U, pending_operations_.size());
-  DCHECK(STATE_UNINITIALIZED == state_ || STATE_FAILURE == state_);
+  DCHECK(state_ == STATE_UNINITIALIZED || state_ == STATE_FAILURE);
   DCHECK(!synchronous_entry_);
   RemoveSelfFromBackend();
 }
@@ -441,7 +442,7 @@ void SimpleEntryImpl::OpenEntryInternal(const CompletionCallback& callback,
   Closure reply = base::Bind(&SimpleEntryImpl::CreationOperationComplete, this,
                              callback, start_time, base::Passed(&sync_entry),
                              base::Passed(&result), out_entry);
-  WorkerPool::PostTaskAndReply(FROM_HERE, task, reply, true);
+  worker_pool_->PostTaskAndReply(FROM_HERE, task, reply);
 }
 
 void SimpleEntryImpl::CreateEntryInternal(const CompletionCallback& callback,
@@ -478,7 +479,7 @@ void SimpleEntryImpl::CreateEntryInternal(const CompletionCallback& callback,
   Closure reply = base::Bind(&SimpleEntryImpl::CreationOperationComplete, this,
                              callback, start_time, base::Passed(&sync_entry),
                              base::Passed(&result), out_entry);
-  WorkerPool::PostTaskAndReply(FROM_HERE, task, reply, true);
+  worker_pool_->PostTaskAndReply(FROM_HERE, task, reply);
 }
 
 void SimpleEntryImpl::CloseInternal() {
@@ -510,7 +511,7 @@ void SimpleEntryImpl::CloseInternal() {
                               base::Passed(&crc32s_to_write));
     Closure reply = base::Bind(&SimpleEntryImpl::CloseOperationComplete, this);
     synchronous_entry_ = NULL;
-    WorkerPool::PostTaskAndReply(FROM_HERE, task, reply, true);
+    worker_pool_->PostTaskAndReply(FROM_HERE, task, reply);
 
     for (int i = 0; i < kSimpleEntryFileCount; ++i) {
       if (!have_written_[i]) {
@@ -565,7 +566,7 @@ void SimpleEntryImpl::ReadDataInternal(int stream_index,
   Closure reply = base::Bind(&SimpleEntryImpl::ReadOperationComplete, this,
                              stream_index, offset, callback,
                              base::Passed(&read_crc32), base::Passed(&result));
-  WorkerPool::PostTaskAndReply(FROM_HERE, task, reply, true);
+  worker_pool_->PostTaskAndReply(FROM_HERE, task, reply);
 }
 
 void SimpleEntryImpl::WriteDataInternal(int stream_index,
@@ -627,7 +628,7 @@ void SimpleEntryImpl::WriteDataInternal(int stream_index,
                             buf_len, truncate, result.get());
   Closure reply = base::Bind(&SimpleEntryImpl::WriteOperationComplete, this,
                              stream_index, callback, base::Passed(&result));
-  WorkerPool::PostTaskAndReply(FROM_HERE, task, reply, true);
+  worker_pool_->PostTaskAndReply(FROM_HERE, task, reply);
 }
 
 void SimpleEntryImpl::CreationOperationComplete(
@@ -732,7 +733,7 @@ void SimpleEntryImpl::ReadOperationComplete(
                                  this, *result, stream_index,
                                  completion_callback,
                                  base::Passed(&new_result));
-      WorkerPool::PostTaskAndReply(FROM_HERE, task, reply, true);
+      worker_pool_->PostTaskAndReply(FROM_HERE, task, reply);
       crc_check_state_[stream_index] = CRC_CHECK_DONE;
       return;
     }
diff --git a/net/disk_cache/simple/simple_entry_impl.h b/net/disk_cache/simple/simple_entry_impl.h
index 75b1c71..c8d3d76 100644
--- a/net/disk_cache/simple/simple_entry_impl.h
+++ b/net/disk_cache/simple/simple_entry_impl.h
@@ -17,7 +17,7 @@
 #include "net/disk_cache/simple/simple_entry_format.h"
 
 namespace base {
-class MessageLoopProxy;
+class TaskRunner;
 }
 
 namespace net {
@@ -219,6 +219,7 @@ class SimpleEntryImpl : public Entry, public base::RefCounted<SimpleEntryImpl>,
   base::ThreadChecker io_thread_checker_;
 
   base::WeakPtr<SimpleBackendImpl> backend_;
+  const scoped_refptr<base::TaskRunner> worker_pool_;
   const base::FilePath path_;
   const std::string key_;
   const uint64 entry_hash_;
diff --git a/net/disk_cache/simple/simple_index_file.cc b/net/disk_cache/simple/simple_index_file.cc
index 6563cbf..0d31e0f 100644
--- a/net/disk_cache/simple/simple_index_file.cc
+++ b/net/disk_cache/simple/simple_index_file.cc
@@ -4,6 +4,8 @@
 
 #include "net/disk_cache/simple/simple_index_file.h"
 
+#include <vector>
+
 #include "base/file_util.h"
 #include "base/files/file_enumerator.h"
 #include "base/hash.h"
@@ -12,7 +14,6 @@
 #include "base/pickle.h"
 #include "base/single_thread_task_runner.h"
 #include "base/threading/thread_restrictions.h"
-#include "base/threading/worker_pool.h"
 #include "net/disk_cache/simple/simple_entry_format.h"
 #include "net/disk_cache/simple/simple_index.h"
 #include "net/disk_cache/simple/simple_synchronous_entry.h"
@@ -106,8 +107,10 @@ bool SimpleIndexFile::IndexMetadata::CheckIndexMetadata() {
 
 SimpleIndexFile::SimpleIndexFile(
     base::SingleThreadTaskRunner* cache_thread,
+    base::TaskRunner* worker_pool,
     const base::FilePath& index_file_directory)
     : cache_thread_(cache_thread),
+      worker_pool_(worker_pool),
       index_file_path_(index_file_directory.AppendASCII("the-real-index")) {}
 
 SimpleIndexFile::~SimpleIndexFile() {}
@@ -115,11 +118,10 @@ SimpleIndexFile::~SimpleIndexFile() {}
 void SimpleIndexFile::LoadIndexEntries(
     scoped_refptr<base::SingleThreadTaskRunner> response_thread,
     const IndexCompletionCallback& completion_callback) {
-  base::WorkerPool::PostTask(
+  worker_pool_->PostTask(
       FROM_HERE,
       base::Bind(&SimpleIndexFile::LoadIndexEntriesInternal,
-                 index_file_path_, response_thread, completion_callback),
-      true);
+                 index_file_path_, response_thread, completion_callback));
 }
 
 void SimpleIndexFile::WriteToDisk(const SimpleIndex::EntrySet& entry_set,
@@ -142,14 +144,13 @@ void SimpleIndexFile::DoomEntrySet(
   scoped_ptr<int> result(new int());
   int* result_p(result.get());
 
-  base::WorkerPool::PostTaskAndReply(
+  worker_pool_->PostTaskAndReply(
       FROM_HERE,
       base::Bind(&SimpleSynchronousEntry::DoomEntrySet,
                  base::Passed(entry_hashes.Pass()), index_file_path_.DirName(),
                  result_p),
       base::Bind(&DoomEntrySetReply, base::Passed(result.Pass()),
-                 reply_callback),
-      true);
+                 reply_callback));
 }
 
 // static
@@ -171,7 +172,7 @@ bool SimpleIndexFile::IsIndexFileStale(const base::FilePath& index_filename) {
 scoped_ptr<SimpleIndex::EntrySet> SimpleIndexFile::LoadFromDisk(
     const base::FilePath& index_filename) {
   std::string contents;
-  if(!file_util::ReadFileToString(index_filename, &contents)) {
+  if (!file_util::ReadFileToString(index_filename, &contents)) {
     LOG(WARNING) << "Could not read Simple Index file.";
     return scoped_ptr<SimpleIndex::EntrySet>();
   }
diff --git a/net/disk_cache/simple/simple_index_file.h b/net/disk_cache/simple/simple_index_file.h
index 3fc218a..9034cac 100644
--- a/net/disk_cache/simple/simple_index_file.h
+++ b/net/disk_cache/simple/simple_index_file.h
@@ -18,6 +18,11 @@
 #include "net/base/net_export.h"
 #include "net/disk_cache/simple/simple_index.h"
 
+namespace base {
+class SingleThreadTaskRunner;
+class TaskRunner;
+}
+
 namespace disk_cache {
 
 const uint64 kSimpleIndexMagicNumber = GG_UINT64_C(0x656e74657220796f);
@@ -62,6 +67,7 @@ class NET_EXPORT_PRIVATE SimpleIndexFile {
       IndexCompletionCallback;
 
   explicit SimpleIndexFile(base::SingleThreadTaskRunner* cache_thread,
+                           base::TaskRunner* worker_pool,
                            const base::FilePath& index_file_directory);
   virtual ~SimpleIndexFile();
 
@@ -118,7 +124,8 @@ class NET_EXPORT_PRIVATE SimpleIndexFile {
     uint32 crc;
   };
 
-  scoped_refptr<base::SingleThreadTaskRunner> cache_thread_;
+  const scoped_refptr<base::SingleThreadTaskRunner> cache_thread_;
+  const scoped_refptr<base::TaskRunner> worker_pool_;
   const base::FilePath index_file_path_;
 
   DISALLOW_COPY_AND_ASSIGN(SimpleIndexFile);
diff --git a/net/disk_cache/simple/simple_index_unittest.cc b/net/disk_cache/simple/simple_index_unittest.cc
index 7f023e8..abda20b 100644
--- a/net/disk_cache/simple/simple_index_unittest.cc
+++ b/net/disk_cache/simple/simple_index_unittest.cc
@@ -9,6 +9,7 @@
 #include "base/pickle.h"
 #include "base/sha1.h"
 #include "base/strings/stringprintf.h"
+#include "base/task_runner.h"
 #include "base/threading/platform_thread.h"
 #include "base/time.h"
 #include "net/disk_cache/simple/simple_index.h"
@@ -43,7 +44,7 @@ class TestSimpleIndexFile : public SimpleIndexFile,
                             public base::SupportsWeakPtr<TestSimpleIndexFile> {
  public:
   TestSimpleIndexFile()
-      : SimpleIndexFile(NULL, base::FilePath()),
+      : SimpleIndexFile(NULL, NULL, base::FilePath()),
         get_index_entries_calls_(0),
         doom_entry_set_calls_(0),
         last_response_thread_(NULL),
@@ -568,7 +569,7 @@ TEST_F(SimpleIndexTest, DiskWritePostponed) {
   index()->Insert("key2");
   index()->UpdateEntrySize("key2", 40);
   EXPECT_TRUE(index()->write_to_disk_timer_.IsRunning());
-  EXPECT_LT(expected_trigger,index()->write_to_disk_timer_.desired_run_time());
+  EXPECT_LT(expected_trigger, index()->write_to_disk_timer_.desired_run_time());
   index()->write_to_disk_timer_.Stop();
 }