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..d2546feea4de8ef6010dc4ad6b51f4d8edecd753
--- /dev/null
+++ b/base/prefs/leveldb_pref_store.cc
@@ -0,0 +1,428 @@
+// 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/task_runner_util.h"
+#include "base/threading/thread_restrictions.h"
+#include "base/time/time.h"
+#include "base/values.h"
+#include "third_party/leveldatabase/env_chromium.h"
+#include "third_party/leveldatabase/src/include/leveldb/db.h"
+#include "third_party/leveldatabase/src/include/leveldb/write_batch.h"
+
+struct LevelDBPrefStore::ReadingResults {
+  bool no_dir;
+  scoped_ptr<leveldb::DB> db;
+  scoped_ptr<PrefValueMap> value_map;
+  PersistentPrefStore::PrefReadError error;
+};
+
+// An instance of this class is created on the UI thread but is used
+// exclusively on the FILE thread.
+class LevelDBPrefStore::FileThreadSerializer {
+ public:
+  FileThreadSerializer(scoped_ptr<leveldb::DB> db) : db_(db.Pass()) {}
+  void WriteToDatabase(
+      scoped_ptr<std::map<std::string, std::string> > keys_to_set,
+      scoped_ptr<std::set<std::string> > keys_to_delete) {
+    DCHECK(keys_to_set->size() > 0 || keys_to_delete->size() > 0);
+    leveldb::WriteBatch batch;
+    for (std::map<std::string, std::string>::iterator iter =
+             keys_to_set->begin();
+         iter != keys_to_set->end();
+         iter++) {
+      batch.Put(iter->first, iter->second);
+    }
+
+    for (std::set<std::string>::iterator iter = keys_to_delete->begin();
+         iter != keys_to_delete->end();
+         iter++) {
+      batch.Delete(*iter);
+    }
+
+    leveldb::Status status = db_->Write(leveldb::WriteOptions(), &batch);
+
+    // 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.
+    // TODO(dgrogan): UMA?
+    DCHECK(status.ok()) << status.ToString();
+  }
+
+ private:
+  scoped_ptr<leveldb::DB> db_;
+  DISALLOW_COPY_AND_ASSIGN(FileThreadSerializer);
+};
+
+bool MoveDirectoryAside(const base::FilePath& path) {
+  base::FilePath bad_path = path.AppendASCII(".bad");
+  if (!base::Move(path, bad_path)) {
+    base::DeleteFile(bad_path, true);
+    return false;
+  }
+  return true;
+}
+
+/* static */
+void LevelDBPrefStore::OpenDB(const base::FilePath& path,
+                              ReadingResults* reading_results) {
+  leveldb::Options options;
+  options.create_if_missing = true;
+  leveldb::DB* db;
+  leveldb::Status status = leveldb::DB::Open(options, path.AsUTF8Unsafe(), &db);
+  if (status.ok()) {
+    reading_results->db.reset(db);
+    reading_results->error = PREF_READ_ERROR_NONE;
+    return;
+  }
+  if (leveldb_env::IsIOError(status)) {
+    reading_results->error = PREF_READ_ERROR_LEVELDB_IO_ERROR;
+    return;
+  }
+  // If it's not an IO error, it's corruption that we can try to repair.
+  status = leveldb::RepairDB(path.AsUTF8Unsafe(), options);
+  bool destroy_succeeded = false;
+  if (!status.ok()) {
+    if (!MoveDirectoryAside(path)) {
+      status = leveldb::DestroyDB(path.AsUTF8Unsafe(), options);
+      if (!status.ok()) {
+        reading_results->error = PREF_READ_ERROR_LEVELDB_DESTROY_FAILED;
+        return;
+      }
+    }
+    destroy_succeeded = true;
+  }
+  // Either repair or destroy succeeded, now try to open again.
+  status = leveldb::DB::Open(options, path.AsUTF8Unsafe(), &db);
+  if (status.ok()) {
+    reading_results->db.reset(db);
+    if (destroy_succeeded) {

[Mattias Nissler (ping if slow), 2014/04/14 10:11:41]

nit: curlies not needed or you could convert lines 106-110 into a single ternary
expression.

[dgrogan, 2014/04/17 01:10:27]

Done.

+      reading_results->error = PREF_READ_ERROR_LEVELDB_DESTROYED_REOPENED;
+    } else {
+      reading_results->error = PREF_READ_ERROR_LEVELDB_REPAIRED_REOPENED;
+    }
+    return;
+  }
+  if (destroy_succeeded) {
+    reading_results->error =
+        PREF_READ_ERROR_LEVELDB_DESTROYED_REOPEN_FAILED;
+    return;
+  }
+  status = leveldb::DestroyDB(path.AsUTF8Unsafe(), options);
+  if (!status.ok()) {
+    reading_results->error =
+        PREF_READ_ERROR_LEVELDB_REPAIRED_REOPEN_FAILED_DESTROY_FAILED;
+    return;
+  }
+  status = leveldb::DB::Open(options, path.AsUTF8Unsafe(), &db);
+  if (!status.ok()) {
+    reading_results->error =
+        PREF_READ_ERROR_LEVELDB_REPAIRED_REOPEN_FAILED_DESTROYED_REOPEN_FAILED;
+    return;
+  }
+  reading_results->error =
+      PREF_READ_ERROR_LEVELDB_REPAIRED_REOPEN_FAILED_DESTROYED_REOPENED;
+  reading_results->db.reset(db);

[Mattias Nissler (ping if slow), 2014/04/14 10:11:41]

Quite some error code proliferation here now, see below for a proposal on how
this could be handled in a better way.

[dgrogan, 2014/04/17 01:10:27]

Yeah, I was dismayed at the explosion of errors...

+}
+
+/* static */
+scoped_ptr<LevelDBPrefStore::ReadingResults> LevelDBPrefStore::DoReading(
+    const base::FilePath& path) {
+  base::ThreadRestrictions::AssertIOAllowed();
+
+  scoped_ptr<ReadingResults> reading_results(new ReadingResults);
+  reading_results->error = PREF_READ_ERROR_NONE;
+
+  reading_results->no_dir = !base::PathExists(path.DirName());
+  OpenDB(path, reading_results.get());
+  if (!reading_results->db.get()) {

[Mattias Nissler (ping if slow), 2014/04/14 10:11:41]

nit: no need for .get() here, scoped_ptr works fine in boolean context.

[dgrogan, 2014/04/17 01:10:27]

Done.

+    DCHECK_NE(PREF_READ_ERROR_NONE, reading_results->error);
+    return reading_results.Pass();
+  }
+
+  reading_results->value_map.reset(new PrefValueMap);
+  scoped_ptr<leveldb::Iterator> it(
+      reading_results->db->NewIterator(leveldb::ReadOptions()));
+  // TODO(dgrogan): Is it really necessary to check it->status() each iteration?
+  for (it->SeekToFirst(); it->Valid() && it->status().ok(); 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) {
+      reading_results->value_map->SetValue(it->key().ToString(), json_value);
+    } else {
+      DLOG(ERROR) << "Invalid json for key " << it->key().ToString()
+                  << ": " << error_message;
+      if (reading_results->error == PREF_READ_ERROR_LEVELDB_REPAIRED_REOPENED ||
+          reading_results->error ==
+              PREF_READ_ERROR_LEVELDB_SOME_DATA_LOST_AFTER_REPAIR) {
+        reading_results->error =
+            PREF_READ_ERROR_LEVELDB_SOME_DATA_LOST_AFTER_REPAIR;
+      } else {
+        reading_results->error = PREF_READ_ERROR_LEVELDB_SOME_DATA_LOST;
+      }
+    }
+  }
+
+  if (!it->status().ok()) {
+    reading_results->error = PREF_READ_ERROR_LEVELDB_ITERATOR_STATUS_NOT_OK;
+    return reading_results.Pass();
+  }
+  return reading_results.Pass();
+}
+
+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),
+      weak_ptr_factory_(this) {}
+
+LevelDBPrefStore::~LevelDBPrefStore() {
+  CommitPendingWrite();
+  sequenced_task_runner_->DeleteSoon(FROM_HERE, serializer_.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::PersistFromUIThread() {
+  if (read_only_)
+    return;
+  DCHECK(serializer_);
+
+  scoped_ptr<std::set<std::string>> keys_to_delete(new std::set<std::string>);

[Mattias Nissler (ping if slow), 2014/04/14 10:11:41]

nit: >> is C++11, which I don't think is allowed in chromium src yet.

[dgrogan, 2014/04/17 01:10:27]

Done.

+  keys_to_delete->swap(keys_to_delete_);
+
+  scoped_ptr<std::map<std::string, std::string> > keys_to_set(
+      new std::map<std::string, std::string>);
+  keys_to_set->swap(keys_to_set_);
+
+  sequenced_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&LevelDBPrefStore::FileThreadSerializer::WriteToDatabase,
+                 base::Unretained(serializer_.get()),
+                 base::Passed(&keys_to_set),

[Mattias Nissler (ping if slow), 2014/04/14 10:11:41]

nit: Could use base::Owned() here and avoid the scoped_ptr in parameter decls
for WriteToDatabase.

[dgrogan, 2014/04/17 01:10:27]

Done.

+                 base::Passed(&keys_to_delete)));
+}
+
+void LevelDBPrefStore::ScheduleWrite() {
+  if (!timer_.IsRunning()) {
+    timer_.Start(FROM_HERE,
+                 base::TimeDelta::FromSeconds(10),
+                 this,
+                 &LevelDBPrefStore::PersistFromUIThread);
+  }
+}
+
+void LevelDBPrefStore::SetValue(const std::string& key, base::Value* value) {
+  SetValueInternal(key, value, true /*notify*/);
+}
+
+void LevelDBPrefStore::SetValueSilently(const std::string& key,
+                                        base::Value* value) {
+  SetValueInternal(key, value, false /*notify*/);
+}
+
+static std::string Serialize(base::Value* value) {
+  std::string value_string;
+  JSONStringValueSerializer serializer(&value_string);
+  bool serialized_ok = serializer.Serialize(*value);
+  DCHECK(serialized_ok);
+  return value_string;
+}
+
+void LevelDBPrefStore::SetValueInternal(const std::string& key,
+                                        base::Value* value,
+                                        bool notify) {
+  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)) {
+    std::string value_string = Serialize(value);
+    prefs_.SetValue(key, new_value.release());
+    MarkForInsertion(key, value_string);
+    if (notify)
+      NotifyObservers(key);
+  }
+}
+
+void LevelDBPrefStore::RemoveValue(const std::string& key) {
+  DCHECK(initialized_);
+  if (prefs_.RemoveValue(key)) {
+    MarkForDeletion(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()) {
+    scoped_ptr<LevelDBPrefStore::ReadingResults> reading_results(
+        new LevelDBPrefStore::ReadingResults);
+    reading_results->error = PREF_READ_ERROR_FILE_NOT_SPECIFIED;
+    OnStorageRead(reading_results.Pass());
+    return PREF_READ_ERROR_FILE_NOT_SPECIFIED;
+  }
+
+  scoped_ptr<LevelDBPrefStore::ReadingResults> reading_results =
+      DoReading(path_);
+  PersistentPrefStore::PrefReadError error = reading_results->error;
+  OnStorageRead(reading_results.Pass());
+  return error;
+}
+
+void LevelDBPrefStore::ReadPrefsAsync(ReadErrorDelegate* error_delegate) {
+  DCHECK_EQ(false, initialized_);
+  error_delegate_.reset(error_delegate);
+  if (path_.empty()) {
+    scoped_ptr<LevelDBPrefStore::ReadingResults> reading_results(
+        new LevelDBPrefStore::ReadingResults);
+    reading_results->error = PREF_READ_ERROR_FILE_NOT_SPECIFIED;
+    OnStorageRead(reading_results.Pass());
+  }
+  PostTaskAndReplyWithResult(sequenced_task_runner_,
+                             FROM_HERE,
+                             base::Bind(&LevelDBPrefStore::DoReading, path_),
+                             base::Bind(&LevelDBPrefStore::OnStorageRead,
+                                        weak_ptr_factory_.GetWeakPtr()));
+}
+
+void LevelDBPrefStore::CommitPendingWrite() {
+  if (timer_.IsRunning()) {
+    timer_.Stop();
+    PersistFromUIThread();
+  }
+}
+
+void LevelDBPrefStore::MarkForDeletion(const std::string& key) {
+  if (read_only_)
+    return;
+  keys_to_delete_.insert(key);
+  // Need to erase in case there's a set operation in the same batch that would
+  // clobber this delete.
+  keys_to_set_.erase(key);
+  ScheduleWrite();
+}
+
+void LevelDBPrefStore::MarkForInsertion(const std::string& key,
+                                        const std::string& value) {
+  if (read_only_)
+    return;
+  keys_to_set_[key] = value;
+  // Need to erase in case there's a delete operation in the same batch that
+  // would clobber this set.
+  keys_to_delete_.erase(key);
+  ScheduleWrite();
+}
+
+void LevelDBPrefStore::ReportValueChanged(const std::string& key) {
+  base::Value* new_value = NULL;
+  DCHECK(prefs_.GetValue(key, &new_value));
+  std::string value_string = Serialize(new_value);
+  MarkForInsertion(key, value_string);
+  NotifyObservers(key);
+}
+
+void LevelDBPrefStore::NotifyObservers(const std::string& key) {
+  FOR_EACH_OBSERVER(PrefStore::Observer, observers_, OnPrefValueChanged(key));
+}
+
+void LevelDBPrefStore::OnStorageRead(
+    scoped_ptr<LevelDBPrefStore::ReadingResults> reading_results) {
+  read_error_ = reading_results->error;
+
+  if (reading_results->no_dir) {
+    FOR_EACH_OBSERVER(
+        PrefStore::Observer, observers_, OnInitializationCompleted(false));
+    return;
+  }
+
+  initialized_ = true;
+
+  switch (read_error_) {
+    case PREF_READ_ERROR_FILE_NOT_SPECIFIED:
+    case PREF_READ_ERROR_LEVELDB_IO_ERROR:
+    case PREF_READ_ERROR_LEVELDB_DESTROY_FAILED:
+    case PREF_READ_ERROR_LEVELDB_DESTROYED_REOPEN_FAILED:
+    case PREF_READ_ERROR_LEVELDB_REPAIRED_REOPEN_FAILED_DESTROY_FAILED:
+    case PREF_READ_ERROR_LEVELDB_REPAIRED_REOPEN_FAILED_DESTROYED_REOPEN_FAILED:
+      DCHECK(reading_results->value_map.get() == NULL);
+      DCHECK(reading_results->db.get() == NULL);
+      read_only_ = true;
+      break;
+    case PREF_READ_ERROR_NONE:
+    case PREF_READ_ERROR_LEVELDB_DESTROYED_REOPENED:
+    case PREF_READ_ERROR_LEVELDB_REPAIRED_REOPENED:
+    case PREF_READ_ERROR_LEVELDB_REPAIRED_REOPEN_FAILED_DESTROYED_REOPENED:
+    case PREF_READ_ERROR_LEVELDB_SOME_DATA_LOST:
+    case PREF_READ_ERROR_LEVELDB_SOME_DATA_LOST_AFTER_REPAIR:

[Mattias Nissler (ping if slow), 2014/04/14 10:11:41]

The error codes here seem like they'd be better served by a bit mask with bits
for destroyed, repaired, open-failed, data-lost. I wonder whether we can just
handle them as such?

It might be cleaner after all to just introduce a histogram for these bits (and
sample all of the applicable bits) and report an report PREF_READ_IO_ERROR to
the caller? WDYT?

[dgrogan, 2014/04/17 01:10:27]

I'd be worried about losing information about which exact code path was taken.
Well, we'd need to report one of at least two error codes to the caller - one to
signify read only and another to signify non read only - so that HandleReadError
can show the proper error message to the user. But I think I understand what you
mean.

I'd be ok with recording to a histogram inside this class and just surfacing one
of the two error codes mentioned above, except that I want to keep separate
histograms for each consumer of this class. I suppose I could add yet another
parameter to the LevelDBPrefStore constructor that would allow the caller to
specify a string to be used in the UMA histogram names.

WDYT?

[Mattias Nissler (ping if slow), 2014/04/17 08:50:50]

If you just binary-or flags and send the result to a UMA histogram, you'll not
lose any information, and you'll still only have log N flags for N code paths :)
That should be relatively simple to implement by working internally with a flags
bitmap and consulting it to determine whether we're read-only or not and what
the PrefReadError to report should be.
That makes sense to me, but what's your justification for separate histograms?
Do we expect different consumers to actually run into different error modes? If
we don't have any indication of this happening in the wild, I'd just keep it
simple for now and use a single UMA histogram name.

+    case PREF_READ_ERROR_LEVELDB_ITERATOR_STATUS_NOT_OK:
+      serializer_.reset(new FileThreadSerializer(reading_results->db.Pass()));
+      prefs_.Swap(reading_results->value_map.get());
+      break;
+    default:
+      NOTREACHED() << "Unknown error: " << read_error_;
+  }
+
+  // TODO(dgrogan): Call pref_filter_->FilterOnLoad
+
+  if (error_delegate_.get() && read_error_ != PREF_READ_ERROR_NONE)
+    error_delegate_->OnError(read_error_);
+
+  FOR_EACH_OBSERVER(
+      PrefStore::Observer, observers_, OnInitializationCompleted(true));
+}