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/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 {
+  ReadingResults() : db_owned(NULL), value_map_owned(NULL) {

[Mattias Nissler (ping if slow), 2014/03/20 09:32:25]

nit: no need for line break.

[dgrogan, 2014/04/12 00:32:45]

Constructor removed.

+  }
+  bool no_dir;
+  leveldb::DB* db_owned;

[Mattias Nissler (ping if slow), 2014/03/20 09:32:25]

Why aren't these scoped_ptrs if they are owned?

[dgrogan, 2014/04/12 00:32:45]

Oops, meant to switch them before sending out. Done now.

+  PrefValueMap* value_map_owned;
+  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);
+};
+
+/* static */
+void LevelDBPrefStore::OpenDB(const base::FilePath& path,
+                              ReadingResults* reading_results) {
+  leveldb::Options options;
+  options.create_if_missing = true;
+  leveldb::Status status = leveldb::DB::Open(
+      options, path.AsUTF8Unsafe(), &reading_results->db_owned);
+  if (status.ok()) {
+    reading_results->error = PersistentPrefStore::PREF_READ_ERROR_NONE;
+    return;
+  }
+  if (leveldb_env::IsIOError(status)) {
+    reading_results->error =
+        PersistentPrefStore::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);

[Mattias Nissler (ping if slow), 2014/03/20 09:32:25]

Is it OK to pass options.create_if_missing = true here?

[dgrogan, 2014/04/12 00:32:45]

Yes, it's ignored by RepairDB.

+  bool destroy_succeeded = false;
+  if (!status.ok()) {
+    // TODO(dgrogan): Should we move the old database aside instead of
+    // destroying it?

[Mattias Nissler (ping if slow), 2014/03/20 09:32:25]

Yes, please. That way, we will have a better chance to get samples from the
field for the corruption case for investigation.

[dgrogan, 2014/04/12 00:32:45]

Done.

+    status = leveldb::DestroyDB(path.AsUTF8Unsafe(), options);

[Mattias Nissler (ping if slow), 2014/03/20 09:32:25]

again, create_if_missing OK here?

[dgrogan, 2014/04/12 00:32:45]

Yep.

+    if (!status.ok()) {
+      reading_results->error =
+          PersistentPrefStore::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(), &reading_results->db_owned);
+  if (!status.ok()) {
+    if (destroy_succeeded)

[Mattias Nissler (ping if slow), 2014/03/20 09:32:25]

nit: I suggest curly braces here; the line breaks below make this hard to read
otherwise.

[dgrogan, 2014/04/12 00:32:45]

Done.

+      reading_results->error =
+          PersistentPrefStore::PREF_READ_ERROR_LEVELDB_DESTROYED_REOPEN_FAILED;
+    else
+      reading_results->error =
+          PersistentPrefStore::PREF_READ_ERROR_LEVELDB_REPAIRED_REOPEN_FAILED;

[Mattias Nissler (ping if slow), 2014/03/20 09:32:25]

Should we destroy and retry in this case?

[dgrogan, 2014/04/12 00:32:45]

Done.

+    return;
+  }
+  if (destroy_succeeded)

[Mattias Nissler (ping if slow), 2014/03/20 09:32:25]

nit: curly braces suggested

[dgrogan, 2014/04/12 00:32:45]

Done.

+    reading_results->error =
+        PersistentPrefStore::PREF_READ_ERROR_LEVELDB_DESTROYED_REOPEN_SUCCESS;
+  else
+    reading_results->error =
+        PersistentPrefStore::PREF_READ_ERROR_LEVELDB_REPAIRED_REOPEN_SUCCESS;
+}
+
+/* static */
+scoped_ptr<LevelDBPrefStore::ReadingResults> LevelDBPrefStore::DoReading(
+    const base::FilePath& path) {
+  base::ThreadRestrictions::AssertIOAllowed();
+
+  scoped_ptr<ReadingResults> reading_results(new ReadingResults);
+
+  reading_results->no_dir = !base::PathExists(path.DirName());
+  OpenDB(path, reading_results.get());
+  if (!reading_results->db_owned)
+    return reading_results.Pass();
+
+  scoped_ptr<PrefValueMap> value_map(new PrefValueMap);
+  scoped_ptr<leveldb::Iterator> it(
+      reading_results->db_owned->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 {
+      // TODO(dgrogan): UMA?

[Mattias Nissler (ping if slow), 2014/03/20 09:32:25]

Hm, good question. Another option would be to crash (i.e. LOG(FATAL)), which is
appropriate if we're confident in level DB catching all corruption internally
and we are OK with the assumption that nobody else messes with our DB.

[dgrogan, 2014/04/12 00:32:45]

I'm confident LevelDB will catch all corruption internally but less so about
someone messing with our DB.

+      NOTREACHED() << error_message;
+    }
+  }
+
+  if (!it->status().ok()) {
+    // TODO(dgrogan): UMA?

[Mattias Nissler (ping if slow), 2014/03/20 09:32:25]

I don't think that's necessary, the pref read error will get histogrammed
anyways, no?

[dgrogan, 2014/04/12 00:32:45]

That's right.

+    reading_results->error = PersistentPrefStore::PREF_READ_ERROR_FILE_OTHER;
+    return reading_results.Pass();
+  }
+  reading_results->value_map_owned = value_map.release();
+  reading_results->error = PersistentPrefStore::PREF_READ_ERROR_NONE;
+  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>);
+  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),
+                 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);
+  keys_to_set_.erase(key);

[Mattias Nissler (ping if slow), 2014/03/20 09:32:25]

Can you put a comment here explaining that the erase is needed so set and delete
operations ending up in the same batch don't clobber each other? Also below.

[dgrogan, 2014/04/12 00:32:45]

Done.

+  ScheduleWrite();
+}
+
+void LevelDBPrefStore::MarkForInsertion(const std::string& key,
+                                        const std::string& value) {
+  if (read_only_)
+    return;
+  keys_to_set_[key] = value;
+  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);
+  keys_to_set_[key] = value_string;

[Mattias Nissler (ping if slow), 2014/03/20 09:32:25]

This should be MarkForInsertion(key, value_string);

[dgrogan, 2014/04/12 00:32:45]

Done.

+  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) {
+  scoped_ptr<leveldb::DB> db(reading_results->db_owned);
+  scoped_ptr<PrefValueMap> value(reading_results->value_map_owned);
+  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_OTHER:
+    case PREF_READ_ERROR_LEVELDB_IO_ERROR:
+    case PREF_READ_ERROR_FILE_NOT_SPECIFIED:
+      DCHECK(value.get() == NULL);
+      DCHECK(db.get() == NULL);
+      read_only_ = true;
+      break;
+    case PREF_READ_ERROR_NONE:
+      serializer_.reset(new FileThreadSerializer(db.Pass()));
+      prefs_.Swap(value.get());
+      break;
+    default:
+      NOTREACHED() << "Unknown error: " << read_error_;

[Mattias Nissler (ping if slow), 2014/03/20 09:32:25]

You have introduced a bunch of new read errors, shouldn't they be handled here?

[dgrogan, 2014/04/12 00:32:45]

Yes, not sure how I missed this. Thanks.

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