Added LevelDBStore

Signed-off-by: Samuel Just <[email protected]>

autogen.sh

@@ -19,4 +19,5 @@ autoconf
 autoheader
 automake -a --add-missing -Wall
 ( cd src/gtest && autoreconf -fvi; )
+( cd src/leveldb && autoreconf -fvi; )
 exit

configure.ac

@@ -11,6 +11,7 @@ AC_PREREQ(2.59)
 AC_INIT([ceph], [0.42.2], [[email protected]])
 
 AC_CONFIG_SUBDIRS([src/gtest])
+AC_CONFIG_SUBDIRS([src/leveldb])
 
 # Environment
 AC_CANONICAL_HOST

src/Makefile.am

@@ -1,6 +1,6 @@
 AUTOMAKE_OPTIONS = gnu
 SUBDIRS = ocf
-DIST_SUBDIRS = gtest ocf
+DIST_SUBDIRS = gtest ocf leveldb
 CLEANFILES =
 bin_PROGRAMS =
 # like bin_PROGRAMS, but these targets are only built for debug builds
@@ -1018,7 +1018,7 @@ libos_la_SOURCES = \
 	os/IndexManager.cc \
 	os/FlatIndex.cc
 libos_la_CXXFLAGS= ${CRYPTO_CXXFLAGS} ${AM_CXXFLAGS}
-libos_la_LIBADD = libglobal.la
+libos_la_LIBADD = libglobal.la leveldb/libleveldb.a
 if WITH_LIBAIO
 libos_la_LIBADD += -laio
 endif

src/os/LevelDBStore.cc

@@ -0,0 +1,114 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+
+#include "LevelDBStore.h"
+#include <set>
+#include <map>
+#include <string>
+#include <tr1/memory>
+#include "leveldb/include/leveldb/db.h"
+#include "leveldb/include/leveldb/write_batch.h"
+#include "leveldb/include/leveldb/slice.h"
+#include <errno.h>
+using std::string;
+
+int LevelDBStore::init(ostream &out)
+{
+  leveldb::Options options;
+  options.create_if_missing = true;
+  leveldb::DB *_db;
+  leveldb::Status status = leveldb::DB::Open(options, path, &_db);
+  db.reset(_db);
+  if (!status.ok()) {
+    out << status.ToString() << std::endl;
+    return -EINVAL;
+  } else
+    return 0;
+}
+
+void LevelDBStore::LevelDBTransactionImpl::set(
+  const string &prefix,
+  const std::map<string, bufferlist> &to_set)
+{
+  for (std::map<string, bufferlist>::const_iterator i = to_set.begin();
+       i != to_set.end();
+       ++i) {
+    buffers.push_back(i->second);
+    buffers.rbegin()->rebuild();
+    bufferlist &bl = *(buffers.rbegin());
+    string key = combine_strings(prefix, i->first);
+    keys.push_back(key);
+    bat.Delete(leveldb::Slice(*(keys.rbegin())));
+    bat.Put(leveldb::Slice(*(keys.rbegin())),
+	    leveldb::Slice(bl.c_str(), bl.length()));
+  }
+}
+void LevelDBStore::LevelDBTransactionImpl::rmkeys(const string &prefix,
+						  const std::set<string> &to_rm)
+{
+  for (std::set<string>::const_iterator i = to_rm.begin();
+       i != to_rm.end();
+       ++i) {
+    string key = combine_strings(prefix, *i);
+    keys.push_back(key);
+    bat.Delete(leveldb::Slice(*(keys.rbegin())));
+  }
+}
+
+void LevelDBStore::LevelDBTransactionImpl::rmkeys_by_prefix(const string &prefix)
+{
+  KeyValueDB::Iterator it = db->get_iterator(prefix);
+  for (it->seek_to_first();
+       it->valid();
+       it->next()) {
+    string key = combine_strings(prefix, it->key());
+    keys.push_back(key);
+    bat.Delete(*(keys.rbegin()));
+  }
+}
+
+int LevelDBStore::get(
+    const string &prefix,
+    const std::set<string> &keys,
+    std::map<string, bufferlist> *out)
+{
+  KeyValueDB::Iterator it = get_iterator(prefix);
+  for (std::set<string>::const_iterator i = keys.begin();
+       i != keys.end();
+       ++i) {
+    it->lower_bound(*i);
+    if (it->valid() && it->key() == *i) {
+      out->insert(make_pair(*i, it->value()));
+    } else if (!it->valid())
+      break;
+  }
+  return 0;
+}
+
+string LevelDBStore::combine_strings(const string &prefix, const string &value)
+{
+  string out = prefix;
+  out.push_back(0);
+  out.append(value);
+  return out;
+}
+
+bufferlist LevelDBStore::to_bufferlist(leveldb::Slice in)
+{
+  bufferlist bl;
+  bl.append(bufferptr(in.data(), in.size()));
+  return bl;
+}
+
+int LevelDBStore::split_key(leveldb::Slice in, string *prefix, string *key)
+{
+  string in_prefix = in.ToString();
+  size_t prefix_len = in_prefix.find('\0');
+  if (prefix_len >= in_prefix.size())
+    return -EINVAL;
+
+  if (prefix)
+    *prefix = string(in_prefix, 0, prefix_len);
+  if (key)
+    *key= string(in_prefix, prefix_len + 1);
+  return 0;
+}

src/os/LevelDBStore.h

@@ -0,0 +1,149 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+#ifndef LEVEL_DB_STORE_H
+#define LEVEL_DB_STORE_H
+
+#include "include/buffer.h"
+#include "KeyValueDB.h"
+#include <set>
+#include <map>
+#include <string>
+#include <tr1/memory>
+#include <boost/scoped_ptr.hpp>
+#include "leveldb/include/leveldb/db.h"
+#include "leveldb/include/leveldb/write_batch.h"
+#include "leveldb/include/leveldb/slice.h"
+
+/**
+ * Uses LevelDB to implement the KeyValueDB interface
+ */
+class LevelDBStore : public KeyValueDB {
+  string path;
+  boost::scoped_ptr<leveldb::DB> db;
+public:
+  LevelDBStore(const string &path) : path(path) {}
+
+  /// Opens underlying db
+  int init(ostream &out);
+
+  class LevelDBTransactionImpl : public KeyValueDB::TransactionImpl {
+  public:
+    leveldb::WriteBatch bat;
+    list<bufferlist> buffers;
+    list<string> keys;
+    LevelDBStore *db;
+
+    LevelDBTransactionImpl(LevelDBStore *db) : db(db) {}
+    void set(
+      const string &prefix,
+      const std::map<string, bufferlist> &to_set
+      );
+    void rmkeys(
+      const string &prefix,
+      const std::set<string> &keys
+      );
+    void rmkeys_by_prefix(
+      const string &prefix
+      );
+  };
+
+  KeyValueDB::Transaction get_transaction() {
+    return std::tr1::shared_ptr< LevelDBTransactionImpl >(
+      new LevelDBTransactionImpl(this));
+  }
+
+  int submit_transaction(KeyValueDB::Transaction t) {
+    LevelDBTransactionImpl * _t =
+      static_cast<LevelDBTransactionImpl *>(t.get());
+    leveldb::Status s = db->Write(leveldb::WriteOptions(), &(_t->bat));
+    return s.ok() ? 0 : -1;
+  }
+
+  int get(
+    const string &prefix,
+    const std::set<string> &key,
+    std::map<string, bufferlist> *out
+    );
+
+  class LevelDBIteratorImpl : public KeyValueDB::IteratorImpl {
+    boost::scoped_ptr<leveldb::Iterator> dbiter;
+    const string prefix;
+  public:
+    LevelDBIteratorImpl(leveldb::Iterator *iter, const string &prefix) :
+      dbiter(iter), prefix(prefix) {}
+    int seek_to_first() {
+      leveldb::Slice slice_prefix(prefix);
+      dbiter->Seek(slice_prefix);
+      return dbiter->status().ok() ? 0 : -1;
+    }
+    int seek_to_last() {
+      string limit = past_prefix(prefix);
+      leveldb::Slice slice_limit(limit);
+	dbiter->Seek(slice_limit);
+      if (!dbiter->Valid()) {
+	dbiter->SeekToLast();
+      } else {
+	dbiter->Prev();
+      }
+      return dbiter->status().ok() ? 0 : -1;
+    }
+    int upper_bound(const string &after) {
+      lower_bound(after);
+      if (valid() && key() == after)
+	next();
+      return dbiter->status().ok() ? 0 : -1;
+    }
+    int lower_bound(const string &to) {
+      string bound = combine_strings(prefix, to);
+      leveldb::Slice slice_bound(bound);
+      dbiter->Seek(slice_bound);
+      return dbiter->status().ok() ? 0 : -1;
+    }
+    bool valid() {
+      return dbiter->Valid() && in_prefix(prefix, dbiter->key());
+    }
+    int next() {
+      if (valid())
+	dbiter->Next();
+      return dbiter->status().ok() ? 0 : -1;
+    }
+    int prev() {
+      if (valid())
+	dbiter->Prev();
+      return dbiter->status().ok() ? 0 : -1;
+    }
+    string key() {
+      string out_key;
+      split_key(dbiter->key(), 0, &out_key);
+      return out_key;
+    }
+    bufferlist value() {
+      return to_bufferlist(dbiter->value());
+    }
+    int status() {
+      return dbiter->status().ok() ? 0 : -1;
+    }
+  };
+  Iterator get_iterator(const string &prefix) {
+    return std::tr1::shared_ptr<LevelDBIteratorImpl>(
+      new LevelDBIteratorImpl(
+	db->NewIterator(leveldb::ReadOptions()),
+	prefix));
+  }
+
+
+  /// Utility
+  static string combine_strings(const string &prefix, const string &value);
+  static int split_key(leveldb::Slice in, string *prefix, string *key);
+  static bufferlist to_bufferlist(leveldb::Slice in);
+  static bool in_prefix(const string &prefix, leveldb::Slice key) {
+    return (key.compare(leveldb::Slice(past_prefix(prefix))) < 0) &&
+      (key.compare(leveldb::Slice(prefix)) > 0);
+  }
+  static string past_prefix(const string prefix) {
+    string limit = prefix;
+    limit.push_back(1);
+    return limit;
+  }
+};
+
+#endif