+++ /dev/null
-// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-// vim: ts=8 sw=2 smarttab
-#ifndef LEVEL_DB_STORE_H
-#define LEVEL_DB_STORE_H
-
-#include "include/types.h"
-#include "include/buffer_fwd.h"
-#include "KeyValueDB.h"
-#include <set>
-#include <map>
-#include <string>
-#include "include/memory.h"
-#include <boost/scoped_ptr.hpp>
-#include "leveldb/db.h"
-#include "leveldb/env.h"
-#include "leveldb/write_batch.h"
-#include "leveldb/slice.h"
-#include "leveldb/cache.h"
-#ifdef HAVE_LEVELDB_FILTER_POLICY
-#include "leveldb/filter_policy.h"
-#endif
-
-#include <errno.h>
-#include "common/errno.h"
-#include "common/dout.h"
-#include "include/assert.h"
-#include "common/Formatter.h"
-#include "common/Cond.h"
-
-#include "common/ceph_context.h"
-
-// reinclude our assert to clobber the system one
-# include "include/assert.h"
-
-class PerfCounters;
-
-enum {
- l_leveldb_first = 34300,
- l_leveldb_gets,
- l_leveldb_txns,
- l_leveldb_get_latency,
- l_leveldb_submit_latency,
- l_leveldb_submit_sync_latency,
- l_leveldb_compact,
- l_leveldb_compact_range,
- l_leveldb_compact_queue_merge,
- l_leveldb_compact_queue_len,
- l_leveldb_last,
-};
-
-extern leveldb::Logger *create_leveldb_ceph_logger();
-
-class CephLevelDBLogger;
-
-/**
- * Uses LevelDB to implement the KeyValueDB interface
- */
-class LevelDBStore : public KeyValueDB {
- CephContext *cct;
- PerfCounters *logger;
- CephLevelDBLogger *ceph_logger;
- string path;
- boost::scoped_ptr<leveldb::Cache> db_cache;
-#ifdef HAVE_LEVELDB_FILTER_POLICY
- boost::scoped_ptr<const leveldb::FilterPolicy> filterpolicy;
-#endif
- boost::scoped_ptr<leveldb::DB> db;
-
- int do_open(ostream &out, bool create_if_missing);
-
- // manage async compactions
- Mutex compact_queue_lock;
- Cond compact_queue_cond;
- list< pair<string,string> > compact_queue;
- bool compact_queue_stop;
- class CompactThread : public Thread {
- LevelDBStore *db;
- public:
- explicit CompactThread(LevelDBStore *d) : db(d) {}
- void *entry() override {
- db->compact_thread_entry();
- return NULL;
- }
- friend class LevelDBStore;
- } compact_thread;
-
- void compact_thread_entry();
-
- void compact_range(const string& start, const string& end) {
- leveldb::Slice cstart(start);
- leveldb::Slice cend(end);
- db->CompactRange(&cstart, &cend);
- }
- void compact_range_async(const string& start, const string& end);
-
-public:
- /// compact the underlying leveldb store
- void compact() override;
-
- /// compact db for all keys with a given prefix
- void compact_prefix(const string& prefix) override {
- compact_range(prefix, past_prefix(prefix));
- }
- void compact_prefix_async(const string& prefix) override {
- compact_range_async(prefix, past_prefix(prefix));
- }
- void compact_range(const string& prefix,
- const string& start, const string& end) override {
- compact_range(combine_strings(prefix, start), combine_strings(prefix, end));
- }
- void compact_range_async(const string& prefix,
- const string& start, const string& end) override {
- compact_range_async(combine_strings(prefix, start),
- combine_strings(prefix, end));
- }
-
-
- /**
- * options_t: Holds options which are minimally interpreted
- * on initialization and then passed through to LevelDB.
- * We transform a couple of these into actual LevelDB
- * structures, but the rest are simply passed through unchanged. See
- * leveldb/options.h for more precise details on each.
- *
- * Set them after constructing the LevelDBStore, but before calling
- * open() or create_and_open().
- */
- struct options_t {
- uint64_t write_buffer_size; /// in-memory write buffer size
- int max_open_files; /// maximum number of files LevelDB can open at once
- uint64_t cache_size; /// size of extra decompressed cache to use
- uint64_t block_size; /// user data per block
- int bloom_size; /// number of bits per entry to put in a bloom filter
- bool compression_enabled; /// whether to use libsnappy compression or not
-
- // don't change these ones. No, seriously
- int block_restart_interval;
- bool error_if_exists;
- bool paranoid_checks;
-
- string log_file;
-
- options_t() :
- write_buffer_size(0), //< 0 means default
- max_open_files(0), //< 0 means default
- cache_size(0), //< 0 means no cache (default)
- block_size(0), //< 0 means default
- bloom_size(0), //< 0 means no bloom filter (default)
- compression_enabled(true), //< set to false for no compression
- block_restart_interval(0), //< 0 means default
- error_if_exists(false), //< set to true if you want to check nonexistence
- paranoid_checks(false) //< set to true if you want paranoid checks
- {}
- } options;
-
- LevelDBStore(CephContext *c, const string &path) :
- cct(c),
- logger(NULL),
- ceph_logger(NULL),
- path(path),
- db_cache(NULL),
-#ifdef HAVE_LEVELDB_FILTER_POLICY
- filterpolicy(NULL),
-#endif
- compact_queue_lock("LevelDBStore::compact_thread_lock"),
- compact_queue_stop(false),
- compact_thread(this),
- options()
- {}
-
- ~LevelDBStore() override;
-
- static int _test_init(const string& dir);
- int init(string option_str="") override;
-
- /// Opens underlying db
- int open(ostream &out) override {
- return do_open(out, false);
- }
- /// Creates underlying db if missing and opens it
- int create_and_open(ostream &out) override {
- return do_open(out, true);
- }
-
- void close() override;
-
- PerfCounters *get_perf_counters() override
- {
- return logger;
- }
-
- class LevelDBTransactionImpl : public KeyValueDB::TransactionImpl {
- public:
- leveldb::WriteBatch bat;
- LevelDBStore *db;
- explicit LevelDBTransactionImpl(LevelDBStore *db) : db(db) {}
- void set(
- const string &prefix,
- const string &k,
- const bufferlist &bl) override;
- using KeyValueDB::TransactionImpl::set;
- void rmkey(
- const string &prefix,
- const string &k) override;
- void rmkeys_by_prefix(
- const string &prefix
- ) override;
- virtual void rm_range_keys(
- const string &prefix,
- const string &start,
- const string &end) override;
-
- using KeyValueDB::TransactionImpl::rmkey;
- };
-
- KeyValueDB::Transaction get_transaction() override {
- return std::make_shared<LevelDBTransactionImpl>(this);
- }
-
- int submit_transaction(KeyValueDB::Transaction t) override;
- int submit_transaction_sync(KeyValueDB::Transaction t) override;
- int get(
- const string &prefix,
- const std::set<string> &key,
- std::map<string, bufferlist> *out
- ) override;
-
- int get(const string &prefix,
- const string &key,
- bufferlist *value) override;
-
- using KeyValueDB::get;
-
- class LevelDBWholeSpaceIteratorImpl :
- public KeyValueDB::WholeSpaceIteratorImpl {
- protected:
- boost::scoped_ptr<leveldb::Iterator> dbiter;
- public:
- explicit LevelDBWholeSpaceIteratorImpl(leveldb::Iterator *iter) :
- dbiter(iter) { }
- ~LevelDBWholeSpaceIteratorImpl() override { }
-
- int seek_to_first() override {
- dbiter->SeekToFirst();
- return dbiter->status().ok() ? 0 : -1;
- }
- int seek_to_first(const string &prefix) override {
- leveldb::Slice slice_prefix(prefix);
- dbiter->Seek(slice_prefix);
- return dbiter->status().ok() ? 0 : -1;
- }
- int seek_to_last() override {
- dbiter->SeekToLast();
- return dbiter->status().ok() ? 0 : -1;
- }
- int seek_to_last(const string &prefix) override {
- 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 &prefix, const string &after) override {
- lower_bound(prefix, after);
- if (valid()) {
- pair<string,string> key = raw_key();
- if (key.first == prefix && key.second == after)
- next();
- }
- return dbiter->status().ok() ? 0 : -1;
- }
- int lower_bound(const string &prefix, const string &to) override {
- string bound = combine_strings(prefix, to);
- leveldb::Slice slice_bound(bound);
- dbiter->Seek(slice_bound);
- return dbiter->status().ok() ? 0 : -1;
- }
- bool valid() override {
- return dbiter->Valid();
- }
- int next() override {
- if (valid())
- dbiter->Next();
- return dbiter->status().ok() ? 0 : -1;
- }
- int prev() override {
- if (valid())
- dbiter->Prev();
- return dbiter->status().ok() ? 0 : -1;
- }
- string key() override {
- string out_key;
- split_key(dbiter->key(), 0, &out_key);
- return out_key;
- }
- pair<string,string> raw_key() override {
- string prefix, key;
- split_key(dbiter->key(), &prefix, &key);
- return make_pair(prefix, key);
- }
- bool raw_key_is_prefixed(const string &prefix) override {
- leveldb::Slice key = dbiter->key();
- if ((key.size() > prefix.length()) && (key[prefix.length()] == '\0')) {
- return memcmp(key.data(), prefix.c_str(), prefix.length()) == 0;
- } else {
- return false;
- }
- }
- bufferlist value() override {
- return to_bufferlist(dbiter->value());
- }
-
- bufferptr value_as_ptr() override {
- leveldb::Slice data = dbiter->value();
- return bufferptr(data.data(), data.size());
- }
-
- int status() override {
- return dbiter->status().ok() ? 0 : -1;
- }
- };
-
- /// 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 string past_prefix(const string &prefix) {
- string limit = prefix;
- limit.push_back(1);
- return limit;
- }
-
- uint64_t get_estimated_size(map<string,uint64_t> &extra) override {
- DIR *store_dir = opendir(path.c_str());
- if (!store_dir) {
- lderr(cct) << __func__ << " something happened opening the store: "
- << cpp_strerror(errno) << dendl;
- return 0;
- }
-
- uint64_t total_size = 0;
- uint64_t sst_size = 0;
- uint64_t log_size = 0;
- uint64_t misc_size = 0;
-
- struct dirent *entry = NULL;
- while ((entry = readdir(store_dir)) != NULL) {
- string n(entry->d_name);
-
- if (n == "." || n == "..")
- continue;
-
- string fpath = path + '/' + n;
- struct stat s;
- int err = stat(fpath.c_str(), &s);
- if (err < 0)
- err = -errno;
- // we may race against leveldb while reading files; this should only
- // happen when those files are being updated, data is being shuffled
- // and files get removed, in which case there's not much of a problem
- // as we'll get to them next time around.
- if (err == -ENOENT) {
- continue;
- }
- if (err < 0) {
- lderr(cct) << __func__ << " error obtaining stats for " << fpath
- << ": " << cpp_strerror(err) << dendl;
- goto err;
- }
-
- size_t pos = n.find_last_of('.');
- if (pos == string::npos) {
- misc_size += s.st_size;
- continue;
- }
-
- string ext = n.substr(pos+1);
- if (ext == "sst") {
- sst_size += s.st_size;
- } else if (ext == "log") {
- log_size += s.st_size;
- } else {
- misc_size += s.st_size;
- }
- }
-
- total_size = sst_size + log_size + misc_size;
-
- extra["sst"] = sst_size;
- extra["log"] = log_size;
- extra["misc"] = misc_size;
- extra["total"] = total_size;
-
-err:
- closedir(store_dir);
- return total_size;
- }
-
-
-protected:
- WholeSpaceIterator _get_iterator() override {
- return std::make_shared<LevelDBWholeSpaceIteratorImpl>(
- db->NewIterator(leveldb::ReadOptions()));
- }
-
-};
-
-#endif
Code Review / stor4nfv.git / blobdiff