Implementation of leveldb-backed PrefStore

base/prefs/leveldb_pref_store.cc

+// Copyright (c) 2012 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "base/prefs/leveldb_pref_store.h"
+
+#include "base/bind.h"
+#include "base/callback.h"
+#include "base/file_util.h"
+#include "base/json/json_string_value_serializer.h"
+#include "base/location.h"
+#include "base/sequenced_task_runner.h"
+#include "base/threading/thread_restrictions.h"
+#include "base/values.h"
+#include "third_party/leveldatabase/env_chromium.h"
+#include "third_party/leveldatabase/src/include/leveldb/db.h"
+
+// This class gets created on the UI thread but used exclusively on the FILE
+// thread.
+class LevelDBPrefStore::FileThreadSerializer {
+ public:
+  FileThreadSerializer(scoped_ptr<leveldb::DB> db) : db_(db.Pass()) {
+  }
+  void WriteValue(const std::string& key, const std::string& value) {
+    leveldb::Status status = db_->Put(leveldb::WriteOptions(), key, value);
+    // DCHECK is fine; the corresponding error is ignored in JsonPrefStore.
+    // There's also no API available to surface the error back up to the caller.
+    DCHECK(status.ok()) << status.ToString();
+  }
+  void RemoveValue(const std::string& key) {
+    leveldb::Status status = db_->Delete(leveldb::WriteOptions(), key);
+    // DCHECK is fine; the corresponding error is ignored in JsonPrefStore.
+    DCHECK(status.ok()) << status.ToString();
+  }
+
+ private:
+  scoped_ptr<leveldb::DB> db_;
+  DISALLOW_COPY_AND_ASSIGN(FileThreadSerializer);
+};
+
+class FileThreadDeserializer
+    : public base::RefCountedThreadSafe<FileThreadDeserializer> {

[Mattias Nissler (ping if slow), 2014/03/05 18:28:15]

See
https://groups.google.com/a/chromium.org/d/msg/chromium-dev/o05a009KHsQ/pyw7a...
for recent coverage of RefCountedThreadSafe. I wouldn't recommend using it in
new code, and agree that clear ownership is usually the better design.

In this case, you most likely want PostTaskAndReply with a static function that
does the reading and passes back the results in the reply.

[dgrogan, 2014/03/06 03:03:40]

Done.

+ public:
+  FileThreadDeserializer(LevelDBPrefStore* delegate,
+                         base::SequencedTaskRunner* sequenced_task_runner)
+      : no_dir_(false),
+        error_(PersistentPrefStore::PREF_READ_ERROR_NONE),
+        delegate_(delegate),
+        sequenced_task_runner_(sequenced_task_runner),
+        origin_loop_proxy_(base::MessageLoopProxy::current()) {
+  }
+
+  void Start(const base::FilePath& path) {
+    DCHECK(origin_loop_proxy_->BelongsToCurrentThread());
+    sequenced_task_runner_->PostTask(
+        FROM_HERE,
+        base::Bind(&FileThreadDeserializer::ReadFileAndReport, this, path));
+  }
+
+  void ReadFileAndReport(const base::FilePath& path) {
+    DCHECK(sequenced_task_runner_->RunsTasksOnCurrentThread());
+
+    value_map_ = DoReading(path, &error_, &no_dir_, &db_);
+
+    origin_loop_proxy_->PostTask(
+        FROM_HERE,
+        base::Bind(&FileThreadDeserializer::ReportOnOriginThread, this));
+  }
+
+  // Reports deserialization result on the origin thread.
+  void ReportOnOriginThread() {
+    DCHECK(origin_loop_proxy_->BelongsToCurrentThread());
+    delegate_->OnStorageRead(db_, value_map_.Pass(), error_, no_dir_);
+  }
+
+  static scoped_ptr<PrefValueMap> DoReading(
+      const base::FilePath& path,
+      PersistentPrefStore::PrefReadError* error,
+      bool* no_dir,
+      leveldb::DB** db) {
+    base::ThreadRestrictions::AssertIOAllowed();
+
+    *no_dir = !base::PathExists(path.DirName());
+
+    leveldb::Options options;
+    options.create_if_missing = true;
+    leveldb::Status status =
+        leveldb::DB::Open(options, path.AsUTF8Unsafe(), db);
+    DCHECK(status.ok()) << status.ToString();
+    if (!status.ok()) {
+      *error = PersistentPrefStore::PREF_READ_ERROR_FILE_OTHER;
+      if (leveldb_env::IsCorruption(status)) {
+        *error = PersistentPrefStore::PREF_READ_ERROR_LEVELDB_CORRUPTION;
+        // TODO(dgrogan): Call RepairDB.
+      } else if (leveldb_env::IsIOError(status)) {
+        *error = PersistentPrefStore::PREF_READ_ERROR_LEVELDB_IO_ERROR;
+      }
+      // TODO(dgrogan): Add UMA histogram?
+      return scoped_ptr<PrefValueMap>();
+    }
+
+    scoped_ptr<PrefValueMap> value_map(new PrefValueMap);
+    scoped_ptr<leveldb::Iterator> it(
+        (*db)->NewIterator(leveldb::ReadOptions()));
+    for (it->SeekToFirst(); it->Valid(); it->Next()) {
+      const std::string value_string = it->value().ToString();
+      JSONStringValueSerializer deserializer(value_string);
+      std::string error_message;
+      int error_code;
+      base::Value* json_value =
+          deserializer.Deserialize(&error_code, &error_message);
+      if (json_value)
+        value_map->SetValue(it->key().ToString(), json_value);
+      else
+        NOTREACHED() << error_message;
+    }
+
+    if (!it->status().ok()) {
+      // TODO(dgrogan): Add error histogram.
+      *error = PersistentPrefStore::PREF_READ_ERROR_FILE_OTHER;
+    }
+    *error = PersistentPrefStore::PREF_READ_ERROR_NONE;
+    return value_map.Pass();
+  }
+
+ private:
+  friend class base::RefCountedThreadSafe<FileThreadDeserializer>;
+  ~FileThreadDeserializer() {}
+
+  bool no_dir_;
+  PersistentPrefStore::PrefReadError error_;
+  scoped_ptr<PrefValueMap> value_map_;
+  leveldb::DB* db_;
+  const scoped_refptr<LevelDBPrefStore> delegate_;
+  const scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner_;
+  const scoped_refptr<base::MessageLoopProxy> origin_loop_proxy_;
+};
+
+LevelDBPrefStore::LevelDBPrefStore(
+    const base::FilePath& filename,
+    base::SequencedTaskRunner* sequenced_task_runner)
+    : path_(filename),
+      sequenced_task_runner_(sequenced_task_runner),
+      original_task_runner_(base::MessageLoopProxy::current()),
+      read_only_(false),
+      initialized_(false),
+      read_error_(PREF_READ_ERROR_OTHER) {
+}
+
+LevelDBPrefStore::~LevelDBPrefStore() {
+  sequenced_task_runner_->DeleteSoon(FROM_HERE, helper_.release());
+}
+
+bool LevelDBPrefStore::GetValue(const std::string& key,
+                                const base::Value** result) const {
+  DCHECK(initialized_);
+  const base::Value* tmp = NULL;
+  if (!prefs_.GetValue(key, &tmp)) {
+    return false;
+  }
+
+  if (result)
+    *result = tmp;
+  return true;
+}
+
+// Callers of GetMutableValue have to also call ReportValueChanged.
+bool LevelDBPrefStore::GetMutableValue(const std::string& key,
+                                       base::Value** result) {
+  DCHECK(initialized_);
+  return prefs_.GetValue(key, result);
+}
+
+void LevelDBPrefStore::AddObserver(PrefStore::Observer* observer) {
+  observers_.AddObserver(observer);
+}
+
+void LevelDBPrefStore::RemoveObserver(PrefStore::Observer* observer) {
+  observers_.RemoveObserver(observer);
+}
+
+bool LevelDBPrefStore::HasObservers() const {
+  return observers_.might_have_observers();
+}
+
+bool LevelDBPrefStore::IsInitializationComplete() const { return initialized_; }
+
+void LevelDBPrefStore::RemoveFromUIThread(const std::string& key) {
+  if (read_only_)
+    return;
+  DCHECK(helper_);
+  sequenced_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&LevelDBPrefStore::FileThreadSerializer::RemoveValue,
+                 base::Unretained(helper_.get()),
+                 key));
+}
+
+void LevelDBPrefStore::PersistFromUIThread(const std::string& key,
+                                           base::Value* value) {
+  if (read_only_)
+    return;
+  std::string value_string;
+  JSONStringValueSerializer serializer(&value_string);
+  bool serialized_ok = serializer.Serialize(*value);
+  DCHECK(serialized_ok);
+
+  DCHECK(helper_);
+  sequenced_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&LevelDBPrefStore::FileThreadSerializer::WriteValue,
+                 base::Unretained(helper_.get()),
+                 key,
+                 value_string));
+}
+
+void LevelDBPrefStore::SetValue(const std::string& key, base::Value* value) {
+  DCHECK(initialized_);
+  DCHECK(value);
+  scoped_ptr<base::Value> new_value(value);
+  base::Value* old_value = NULL;
+  bool found = prefs_.GetValue(key, &old_value);
+  if (!found || !value->Equals(old_value)) {
+    PersistFromUIThread(key, new_value.get());
+    prefs_.SetValue(key, new_value.release());
+    NotifyObservers(key);
+  }
+}
+
+void LevelDBPrefStore::SetValueSilently(const std::string& key,
+                                        base::Value* value) {
+  DCHECK(initialized_);
+  DCHECK(value);
+  scoped_ptr<base::Value> new_value(value);
+  base::Value* old_value = NULL;
+  prefs_.GetValue(key, &old_value);
+  if (!old_value || !value->Equals(old_value)) {
+    PersistFromUIThread(key, new_value.get());
+    prefs_.SetValue(key, new_value.release());
+  }
+}
+
+void LevelDBPrefStore::RemoveValue(const std::string& key) {
+  DCHECK(initialized_);
+  if (prefs_.RemoveValue(key)) {
+    RemoveFromUIThread(key);
+    NotifyObservers(key);
+  }
+}
+
+bool LevelDBPrefStore::ReadOnly() const { return read_only_; }
+
+PersistentPrefStore::PrefReadError LevelDBPrefStore::GetReadError() const {
+  return read_error_;
+}
+
+PersistentPrefStore::PrefReadError LevelDBPrefStore::ReadPrefs() {
+  DCHECK(!initialized_);
+  if (path_.empty()) {
+    OnStorageRead(NULL,
+                  scoped_ptr<PrefValueMap>(),
+                  PREF_READ_ERROR_FILE_NOT_SPECIFIED,
+                  false);
+    return PREF_READ_ERROR_FILE_NOT_SPECIFIED;
+  }
+
+  PrefReadError error;
+  bool no_dir;
+  leveldb::DB* db = NULL;
+  scoped_ptr<PrefValueMap> map =
+      FileThreadDeserializer::DoReading(path_, &error, &no_dir, &db);
+  OnStorageRead(db, map.Pass(), error, no_dir);
+  return error;
+}
+
+void LevelDBPrefStore::ReadPrefsAsync(ReadErrorDelegate* error_delegate) {
+  DCHECK_EQ(false, initialized_);
+  error_delegate_.reset(error_delegate);
+  if (path_.empty()) {
+    OnStorageRead(NULL,
+                  scoped_ptr<PrefValueMap>(),
+                  PREF_READ_ERROR_FILE_NOT_SPECIFIED,
+                  false);
+    return;
+  }
+  scoped_refptr<FileThreadDeserializer> deserializer(
+      new FileThreadDeserializer(this, sequenced_task_runner_.get()));
+  deserializer->Start(path_);
+}
+
+void LevelDBPrefStore::CommitPendingWrite() {
+  // No-op, key-value pairs are committed as they are set.
+  // TODO(dgrogan): If this doesn't perform well because, for example, the
+  // extensions prefs are changed frequently then we'll have to track a set of
+  // keys that have changed and write them to disk every k seconds.
+}
+
+void LevelDBPrefStore::ReportValueChanged(const std::string& key) {
+  base::Value* new_value = NULL;
+  DCHECK(prefs_.GetValue(key, &new_value));
+  PersistFromUIThread(key, new_value);
+  NotifyObservers(key);
+}
+
+void LevelDBPrefStore::NotifyObservers(const std::string& key) {
+  FOR_EACH_OBSERVER(PrefStore::Observer, observers_, OnPrefValueChanged(key));
+}
+
+void LevelDBPrefStore::OnStorageRead(leveldb::DB* db_owned,
+                                     scoped_ptr<PrefValueMap> value,
+                                     PersistentPrefStore::PrefReadError error,
+                                     bool no_dir) {
+  scoped_ptr<leveldb::DB> db(db_owned);
+  read_error_ = error;
+
+  if (no_dir) {
+    FOR_EACH_OBSERVER(
+        PrefStore::Observer, observers_, OnInitializationCompleted(false));
+    return;
+  }
+
+  initialized_ = true;
+
+  switch (error) {
+    case PREF_READ_ERROR_FILE_OTHER:
+    case PREF_READ_ERROR_LEVELDB_IO_ERROR:
+    case PREF_READ_ERROR_LEVELDB_CORRUPTION:
+    case PREF_READ_ERROR_FILE_NOT_SPECIFIED:
+      DCHECK(db.get() == NULL);
+      DCHECK(value.get() == NULL);
+      read_only_ = true;
+      break;
+    case PREF_READ_ERROR_NONE:
+      helper_.reset(new FileThreadSerializer(db.Pass()));
+      prefs_.Swap(value.get());
+      break;
+    default:
+      NOTREACHED() << "Unknown error: " << error;
+  }
+
+  // TODO(dgrogan): Call pref_filter_->FilterOnLoad
+
+  if (error_delegate_.get() && error != PREF_READ_ERROR_NONE)
+    error_delegate_->OnError(error);
+
+  FOR_EACH_OBSERVER(
+      PrefStore::Observer, observers_, OnInitializationCompleted(true));
+}