Extract protobuf database into a new 'leveldb_proto' component

components/leveldb_proto/core/proto_database_impl.h

+// Copyright 2014 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.
+
+#ifndef COMPONENTS_LEVELDB_PROTO_CORE_PROTO_DATABASE_IMPL_H_
+#define COMPONENTS_LEVELDB_PROTO_CORE_PROTO_DATABASE_IMPL_H_
+
+#include <string>
+#include <vector>
+
+#include "base/bind.h"
+#include "base/file_util.h"
+#include "base/files/file_path.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/message_loop/message_loop.h"
+#include "base/sequenced_task_runner.h"
+#include "base/strings/string_util.h"
+#include "base/threading/sequenced_worker_pool.h"
+#include "base/threading/thread_checker.h"
+#include "base/threading/thread_collision_warner.h"
+#include "components/leveldb_proto/core/proto_database.h"
+#include "third_party/leveldatabase/src/include/leveldb/db.h"
+#include "third_party/leveldatabase/src/include/leveldb/iterator.h"
+#include "third_party/leveldatabase/src/include/leveldb/options.h"
+#include "third_party/leveldatabase/src/include/leveldb/slice.h"
+#include "third_party/leveldatabase/src/include/leveldb/status.h"
+#include "third_party/leveldatabase/src/include/leveldb/write_batch.h"
+
+namespace leveldb_proto {
+
+typedef std::vector<std::pair<std::string, std::string> > KeyValueVector;
+typedef std::vector<std::string> KeyVector;
+
+// When the ProtoDatabaseImpl instance is deleted, in-progress asynchronous
+// operations will be completed and the corresponding callbacks will be called.
+// Construction/calls/destruction should all happen on the same thread.
+template <typename T>
+class ProtoDatabaseImpl : public ProtoDatabase<T> {
+ public:
+  // The underlying database. Calls to this type may be blocking.
+  class Database {

[cjhopman, 2014/06/16 22:44:44]

Since this is an inner class of ProtoDatabaseImpl it still depends on the type T
(I would hope the compiler is smart enough to realize that the generated code is
independent of the type, but idk and it probably still needs to do a large part
of the code generation for each different type).

The implementation of this and LevelDB don't refer to the type (or to
ProtoDatabaseImpl) so they can be moved out of this class (though then you need
to come up with a good name for them).

[Mathieu, 2014/06/17 17:40:59]

Per discussion, I removed Database and kept LevelDB, now in a separate file.
Works well, and there are no circular dependencies.

+   public:
+    virtual bool Init(const base::FilePath& database_dir) = 0;
+    virtual bool Save(const KeyValueVector& pairs_to_save,
+                      const KeyVector& keys_to_remove) = 0;
+    virtual bool Load(std::vector<std::string>* entries) = 0;
+    virtual ~Database() {}
+  };
+
+  // Once constructed, function calls and destruction should all occur on the
+  // same thread (not necessarily the same as the constructor).
+  class LevelDB : public Database {
+   public:
+    LevelDB();
+    virtual ~LevelDB();
+    virtual bool Init(const base::FilePath& database_dir) OVERRIDE;
+    virtual bool Save(const KeyValueVector& pairs_to_save,
+                      const KeyVector& keys_to_remove) OVERRIDE;
+    virtual bool Load(std::vector<std::string>* entries) OVERRIDE;
+
+   private:
+    DFAKE_MUTEX(thread_checker_);
+    scoped_ptr<leveldb::DB> db_;
+  };
+
+  // All blocking calls/disk access will happen on the provided |task_runner|.
+  explicit ProtoDatabaseImpl(
+      scoped_refptr<base::SequencedTaskRunner> task_runner);
+
+  virtual ~ProtoDatabaseImpl();
+
+  // ProtoDatabase implementation.
+  // TODO(cjhopman): Perhaps Init() shouldn't be exposed to users and not just
+  //     part of the constructor
+  virtual void Init(const base::FilePath& database_dir,
+                    typename ProtoDatabase<T>::InitCallback callback) OVERRIDE;
+  virtual void UpdateEntries(
+      scoped_ptr<typename ProtoDatabase<T>::KeyEntryVector> entries_to_save,
+      scoped_ptr<KeyVector> keys_to_remove,
+      typename ProtoDatabase<T>::UpdateCallback callback) OVERRIDE;
+  virtual void LoadEntries(
+      typename ProtoDatabase<T>::LoadCallback callback) OVERRIDE;
+
+  // Allow callers to provide their own Database implementation.
+  void InitWithDatabase(scoped_ptr<Database> database,
+                        const base::FilePath& database_dir,
+                        typename ProtoDatabase<T>::InitCallback callback);
+
+ private:
+  base::ThreadChecker thread_checker_;
+
+  // Used to run blocking tasks in-order.
+  scoped_refptr<base::SequencedTaskRunner> task_runner_;
+
+  scoped_ptr<Database> db_;
+
+  DISALLOW_COPY_AND_ASSIGN(ProtoDatabaseImpl);
+};
+
+template <typename T>
+ProtoDatabaseImpl<T>::LevelDB::LevelDB() {}
+
+template <typename T>
+ProtoDatabaseImpl<T>::LevelDB::~LevelDB() {
+  DFAKE_SCOPED_LOCK(thread_checker_);
+}
+
+template <typename T>
+bool ProtoDatabaseImpl<T>::LevelDB::Init(const base::FilePath& database_dir) {
+  DFAKE_SCOPED_LOCK(thread_checker_);
+
+  leveldb::Options options;
+  options.create_if_missing = true;
+  options.max_open_files = 0;  // Use minimum.
+
+  std::string path = database_dir.AsUTF8Unsafe();
+
+  leveldb::DB* db = NULL;
+  leveldb::Status status = leveldb::DB::Open(options, path, &db);
+  if (status.IsCorruption()) {
+    base::DeleteFile(database_dir, true);
+    status = leveldb::DB::Open(options, path, &db);
+  }
+
+  if (status.ok()) {
+    CHECK(db);
+    db_.reset(db);
+    return true;
+  }
+
+  LOG(WARNING) << "Unable to open " << database_dir.value() << ": "
+               << status.ToString();
+  return false;
+}
+
+template <typename T>
+bool ProtoDatabaseImpl<T>::LevelDB::Save(const KeyValueVector& entries_to_save,
+                                         const KeyVector& keys_to_remove) {
+  DFAKE_SCOPED_LOCK(thread_checker_);
+
+  leveldb::WriteBatch updates;
+  for (KeyValueVector::const_iterator it = entries_to_save.begin();
+       it != entries_to_save.end(); ++it) {
+    updates.Put(leveldb::Slice(it->first), leveldb::Slice(it->second));
+  }
+  for (KeyVector::const_iterator it = keys_to_remove.begin();
+       it != keys_to_remove.end(); ++it) {
+    updates.Delete(leveldb::Slice(*it));
+  }
+
+  leveldb::WriteOptions options;
+  options.sync = true;
+  leveldb::Status status = db_->Write(options, &updates);
+  if (status.ok()) return true;
+
+  DLOG(WARNING) << "Failed writing leveldb_proto entries: "
+                << status.ToString();
+  return false;
+}
+
+template <typename T>
+bool ProtoDatabaseImpl<T>::LevelDB::Load(std::vector<std::string>* entries) {
+  DFAKE_SCOPED_LOCK(thread_checker_);
+
+  leveldb::ReadOptions options;
+  scoped_ptr<leveldb::Iterator> db_iterator(db_->NewIterator(options));
+  for (db_iterator->SeekToFirst(); db_iterator->Valid(); db_iterator->Next()) {
+    leveldb::Slice value_slice = db_iterator->value();
+    std::string entry(value_slice.data(), value_slice.size());
+    entries->push_back(entry);
+  }
+  return true;
+}
+
+namespace {
+
+template <typename T>
+void RunInitCallback(typename ProtoDatabase<T>::InitCallback callback,
+                     const bool* success) {
+  callback.Run(*success);
+}
+
+template <typename T>
+void RunUpdateCallback(typename ProtoDatabase<T>::UpdateCallback callback,
+                       const bool* success) {
+  callback.Run(*success);
+}
+
+template <typename T>
+void RunLoadCallback(typename ProtoDatabase<T>::LoadCallback callback,
+                     const bool* success, scoped_ptr<std::vector<T> > entries) {
+  callback.Run(*success, entries.Pass());
+}
+
+template <typename T>
+void InitFromTaskRunner(typename ProtoDatabaseImpl<T>::Database* database,
+                        const base::FilePath& database_dir, bool* success) {
+  DCHECK(success);
+
+  // TODO(cjhopman): Histogram for database size.
+  *success = database->Init(database_dir);
+}
+
+template <typename T>
+void UpdateEntriesFromTaskRunner(
+    typename ProtoDatabaseImpl<T>::Database* database,
+    scoped_ptr<typename ProtoDatabase<T>::KeyEntryVector> entries_to_save,
+    scoped_ptr<KeyVector> keys_to_remove, bool* success) {
+  DCHECK(success);
+  // Serialize the values from Proto to string before passing on to database.
+  KeyValueVector pairs_to_save;
+  for (typename ProtoDatabase<T>::KeyEntryVector::iterator it =
+           entries_to_save->begin();
+       it != entries_to_save->end(); ++it) {
+    pairs_to_save.push_back(
+        std::make_pair(it->first, it->second.SerializeAsString()));
+  }
+  *success = database->Save(pairs_to_save, *keys_to_remove);
+}
+
+template <typename T>
+void LoadEntriesFromTaskRunner(
+    typename ProtoDatabaseImpl<T>::Database* database, std::vector<T>* entries,
+    bool* success) {
+  DCHECK(success);
+  DCHECK(entries);
+
+  entries->clear();
+  std::vector<std::string> loaded_entries;
+  *success = database->Load(&loaded_entries);
+  for (std::vector<std::string>::iterator it = loaded_entries.begin();
+       it != loaded_entries.end(); ++it) {
+    T entry;
+    if (!entry.ParseFromString(*it)) {
+      DLOG(WARNING) << "Unable to parse leveldb_proto entry " << *it;
+      // TODO(cjhopman): Decide what to do about un-parseable entries.
+    }
+    entries->push_back(entry);
+  }
+}
+
+}  // namespace
+
+template <typename T>
+ProtoDatabaseImpl<T>::ProtoDatabaseImpl(
+    scoped_refptr<base::SequencedTaskRunner> task_runner)
+    : task_runner_(task_runner) {}
+
+template <typename T>
+ProtoDatabaseImpl<T>::~ProtoDatabaseImpl() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  if (!task_runner_->DeleteSoon(FROM_HERE, db_.release())) {
+    DLOG(WARNING) << "DOM distiller database will not be deleted.";
+  }
+}
+
+template <typename T>
+void ProtoDatabaseImpl<T>::Init(
+    const base::FilePath& database_dir,
+    typename ProtoDatabase<T>::InitCallback callback) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  InitWithDatabase(scoped_ptr<Database>(new LevelDB()), database_dir, callback);
+}
+
+template <typename T>
+void ProtoDatabaseImpl<T>::InitWithDatabase(
+    scoped_ptr<Database> database, const base::FilePath& database_dir,
+    typename ProtoDatabase<T>::InitCallback callback) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(!db_);
+  DCHECK(database);
+  db_.reset(database.release());
+  bool* success = new bool(false);
+  task_runner_->PostTaskAndReply(
+      FROM_HERE, base::Bind(InitFromTaskRunner<T>, base::Unretained(db_.get()),
+                            database_dir, success),
+      base::Bind(RunInitCallback<T>, callback, base::Owned(success)));
+}
+
+template <typename T>
+void ProtoDatabaseImpl<T>::UpdateEntries(
+    scoped_ptr<typename ProtoDatabase<T>::KeyEntryVector> entries_to_save,
+    scoped_ptr<KeyVector> keys_to_remove,
+    typename ProtoDatabase<T>::UpdateCallback callback) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  bool* success = new bool(false);
+  task_runner_->PostTaskAndReply(
+      FROM_HERE,
+      base::Bind(UpdateEntriesFromTaskRunner<T>, base::Unretained(db_.get()),
+                 base::Passed(&entries_to_save), base::Passed(&keys_to_remove),
+                 success),
+      base::Bind(RunUpdateCallback<T>, callback, base::Owned(success)));
+}
+
+template <typename T>
+void ProtoDatabaseImpl<T>::LoadEntries(
+    typename ProtoDatabase<T>::LoadCallback callback) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  bool* success = new bool(false);
+
+  scoped_ptr<std::vector<T> > entries(new std::vector<T>());
+  // Get this pointer before entries is base::Passed() so we can use it below.
+  std::vector<T>* entries_ptr = entries.get();
+
+  task_runner_->PostTaskAndReply(
+      FROM_HERE, base::Bind(LoadEntriesFromTaskRunner<T>,
+                            base::Unretained(db_.get()), entries_ptr, success),
+      base::Bind(RunLoadCallback<T>, callback, base::Owned(success),
+                 base::Passed(&entries)));
+}
+
+}  // namespace leveldb_proto
+
+#endif  // COMPONENTS_LEVELDB_PROTO_CORE_PROTO_DATABASE_IMPL_H_