Implementation of leveldb-backed PrefStore

base/prefs/leveldb_pref_store.cc

Index: base/prefs/leveldb_pref_store.cc
diff --git a/base/prefs/leveldb_pref_store.cc b/base/prefs/leveldb_pref_store.cc
new file mode 100644
index 0000000000000000000000000000000000000000..e698550ddeac2d6b95c1bc17a733fafa5be59e9d
--- /dev/null
+++ b/base/prefs/leveldb_pref_store.cc
@@ -0,0 +1,271 @@
+// 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/values.h"
+#include "third_party/leveldatabase/env_chromium.h"
+#include "third_party/leveldatabase/src/include/leveldb/db.h"
+

[Mattias Nissler (ping if slow), 2014/02/27 08:49:23]

This should have a comment saying that it gets created on the UI thread but then
operated exclusively on the file thread.

[dgrogan, 2014/03/05 03:06:58]

Done.

+class LevelDBPrefStore::Helper {
+ public:
+  Helper(scoped_ptr<leveldb::DB> db) : db_(db.Pass()) {
+  }
+  void PersistOnFileThread(const std::string& key, const std::string& value) {

[Mattias Nissler (ping if slow), 2014/02/27 08:49:23]

I'd just name the functions here WriteValue() and RemoveValue() or similar. By
virtue of being defined on the helper object it's clear to me that they run on 
the file thread.

[dgrogan, 2014/03/05 03:06:58]

Done.

+    leveldb::Status status = db_->Put(leveldb::WriteOptions(), key, value);
+    // DCHECK is fine; the corresponding error is ignored in json_pref_store.
+    // There's also no API available to surface the error back up to the caller.
+    DCHECK(status.ok()) << status.ToString();
+  }
+  void RemoveOnFileThread(const std::string& key) {
+    leveldb::Status status = db_->Delete(leveldb::WriteOptions(), key);
+    // DCHECK is fine; the corresponding error is ignored in json_pref_store.
+    DCHECK(status.ok()) << status.ToString();
+  }
+
+ private:
+  scoped_ptr<leveldb::DB> db_;

[Mattias Nissler (ping if slow), 2014/02/27 08:49:23]

nit: DISALLOW_COPY_AND_ASSIGN

[dgrogan, 2014/03/05 03:06:58]

Done.

+};
+
+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;

[Mattias Nissler (ping if slow), 2014/02/27 08:49:23]

nit: DCHECK(helper_)

[dgrogan, 2014/03/05 03:06:58]

Done.

+  sequenced_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&LevelDBPrefStore::Helper::RemoveOnFileThread,
+                 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);
+

[Mattias Nissler (ping if slow), 2014/02/27 08:49:23]

nit: DCHECK(helper_)

[dgrogan, 2014/03/05 03:06:58]

Done.

+  sequenced_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&LevelDBPrefStore::Helper::PersistOnFileThread,
+                 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(PREF_READ_ERROR_FILE_NOT_SPECIFIED, false);
+    return PREF_READ_ERROR_FILE_NOT_SPECIFIED;
+  }
+
+  bool no_dir = !base::PathExists(path_.DirName());
+
+  leveldb::DB* db;
+  leveldb::Options options;
+  options.create_if_missing = true;
+  leveldb::Status status =
+      leveldb::DB::Open(options, path_.AsUTF8Unsafe(), &db);
+  helper_.reset(new Helper(make_scoped_ptr(db)));

[dgrogan, 2014/02/27 03:19:48]

This seems sloppy, reading from disk on the UI thread, then passing over the DB
handle to live on the File thread.

[Mattias Nissler (ping if slow), 2014/02/27 08:49:23]

If you want, add a comment saying that this file operation is intentionally
running on the UI thread during browser startup and that code running later will
have to go through ReadPrefsAsync().

Btw., this is enforced by ThreadRestrictions which will assert for file
operations on the UI thread after browser startup. I hope it does so too for
leveldb operations, but might be good to check. If it doesn't assert, I suggest
putting an explicit ThreadRestrictions::AssertIOAllowed() here.

[dgrogan, 2014/03/05 03:06:58]

Comment and ThreadRestrictions::AssertIOAllowed added. LevelDB does not have
ThreadRestrictions::AssertIOAllowed in its functions that touch disk.

+  DCHECK(status.ok()) << status.ToString();
+  if (!status.ok()) {
+    PrefReadError error = PREF_READ_ERROR_FILE_OTHER;
+    if (leveldb_env::IsCorruption(status)) {
+      error = PREF_READ_ERROR_LEVELDB_CORRUPTION;
+      // TODO(dgrogan): Call RepairDB.
+    } else if (leveldb_env::IsIOError(status)) {
+      error = PREF_READ_ERROR_LEVELDB_IO_ERROR;
+    }
+    // TODO(dgrogan): Add UMA histogram?
+    OnStorageRead(error, no_dir);
+    return error;
+  }
+
+  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)
+      prefs_.SetValue(it->key().ToString(), json_value);
+    else
+      NOTREACHED() << error_message;
+  }
+
+  PersistentPrefStore::PrefReadError error = PREF_READ_ERROR_NONE;
+  if (!it->status().ok()) {
+    // TODO(dgrogan): Add error histogram.
+    error = PREF_READ_ERROR_FILE_OTHER;
+  }
+  OnStorageRead(error, no_dir);
+  return error;
+}
+
+void LevelDBPrefStore::ReadPrefsAsync(ReadErrorDelegate* error_delegate) {
+  DCHECK_EQ(false, initialized_);
+  error_delegate_.reset(error_delegate);
+  if (path_.empty()) {
+    OnStorageRead(PREF_READ_ERROR_FILE_NOT_SPECIFIED, false);
+    return;
+  }
+
+  sequenced_task_runner_->PostTask(FROM_HERE, base::Bind(
+      base::IgnoreResult(&LevelDBPrefStore::ReadPrefs), this));

[Mattias Nissler (ping if slow), 2014/02/27 08:49:23]

This needs more love, you can't just updated |prefs_| from a different thread.

[dgrogan, 2014/03/05 03:06:58]

I resurrected FileThreadDeserializer to go with a tried-and-true approach.

+}
+
+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(PersistentPrefStore::PrefReadError error,
+                                     bool no_dir) {
+  if (!original_task_runner_->RunsTasksOnCurrentThread()) {
+    // When db is opened async, this function needs to be called on original
+    // thread.
+    original_task_runner_->PostTask(FROM_HERE,
+        base::Bind(&LevelDBPrefStore::OnStorageRead, this, error, no_dir));
+  }
+  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:
+      read_only_ = true;
+      break;
+    case PREF_READ_ERROR_NONE:
+      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));
+}