Moved JsonPrefStore to use SequencedWorkerPool instead of FILE thread…

…. The pool also ensures that the same file requests are written in order received and that they block on shutdown.

BUG=153367
TEST=existing unit/browser tests

Review URL: https://codereview.chromium.org/11027070

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

base/files/important_file_writer.cc

@@ -13,7 +13,7 @@
 #include "base/file_path.h"
 #include "base/file_util.h"
 #include "base/logging.h"
-#include "base/message_loop_proxy.h"
+#include "base/task_runner.h"
 #include "base/metrics/histogram.h"
 #include "base/string_number_conversions.h"
 #include "base/threading/thread.h"
@@ -90,14 +90,14 @@ void WriteToDiskTask(const FilePath& path, const std::string& data) {
 }  // namespace
 
 ImportantFileWriter::ImportantFileWriter(
-    const FilePath& path, MessageLoopProxy* file_message_loop_proxy)
+    const FilePath& path, base::SequencedTaskRunner* task_runner)
         : path_(path),
-          file_message_loop_proxy_(file_message_loop_proxy),
+          task_runner_(task_runner),
           serializer_(NULL),
           commit_interval_(TimeDelta::FromMilliseconds(
               kDefaultCommitIntervalMs)) {
   DCHECK(CalledOnValidThread());
-  DCHECK(file_message_loop_proxy_.get());
+  DCHECK(task_runner_.get());
 }
 
 ImportantFileWriter::~ImportantFileWriter() {
@@ -122,8 +122,8 @@ void ImportantFileWriter::WriteNow(const std::string& data) {
   if (HasPendingWrite())
     timer_.Stop();
 
-  if (!file_message_loop_proxy_->PostTask(
-      FROM_HERE, MakeCriticalClosure(Bind(&WriteToDiskTask, path_, data)))) {
+  if (!task_runner_->PostTask(FROM_HERE,
+          MakeCriticalClosure(Bind(&WriteToDiskTask, path_, data)))) {
     // Posting the task to background message loop is not expected
     // to fail, but if it does, avoid losing data and just hit the disk
     // on the current thread.

base/files/important_file_writer.h

@@ -17,7 +17,7 @@
 
 namespace base {
 
-class MessageLoopProxy;
+class SequencedTaskRunner;
 class Thread;
 
 // Helper to ensure that a file won't be corrupted by the write (for example on
@@ -53,11 +53,11 @@ class BASE_EXPORT ImportantFileWriter : public NonThreadSafe {
 
   // Initialize the writer.
   // |path| is the name of file to write.
-  // |file_message_loop_proxy| is the MessageLoopProxy for a thread on which
-  // file I/O can be done.
+  // |task_runner| is the SequencedTaskRunner instance where on which we will
+  // execute file I/O operations.
   // All non-const methods, ctor and dtor must be called on the same thread.
   ImportantFileWriter(const FilePath& path,
-                      MessageLoopProxy* file_message_loop_proxy);
+                      base::SequencedTaskRunner* task_runner);
 
   // You have to ensure that there are no pending writes at the moment
   // of destruction.
@@ -96,8 +96,8 @@ class BASE_EXPORT ImportantFileWriter : public NonThreadSafe {
   // Path being written to.
   const FilePath path_;
 
-  // MessageLoopProxy for the thread on which file I/O can be done.
-  scoped_refptr<MessageLoopProxy> file_message_loop_proxy_;
+  // TaskRunner for the thread on which file I/O can be done.
+  const scoped_refptr<base::SequencedTaskRunner> task_runner_;
 
   // Timer used to schedule commit after ScheduleWrite.
   OneShotTimer<ImportantFileWriter> timer_;

base/prefs/json_pref_store.cc

@@ -13,6 +13,8 @@
 #include "base/json/json_string_value_serializer.h"
 #include "base/memory/ref_counted.h"
 #include "base/message_loop_proxy.h"
+#include "base/sequenced_task_runner.h"
+#include "base/threading/sequenced_worker_pool.h"
 #include "base/values.h"
 
 namespace {
@@ -26,25 +28,25 @@ class FileThreadDeserializer
     : public base::RefCountedThreadSafe<FileThreadDeserializer> {
  public:
   FileThreadDeserializer(JsonPrefStore* delegate,
-                         base::MessageLoopProxy* file_loop_proxy)
+                         base::SequencedTaskRunner* sequenced_task_runner)
       : no_dir_(false),
         error_(PersistentPrefStore::PREF_READ_ERROR_NONE),
         delegate_(delegate),
-        file_loop_proxy_(file_loop_proxy),
+        sequenced_task_runner_(sequenced_task_runner),
         origin_loop_proxy_(base::MessageLoopProxy::current()) {
   }
 
   void Start(const FilePath& path) {
     DCHECK(origin_loop_proxy_->BelongsToCurrentThread());
-    file_loop_proxy_->PostTask(
+    sequenced_task_runner_->PostTask(
         FROM_HERE,
         base::Bind(&FileThreadDeserializer::ReadFileAndReport,
                    this, path));
   }
 
-  // Deserializes JSON on the file thread.
+  // Deserializes JSON on the sequenced task runner.
   void ReadFileAndReport(const FilePath& path) {
-    DCHECK(file_loop_proxy_->BelongsToCurrentThread());
+    DCHECK(sequenced_task_runner_->RunsTasksOnCurrentThread());
 
     value_.reset(DoReading(path, &error_, &no_dir_));
 
@@ -84,9 +86,9 @@ class FileThreadDeserializer
   bool no_dir_;
   PersistentPrefStore::PrefReadError error_;
   scoped_ptr<Value> value_;
-  scoped_refptr<JsonPrefStore> delegate_;
-  scoped_refptr<base::MessageLoopProxy> file_loop_proxy_;
-  scoped_refptr<base::MessageLoopProxy> origin_loop_proxy_;
+  const scoped_refptr<JsonPrefStore> delegate_;
+  const scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner_;
+  const scoped_refptr<base::MessageLoopProxy> origin_loop_proxy_;
 };
 
 // static
@@ -98,7 +100,8 @@ void FileThreadDeserializer::HandleErrors(
     PersistentPrefStore::PrefReadError* error) {
   *error = PersistentPrefStore::PREF_READ_ERROR_NONE;
   if (!value) {
-    DVLOG(1) << "Error while loading JSON file: " << error_msg;
+    DVLOG(1) << "Error while loading JSON file: " << error_msg
+             << ", file: " << path.value();
     switch (error_code) {
       case JSONFileValueSerializer::JSON_ACCESS_DENIED:
         *error = PersistentPrefStore::PREF_READ_ERROR_ACCESS_DENIED;
@@ -137,13 +140,23 @@ void FileThreadDeserializer::HandleErrors(
 
 }  // namespace
 
+scoped_refptr<base::SequencedTaskRunner> JsonPrefStore::GetTaskRunnerForFile(
+    const FilePath& filename,
+    base::SequencedWorkerPool* worker_pool) {
+  std::string token("json_pref_store-");
+  token.append(filename.AsUTF8Unsafe());
+  return worker_pool->GetSequencedTaskRunnerWithShutdownBehavior(
+      worker_pool->GetNamedSequenceToken(token),
+      base::SequencedWorkerPool::BLOCK_SHUTDOWN);
+}
+
 JsonPrefStore::JsonPrefStore(const FilePath& filename,
-                             base::MessageLoopProxy* file_message_loop_proxy)
+                             base::SequencedTaskRunner* sequenced_task_runner)
     : path_(filename),
-      file_message_loop_proxy_(file_message_loop_proxy),
+      sequenced_task_runner_(sequenced_task_runner),
       prefs_(new DictionaryValue()),
       read_only_(false),
-      writer_(filename, file_message_loop_proxy),
+      writer_(filename, sequenced_task_runner),
       error_delegate_(NULL),
       initialized_(false),
       read_error_(PREF_READ_ERROR_OTHER) {
@@ -245,7 +258,7 @@ void JsonPrefStore::ReadPrefsAsync(ReadErrorDelegate *error_delegate) {
   // Start async reading of the preferences file. It will delete itself
   // in the end.
   scoped_refptr<FileThreadDeserializer> deserializer(
-      new FileThreadDeserializer(this, file_message_loop_proxy_.get()));
+      new FileThreadDeserializer(this, sequenced_task_runner_.get()));
   deserializer->Start(path_);
 }
 

base/prefs/json_pref_store.h

@@ -20,7 +20,8 @@
 
 namespace base {
 class DictionaryValue;
-class MessageLoopProxy;
+class SequencedWorkerPool;
+class SequencedTaskRunner;
 class Value;
 }
 
@@ -31,10 +32,16 @@ class BASE_PREFS_EXPORT JsonPrefStore
     : public PersistentPrefStore,
       public base::ImportantFileWriter::DataSerializer {
  public:
-  // |file_message_loop_proxy| is the MessageLoopProxy for a thread on which
-  // file I/O can be done.
+  // Returns instance of SequencedTaskRunner which guarantees that file
+  // operations on the same file will be executed in sequenced order.
+  static scoped_refptr<base::SequencedTaskRunner> GetTaskRunnerForFile(
+      const FilePath& pref_filename,
+      base::SequencedWorkerPool* worker_pool);
+
+  // |sequenced_task_runner| is must be a shutdown-blocking task runner, ideally
+  // created by GetTaskRunnerForFile() method above.
   JsonPrefStore(const FilePath& pref_filename,
-                base::MessageLoopProxy* file_message_loop_proxy);
+                base::SequencedTaskRunner* sequenced_task_runner);
 
   // PrefStore overrides:
   virtual ReadResult GetValue(const std::string& key,
@@ -72,7 +79,7 @@ class BASE_PREFS_EXPORT JsonPrefStore
   virtual bool SerializeData(std::string* output) OVERRIDE;
 
   FilePath path_;
-  scoped_refptr<base::MessageLoopProxy> file_message_loop_proxy_;
+  const scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner_;
 
   scoped_ptr<base::DictionaryValue> prefs_;
 

base/prefs/json_pref_store_unittest.cc

@@ -5,13 +5,12 @@
 #include "base/file_util.h"
 #include "base/memory/ref_counted.h"
 #include "base/memory/scoped_ptr.h"
-#include "base/message_loop.h"
-#include "base/message_loop_proxy.h"
 #include "base/path_service.h"
 #include "base/prefs/json_pref_store.h"
 #include "base/scoped_temp_dir.h"
 #include "base/string_number_conversions.h"
 #include "base/string_util.h"
+#include "base/threading/sequenced_worker_pool.h"
 #include "base/threading/thread.h"
 #include "base/utf_string_conversions.h"
 #include "base/values.h"
@@ -37,9 +36,7 @@ class MockReadErrorDelegate : public PersistentPrefStore::ReadErrorDelegate {
 
 class JsonPrefStoreTest : public testing::Test {
  protected:
-  virtual void SetUp() {
-    message_loop_proxy_ = base::MessageLoopProxy::current();
-
+  virtual void SetUp() OVERRIDE {
     ASSERT_TRUE(temp_dir_.CreateUniqueTempDir());
 
     ASSERT_TRUE(PathService::Get(chrome::DIR_TEST_DATA, &data_dir_));
@@ -53,15 +50,15 @@ class JsonPrefStoreTest : public testing::Test {
   FilePath data_dir_;
   // A message loop that we can use as the file thread message loop.
   MessageLoop message_loop_;
-  scoped_refptr<base::MessageLoopProxy> message_loop_proxy_;
 };
 
 // Test fallback behavior for a nonexistent file.
 TEST_F(JsonPrefStoreTest, NonExistentFile) {
   FilePath bogus_input_file = data_dir_.AppendASCII("read.txt");
   ASSERT_FALSE(file_util::PathExists(bogus_input_file));
   scoped_refptr<JsonPrefStore> pref_store =
-      new JsonPrefStore(bogus_input_file, message_loop_proxy_.get());
+      new JsonPrefStore(
+          bogus_input_file, message_loop_.message_loop_proxy());
   EXPECT_EQ(PersistentPrefStore::PREF_READ_ERROR_NO_FILE,
             pref_store->ReadPrefs());
   EXPECT_FALSE(pref_store->ReadOnly());
@@ -73,7 +70,8 @@ TEST_F(JsonPrefStoreTest, InvalidFile) {
   FilePath invalid_file = temp_dir_.path().AppendASCII("invalid.json");
   ASSERT_TRUE(file_util::CopyFile(invalid_file_original, invalid_file));
   scoped_refptr<JsonPrefStore> pref_store =
-      new JsonPrefStore(invalid_file, message_loop_proxy_.get());
+      new JsonPrefStore(
+          invalid_file, message_loop_.message_loop_proxy());
   EXPECT_EQ(PersistentPrefStore::PREF_READ_ERROR_JSON_PARSE,
             pref_store->ReadPrefs());
   EXPECT_FALSE(pref_store->ReadOnly());
@@ -88,7 +86,7 @@ TEST_F(JsonPrefStoreTest, InvalidFile) {
 
 // This function is used to avoid code duplication while testing synchronous and
 // asynchronous version of the JsonPrefStore loading.
-void RunBasicJsonPrefStoreTest(JsonPrefStore *pref_store,
+void RunBasicJsonPrefStoreTest(JsonPrefStore* pref_store,
                                const FilePath& output_file,
                                const FilePath& golden_output_file) {
   const char kNewWindowsInTabs[] = "tabs.new_windows_in_tabs";
@@ -166,7 +164,8 @@ TEST_F(JsonPrefStoreTest, Basic) {
   FilePath input_file = temp_dir_.path().AppendASCII("write.json");
   ASSERT_TRUE(file_util::PathExists(input_file));
   scoped_refptr<JsonPrefStore> pref_store =
-      new JsonPrefStore(input_file, message_loop_proxy_.get());
+      new JsonPrefStore(
+          input_file, message_loop_.message_loop_proxy());
   ASSERT_EQ(PersistentPrefStore::PREF_READ_ERROR_NONE, pref_store->ReadPrefs());
   ASSERT_FALSE(pref_store->ReadOnly());
 
@@ -193,21 +192,24 @@ TEST_F(JsonPrefStoreTest, BasicAsync) {
   FilePath input_file = temp_dir_.path().AppendASCII("write.json");
   ASSERT_TRUE(file_util::PathExists(input_file));
   scoped_refptr<JsonPrefStore> pref_store =
-      new JsonPrefStore(input_file, message_loop_proxy_.get());
+      new JsonPrefStore(
+          input_file, message_loop_.message_loop_proxy());
 
-  MockPrefStoreObserver mock_observer;
-  pref_store->AddObserver(&mock_observer);
+  {
+    MockPrefStoreObserver mock_observer;
+    pref_store->AddObserver(&mock_observer);
 
-  MockReadErrorDelegate *mock_error_delegate = new MockReadErrorDelegate;
-  pref_store->ReadPrefsAsync(mock_error_delegate);
+    MockReadErrorDelegate* mock_error_delegate = new MockReadErrorDelegate;
+    pref_store->ReadPrefsAsync(mock_error_delegate);
 
-  EXPECT_CALL(mock_observer, OnInitializationCompleted(true)).Times(1);
-  EXPECT_CALL(*mock_error_delegate,
-              OnError(PersistentPrefStore::PREF_READ_ERROR_NONE)).Times(0);
-  message_loop_.RunUntilIdle();
-  pref_store->RemoveObserver(&mock_observer);
+    EXPECT_CALL(mock_observer, OnInitializationCompleted(true)).Times(1);
+    EXPECT_CALL(*mock_error_delegate,
+                OnError(PersistentPrefStore::PREF_READ_ERROR_NONE)).Times(0);
+    message_loop_.RunUntilIdle();
+    pref_store->RemoveObserver(&mock_observer);
 
-  ASSERT_FALSE(pref_store->ReadOnly());
+    ASSERT_FALSE(pref_store->ReadOnly());
+  }
 
   // The JSON file looks like this:
   // {
@@ -229,7 +231,8 @@ TEST_F(JsonPrefStoreTest, AsyncNonExistingFile) {
   FilePath bogus_input_file = data_dir_.AppendASCII("read.txt");
   ASSERT_FALSE(file_util::PathExists(bogus_input_file));
   scoped_refptr<JsonPrefStore> pref_store =
-      new JsonPrefStore(bogus_input_file, message_loop_proxy_.get());
+      new JsonPrefStore(
+          bogus_input_file, message_loop_.message_loop_proxy());
   MockPrefStoreObserver mock_observer;
   pref_store->AddObserver(&mock_observer);
 
@@ -255,7 +258,8 @@ TEST_F(JsonPrefStoreTest, NeedsEmptyValue) {
   // Test that the persistent value can be loaded.
   ASSERT_TRUE(file_util::PathExists(pref_file));
   scoped_refptr<JsonPrefStore> pref_store =
-      new JsonPrefStore(pref_file, message_loop_proxy_.get());
+      new JsonPrefStore(
+          pref_file, message_loop_.message_loop_proxy());
   ASSERT_EQ(PersistentPrefStore::PREF_READ_ERROR_NONE, pref_store->ReadPrefs());
   ASSERT_FALSE(pref_store->ReadOnly());
 

chrome/browser/automation/testing_automation_provider.cc

@@ -19,6 +19,7 @@
 #include "base/path_service.h"
 #include "base/process.h"
 #include "base/process_util.h"
+#include "base/sequenced_task_runner.h"
 #include "base/stringprintf.h"
 #include "base/threading/thread_restrictions.h"
 #include "base/time.h"
@@ -1288,7 +1289,9 @@ void TestingAutomationProvider::GetBookmarksAsJSON(
     return;
   }
   scoped_refptr<BookmarkStorage> storage(
-      new BookmarkStorage(browser->profile(), bookmark_model));
+      new BookmarkStorage(browser->profile(),
+                          bookmark_model,
+                          browser->profile()->GetIOTaskRunner()));
   if (!storage->SerializeData(&bookmarks_as_json)) {
     reply.SendError("Failed to serialize bookmarks");
     return;

chrome/browser/bookmarks/bookmark_model.cc

@@ -11,6 +11,7 @@
 #include "base/bind_helpers.h"
 #include "base/json/json_string_value_serializer.h"
 #include "base/memory/scoped_vector.h"
+#include "base/sequenced_task_runner.h"
 #include "base/string_util.h"
 #include "base/values.h"
 #include "build/build_config.h"
@@ -247,7 +248,7 @@ void BookmarkModel::Load() {
                  content::Source<Profile>(profile_));
 
   // Load the bookmarks. BookmarkStorage notifies us when done.
-  store_ = new BookmarkStorage(profile_, this);
+  store_ = new BookmarkStorage(profile_, this, profile_->GetIOTaskRunner());
   store_->LoadBookmarks(CreateLoadDetails());
 }