--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include <set>
+#include <map>
+#include <string>
+#include <memory>
+#include <errno.h>
+#include <unistd.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+
+#include "rocksdb/db.h"
+#include "rocksdb/table.h"
+#include "rocksdb/env.h"
+#include "rocksdb/slice.h"
+#include "rocksdb/cache.h"
+#include "rocksdb/filter_policy.h"
+#include "rocksdb/utilities/convenience.h"
+#include "rocksdb/merge_operator.h"
+using std::string;
+#include "common/perf_counters.h"
+#include "common/debug.h"
+#include "include/str_list.h"
+#include "include/stringify.h"
+#include "include/str_map.h"
+#include "KeyValueDB.h"
+#include "RocksDBStore.h"
+
+#include "common/debug.h"
+
+#define dout_context cct
+#define dout_subsys ceph_subsys_rocksdb
+#undef dout_prefix
+#define dout_prefix *_dout << "rocksdb: "
+
+static rocksdb::SliceParts prepare_sliceparts(const bufferlist &bl,
+ vector<rocksdb::Slice> *slices)
+{
+ unsigned n = 0;
+ for (auto& buf : bl.buffers()) {
+ (*slices)[n].data_ = buf.c_str();
+ (*slices)[n].size_ = buf.length();
+ n++;
+ }
+ return rocksdb::SliceParts(slices->data(), slices->size());
+}
+
+//
+// One of these per rocksdb instance, implements the merge operator prefix stuff
+//
+class RocksDBStore::MergeOperatorRouter : public rocksdb::AssociativeMergeOperator {
+ RocksDBStore& store;
+ public:
+ const char *Name() const override {
+ // Construct a name that rocksDB will validate against. We want to
+ // do this in a way that doesn't constrain the ordering of calls
+ // to set_merge_operator, so sort the merge operators and then
+ // construct a name from all of those parts.
+ store.assoc_name.clear();
+ map<std::string,std::string> names;
+ for (auto& p : store.merge_ops) names[p.first] = p.second->name();
+ for (auto& p : names) {
+ store.assoc_name += '.';
+ store.assoc_name += p.first;
+ store.assoc_name += ':';
+ store.assoc_name += p.second;
+ }
+ return store.assoc_name.c_str();
+ }
+
+ MergeOperatorRouter(RocksDBStore &_store) : store(_store) {}
+
+ bool Merge(const rocksdb::Slice& key,
+ const rocksdb::Slice* existing_value,
+ const rocksdb::Slice& value,
+ std::string* new_value,
+ rocksdb::Logger* logger) const override {
+ // Check each prefix
+ for (auto& p : store.merge_ops) {
+ if (p.first.compare(0, p.first.length(),
+ key.data(), p.first.length()) == 0 &&
+ key.data()[p.first.length()] == 0) {
+ if (existing_value) {
+ p.second->merge(existing_value->data(), existing_value->size(),
+ value.data(), value.size(),
+ new_value);
+ } else {
+ p.second->merge_nonexistent(value.data(), value.size(), new_value);
+ }
+ break;
+ }
+ }
+ return true; // OK :)
+ }
+
+};
+
+int RocksDBStore::set_merge_operator(
+ const string& prefix,
+ std::shared_ptr<KeyValueDB::MergeOperator> mop)
+{
+ // If you fail here, it's because you can't do this on an open database
+ assert(db == nullptr);
+ merge_ops.push_back(std::make_pair(prefix,mop));
+ return 0;
+}
+
+class CephRocksdbLogger : public rocksdb::Logger {
+ CephContext *cct;
+public:
+ explicit CephRocksdbLogger(CephContext *c) : cct(c) {
+ cct->get();
+ }
+ ~CephRocksdbLogger() override {
+ cct->put();
+ }
+
+ // Write an entry to the log file with the specified format.
+ void Logv(const char* format, va_list ap) override {
+ Logv(rocksdb::INFO_LEVEL, format, ap);
+ }
+
+ // Write an entry to the log file with the specified log level
+ // and format. Any log with level under the internal log level
+ // of *this (see @SetInfoLogLevel and @GetInfoLogLevel) will not be
+ // printed.
+ void Logv(const rocksdb::InfoLogLevel log_level, const char* format,
+ va_list ap) override {
+ int v = rocksdb::NUM_INFO_LOG_LEVELS - log_level - 1;
+ dout(v);
+ char buf[65536];
+ vsnprintf(buf, sizeof(buf), format, ap);
+ *_dout << buf << dendl;
+ }
+};
+
+rocksdb::Logger *create_rocksdb_ceph_logger()
+{
+ return new CephRocksdbLogger(g_ceph_context);
+}
+
+static int string2bool(const string &val, bool &b_val)
+{
+ if (strcasecmp(val.c_str(), "false") == 0) {
+ b_val = false;
+ return 0;
+ } else if (strcasecmp(val.c_str(), "true") == 0) {
+ b_val = true;
+ return 0;
+ } else {
+ std::string err;
+ int b = strict_strtol(val.c_str(), 10, &err);
+ if (!err.empty())
+ return -EINVAL;
+ b_val = !!b;
+ return 0;
+ }
+}
+
+int RocksDBStore::tryInterpret(const string &key, const string &val, rocksdb::Options &opt)
+{
+ if (key == "compaction_threads") {
+ std::string err;
+ int f = strict_sistrtoll(val.c_str(), &err);
+ if (!err.empty())
+ return -EINVAL;
+ //Low priority threadpool is used for compaction
+ opt.env->SetBackgroundThreads(f, rocksdb::Env::Priority::LOW);
+ } else if (key == "flusher_threads") {
+ std::string err;
+ int f = strict_sistrtoll(val.c_str(), &err);
+ if (!err.empty())
+ return -EINVAL;
+ //High priority threadpool is used for flusher
+ opt.env->SetBackgroundThreads(f, rocksdb::Env::Priority::HIGH);
+ } else if (key == "compact_on_mount") {
+ int ret = string2bool(val, compact_on_mount);
+ if (ret != 0)
+ return ret;
+ } else if (key == "disableWAL") {
+ int ret = string2bool(val, disableWAL);
+ if (ret != 0)
+ return ret;
+ } else {
+ //unrecognize config options.
+ return -EINVAL;
+ }
+ return 0;
+}
+
+int RocksDBStore::ParseOptionsFromString(const string &opt_str, rocksdb::Options &opt)
+{
+ map<string, string> str_map;
+ int r = get_str_map(opt_str, &str_map, ",\n;");
+ if (r < 0)
+ return r;
+ map<string, string>::iterator it;
+ for(it = str_map.begin(); it != str_map.end(); ++it) {
+ string this_opt = it->first + "=" + it->second;
+ rocksdb::Status status = rocksdb::GetOptionsFromString(opt, this_opt , &opt);
+ if (!status.ok()) {
+ //unrecognized by rocksdb, try to interpret by ourselves.
+ r = tryInterpret(it->first, it->second, opt);
+ if (r < 0) {
+ derr << status.ToString() << dendl;
+ return -EINVAL;
+ }
+ }
+ lgeneric_dout(cct, 0) << " set rocksdb option " << it->first
+ << " = " << it->second << dendl;
+ }
+ return 0;
+}
+
+int RocksDBStore::init(string _options_str)
+{
+ options_str = _options_str;
+ rocksdb::Options opt;
+ //try parse options
+ if (options_str.length()) {
+ int r = ParseOptionsFromString(options_str, opt);
+ if (r != 0) {
+ return -EINVAL;
+ }
+ }
+ return 0;
+}
+
+int RocksDBStore::create_and_open(ostream &out)
+{
+ if (env) {
+ unique_ptr<rocksdb::Directory> dir;
+ env->NewDirectory(path, &dir);
+ } else {
+ int r = ::mkdir(path.c_str(), 0755);
+ if (r < 0)
+ r = -errno;
+ if (r < 0 && r != -EEXIST) {
+ derr << __func__ << " failed to create " << path << ": " << cpp_strerror(r)
+ << dendl;
+ return r;
+ }
+ }
+ return do_open(out, true);
+}
+
+int RocksDBStore::do_open(ostream &out, bool create_if_missing)
+{
+ rocksdb::Options opt;
+ rocksdb::Status status;
+
+ if (options_str.length()) {
+ int r = ParseOptionsFromString(options_str, opt);
+ if (r != 0) {
+ return -EINVAL;
+ }
+ }
+
+ if (g_conf->rocksdb_perf) {
+ dbstats = rocksdb::CreateDBStatistics();
+ opt.statistics = dbstats;
+ }
+
+ opt.create_if_missing = create_if_missing;
+ if (g_conf->rocksdb_separate_wal_dir) {
+ opt.wal_dir = path + ".wal";
+ }
+ if (g_conf->get_val<std::string>("rocksdb_db_paths").length()) {
+ list<string> paths;
+ get_str_list(g_conf->get_val<std::string>("rocksdb_db_paths"), "; \t", paths);
+ for (auto& p : paths) {
+ size_t pos = p.find(',');
+ if (pos == std::string::npos) {
+ derr << __func__ << " invalid db path item " << p << " in "
+ << g_conf->get_val<std::string>("rocksdb_db_paths") << dendl;
+ return -EINVAL;
+ }
+ string path = p.substr(0, pos);
+ string size_str = p.substr(pos + 1);
+ uint64_t size = atoll(size_str.c_str());
+ if (!size) {
+ derr << __func__ << " invalid db path item " << p << " in "
+ << g_conf->get_val<std::string>("rocksdb_db_paths") << dendl;
+ return -EINVAL;
+ }
+ opt.db_paths.push_back(rocksdb::DbPath(path, size));
+ dout(10) << __func__ << " db_path " << path << " size " << size << dendl;
+ }
+ }
+
+ if (g_conf->rocksdb_log_to_ceph_log) {
+ opt.info_log.reset(new CephRocksdbLogger(g_ceph_context));
+ }
+
+ if (priv) {
+ dout(10) << __func__ << " using custom Env " << priv << dendl;
+ opt.env = static_cast<rocksdb::Env*>(priv);
+ }
+
+ // caches
+ if (!set_cache_flag) {
+ cache_size = g_conf->rocksdb_cache_size;
+ }
+ uint64_t row_cache_size = cache_size * g_conf->rocksdb_cache_row_ratio;
+ uint64_t block_cache_size = cache_size - row_cache_size;
+
+ if (block_cache_size == 0) {
+ // disable block cache
+ dout(10) << __func__ << " block_cache_size " << block_cache_size
+ << ", setting no_block_cache " << dendl;
+ bbt_opts.no_block_cache = true;
+ } else {
+ if (g_conf->rocksdb_cache_type == "lru") {
+ bbt_opts.block_cache = rocksdb::NewLRUCache(
+ block_cache_size,
+ g_conf->rocksdb_cache_shard_bits);
+ } else if (g_conf->rocksdb_cache_type == "clock") {
+ bbt_opts.block_cache = rocksdb::NewClockCache(
+ block_cache_size,
+ g_conf->rocksdb_cache_shard_bits);
+ } else {
+ derr << "unrecognized rocksdb_cache_type '" << g_conf->rocksdb_cache_type
+ << "'" << dendl;
+ return -EINVAL;
+ }
+ }
+ bbt_opts.block_size = g_conf->rocksdb_block_size;
+
+ if (row_cache_size > 0)
+ opt.row_cache = rocksdb::NewLRUCache(row_cache_size,
+ g_conf->rocksdb_cache_shard_bits);
+ uint64_t bloom_bits = g_conf->get_val<uint64_t>("rocksdb_bloom_bits_per_key");
+ if (bloom_bits > 0) {
+ dout(10) << __func__ << " set bloom filter bits per key to "
+ << bloom_bits << dendl;
+ bbt_opts.filter_policy.reset(rocksdb::NewBloomFilterPolicy(bloom_bits));
+ }
+ if (g_conf->get_val<std::string>("rocksdb_index_type") == "binary_search")
+ bbt_opts.index_type = rocksdb::BlockBasedTableOptions::IndexType::kBinarySearch;
+ if (g_conf->get_val<std::string>("rocksdb_index_type") == "hash_search")
+ bbt_opts.index_type = rocksdb::BlockBasedTableOptions::IndexType::kHashSearch;
+ if (g_conf->get_val<std::string>("rocksdb_index_type") == "two_level")
+ bbt_opts.index_type = rocksdb::BlockBasedTableOptions::IndexType::kTwoLevelIndexSearch;
+ bbt_opts.cache_index_and_filter_blocks =
+ g_conf->get_val<bool>("rocksdb_cache_index_and_filter_blocks");
+ bbt_opts.cache_index_and_filter_blocks_with_high_priority =
+ g_conf->get_val<bool>("rocksdb_cache_index_and_filter_blocks_with_high_priority");
+ bbt_opts.partition_filters = g_conf->get_val<bool>("rocksdb_partition_filters");
+ if (g_conf->get_val<uint64_t>("rocksdb_metadata_block_size") > 0)
+ bbt_opts.metadata_block_size = g_conf->get_val<uint64_t>("rocksdb_metadata_block_size");
+ bbt_opts.pin_l0_filter_and_index_blocks_in_cache =
+ g_conf->get_val<bool>("rocksdb_pin_l0_filter_and_index_blocks_in_cache");
+
+ opt.table_factory.reset(rocksdb::NewBlockBasedTableFactory(bbt_opts));
+ dout(10) << __func__ << " block size " << g_conf->rocksdb_block_size
+ << ", block_cache size " << prettybyte_t(block_cache_size)
+ << ", row_cache size " << prettybyte_t(row_cache_size)
+ << "; shards "
+ << (1 << g_conf->rocksdb_cache_shard_bits)
+ << ", type " << g_conf->rocksdb_cache_type
+ << dendl;
+
+ opt.merge_operator.reset(new MergeOperatorRouter(*this));
+ status = rocksdb::DB::Open(opt, path, &db);
+ if (!status.ok()) {
+ derr << status.ToString() << dendl;
+ return -EINVAL;
+ }
+
+ PerfCountersBuilder plb(g_ceph_context, "rocksdb", l_rocksdb_first, l_rocksdb_last);
+ plb.add_u64_counter(l_rocksdb_gets, "get", "Gets");
+ plb.add_u64_counter(l_rocksdb_txns, "submit_transaction", "Submit transactions");
+ plb.add_u64_counter(l_rocksdb_txns_sync, "submit_transaction_sync", "Submit transactions sync");
+ plb.add_time_avg(l_rocksdb_get_latency, "get_latency", "Get latency");
+ plb.add_time_avg(l_rocksdb_submit_latency, "submit_latency", "Submit Latency");
+ plb.add_time_avg(l_rocksdb_submit_sync_latency, "submit_sync_latency", "Submit Sync Latency");
+ plb.add_u64_counter(l_rocksdb_compact, "compact", "Compactions");
+ plb.add_u64_counter(l_rocksdb_compact_range, "compact_range", "Compactions by range");
+ plb.add_u64_counter(l_rocksdb_compact_queue_merge, "compact_queue_merge", "Mergings of ranges in compaction queue");
+ plb.add_u64(l_rocksdb_compact_queue_len, "compact_queue_len", "Length of compaction queue");
+ plb.add_time_avg(l_rocksdb_write_wal_time, "rocksdb_write_wal_time", "Rocksdb write wal time");
+ plb.add_time_avg(l_rocksdb_write_memtable_time, "rocksdb_write_memtable_time", "Rocksdb write memtable time");
+ plb.add_time_avg(l_rocksdb_write_delay_time, "rocksdb_write_delay_time", "Rocksdb write delay time");
+ plb.add_time_avg(l_rocksdb_write_pre_and_post_process_time,
+ "rocksdb_write_pre_and_post_time", "total time spent on writing a record, excluding write process");
+ logger = plb.create_perf_counters();
+ cct->get_perfcounters_collection()->add(logger);
+
+ if (compact_on_mount) {
+ derr << "Compacting rocksdb store..." << dendl;
+ compact();
+ derr << "Finished compacting rocksdb store" << dendl;
+ }
+ return 0;
+}
+
+int RocksDBStore::_test_init(const string& dir)
+{
+ rocksdb::Options options;
+ options.create_if_missing = true;
+ rocksdb::DB *db;
+ rocksdb::Status status = rocksdb::DB::Open(options, dir, &db);
+ delete db;
+ db = nullptr;
+ return status.ok() ? 0 : -EIO;
+}
+
+RocksDBStore::~RocksDBStore()
+{
+ close();
+ delete logger;
+
+ // Ensure db is destroyed before dependent db_cache and filterpolicy
+ delete db;
+ db = nullptr;
+
+ if (priv) {
+ delete static_cast<rocksdb::Env*>(priv);
+ }
+}
+
+void RocksDBStore::close()
+{
+ // stop compaction thread
+ compact_queue_lock.Lock();
+ if (compact_thread.is_started()) {
+ compact_queue_stop = true;
+ compact_queue_cond.Signal();
+ compact_queue_lock.Unlock();
+ compact_thread.join();
+ } else {
+ compact_queue_lock.Unlock();
+ }
+
+ if (logger)
+ cct->get_perfcounters_collection()->remove(logger);
+}
+
+void RocksDBStore::split_stats(const std::string &s, char delim, std::vector<std::string> &elems) {
+ std::stringstream ss;
+ ss.str(s);
+ std::string item;
+ while (std::getline(ss, item, delim)) {
+ elems.push_back(item);
+ }
+}
+
+void RocksDBStore::get_statistics(Formatter *f)
+{
+ if (!g_conf->rocksdb_perf) {
+ dout(20) << __func__ << "RocksDB perf is disabled, can't probe for stats"
+ << dendl;
+ return;
+ }
+
+ if (g_conf->rocksdb_collect_compaction_stats) {
+ std::string stat_str;
+ bool status = db->GetProperty("rocksdb.stats", &stat_str);
+ if (status) {
+ f->open_object_section("rocksdb_statistics");
+ f->dump_string("rocksdb_compaction_statistics", "");
+ vector<string> stats;
+ split_stats(stat_str, '\n', stats);
+ for (auto st :stats) {
+ f->dump_string("", st);
+ }
+ f->close_section();
+ }
+ }
+ if (g_conf->rocksdb_collect_extended_stats) {
+ if (dbstats) {
+ f->open_object_section("rocksdb_extended_statistics");
+ string stat_str = dbstats->ToString();
+ vector<string> stats;
+ split_stats(stat_str, '\n', stats);
+ f->dump_string("rocksdb_extended_statistics", "");
+ for (auto st :stats) {
+ f->dump_string(".", st);
+ }
+ f->close_section();
+ }
+ f->open_object_section("rocksdbstore_perf_counters");
+ logger->dump_formatted(f,0);
+ f->close_section();
+ }
+ if (g_conf->rocksdb_collect_memory_stats) {
+ f->open_object_section("rocksdb_memtable_statistics");
+ std::string str(stringify(bbt_opts.block_cache->GetUsage()));
+ f->dump_string("block_cache_usage", str.data());
+ str.clear();
+ str.append(stringify(bbt_opts.block_cache->GetPinnedUsage()));
+ f->dump_string("block_cache_pinned_blocks_usage", str);
+ str.clear();
+ db->GetProperty("rocksdb.cur-size-all-mem-tables", &str);
+ f->dump_string("rocksdb_memtable_usage", str);
+ f->close_section();
+ }
+}
+
+int RocksDBStore::submit_transaction(KeyValueDB::Transaction t)
+{
+ utime_t start = ceph_clock_now();
+ // enable rocksdb breakdown
+ // considering performance overhead, default is disabled
+ if (g_conf->rocksdb_perf) {
+ rocksdb::SetPerfLevel(rocksdb::PerfLevel::kEnableTimeExceptForMutex);
+ rocksdb::perf_context.Reset();
+ }
+
+ RocksDBTransactionImpl * _t =
+ static_cast<RocksDBTransactionImpl *>(t.get());
+ rocksdb::WriteOptions woptions;
+ woptions.disableWAL = disableWAL;
+ lgeneric_subdout(cct, rocksdb, 30) << __func__;
+ RocksWBHandler bat_txc;
+ _t->bat.Iterate(&bat_txc);
+ *_dout << " Rocksdb transaction: " << bat_txc.seen << dendl;
+
+ rocksdb::Status s = db->Write(woptions, &_t->bat);
+ if (!s.ok()) {
+ RocksWBHandler rocks_txc;
+ _t->bat.Iterate(&rocks_txc);
+ derr << __func__ << " error: " << s.ToString() << " code = " << s.code()
+ << " Rocksdb transaction: " << rocks_txc.seen << dendl;
+ }
+ utime_t lat = ceph_clock_now() - start;
+
+ if (g_conf->rocksdb_perf) {
+ utime_t write_memtable_time;
+ utime_t write_delay_time;
+ utime_t write_wal_time;
+ utime_t write_pre_and_post_process_time;
+ write_wal_time.set_from_double(
+ static_cast<double>(rocksdb::perf_context.write_wal_time)/1000000000);
+ write_memtable_time.set_from_double(
+ static_cast<double>(rocksdb::perf_context.write_memtable_time)/1000000000);
+ write_delay_time.set_from_double(
+ static_cast<double>(rocksdb::perf_context.write_delay_time)/1000000000);
+ write_pre_and_post_process_time.set_from_double(
+ static_cast<double>(rocksdb::perf_context.write_pre_and_post_process_time)/1000000000);
+ logger->tinc(l_rocksdb_write_memtable_time, write_memtable_time);
+ logger->tinc(l_rocksdb_write_delay_time, write_delay_time);
+ logger->tinc(l_rocksdb_write_wal_time, write_wal_time);
+ logger->tinc(l_rocksdb_write_pre_and_post_process_time, write_pre_and_post_process_time);
+ }
+
+ logger->inc(l_rocksdb_txns);
+ logger->tinc(l_rocksdb_submit_latency, lat);
+
+ return s.ok() ? 0 : -1;
+}
+
+int RocksDBStore::submit_transaction_sync(KeyValueDB::Transaction t)
+{
+ utime_t start = ceph_clock_now();
+ // enable rocksdb breakdown
+ // considering performance overhead, default is disabled
+ if (g_conf->rocksdb_perf) {
+ rocksdb::SetPerfLevel(rocksdb::PerfLevel::kEnableTimeExceptForMutex);
+ rocksdb::perf_context.Reset();
+ }
+
+ RocksDBTransactionImpl * _t =
+ static_cast<RocksDBTransactionImpl *>(t.get());
+ rocksdb::WriteOptions woptions;
+ woptions.sync = true;
+ woptions.disableWAL = disableWAL;
+ lgeneric_subdout(cct, rocksdb, 30) << __func__;
+ RocksWBHandler bat_txc;
+ _t->bat.Iterate(&bat_txc);
+ *_dout << " Rocksdb transaction: " << bat_txc.seen << dendl;
+
+ rocksdb::Status s = db->Write(woptions, &_t->bat);
+ if (!s.ok()) {
+ RocksWBHandler rocks_txc;
+ _t->bat.Iterate(&rocks_txc);
+ derr << __func__ << " error: " << s.ToString() << " code = " << s.code()
+ << " Rocksdb transaction: " << rocks_txc.seen << dendl;
+ }
+ utime_t lat = ceph_clock_now() - start;
+
+ if (g_conf->rocksdb_perf) {
+ utime_t write_memtable_time;
+ utime_t write_delay_time;
+ utime_t write_wal_time;
+ utime_t write_pre_and_post_process_time;
+ write_wal_time.set_from_double(
+ static_cast<double>(rocksdb::perf_context.write_wal_time)/1000000000);
+ write_memtable_time.set_from_double(
+ static_cast<double>(rocksdb::perf_context.write_memtable_time)/1000000000);
+ write_delay_time.set_from_double(
+ static_cast<double>(rocksdb::perf_context.write_delay_time)/1000000000);
+ write_pre_and_post_process_time.set_from_double(
+ static_cast<double>(rocksdb::perf_context.write_pre_and_post_process_time)/1000000000);
+ logger->tinc(l_rocksdb_write_memtable_time, write_memtable_time);
+ logger->tinc(l_rocksdb_write_delay_time, write_delay_time);
+ logger->tinc(l_rocksdb_write_wal_time, write_wal_time);
+ logger->tinc(l_rocksdb_write_pre_and_post_process_time, write_pre_and_post_process_time);
+ }
+
+ logger->inc(l_rocksdb_txns_sync);
+ logger->tinc(l_rocksdb_submit_sync_latency, lat);
+
+ return s.ok() ? 0 : -1;
+}
+
+RocksDBStore::RocksDBTransactionImpl::RocksDBTransactionImpl(RocksDBStore *_db)
+{
+ db = _db;
+}
+
+void RocksDBStore::RocksDBTransactionImpl::set(
+ const string &prefix,
+ const string &k,
+ const bufferlist &to_set_bl)
+{
+ string key = combine_strings(prefix, k);
+
+ // bufferlist::c_str() is non-constant, so we can't call c_str()
+ if (to_set_bl.is_contiguous() && to_set_bl.length() > 0) {
+ bat.Put(rocksdb::Slice(key),
+ rocksdb::Slice(to_set_bl.buffers().front().c_str(),
+ to_set_bl.length()));
+ } else {
+ rocksdb::Slice key_slice(key);
+ vector<rocksdb::Slice> value_slices(to_set_bl.buffers().size());
+ bat.Put(nullptr, rocksdb::SliceParts(&key_slice, 1),
+ prepare_sliceparts(to_set_bl, &value_slices));
+ }
+}
+
+void RocksDBStore::RocksDBTransactionImpl::set(
+ const string &prefix,
+ const char *k, size_t keylen,
+ const bufferlist &to_set_bl)
+{
+ string key;
+ combine_strings(prefix, k, keylen, &key);
+
+ // bufferlist::c_str() is non-constant, so we can't call c_str()
+ if (to_set_bl.is_contiguous() && to_set_bl.length() > 0) {
+ bat.Put(rocksdb::Slice(key),
+ rocksdb::Slice(to_set_bl.buffers().front().c_str(),
+ to_set_bl.length()));
+ } else {
+ rocksdb::Slice key_slice(key);
+ vector<rocksdb::Slice> value_slices(to_set_bl.buffers().size());
+ bat.Put(nullptr, rocksdb::SliceParts(&key_slice, 1),
+ prepare_sliceparts(to_set_bl, &value_slices));
+ }
+}
+
+void RocksDBStore::RocksDBTransactionImpl::rmkey(const string &prefix,
+ const string &k)
+{
+ bat.Delete(combine_strings(prefix, k));
+}
+
+void RocksDBStore::RocksDBTransactionImpl::rmkey(const string &prefix,
+ const char *k,
+ size_t keylen)
+{
+ string key;
+ combine_strings(prefix, k, keylen, &key);
+ bat.Delete(key);
+}
+
+void RocksDBStore::RocksDBTransactionImpl::rm_single_key(const string &prefix,
+ const string &k)
+{
+ bat.SingleDelete(combine_strings(prefix, k));
+}
+
+void RocksDBStore::RocksDBTransactionImpl::rmkeys_by_prefix(const string &prefix)
+{
+ if (db->enable_rmrange) {
+ string endprefix = prefix;
+ endprefix.push_back('\x01');
+ bat.DeleteRange(combine_strings(prefix, string()),
+ combine_strings(endprefix, string()));
+ } else {
+ KeyValueDB::Iterator it = db->get_iterator(prefix);
+ for (it->seek_to_first();
+ it->valid();
+ it->next()) {
+ bat.Delete(combine_strings(prefix, it->key()));
+ }
+ }
+}
+
+void RocksDBStore::RocksDBTransactionImpl::rm_range_keys(const string &prefix,
+ const string &start,
+ const string &end)
+{
+ if (db->enable_rmrange) {
+ bat.DeleteRange(combine_strings(prefix, start), combine_strings(prefix, end));
+ } else {
+ auto it = db->get_iterator(prefix);
+ it->lower_bound(start);
+ while (it->valid()) {
+ if (it->key() >= end) {
+ break;
+ }
+ bat.Delete(combine_strings(prefix, it->key()));
+ it->next();
+ }
+ }
+}
+
+void RocksDBStore::RocksDBTransactionImpl::merge(
+ const string &prefix,
+ const string &k,
+ const bufferlist &to_set_bl)
+{
+ string key = combine_strings(prefix, k);
+
+ // bufferlist::c_str() is non-constant, so we can't call c_str()
+ if (to_set_bl.is_contiguous() && to_set_bl.length() > 0) {
+ bat.Merge(rocksdb::Slice(key),
+ rocksdb::Slice(to_set_bl.buffers().front().c_str(),
+ to_set_bl.length()));
+ } else {
+ // make a copy
+ rocksdb::Slice key_slice(key);
+ vector<rocksdb::Slice> value_slices(to_set_bl.buffers().size());
+ bat.Merge(nullptr, rocksdb::SliceParts(&key_slice, 1),
+ prepare_sliceparts(to_set_bl, &value_slices));
+ }
+}
+
+//gets will bypass RocksDB row cache, since it uses iterator
+int RocksDBStore::get(
+ const string &prefix,
+ const std::set<string> &keys,
+ std::map<string, bufferlist> *out)
+{
+ utime_t start = ceph_clock_now();
+ for (std::set<string>::const_iterator i = keys.begin();
+ i != keys.end(); ++i) {
+ std::string value;
+ std::string bound = combine_strings(prefix, *i);
+ auto status = db->Get(rocksdb::ReadOptions(), rocksdb::Slice(bound), &value);
+ if (status.ok()) {
+ (*out)[*i].append(value);
+ } else if (status.IsIOError()) {
+ ceph_abort_msg(cct, status.ToString());
+ }
+
+ }
+ utime_t lat = ceph_clock_now() - start;
+ logger->inc(l_rocksdb_gets);
+ logger->tinc(l_rocksdb_get_latency, lat);
+ return 0;
+}
+
+int RocksDBStore::get(
+ const string &prefix,
+ const string &key,
+ bufferlist *out)
+{
+ assert(out && (out->length() == 0));
+ utime_t start = ceph_clock_now();
+ int r = 0;
+ string value, k;
+ rocksdb::Status s;
+ k = combine_strings(prefix, key);
+ s = db->Get(rocksdb::ReadOptions(), rocksdb::Slice(k), &value);
+ if (s.ok()) {
+ out->append(value);
+ } else if (s.IsNotFound()) {
+ r = -ENOENT;
+ } else {
+ ceph_abort_msg(cct, s.ToString());
+ }
+ utime_t lat = ceph_clock_now() - start;
+ logger->inc(l_rocksdb_gets);
+ logger->tinc(l_rocksdb_get_latency, lat);
+ return r;
+}
+
+int RocksDBStore::get(
+ const string& prefix,
+ const char *key,
+ size_t keylen,
+ bufferlist *out)
+{
+ assert(out && (out->length() == 0));
+ utime_t start = ceph_clock_now();
+ int r = 0;
+ string value, k;
+ combine_strings(prefix, key, keylen, &k);
+ rocksdb::Status s;
+ s = db->Get(rocksdb::ReadOptions(), rocksdb::Slice(k), &value);
+ if (s.ok()) {
+ out->append(value);
+ } else if (s.IsNotFound()) {
+ r = -ENOENT;
+ } else {
+ ceph_abort_msg(cct, s.ToString());
+ }
+ utime_t lat = ceph_clock_now() - start;
+ logger->inc(l_rocksdb_gets);
+ logger->tinc(l_rocksdb_get_latency, lat);
+ return r;
+}
+
+int RocksDBStore::split_key(rocksdb::Slice in, string *prefix, string *key)
+{
+ size_t prefix_len = 0;
+
+ // Find separator inside Slice
+ char* separator = (char*) memchr(in.data(), 0, in.size());
+ if (separator == NULL)
+ return -EINVAL;
+ prefix_len = size_t(separator - in.data());
+ if (prefix_len >= in.size())
+ return -EINVAL;
+
+ // Fetch prefix and/or key directly from Slice
+ if (prefix)
+ *prefix = string(in.data(), prefix_len);
+ if (key)
+ *key = string(separator+1, in.size()-prefix_len-1);
+ return 0;
+}
+
+void RocksDBStore::compact()
+{
+ logger->inc(l_rocksdb_compact);
+ rocksdb::CompactRangeOptions options;
+ db->CompactRange(options, nullptr, nullptr);
+}
+
+
+void RocksDBStore::compact_thread_entry()
+{
+ compact_queue_lock.Lock();
+ while (!compact_queue_stop) {
+ while (!compact_queue.empty()) {
+ pair<string,string> range = compact_queue.front();
+ compact_queue.pop_front();
+ logger->set(l_rocksdb_compact_queue_len, compact_queue.size());
+ compact_queue_lock.Unlock();
+ logger->inc(l_rocksdb_compact_range);
+ compact_range(range.first, range.second);
+ compact_queue_lock.Lock();
+ continue;
+ }
+ compact_queue_cond.Wait(compact_queue_lock);
+ }
+ compact_queue_lock.Unlock();
+}
+
+void RocksDBStore::compact_range_async(const string& start, const string& end)
+{
+ Mutex::Locker l(compact_queue_lock);
+
+ // try to merge adjacent ranges. this is O(n), but the queue should
+ // be short. note that we do not cover all overlap cases and merge
+ // opportunities here, but we capture the ones we currently need.
+ list< pair<string,string> >::iterator p = compact_queue.begin();
+ while (p != compact_queue.end()) {
+ if (p->first == start && p->second == end) {
+ // dup; no-op
+ return;
+ }
+ if (p->first <= end && p->first > start) {
+ // merge with existing range to the right
+ compact_queue.push_back(make_pair(start, p->second));
+ compact_queue.erase(p);
+ logger->inc(l_rocksdb_compact_queue_merge);
+ break;
+ }
+ if (p->second >= start && p->second < end) {
+ // merge with existing range to the left
+ compact_queue.push_back(make_pair(p->first, end));
+ compact_queue.erase(p);
+ logger->inc(l_rocksdb_compact_queue_merge);
+ break;
+ }
+ ++p;
+ }
+ if (p == compact_queue.end()) {
+ // no merge, new entry.
+ compact_queue.push_back(make_pair(start, end));
+ logger->set(l_rocksdb_compact_queue_len, compact_queue.size());
+ }
+ compact_queue_cond.Signal();
+ if (!compact_thread.is_started()) {
+ compact_thread.create("rstore_compact");
+ }
+}
+bool RocksDBStore::check_omap_dir(string &omap_dir)
+{
+ rocksdb::Options options;
+ options.create_if_missing = true;
+ rocksdb::DB *db;
+ rocksdb::Status status = rocksdb::DB::Open(options, omap_dir, &db);
+ delete db;
+ db = nullptr;
+ return status.ok();
+}
+void RocksDBStore::compact_range(const string& start, const string& end)
+{
+ rocksdb::CompactRangeOptions options;
+ rocksdb::Slice cstart(start);
+ rocksdb::Slice cend(end);
+ db->CompactRange(options, &cstart, &cend);
+}
+RocksDBStore::RocksDBWholeSpaceIteratorImpl::~RocksDBWholeSpaceIteratorImpl()
+{
+ delete dbiter;
+}
+int RocksDBStore::RocksDBWholeSpaceIteratorImpl::seek_to_first()
+{
+ dbiter->SeekToFirst();
+ assert(!dbiter->status().IsIOError());
+ return dbiter->status().ok() ? 0 : -1;
+}
+int RocksDBStore::RocksDBWholeSpaceIteratorImpl::seek_to_first(const string &prefix)
+{
+ rocksdb::Slice slice_prefix(prefix);
+ dbiter->Seek(slice_prefix);
+ assert(!dbiter->status().IsIOError());
+ return dbiter->status().ok() ? 0 : -1;
+}
+int RocksDBStore::RocksDBWholeSpaceIteratorImpl::seek_to_last()
+{
+ dbiter->SeekToLast();
+ assert(!dbiter->status().IsIOError());
+ return dbiter->status().ok() ? 0 : -1;
+}
+int RocksDBStore::RocksDBWholeSpaceIteratorImpl::seek_to_last(const string &prefix)
+{
+ string limit = past_prefix(prefix);
+ rocksdb::Slice slice_limit(limit);
+ dbiter->Seek(slice_limit);
+
+ if (!dbiter->Valid()) {
+ dbiter->SeekToLast();
+ } else {
+ dbiter->Prev();
+ }
+ return dbiter->status().ok() ? 0 : -1;
+}
+int RocksDBStore::RocksDBWholeSpaceIteratorImpl::upper_bound(const string &prefix, const string &after)
+{
+ 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 RocksDBStore::RocksDBWholeSpaceIteratorImpl::lower_bound(const string &prefix, const string &to)
+{
+ string bound = combine_strings(prefix, to);
+ rocksdb::Slice slice_bound(bound);
+ dbiter->Seek(slice_bound);
+ return dbiter->status().ok() ? 0 : -1;
+}
+bool RocksDBStore::RocksDBWholeSpaceIteratorImpl::valid()
+{
+ return dbiter->Valid();
+}
+int RocksDBStore::RocksDBWholeSpaceIteratorImpl::next()
+{
+ if (valid()) {
+ dbiter->Next();
+ }
+ assert(!dbiter->status().IsIOError());
+ return dbiter->status().ok() ? 0 : -1;
+}
+int RocksDBStore::RocksDBWholeSpaceIteratorImpl::prev()
+{
+ if (valid()) {
+ dbiter->Prev();
+ }
+ assert(!dbiter->status().IsIOError());
+ return dbiter->status().ok() ? 0 : -1;
+}
+string RocksDBStore::RocksDBWholeSpaceIteratorImpl::key()
+{
+ string out_key;
+ split_key(dbiter->key(), 0, &out_key);
+ return out_key;
+}
+pair<string,string> RocksDBStore::RocksDBWholeSpaceIteratorImpl::raw_key()
+{
+ string prefix, key;
+ split_key(dbiter->key(), &prefix, &key);
+ return make_pair(prefix, key);
+}
+
+bool RocksDBStore::RocksDBWholeSpaceIteratorImpl::raw_key_is_prefixed(const string &prefix) {
+ // Look for "prefix\0" right in rocksb::Slice
+ rocksdb::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 RocksDBStore::RocksDBWholeSpaceIteratorImpl::value()
+{
+ return to_bufferlist(dbiter->value());
+}
+
+size_t RocksDBStore::RocksDBWholeSpaceIteratorImpl::key_size()
+{
+ return dbiter->key().size();
+}
+
+size_t RocksDBStore::RocksDBWholeSpaceIteratorImpl::value_size()
+{
+ return dbiter->value().size();
+}
+
+bufferptr RocksDBStore::RocksDBWholeSpaceIteratorImpl::value_as_ptr()
+{
+ rocksdb::Slice val = dbiter->value();
+ return bufferptr(val.data(), val.size());
+}
+
+int RocksDBStore::RocksDBWholeSpaceIteratorImpl::status()
+{
+ return dbiter->status().ok() ? 0 : -1;
+}
+
+string RocksDBStore::past_prefix(const string &prefix)
+{
+ string limit = prefix;
+ limit.push_back(1);
+ return limit;
+}
+
+RocksDBStore::WholeSpaceIterator RocksDBStore::_get_iterator()
+{
+ return std::make_shared<RocksDBWholeSpaceIteratorImpl>(
+ db->NewIterator(rocksdb::ReadOptions()));
+}
+
Code Review / stor4nfv.git / blobdiff