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 {
+  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(
+      std::map<std::string, std::string>* keys_to_set,
+      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)
+      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);
+}
+
+/* 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) {
+    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>);
+  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::Owned(keys_to_set.release()),
+                 base::Owned(keys_to_delete.release())));
+}
+
+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:
+    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));
+}