#include <map>
#include <string>
#include <memory>
+#if __has_include(<filesystem>)
+#include <filesystem>
+namespace fs = std::filesystem;
+#elif __has_include(<experimental/filesystem>)
+#include <experimental/filesystem>
+namespace fs = std::experimental::filesystem;
+#endif
#include <errno.h>
#include <unistd.h>
#include <sys/types.h>
#include "rocksdb/utilities/convenience.h"
#include "rocksdb/merge_operator.h"
-using std::string;
#include "common/perf_counters.h"
#include "common/PriorityCache.h"
#include "include/common_fwd.h"
+#include "include/scope_guard.h"
#include "include/str_list.h"
#include "include/stringify.h"
#include "include/str_map.h"
#undef dout_prefix
#define dout_prefix *_dout << "rocksdb: "
+using std::function;
+using std::list;
+using std::map;
+using std::ostream;
+using std::pair;
+using std::set;
+using std::string;
+using std::unique_ptr;
+using std::vector;
+
+using ceph::bufferlist;
+using ceph::bufferptr;
+using ceph::Formatter;
+
+static const char* sharding_def_dir = "sharding";
+static const char* sharding_def_file = "sharding/def";
+static const char* sharding_recreate = "sharding/recreate_columns";
+static const char* resharding_column_lock = "reshardingXcommencingXlocked";
+
static bufferlist to_bufferlist(rocksdb::Slice in) {
bufferlist bl;
bl.append(bufferptr(in.data(), in.size()));
for (auto& p : store.merge_ops) {
names[p.first] = p.second->name();
}
- for (auto& p : store.cf_handles) {
- names.erase(p.first);
- }
for (auto& p : names) {
store.assoc_name += '.';
store.assoc_name += p.first;
return 0;
}
}
-
+
+namespace rocksdb {
+extern std::string trim(const std::string& str);
+}
+
+// this function is a modification of rocksdb's StringToMap:
+// 1) accepts ' \n ; as separators
+// 2) leaves compound options with enclosing { and }
+rocksdb::Status StringToMap(const std::string& opts_str,
+ std::unordered_map<std::string, std::string>* opts_map)
+{
+ using rocksdb::Status;
+ using rocksdb::trim;
+ assert(opts_map);
+ // Example:
+ // opts_str = "write_buffer_size=1024;max_write_buffer_number=2;"
+ // "nested_opt={opt1=1;opt2=2};max_bytes_for_level_base=100"
+ size_t pos = 0;
+ std::string opts = trim(opts_str);
+ while (pos < opts.size()) {
+ size_t eq_pos = opts.find('=', pos);
+ if (eq_pos == std::string::npos) {
+ return Status::InvalidArgument("Mismatched key value pair, '=' expected");
+ }
+ std::string key = trim(opts.substr(pos, eq_pos - pos));
+ if (key.empty()) {
+ return Status::InvalidArgument("Empty key found");
+ }
+
+ // skip space after '=' and look for '{' for possible nested options
+ pos = eq_pos + 1;
+ while (pos < opts.size() && isspace(opts[pos])) {
+ ++pos;
+ }
+ // Empty value at the end
+ if (pos >= opts.size()) {
+ (*opts_map)[key] = "";
+ break;
+ }
+ if (opts[pos] == '{') {
+ int count = 1;
+ size_t brace_pos = pos + 1;
+ while (brace_pos < opts.size()) {
+ if (opts[brace_pos] == '{') {
+ ++count;
+ } else if (opts[brace_pos] == '}') {
+ --count;
+ if (count == 0) {
+ break;
+ }
+ }
+ ++brace_pos;
+ }
+ // found the matching closing brace
+ if (count == 0) {
+ //include both '{' and '}'
+ (*opts_map)[key] = trim(opts.substr(pos, brace_pos - pos + 1));
+ // skip all whitespace and move to the next ';,'
+ // brace_pos points to the matching '}'
+ pos = brace_pos + 1;
+ while (pos < opts.size() && isspace(opts[pos])) {
+ ++pos;
+ }
+ if (pos < opts.size() && opts[pos] != ';' && opts[pos] != ',') {
+ return Status::InvalidArgument(
+ "Unexpected chars after nested options");
+ }
+ ++pos;
+ } else {
+ return Status::InvalidArgument(
+ "Mismatched curly braces for nested options");
+ }
+ } else {
+ size_t sc_pos = opts.find_first_of(",;", pos);
+ if (sc_pos == std::string::npos) {
+ (*opts_map)[key] = trim(opts.substr(pos));
+ // It either ends with a trailing , ; or the last key-value pair
+ break;
+ } else {
+ (*opts_map)[key] = trim(opts.substr(pos, sc_pos - pos));
+ }
+ pos = sc_pos + 1;
+ }
+ }
+ return Status::OK();
+}
+
int RocksDBStore::tryInterpret(const string &key, const string &val, rocksdb::Options &opt)
{
if (key == "compaction_threads") {
{
// keep aligned with func tryInterpret
const set<string> need_interp_keys = {"compaction_threads", "flusher_threads", "compact_on_mount", "disableWAL"};
+ int r;
+ rocksdb::Status status;
+ std::unordered_map<std::string, std::string> str_map;
+ status = StringToMap(opt_str, &str_map);
+ if (!status.ok()) {
+ dout(5) << __func__ << " error '" << status.getState() <<
+ "' while parsing options '" << opt_str << "'" << dendl;
+ return -EINVAL;
+ }
- 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) {
+ for (auto 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);
return -EINVAL;
}
}
- lgeneric_dout(cct, 0) << " set rocksdb option " << it->first
+ lgeneric_dout(cct, 1) << " set rocksdb option " << it->first
<< " = " << it->second << dendl;
}
return 0;
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;
+ if (!fs::exists(path)) {
+ std::error_code ec;
+ if (!fs::create_directory(path, ec)) {
+ derr << __func__ << " failed to create " << path
+ << ": " << ec.message() << dendl;
+ return -ec.value();
+ }
+ fs::permissions(path,
+ fs::perms::owner_all |
+ fs::perms::group_read | fs::perms::group_exec |
+ fs::perms::others_read | fs::perms::others_exec);
}
}
return 0;
}
int RocksDBStore::install_cf_mergeop(
- const string &cf_name,
+ const string &key_prefix,
rocksdb::ColumnFamilyOptions *cf_opt)
{
ceph_assert(cf_opt != nullptr);
cf_opt->merge_operator.reset();
for (auto& i : merge_ops) {
- if (i.first == cf_name) {
+ if (i.first == key_prefix) {
cf_opt->merge_operator.reset(new MergeOperatorLinker(i.second));
}
}
}
int RocksDBStore::create_and_open(ostream &out,
- const vector<ColumnFamily>& cfs)
+ const std::string& cfs)
{
int r = create_db_dir();
if (r < 0)
return r;
- if (cfs.empty()) {
- return do_open(out, true, false, nullptr);
+ return do_open(out, true, false, cfs);
+}
+
+std::shared_ptr<rocksdb::Cache> RocksDBStore::create_block_cache(
+ const std::string& cache_type, size_t cache_size, double cache_prio_high) {
+ std::shared_ptr<rocksdb::Cache> cache;
+ auto shard_bits = cct->_conf->rocksdb_cache_shard_bits;
+ if (cache_type == "binned_lru") {
+ cache = rocksdb_cache::NewBinnedLRUCache(cct, cache_size, shard_bits, false, cache_prio_high);
+ } else if (cache_type == "lru") {
+ cache = rocksdb::NewLRUCache(cache_size, shard_bits);
+ } else if (cache_type == "clock") {
+ cache = rocksdb::NewClockCache(cache_size, shard_bits);
+ if (!cache) {
+ derr << "rocksdb_cache_type '" << cache
+ << "' chosen, but RocksDB not compiled with LibTBB. "
+ << dendl;
+ }
} else {
- return do_open(out, true, false, &cfs);
+ derr << "unrecognized rocksdb_cache_type '" << cache_type << "'" << dendl;
}
+ return cache;
}
int RocksDBStore::load_rocksdb_options(bool create_if_missing, rocksdb::Options& opt)
}
}
- if (g_conf()->rocksdb_perf) {
+ if (cct->_conf->rocksdb_perf) {
dbstats = rocksdb::CreateDBStatistics();
opt.statistics = dbstats;
}
return -e.code().value();
}
- if (g_conf()->rocksdb_log_to_ceph_log) {
- opt.info_log.reset(new CephRocksdbLogger(g_ceph_context));
+ if (cct->_conf->rocksdb_log_to_ceph_log) {
+ opt.info_log.reset(new CephRocksdbLogger(cct));
}
if (priv) {
dout(10) << __func__ << " using custom Env " << priv << dendl;
opt.env = static_cast<rocksdb::Env*>(priv);
+ } else {
+ env = opt.env;
}
opt.env->SetAllowNonOwnerAccess(false);
// caches
if (!set_cache_flag) {
- cache_size = g_conf()->rocksdb_cache_size;
+ cache_size = cct->_conf->rocksdb_cache_size;
}
- uint64_t row_cache_size = cache_size * g_conf()->rocksdb_cache_row_ratio;
+ uint64_t row_cache_size = cache_size * cct->_conf->rocksdb_cache_row_ratio;
uint64_t block_cache_size = cache_size - row_cache_size;
- if (g_conf()->rocksdb_cache_type == "binned_lru") {
- bbt_opts.block_cache = rocksdb_cache::NewBinnedLRUCache(
- cct,
- block_cache_size,
- g_conf()->rocksdb_cache_shard_bits);
- } 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);
- if (!bbt_opts.block_cache) {
- derr << "rocksdb_cache_type '" << g_conf()->rocksdb_cache_type
- << "' chosen, but RocksDB not compiled with LibTBB. "
- << dendl;
- return -EINVAL;
- }
- } else {
- derr << "unrecognized rocksdb_cache_type '" << g_conf()->rocksdb_cache_type
- << "'" << dendl;
+ bbt_opts.block_cache = create_block_cache(cct->_conf->rocksdb_cache_type, block_cache_size);
+ if (!bbt_opts.block_cache) {
return -EINVAL;
}
- bbt_opts.block_size = g_conf()->rocksdb_block_size;
+ bbt_opts.block_size = cct->_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");
+ cct->_conf->rocksdb_cache_shard_bits);
+ uint64_t bloom_bits = cct->_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));
}
using std::placeholders::_1;
- if (g_conf().with_val<std::string>("rocksdb_index_type",
+ if (cct->_conf.with_val<std::string>("rocksdb_index_type",
std::bind(std::equal_to<std::string>(), _1,
"binary_search")))
bbt_opts.index_type = rocksdb::BlockBasedTableOptions::IndexType::kBinarySearch;
- if (g_conf().with_val<std::string>("rocksdb_index_type",
+ if (cct->_conf.with_val<std::string>("rocksdb_index_type",
std::bind(std::equal_to<std::string>(), _1,
"hash_search")))
bbt_opts.index_type = rocksdb::BlockBasedTableOptions::IndexType::kHashSearch;
- if (g_conf().with_val<std::string>("rocksdb_index_type",
+ if (cct->_conf.with_val<std::string>("rocksdb_index_type",
std::bind(std::equal_to<std::string>(), _1,
"two_level")))
bbt_opts.index_type = rocksdb::BlockBasedTableOptions::IndexType::kTwoLevelIndexSearch;
if (!bbt_opts.no_block_cache) {
bbt_opts.cache_index_and_filter_blocks =
- g_conf().get_val<bool>("rocksdb_cache_index_and_filter_blocks");
+ cct->_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");
+ cct->_conf.get_val<bool>("rocksdb_cache_index_and_filter_blocks_with_high_priority");
bbt_opts.pin_l0_filter_and_index_blocks_in_cache =
- g_conf().get_val<bool>("rocksdb_pin_l0_filter_and_index_blocks_in_cache");
+ cct->_conf.get_val<bool>("rocksdb_pin_l0_filter_and_index_blocks_in_cache");
}
- bbt_opts.partition_filters = g_conf().get_val<bool>("rocksdb_partition_filters");
- if (g_conf().get_val<Option::size_t>("rocksdb_metadata_block_size") > 0)
- bbt_opts.metadata_block_size = g_conf().get_val<Option::size_t>("rocksdb_metadata_block_size");
+ bbt_opts.partition_filters = cct->_conf.get_val<bool>("rocksdb_partition_filters");
+ if (cct->_conf.get_val<Option::size_t>("rocksdb_metadata_block_size") > 0)
+ bbt_opts.metadata_block_size = cct->_conf.get_val<Option::size_t>("rocksdb_metadata_block_size");
opt.table_factory.reset(rocksdb::NewBlockBasedTableFactory(bbt_opts));
- dout(10) << __func__ << " block size " << g_conf()->rocksdb_block_size
+ dout(10) << __func__ << " block size " << cct->_conf->rocksdb_block_size
<< ", block_cache size " << byte_u_t(block_cache_size)
<< ", row_cache size " << byte_u_t(row_cache_size)
<< "; shards "
- << (1 << g_conf()->rocksdb_cache_shard_bits)
- << ", type " << g_conf()->rocksdb_cache_type
+ << (1 << cct->_conf->rocksdb_cache_shard_bits)
+ << ", type " << cct->_conf->rocksdb_cache_type
<< dendl;
opt.merge_operator.reset(new MergeOperatorRouter(*this));
+ comparator = opt.comparator;
+ return 0;
+}
+
+void RocksDBStore::add_column_family(const std::string& cf_name, uint32_t hash_l, uint32_t hash_h,
+ size_t shard_idx, rocksdb::ColumnFamilyHandle *handle) {
+ dout(10) << __func__ << " column_name=" << cf_name << " shard_idx=" << shard_idx <<
+ " hash_l=" << hash_l << " hash_h=" << hash_h << " handle=" << (void*) handle << dendl;
+ bool exists = cf_handles.count(cf_name) > 0;
+ auto& column = cf_handles[cf_name];
+ if (exists) {
+ ceph_assert(hash_l == column.hash_l);
+ ceph_assert(hash_h == column.hash_h);
+ } else {
+ ceph_assert(hash_l < hash_h);
+ column.hash_l = hash_l;
+ column.hash_h = hash_h;
+ }
+ if (column.handles.size() <= shard_idx)
+ column.handles.resize(shard_idx + 1);
+ column.handles[shard_idx] = handle;
+ cf_ids_to_prefix.emplace(handle->GetID(), cf_name);
+}
+
+bool RocksDBStore::is_column_family(const std::string& prefix) {
+ return cf_handles.count(prefix);
+}
+
+rocksdb::ColumnFamilyHandle *RocksDBStore::get_cf_handle(const std::string& prefix, const std::string& key) {
+ auto iter = cf_handles.find(prefix);
+ if (iter == cf_handles.end()) {
+ return nullptr;
+ } else {
+ if (iter->second.handles.size() == 1) {
+ return iter->second.handles[0];
+ } else {
+ uint32_t hash_l = std::min<uint32_t>(iter->second.hash_l, key.size());
+ uint32_t hash_h = std::min<uint32_t>(iter->second.hash_h, key.size());
+ uint32_t hash = ceph_str_hash_rjenkins(&key[hash_l], hash_h - hash_l);
+ return iter->second.handles[hash % iter->second.handles.size()];
+ }
+ }
+}
+
+rocksdb::ColumnFamilyHandle *RocksDBStore::get_cf_handle(const std::string& prefix, const char* key, size_t keylen) {
+ auto iter = cf_handles.find(prefix);
+ if (iter == cf_handles.end()) {
+ return nullptr;
+ } else {
+ if (iter->second.handles.size() == 1) {
+ return iter->second.handles[0];
+ } else {
+ uint32_t hash_l = std::min<uint32_t>(iter->second.hash_l, keylen);
+ uint32_t hash_h = std::min<uint32_t>(iter->second.hash_h, keylen);
+ uint32_t hash = ceph_str_hash_rjenkins(&key[hash_l], hash_h - hash_l);
+ return iter->second.handles[hash % iter->second.handles.size()];
+ }
+ }
+}
+
+/**
+ * Definition of sharding:
+ * space-separated list of: column_def [ '=' options ]
+ * column_def := column_name '(' shard_count ')'
+ * column_def := column_name '(' shard_count ',' hash_begin '-' ')'
+ * column_def := column_name '(' shard_count ',' hash_begin '-' hash_end ')'
+ * I=write_buffer_size=1048576 O(6) m(7,10-) prefix(4,0-10)=disable_auto_compactions=true,max_bytes_for_level_base=1048576
+ */
+bool RocksDBStore::parse_sharding_def(const std::string_view text_def_in,
+ std::vector<ColumnFamily>& sharding_def,
+ char const* *error_position,
+ std::string *error_msg)
+{
+ std::string_view text_def = text_def_in;
+ char const* error_position_local = nullptr;
+ std::string error_msg_local;
+ if (error_position == nullptr) {
+ error_position = &error_position_local;
+ }
+ *error_position = nullptr;
+ if (error_msg == nullptr) {
+ error_msg = &error_msg_local;
+ error_msg->clear();
+ }
+
+ sharding_def.clear();
+ while (!text_def.empty()) {
+ std::string_view options;
+ std::string_view name;
+ size_t shard_cnt = 1;
+ uint32_t l_bound = 0;
+ uint32_t h_bound = std::numeric_limits<uint32_t>::max();
+
+ std::string_view column_def;
+ size_t spos = text_def.find(' ');
+ if (spos == std::string_view::npos) {
+ column_def = text_def;
+ text_def = std::string_view(text_def.end(), 0);
+ } else {
+ column_def = text_def.substr(0, spos);
+ text_def = text_def.substr(spos + 1);
+ }
+ size_t eqpos = column_def.find('=');
+ if (eqpos != std::string_view::npos) {
+ options = column_def.substr(eqpos + 1);
+ column_def = column_def.substr(0, eqpos);
+ }
+
+ size_t bpos = column_def.find('(');
+ if (bpos != std::string_view::npos) {
+ name = column_def.substr(0, bpos);
+ const char* nptr = &column_def[bpos + 1];
+ char* endptr;
+ shard_cnt = strtol(nptr, &endptr, 10);
+ if (nptr == endptr) {
+ *error_position = nptr;
+ *error_msg = "expecting integer";
+ break;
+ }
+ nptr = endptr;
+ if (*nptr == ',') {
+ nptr++;
+ l_bound = strtol(nptr, &endptr, 10);
+ if (nptr == endptr) {
+ *error_position = nptr;
+ *error_msg = "expecting integer";
+ break;
+ }
+ nptr = endptr;
+ if (*nptr != '-') {
+ *error_position = nptr;
+ *error_msg = "expecting '-'";
+ break;
+ }
+ nptr++;
+ h_bound = strtol(nptr, &endptr, 10);
+ if (nptr == endptr) {
+ h_bound = std::numeric_limits<uint32_t>::max();
+ }
+ nptr = endptr;
+ }
+ if (*nptr != ')') {
+ *error_position = nptr;
+ *error_msg = "expecting ')'";
+ break;
+ }
+ } else {
+ name = column_def;
+ }
+ sharding_def.emplace_back(std::string(name), shard_cnt, std::string(options), l_bound, h_bound);
+ }
+ return *error_position == nullptr;
+}
+
+void RocksDBStore::sharding_def_to_columns(const std::vector<ColumnFamily>& sharding_def,
+ std::vector<std::string>& columns)
+{
+ columns.clear();
+ for (size_t i = 0; i < sharding_def.size(); i++) {
+ if (sharding_def[i].shard_cnt == 1) {
+ columns.push_back(sharding_def[i].name);
+ } else {
+ for (size_t j = 0; j < sharding_def[i].shard_cnt; j++) {
+ columns.push_back(sharding_def[i].name + "-" + to_string(j));
+ }
+ }
+ }
+}
+
+int RocksDBStore::create_shards(const rocksdb::Options& opt,
+ const std::vector<ColumnFamily>& sharding_def)
+{
+ for (auto& p : sharding_def) {
+ // copy default CF settings, block cache, merge operators as
+ // the base for new CF
+ rocksdb::ColumnFamilyOptions cf_opt(opt);
+ // user input options will override the base options
+ std::unordered_map<std::string, std::string> column_opts_map;
+ std::string block_cache_opts;
+ int r = extract_block_cache_options(p.options, &column_opts_map, &block_cache_opts);
+ if (r != 0) {
+ derr << __func__ << " failed to parse options; column family=" << p.name <<
+ " options=" << p.options << dendl;
+ return -EINVAL;
+ }
+ rocksdb::Status status;
+ status = rocksdb::GetColumnFamilyOptionsFromMap(cf_opt, column_opts_map, &cf_opt);
+ if (!status.ok()) {
+ derr << __func__ << " invalid db options; column family="
+ << p.name << " options=" << p.options << dendl;
+ return -EINVAL;
+ }
+ install_cf_mergeop(p.name, &cf_opt);
+ for (size_t idx = 0; idx < p.shard_cnt; idx++) {
+ std::string cf_name;
+ if (p.shard_cnt == 1)
+ cf_name = p.name;
+ else
+ cf_name = p.name + "-" + to_string(idx);
+ rocksdb::ColumnFamilyHandle *cf;
+ status = db->CreateColumnFamily(cf_opt, cf_name, &cf);
+ if (!status.ok()) {
+ derr << __func__ << " Failed to create rocksdb column family: "
+ << cf_name << dendl;
+ return -EINVAL;
+ }
+ // store the new CF handle
+ add_column_family(p.name, p.hash_l, p.hash_h, idx, cf);
+ }
+ }
+ return 0;
+}
+
+int RocksDBStore::apply_sharding(const rocksdb::Options& opt,
+ const std::string& sharding_text)
+{
+ // create and open column families
+ if (!sharding_text.empty()) {
+ bool b;
+ int r;
+ rocksdb::Status status;
+ std::vector<ColumnFamily> sharding_def;
+ char const* error_position;
+ std::string error_msg;
+ b = parse_sharding_def(sharding_text, sharding_def, &error_position, &error_msg);
+ if (!b) {
+ dout(1) << __func__ << " bad sharding: " << dendl;
+ dout(1) << __func__ << sharding_text << dendl;
+ dout(1) << __func__ << std::string(error_position - &sharding_text[0], ' ') << "^" << error_msg << dendl;
+ return -EINVAL;
+ }
+ r = create_shards(opt, sharding_def);
+ if (r != 0 ) {
+ derr << __func__ << " create_shards failed error=" << r << dendl;
+ return r;
+ }
+ opt.env->CreateDir(sharding_def_dir);
+ status = rocksdb::WriteStringToFile(opt.env, sharding_text,
+ sharding_def_file, true);
+ if (!status.ok()) {
+ derr << __func__ << " cannot write to " << sharding_def_file << dendl;
+ return -EIO;
+ }
+ } else {
+ opt.env->DeleteFile(sharding_def_file);
+ }
+ return 0;
+}
+// linking to rocksdb function defined in options_helper.cc
+// it can parse nested params like "nested_opt={opt1=1;opt2=2}"
+
+extern rocksdb::Status rocksdb::StringToMap(const std::string& opts_str,
+ std::unordered_map<std::string, std::string>* opts_map);
+
+int RocksDBStore::extract_block_cache_options(const std::string& opts_str,
+ std::unordered_map<std::string, std::string>* column_opts_map,
+ std::string* block_cache_opt)
+{
+ dout(5) << __func__ << " opts_str=" << opts_str << dendl;
+ rocksdb::Status status = rocksdb::StringToMap(opts_str, column_opts_map);
+ if (!status.ok()) {
+ dout(5) << __func__ << " error '" << status.getState() <<
+ "' while parsing options '" << opts_str << "'" << dendl;
+ return -EINVAL;
+ }
+ //extract "block_cache" option
+ if (auto it = column_opts_map->find("block_cache"); it != column_opts_map->end()) {
+ *block_cache_opt = it->second;
+ column_opts_map->erase(it);
+ } else {
+ block_cache_opt->clear();
+ }
+ return 0;
+}
+
+
+
+int RocksDBStore::verify_sharding(const rocksdb::Options& opt,
+ std::vector<rocksdb::ColumnFamilyDescriptor>& existing_cfs,
+ std::vector<std::pair<size_t, RocksDBStore::ColumnFamily> >& existing_cfs_shard,
+ std::vector<rocksdb::ColumnFamilyDescriptor>& missing_cfs,
+ std::vector<std::pair<size_t, RocksDBStore::ColumnFamily> >& missing_cfs_shard)
+{
+ rocksdb::Status status;
+ std::string stored_sharding_text;
+ status = opt.env->FileExists(sharding_def_file);
+ if (status.ok()) {
+ status = rocksdb::ReadFileToString(opt.env,
+ sharding_def_file,
+ &stored_sharding_text);
+ if(!status.ok()) {
+ derr << __func__ << " cannot read from " << sharding_def_file << dendl;
+ return -EIO;
+ }
+ dout(20) << __func__ << " sharding=" << stored_sharding_text << dendl;
+ } else {
+ dout(30) << __func__ << " no sharding" << dendl;
+ //no "sharding_def" present
+ }
+ //check if sharding_def matches stored_sharding_def
+ std::vector<ColumnFamily> stored_sharding_def;
+ parse_sharding_def(stored_sharding_text, stored_sharding_def);
+
+ std::sort(stored_sharding_def.begin(), stored_sharding_def.end(),
+ [](ColumnFamily& a, ColumnFamily& b) { return a.name < b.name; } );
+
+ std::vector<string> rocksdb_cfs;
+ status = rocksdb::DB::ListColumnFamilies(rocksdb::DBOptions(opt),
+ path, &rocksdb_cfs);
+ if (!status.ok()) {
+ return -EIO;
+ }
+ dout(5) << __func__ << " column families from rocksdb: " << rocksdb_cfs << dendl;
+
+ auto emplace_cf = [&] (const RocksDBStore::ColumnFamily& column,
+ int32_t shard_id,
+ const std::string& shard_name,
+ const rocksdb::ColumnFamilyOptions& opt) {
+ if (std::find(rocksdb_cfs.begin(), rocksdb_cfs.end(), shard_name) != rocksdb_cfs.end()) {
+ existing_cfs.emplace_back(shard_name, opt);
+ existing_cfs_shard.emplace_back(shard_id, column);
+ } else {
+ missing_cfs.emplace_back(shard_name, opt);
+ missing_cfs_shard.emplace_back(shard_id, column);
+ }
+ };
+
+ for (auto& column : stored_sharding_def) {
+ rocksdb::ColumnFamilyOptions cf_opt(opt);
+ std::unordered_map<std::string, std::string> options_map;
+ std::string block_cache_opt;
+
+ int r = extract_block_cache_options(column.options, &options_map, &block_cache_opt);
+ if (r != 0) {
+ derr << __func__ << " failed to parse options; column family=" << column.name <<
+ " options=" << column.options << dendl;
+ return -EINVAL;
+ }
+ status = rocksdb::GetColumnFamilyOptionsFromMap(cf_opt, options_map, &cf_opt);
+ if (!status.ok()) {
+ derr << __func__ << " invalid db column family options for CF '"
+ << column.name << "': " << column.options << dendl;
+ derr << __func__ << " error = '" << status.getState() << "'" << dendl;
+ return -EINVAL;
+ }
+ install_cf_mergeop(column.name, &cf_opt);
+
+ if (!block_cache_opt.empty()) {
+ std::unordered_map<std::string, std::string> cache_options_map;
+ status = rocksdb::StringToMap(block_cache_opt, &cache_options_map);
+ if (!status.ok()) {
+ derr << __func__ << " invalid block cache options; column=" << column.name <<
+ " options=" << block_cache_opt << dendl;
+ derr << __func__ << " error = '" << status.getState() << "'" << dendl;
+ return -EINVAL;
+ }
+ bool require_new_block_cache = false;
+ std::string cache_type = cct->_conf->rocksdb_cache_type;
+ if (const auto it = cache_options_map.find("type"); it !=cache_options_map.end()) {
+ cache_type = it->second;
+ cache_options_map.erase(it);
+ require_new_block_cache = true;
+ }
+ size_t cache_size = cct->_conf->rocksdb_cache_size;
+ if (auto it = cache_options_map.find("size"); it !=cache_options_map.end()) {
+ std::string error;
+ cache_size = strict_iecstrtoll(it->second.c_str(), &error);
+ if (!error.empty()) {
+ derr << __func__ << " invalid size: '" << it->second << "'" << dendl;
+ }
+ cache_options_map.erase(it);
+ require_new_block_cache = true;
+ }
+ double high_pri_pool_ratio = 0.0;
+ if (auto it = cache_options_map.find("high_ratio"); it !=cache_options_map.end()) {
+ std::string error;
+ high_pri_pool_ratio = strict_strtod(it->second.c_str(), &error);
+ if (!error.empty()) {
+ derr << __func__ << " invalid high_pri (float): '" << it->second << "'" << dendl;
+ }
+ cache_options_map.erase(it);
+ require_new_block_cache = true;
+ }
+
+ rocksdb::BlockBasedTableOptions column_bbt_opts;
+ status = GetBlockBasedTableOptionsFromMap(bbt_opts, cache_options_map, &column_bbt_opts);
+ if (!status.ok()) {
+ derr << __func__ << " invalid block cache options; column=" << column.name <<
+ " options=" << block_cache_opt << dendl;
+ derr << __func__ << " error = '" << status.getState() << "'" << dendl;
+ return -EINVAL;
+ }
+ std::shared_ptr<rocksdb::Cache> block_cache;
+ if (column_bbt_opts.no_block_cache) {
+ // clear all settings except no_block_cache
+ // rocksdb does not like then
+ column_bbt_opts = rocksdb::BlockBasedTableOptions();
+ column_bbt_opts.no_block_cache = true;
+ } else {
+ if (require_new_block_cache) {
+ block_cache = create_block_cache(cache_type, cache_size, high_pri_pool_ratio);
+ if (!block_cache) {
+ dout(5) << __func__ << " failed to create block cache for params: " << block_cache_opt << dendl;
+ return -EINVAL;
+ }
+ } else {
+ block_cache = bbt_opts.block_cache;
+ }
+ }
+ column_bbt_opts.block_cache = block_cache;
+ cf_bbt_opts[column.name] = column_bbt_opts;
+ cf_opt.table_factory.reset(NewBlockBasedTableFactory(cf_bbt_opts[column.name]));
+ }
+ if (column.shard_cnt == 1) {
+ emplace_cf(column, 0, column.name, cf_opt);
+ } else {
+ for (size_t i = 0; i < column.shard_cnt; i++) {
+ std::string cf_name = column.name + "-" + to_string(i);
+ emplace_cf(column, i, cf_name, cf_opt);
+ }
+ }
+ }
+ existing_cfs.emplace_back("default", opt);
+ if (existing_cfs.size() != rocksdb_cfs.size()) {
+ std::vector<std::string> columns_from_stored;
+ sharding_def_to_columns(stored_sharding_def, columns_from_stored);
+ derr << __func__ << " extra columns in rocksdb. rocksdb columns = " << rocksdb_cfs
+ << " target columns = " << columns_from_stored << dendl;
+ return -EIO;
+ }
return 0;
}
+std::ostream& operator<<(std::ostream& out, const RocksDBStore::ColumnFamily& cf)
+{
+ out << "(";
+ out << cf.name;
+ out << ",";
+ out << cf.shard_cnt;
+ out << ",";
+ out << cf.hash_l;
+ out << "-";
+ if (cf.hash_h != std::numeric_limits<uint32_t>::max()) {
+ out << cf.hash_h;
+ }
+ out << ",";
+ out << cf.options;
+ out << ")";
+ return out;
+}
+
int RocksDBStore::do_open(ostream &out,
bool create_if_missing,
bool open_readonly,
- const vector<ColumnFamily>* cfs)
+ const std::string& sharding_text)
{
ceph_assert(!(create_if_missing && open_readonly));
rocksdb::Options opt;
derr << status.ToString() << dendl;
return -EINVAL;
}
- // create and open column families
- if (cfs) {
- for (auto& p : *cfs) {
- // copy default CF settings, block cache, merge operators as
- // the base for new CF
- rocksdb::ColumnFamilyOptions cf_opt(opt);
- // user input options will override the base options
- status = rocksdb::GetColumnFamilyOptionsFromString(
- cf_opt, p.option, &cf_opt);
- if (!status.ok()) {
- derr << __func__ << " invalid db column family option string for CF: "
- << p.name << dendl;
- return -EINVAL;
- }
- install_cf_mergeop(p.name, &cf_opt);
- rocksdb::ColumnFamilyHandle *cf;
- status = db->CreateColumnFamily(cf_opt, p.name, &cf);
- if (!status.ok()) {
- derr << __func__ << " Failed to create rocksdb column family: "
- << p.name << dendl;
- return -EINVAL;
- }
- // store the new CF handle
- add_column_family(p.name, static_cast<void*>(cf));
- }
+ r = apply_sharding(opt, sharding_text);
+ if (r < 0) {
+ return r;
}
default_cf = db->DefaultColumnFamily();
} else {
- std::vector<string> existing_cfs;
- status = rocksdb::DB::ListColumnFamilies(
- rocksdb::DBOptions(opt),
- path,
- &existing_cfs);
- dout(1) << __func__ << " column families: " << existing_cfs << dendl;
+ std::vector<rocksdb::ColumnFamilyDescriptor> existing_cfs;
+ std::vector<std::pair<size_t, RocksDBStore::ColumnFamily> > existing_cfs_shard;
+ std::vector<rocksdb::ColumnFamilyDescriptor> missing_cfs;
+ std::vector<std::pair<size_t, RocksDBStore::ColumnFamily> > missing_cfs_shard;
+
+ r = verify_sharding(opt,
+ existing_cfs, existing_cfs_shard,
+ missing_cfs, missing_cfs_shard);
+ if (r < 0) {
+ return r;
+ }
+ std::string sharding_recreate_text;
+ status = rocksdb::ReadFileToString(opt.env,
+ sharding_recreate,
+ &sharding_recreate_text);
+ bool recreate_mode = status.ok() && sharding_recreate_text == "1";
+
+ ceph_assert(!recreate_mode || !open_readonly);
+ if (recreate_mode == false && missing_cfs.size() != 0) {
+ // We do not accept when there are missing column families, except case that we are during resharding.
+ // We can get into this case if resharding was interrupted. It gives a chance to continue.
+ // Opening DB is only allowed in read-only mode.
+ if (open_readonly == false &&
+ std::find_if(missing_cfs.begin(), missing_cfs.end(),
+ [](const rocksdb::ColumnFamilyDescriptor& c) { return c.name == resharding_column_lock; }
+ ) != missing_cfs.end()) {
+ derr << __func__ << " missing column families: " << missing_cfs_shard << dendl;
+ return -EIO;
+ }
+ }
+
if (existing_cfs.empty()) {
// no column families
if (open_readonly) {
- status = rocksdb::DB::Open(opt, path, &db);
+ status = rocksdb::DB::OpenForReadOnly(opt, path, &db);
} else {
- status = rocksdb::DB::OpenForReadOnly(opt, path, &db);
+ status = rocksdb::DB::Open(opt, path, &db);
}
if (!status.ok()) {
derr << status.ToString() << dendl;
}
default_cf = db->DefaultColumnFamily();
} else {
- // we cannot change column families for a created database. so, map
- // what options we are given to whatever cf's already exist.
- std::vector<rocksdb::ColumnFamilyDescriptor> column_families;
- for (auto& n : existing_cfs) {
- // copy default CF settings, block cache, merge operators as
- // the base for new CF
- rocksdb::ColumnFamilyOptions cf_opt(opt);
- bool found = false;
- if (cfs) {
- for (auto& i : *cfs) {
- if (i.name == n) {
- found = true;
- status = rocksdb::GetColumnFamilyOptionsFromString(
- cf_opt, i.option, &cf_opt);
- if (!status.ok()) {
- derr << __func__ << " invalid db column family options for CF '"
- << i.name << "': " << i.option << dendl;
- return -EINVAL;
- }
- }
- }
- }
- if (n != rocksdb::kDefaultColumnFamilyName) {
- install_cf_mergeop(n, &cf_opt);
- }
- column_families.push_back(rocksdb::ColumnFamilyDescriptor(n, cf_opt));
- if (!found && n != rocksdb::kDefaultColumnFamilyName) {
- dout(1) << __func__ << " column family '" << n
- << "' exists but not expected" << dendl;
- }
- }
std::vector<rocksdb::ColumnFamilyHandle*> handles;
if (open_readonly) {
status = rocksdb::DB::OpenForReadOnly(rocksdb::DBOptions(opt),
- path, column_families,
+ path, existing_cfs,
&handles, &db);
} else {
status = rocksdb::DB::Open(rocksdb::DBOptions(opt),
- path, column_families, &handles, &db);
+ path, existing_cfs, &handles, &db);
}
if (!status.ok()) {
derr << status.ToString() << dendl;
return -EINVAL;
}
- for (unsigned i = 0; i < existing_cfs.size(); ++i) {
- if (existing_cfs[i] == rocksdb::kDefaultColumnFamilyName) {
- default_cf = handles[i];
- must_close_default_cf = true;
- } else {
- add_column_family(existing_cfs[i], static_cast<void*>(handles[i]));
+ ceph_assert(existing_cfs.size() == existing_cfs_shard.size() + 1);
+ ceph_assert(handles.size() == existing_cfs.size());
+ dout(10) << __func__ << " existing_cfs=" << existing_cfs.size() << dendl;
+ for (size_t i = 0; i < existing_cfs_shard.size(); i++) {
+ add_column_family(existing_cfs_shard[i].second.name,
+ existing_cfs_shard[i].second.hash_l,
+ existing_cfs_shard[i].second.hash_h,
+ existing_cfs_shard[i].first,
+ handles[i]);
+ }
+ default_cf = handles[handles.size() - 1];
+ must_close_default_cf = true;
+
+ if (missing_cfs.size() > 0 &&
+ std::find_if(missing_cfs.begin(), missing_cfs.end(),
+ [](const rocksdb::ColumnFamilyDescriptor& c) { return c.name == resharding_column_lock; }
+ ) == missing_cfs.end())
+ {
+ dout(10) << __func__ << " missing_cfs=" << missing_cfs.size() << dendl;
+ ceph_assert(recreate_mode);
+ ceph_assert(missing_cfs.size() == missing_cfs_shard.size());
+ for (size_t i = 0; i < missing_cfs.size(); i++) {
+ rocksdb::ColumnFamilyHandle *cf;
+ status = db->CreateColumnFamily(missing_cfs[i].options, missing_cfs[i].name, &cf);
+ if (!status.ok()) {
+ derr << __func__ << " Failed to create rocksdb column family: "
+ << missing_cfs[i].name << dendl;
+ return -EINVAL;
+ }
+ add_column_family(missing_cfs_shard[i].second.name,
+ missing_cfs_shard[i].second.hash_l,
+ missing_cfs_shard[i].second.hash_h,
+ missing_cfs_shard[i].first,
+ cf);
}
+ opt.env->DeleteFile(sharding_recreate);
}
}
}
ceph_assert(default_cf != nullptr);
- PerfCountersBuilder plb(g_ceph_context, "rocksdb", l_rocksdb_first, l_rocksdb_last);
+ PerfCountersBuilder plb(cct, "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");
RocksDBStore::~RocksDBStore()
{
close();
- delete logger;
-
- // Ensure db is destroyed before dependent db_cache and filterpolicy
- for (auto& p : cf_handles) {
- db->DestroyColumnFamilyHandle(
- static_cast<rocksdb::ColumnFamilyHandle*>(p.second));
- p.second = nullptr;
- }
- if (must_close_default_cf) {
- db->DestroyColumnFamilyHandle(default_cf);
- must_close_default_cf = false;
- }
- default_cf = nullptr;
- delete db;
- db = nullptr;
-
if (priv) {
delete static_cast<rocksdb::Env*>(priv);
}
compact_queue_cond.notify_all();
compact_queue_lock.unlock();
compact_thread.join();
- dout(1) << __func__ << " compaction thread to stopped" << dendl;
+ dout(1) << __func__ << " compaction thread to stopped" << dendl;
} else {
compact_queue_lock.unlock();
}
- if (logger)
+ if (logger) {
cct->get_perfcounters_collection()->remove(logger);
+ delete logger;
+ logger = nullptr;
+ }
+
+ // Ensure db is destroyed before dependent db_cache and filterpolicy
+ for (auto& p : cf_handles) {
+ for (size_t i = 0; i < p.second.handles.size(); i++) {
+ db->DestroyColumnFamilyHandle(p.second.handles[i]);
+ }
+ }
+ cf_handles.clear();
+ if (must_close_default_cf) {
+ db->DestroyColumnFamilyHandle(default_cf);
+ must_close_default_cf = false;
+ }
+ default_cf = nullptr;
+ delete db;
+ db = nullptr;
}
int RocksDBStore::repair(std::ostream &out)
{
+ rocksdb::Status status;
rocksdb::Options opt;
int r = load_rocksdb_options(false, opt);
if (r) {
out << "load rocksdb options failed" << std::endl;
return r;
}
- rocksdb::Status status = rocksdb::RepairDB(path, opt);
+ //need to save sharding definition, repairDB will delete files it does not know
+ std::string stored_sharding_text;
+ status = opt.env->FileExists(sharding_def_file);
if (status.ok()) {
+ status = rocksdb::ReadFileToString(opt.env,
+ sharding_def_file,
+ &stored_sharding_text);
+ if (!status.ok()) {
+ stored_sharding_text.clear();
+ }
+ }
+ dout(10) << __func__ << " stored_sharding: " << stored_sharding_text << dendl;
+ status = rocksdb::RepairDB(path, opt);
+ bool repaired = status.ok();
+ if (!stored_sharding_text.empty()) {
+ //recreate markers even if repair failed
+ opt.env->CreateDir(sharding_def_dir);
+ status = rocksdb::WriteStringToFile(opt.env, stored_sharding_text,
+ sharding_def_file, true);
+ if (!status.ok()) {
+ derr << __func__ << " cannot write to " << sharding_def_file << dendl;
+ return -1;
+ }
+ status = rocksdb::WriteStringToFile(opt.env, "1",
+ sharding_recreate, true);
+ if (!status.ok()) {
+ derr << __func__ << " cannot write to " << sharding_recreate << dendl;
+ return -1;
+ }
+ // fiinalize sharding recreate
+ if (do_open(out, false, false)) {
+ derr << __func__ << " cannot finalize repair" << dendl;
+ return -1;
+ }
+ close();
+ }
+
+ if (repaired && status.ok()) {
return 0;
} else {
out << "repair rocksdb failed : " << status.ToString() << std::endl;
- return 1;
+ return -1;
}
}
int64_t RocksDBStore::estimate_prefix_size(const string& prefix,
const string& key_prefix)
{
- auto cf = get_cf_handle(prefix);
uint64_t size = 0;
uint8_t flags =
//rocksdb::DB::INCLUDE_MEMTABLES | // do not include memtables...
rocksdb::DB::INCLUDE_FILES;
- if (cf) {
- string start = key_prefix + string(1, '\x00');
- string limit = key_prefix + string("\xff\xff\xff\xff");
- rocksdb::Range r(start, limit);
- db->GetApproximateSizes(cf, &r, 1, &size, flags);
+ auto p_iter = cf_handles.find(prefix);
+ if (p_iter != cf_handles.end()) {
+ for (auto cf : p_iter->second.handles) {
+ uint64_t s = 0;
+ string start = key_prefix + string(1, '\x00');
+ string limit = key_prefix + string("\xff\xff\xff\xff");
+ rocksdb::Range r(start, limit);
+ db->GetApproximateSizes(cf, &r, 1, &s, flags);
+ size += s;
+ }
} else {
string start = combine_strings(prefix , key_prefix);
string limit = combine_strings(prefix , key_prefix + "\xff\xff\xff\xff");
void RocksDBStore::get_statistics(Formatter *f)
{
- if (!g_conf()->rocksdb_perf) {
+ if (!cct->_conf->rocksdb_perf) {
dout(20) << __func__ << " RocksDB perf is disabled, can't probe for stats"
<< dendl;
return;
}
- if (g_conf()->rocksdb_collect_compaction_stats) {
+ if (cct->_conf->rocksdb_collect_compaction_stats) {
std::string stat_str;
bool status = db->GetProperty("rocksdb.stats", &stat_str);
if (status) {
f->close_section();
}
}
- if (g_conf()->rocksdb_collect_extended_stats) {
+ if (cct->_conf->rocksdb_collect_extended_stats) {
if (dbstats) {
f->open_object_section("rocksdb_extended_statistics");
string stat_str = dbstats->ToString();
logger->dump_formatted(f,0);
f->close_section();
}
- if (g_conf()->rocksdb_collect_memory_stats) {
+ if (cct->_conf->rocksdb_collect_memory_stats) {
f->open_object_section("rocksdb_memtable_statistics");
std::string str;
if (!bbt_opts.no_block_cache) {
}
}
+struct RocksDBStore::RocksWBHandler: public rocksdb::WriteBatch::Handler {
+ RocksWBHandler(const RocksDBStore& db) : db(db) {}
+ const RocksDBStore& db;
+ std::stringstream seen;
+ int num_seen = 0;
+
+ void dump(const char* op_name,
+ uint32_t column_family_id,
+ const rocksdb::Slice& key_in,
+ const rocksdb::Slice* value = nullptr) {
+ string prefix;
+ string key;
+ ssize_t size = value ? value->size() : -1;
+ seen << std::endl << op_name << "(";
+
+ if (column_family_id == 0) {
+ db.split_key(key_in, &prefix, &key);
+ } else {
+ auto it = db.cf_ids_to_prefix.find(column_family_id);
+ ceph_assert(it != db.cf_ids_to_prefix.end());
+ prefix = it->second;
+ key = key_in.ToString();
+ }
+ seen << " prefix = " << prefix;
+ seen << " key = " << pretty_binary_string(key);
+ if (size != -1)
+ seen << " value size = " << std::to_string(size);
+ seen << ")";
+ num_seen++;
+ }
+ void Put(const rocksdb::Slice& key,
+ const rocksdb::Slice& value) override {
+ dump("Put", 0, key, &value);
+ }
+ rocksdb::Status PutCF(uint32_t column_family_id, const rocksdb::Slice& key,
+ const rocksdb::Slice& value) override {
+ dump("PutCF", column_family_id, key, &value);
+ return rocksdb::Status::OK();
+ }
+ void SingleDelete(const rocksdb::Slice& key) override {
+ dump("SingleDelete", 0, key);
+ }
+ rocksdb::Status SingleDeleteCF(uint32_t column_family_id, const rocksdb::Slice& key) override {
+ dump("SingleDeleteCF", column_family_id, key);
+ return rocksdb::Status::OK();
+ }
+ void Delete(const rocksdb::Slice& key) override {
+ dump("Delete", 0, key);
+ }
+ rocksdb::Status DeleteCF(uint32_t column_family_id, const rocksdb::Slice& key) override {
+ dump("DeleteCF", column_family_id, key);
+ return rocksdb::Status::OK();
+ }
+ void Merge(const rocksdb::Slice& key,
+ const rocksdb::Slice& value) override {
+ dump("Merge", 0, key, &value);
+ }
+ rocksdb::Status MergeCF(uint32_t column_family_id, const rocksdb::Slice& key,
+ const rocksdb::Slice& value) override {
+ dump("MergeCF", column_family_id, key, &value);
+ return rocksdb::Status::OK();
+ }
+ bool Continue() override { return num_seen < 50; }
+};
+
int RocksDBStore::submit_common(rocksdb::WriteOptions& woptions, KeyValueDB::Transaction t)
{
// enable rocksdb breakdown
// considering performance overhead, default is disabled
- if (g_conf()->rocksdb_perf) {
+ if (cct->_conf->rocksdb_perf) {
rocksdb::SetPerfLevel(rocksdb::PerfLevel::kEnableTimeExceptForMutex);
rocksdb::get_perf_context()->Reset();
}
static_cast<RocksDBTransactionImpl *>(t.get());
woptions.disableWAL = disableWAL;
lgeneric_subdout(cct, rocksdb, 30) << __func__;
- RocksWBHandler bat_txc;
+ RocksWBHandler bat_txc(*this);
_t->bat.Iterate(&bat_txc);
- *_dout << " Rocksdb transaction: " << bat_txc.seen << dendl;
+ *_dout << " Rocksdb transaction: " << bat_txc.seen.str() << dendl;
rocksdb::Status s = db->Write(woptions, &_t->bat);
if (!s.ok()) {
- RocksWBHandler rocks_txc;
+ RocksWBHandler rocks_txc(*this);
_t->bat.Iterate(&rocks_txc);
derr << __func__ << " error: " << s.ToString() << " code = " << s.code()
- << " Rocksdb transaction: " << rocks_txc.seen << dendl;
+ << " Rocksdb transaction: " << rocks_txc.seen.str() << dendl;
}
- if (g_conf()->rocksdb_perf) {
+ if (cct->_conf->rocksdb_perf) {
utime_t write_memtable_time;
utime_t write_delay_time;
utime_t write_wal_time;
int result = submit_common(woptions, t);
utime_t lat = ceph_clock_now() - start;
- logger->inc(l_rocksdb_txns);
logger->tinc(l_rocksdb_submit_latency, lat);
return result;
int result = submit_common(woptions, t);
utime_t lat = ceph_clock_now() - start;
- logger->inc(l_rocksdb_txns_sync);
logger->tinc(l_rocksdb_submit_sync_latency, lat);
return result;
const string &k,
const bufferlist &to_set_bl)
{
- auto cf = db->get_cf_handle(prefix);
+ auto cf = db->get_cf_handle(prefix, k);
if (cf) {
put_bat(bat, cf, k, to_set_bl);
} else {
const char *k, size_t keylen,
const bufferlist &to_set_bl)
{
- auto cf = db->get_cf_handle(prefix);
+ auto cf = db->get_cf_handle(prefix, k, keylen);
if (cf) {
string key(k, keylen); // fixme?
put_bat(bat, cf, key, to_set_bl);
void RocksDBStore::RocksDBTransactionImpl::rmkey(const string &prefix,
const string &k)
{
- auto cf = db->get_cf_handle(prefix);
+ auto cf = db->get_cf_handle(prefix, k);
if (cf) {
bat.Delete(cf, rocksdb::Slice(k));
} else {
const char *k,
size_t keylen)
{
- auto cf = db->get_cf_handle(prefix);
+ auto cf = db->get_cf_handle(prefix, k, keylen);
if (cf) {
bat.Delete(cf, rocksdb::Slice(k, keylen));
} else {
void RocksDBStore::RocksDBTransactionImpl::rm_single_key(const string &prefix,
const string &k)
{
- auto cf = db->get_cf_handle(prefix);
+ auto cf = db->get_cf_handle(prefix, k);
if (cf) {
bat.SingleDelete(cf, k);
} else {
void RocksDBStore::RocksDBTransactionImpl::rmkeys_by_prefix(const string &prefix)
{
- auto cf = db->get_cf_handle(prefix);
- uint64_t cnt = db->delete_range_threshold;
- bat.SetSavePoint();
- auto it = db->get_iterator(prefix);
- for (it->seek_to_first(); it->valid(); it->next()) {
- if (!cnt) {
- bat.RollbackToSavePoint();
- if (cf) {
- string endprefix = "\xff\xff\xff\xff"; // FIXME: this is cheating...
- bat.DeleteRange(cf, string(), endprefix);
- } else {
- string endprefix = prefix;
+ auto p_iter = db->cf_handles.find(prefix);
+ if (p_iter == db->cf_handles.end()) {
+ uint64_t cnt = db->delete_range_threshold;
+ bat.SetSavePoint();
+ auto it = db->get_iterator(prefix);
+ for (it->seek_to_first(); it->valid() && (--cnt) != 0; it->next()) {
+ bat.Delete(db->default_cf, combine_strings(prefix, it->key()));
+ }
+ if (cnt == 0) {
+ bat.RollbackToSavePoint();
+ string endprefix = prefix;
endprefix.push_back('\x01');
bat.DeleteRange(db->default_cf,
combine_strings(prefix, string()),
combine_strings(endprefix, string()));
- }
- return;
- }
- if (cf) {
- bat.Delete(cf, rocksdb::Slice(it->key()));
} else {
- bat.Delete(db->default_cf, combine_strings(prefix, it->key()));
+ bat.PopSavePoint();
+ }
+ } else {
+ ceph_assert(p_iter->second.handles.size() >= 1);
+ for (auto cf : p_iter->second.handles) {
+ uint64_t cnt = db->delete_range_threshold;
+ bat.SetSavePoint();
+ auto it = db->new_shard_iterator(cf);
+ for (it->SeekToFirst(); it->Valid() && (--cnt) != 0; it->Next()) {
+ bat.Delete(cf, it->key());
+ }
+ if (cnt == 0) {
+ bat.RollbackToSavePoint();
+ string endprefix = "\xff\xff\xff\xff"; // FIXME: this is cheating...
+ bat.DeleteRange(cf, string(), endprefix);
+ } else {
+ bat.PopSavePoint();
+ }
}
- --cnt;
}
- bat.PopSavePoint();
}
void RocksDBStore::RocksDBTransactionImpl::rm_range_keys(const string &prefix,
const string &start,
const string &end)
{
- auto cf = db->get_cf_handle(prefix);
-
- uint64_t cnt = db->delete_range_threshold;
- auto it = db->get_iterator(prefix);
- bat.SetSavePoint();
- it->lower_bound(start);
- while (it->valid()) {
- if (it->key() >= end) {
- break;
+ auto p_iter = db->cf_handles.find(prefix);
+ if (p_iter == db->cf_handles.end()) {
+ uint64_t cnt = db->delete_range_threshold;
+ bat.SetSavePoint();
+ auto it = db->get_iterator(prefix);
+ for (it->lower_bound(start);
+ it->valid() && db->comparator->Compare(it->key(), end) < 0 && (--cnt) != 0;
+ it->next()) {
+ bat.Delete(db->default_cf, combine_strings(prefix, it->key()));
}
- if (!cnt) {
+ if (cnt == 0) {
bat.RollbackToSavePoint();
- if (cf) {
- bat.DeleteRange(cf, rocksdb::Slice(start), rocksdb::Slice(end));
+ bat.DeleteRange(db->default_cf,
+ rocksdb::Slice(combine_strings(prefix, start)),
+ rocksdb::Slice(combine_strings(prefix, end)));
+ } else {
+ bat.PopSavePoint();
+ }
+ } else {
+ ceph_assert(p_iter->second.handles.size() >= 1);
+ for (auto cf : p_iter->second.handles) {
+ uint64_t cnt = db->delete_range_threshold;
+ bat.SetSavePoint();
+ rocksdb::Iterator* it = db->new_shard_iterator(cf);
+ ceph_assert(it != nullptr);
+ for (it->Seek(start);
+ it->Valid() && db->comparator->Compare(it->key(), end) < 0 && (--cnt) != 0;
+ it->Next()) {
+ bat.Delete(cf, it->key());
+ }
+ if (cnt == 0) {
+ bat.RollbackToSavePoint();
+ bat.DeleteRange(cf, rocksdb::Slice(start), rocksdb::Slice(end));
} else {
- bat.DeleteRange(db->default_cf,
- rocksdb::Slice(combine_strings(prefix, start)),
- rocksdb::Slice(combine_strings(prefix, end)));
+ bat.PopSavePoint();
}
- return;
+ delete it;
}
- if (cf) {
- bat.Delete(cf, rocksdb::Slice(it->key()));
- } else {
- bat.Delete(db->default_cf, combine_strings(prefix, it->key()));
- }
- it->next();
- --cnt;
}
- bat.PopSavePoint();
}
void RocksDBStore::RocksDBTransactionImpl::merge(
const string &k,
const bufferlist &to_set_bl)
{
- auto cf = db->get_cf_handle(prefix);
+ auto cf = db->get_cf_handle(prefix, k);
if (cf) {
// bufferlist::c_str() is non-constant, so we can't call c_str()
if (to_set_bl.is_contiguous() && to_set_bl.length() > 0) {
const std::set<string> &keys,
std::map<string, bufferlist> *out)
{
+ rocksdb::PinnableSlice value;
utime_t start = ceph_clock_now();
- auto cf = get_cf_handle(prefix);
- if (cf) {
+ if (cf_handles.count(prefix) > 0) {
for (auto& key : keys) {
- std::string value;
+ auto cf_handle = get_cf_handle(prefix, key);
auto status = db->Get(rocksdb::ReadOptions(),
- cf,
+ cf_handle,
rocksdb::Slice(key),
&value);
if (status.ok()) {
- (*out)[key].append(value);
+ (*out)[key].append(value.data(), value.size());
} else if (status.IsIOError()) {
ceph_abort_msg(status.getState());
}
+ value.Reset();
}
} else {
for (auto& key : keys) {
- std::string value;
string k = combine_strings(prefix, key);
auto status = db->Get(rocksdb::ReadOptions(),
default_cf,
rocksdb::Slice(k),
&value);
if (status.ok()) {
- (*out)[key].append(value);
+ (*out)[key].append(value.data(), value.size());
} else if (status.IsIOError()) {
ceph_abort_msg(status.getState());
}
+ value.Reset();
}
}
utime_t lat = ceph_clock_now() - start;
ceph_assert(out && (out->length() == 0));
utime_t start = ceph_clock_now();
int r = 0;
- string value;
+ rocksdb::PinnableSlice value;
rocksdb::Status s;
- auto cf = get_cf_handle(prefix);
+ auto cf = get_cf_handle(prefix, key);
if (cf) {
s = db->Get(rocksdb::ReadOptions(),
cf,
&value);
}
if (s.ok()) {
- out->append(value);
+ out->append(value.data(), value.size());
} else if (s.IsNotFound()) {
r = -ENOENT;
} else {
ceph_assert(out && (out->length() == 0));
utime_t start = ceph_clock_now();
int r = 0;
- string value;
+ rocksdb::PinnableSlice value;
rocksdb::Status s;
- auto cf = get_cf_handle(prefix);
+ auto cf = get_cf_handle(prefix, key, keylen);
if (cf) {
s = db->Get(rocksdb::ReadOptions(),
cf,
&value);
}
if (s.ok()) {
- out->append(value);
+ out->append(value.data(), value.size());
} else if (s.IsNotFound()) {
r = -ENOENT;
} else {
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)
rocksdb::CompactRangeOptions options;
db->CompactRange(options, default_cf, nullptr, nullptr);
for (auto cf : cf_handles) {
- db->CompactRange(
- options,
- static_cast<rocksdb::ColumnFamilyHandle*>(cf.second),
- nullptr, nullptr);
+ for (auto shard_cf : cf.second.handles) {
+ db->CompactRange(
+ options,
+ shard_cf,
+ nullptr, nullptr);
+ }
}
}
-
void RocksDBStore::compact_thread_entry()
{
std::unique_lock l{compact_queue_lock};
dout(10) << __func__ << " enter" << dendl;
while (!compact_queue_stop) {
if (!compact_queue.empty()) {
- pair<string,string> range = compact_queue.front();
+ auto range = compact_queue.front();
compact_queue.pop_front();
logger->set(l_rocksdb_compact_queue_len, compact_queue.size());
l.unlock();
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);
+ string prefix_start, key_start;
+ string prefix_end, key_end;
+ string key_highest = "\xff\xff\xff\xff"; //cheating
+ string key_lowest = "";
+
+ auto compact_range = [&] (const decltype(cf_handles)::iterator column_it,
+ const std::string& start,
+ const std::string& end) {
+ rocksdb::Slice cstart(start);
+ rocksdb::Slice cend(end);
+ for (const auto& shard_it : column_it->second.handles) {
+ db->CompactRange(options, shard_it, &cstart, &cend);
+ }
+ };
+ db->CompactRange(options, default_cf, &cstart, &cend);
+ split_key(cstart, &prefix_start, &key_start);
+ split_key(cend, &prefix_end, &key_end);
+ if (prefix_start == prefix_end) {
+ const auto& column = cf_handles.find(prefix_start);
+ if (column != cf_handles.end()) {
+ compact_range(column, key_start, key_end);
+ }
+ } else {
+ auto column = cf_handles.find(prefix_start);
+ if (column != cf_handles.end()) {
+ compact_range(column, key_start, key_highest);
+ ++column;
+ }
+ const auto& column_end = cf_handles.find(prefix_end);
+ while (column != column_end) {
+ compact_range(column, key_lowest, key_highest);
+ column++;
+ }
+ if (column != cf_handles.end()) {
+ compact_range(column, key_lowest, key_end);
+ }
+ }
}
RocksDBStore::RocksDBWholeSpaceIteratorImpl::~RocksDBWholeSpaceIteratorImpl()
return limit;
}
-RocksDBStore::WholeSpaceIterator RocksDBStore::get_wholespace_iterator()
-{
- return std::make_shared<RocksDBWholeSpaceIteratorImpl>(
- db->NewIterator(rocksdb::ReadOptions(), default_cf));
-}
-
class CFIteratorImpl : public KeyValueDB::IteratorImpl {
protected:
string prefix;
}
};
-KeyValueDB::Iterator RocksDBStore::get_iterator(const std::string& prefix)
+
+//merge column iterators and rest iterator
+class WholeMergeIteratorImpl : public KeyValueDB::WholeSpaceIteratorImpl {
+private:
+ RocksDBStore* db;
+ KeyValueDB::WholeSpaceIterator main;
+ std::map<std::string, KeyValueDB::Iterator> shards;
+ std::map<std::string, KeyValueDB::Iterator>::iterator current_shard;
+ enum {on_main, on_shard} smaller;
+
+public:
+ WholeMergeIteratorImpl(RocksDBStore* db)
+ : db(db)
+ , main(db->get_default_cf_iterator())
+ {
+ for (auto& e : db->cf_handles) {
+ shards.emplace(e.first, db->get_iterator(e.first));
+ }
+ }
+
+ // returns true if value in main is smaller then in shards
+ // invalid is larger then actual value
+ bool is_main_smaller() {
+ if (main->valid()) {
+ if (current_shard != shards.end()) {
+ auto main_rk = main->raw_key();
+ ceph_assert(current_shard->second->valid());
+ auto shards_rk = current_shard->second->raw_key();
+ if (main_rk.first < shards_rk.first)
+ return true;
+ if (main_rk.first > shards_rk.first)
+ return false;
+ return main_rk.second < shards_rk.second;
+ } else {
+ return true;
+ }
+ } else {
+ if (current_shard != shards.end()) {
+ return false;
+ } else {
+ //this means that neither is valid
+ //we select main to be smaller, so valid() will signal properly
+ return true;
+ }
+ }
+ }
+
+ int seek_to_first() override {
+ int r0 = main->seek_to_first();
+ int r1 = 0;
+ // find first shard that has some data
+ current_shard = shards.begin();
+ while (current_shard != shards.end()) {
+ r1 = current_shard->second->seek_to_first();
+ if (r1 != 0 || current_shard->second->valid()) {
+ //this is the first shard that will yield some keys
+ break;
+ }
+ ++current_shard;
+ }
+ smaller = is_main_smaller() ? on_main : on_shard;
+ return r0 == 0 && r1 == 0 ? 0 : -1;
+ }
+
+ int seek_to_first(const std::string &prefix) override {
+ int r0 = main->seek_to_first(prefix);
+ int r1 = 0;
+ // find first shard that has some data
+ current_shard = shards.lower_bound(prefix);
+ while (current_shard != shards.end()) {
+ r1 = current_shard->second->seek_to_first();
+ if (r1 != 0 || current_shard->second->valid()) {
+ //this is the first shard that will yield some keys
+ break;
+ }
+ ++current_shard;
+ }
+ smaller = is_main_smaller() ? on_main : on_shard;
+ return r0 == 0 && r1 == 0 ? 0 : -1;
+ };
+
+ int seek_to_last() override {
+ int r0 = main->seek_to_last();
+ int r1 = 0;
+ r1 = shards_seek_to_last();
+ //if we have 2 candidates, we need to select
+ if (main->valid()) {
+ if (shards_valid()) {
+ if (is_main_smaller()) {
+ smaller = on_shard;
+ main->next();
+ } else {
+ smaller = on_main;
+ shards_next();
+ }
+ } else {
+ smaller = on_main;
+ }
+ } else {
+ if (shards_valid()) {
+ smaller = on_shard;
+ } else {
+ smaller = on_main;
+ }
+ }
+ return r0 == 0 && r1 == 0 ? 0 : -1;
+ }
+
+ int seek_to_last(const std::string &prefix) override {
+ int r0 = main->seek_to_last(prefix);
+ int r1 = 0;
+ // find last shard that has some data
+ bool found = false;
+ current_shard = shards.lower_bound(prefix);
+ while (current_shard != shards.begin()) {
+ r1 = current_shard->second->seek_to_last();
+ if (r1 != 0)
+ break;
+ if (current_shard->second->valid()) {
+ found = true;
+ break;
+ }
+ }
+ //if we have 2 candidates, we need to select
+ if (main->valid() && found) {
+ if (is_main_smaller()) {
+ main->next();
+ } else {
+ shards_next();
+ }
+ }
+ if (!found) {
+ //set shards state that properly represents eof
+ current_shard = shards.end();
+ }
+ smaller = is_main_smaller() ? on_main : on_shard;
+ return r0 == 0 && r1 == 0 ? 0 : -1;
+ }
+
+ int upper_bound(const std::string &prefix, const std::string &after) override {
+ int r0 = main->upper_bound(prefix, after);
+ int r1 = 0;
+ if (r0 != 0)
+ return r0;
+ current_shard = shards.lower_bound(prefix);
+ if (current_shard != shards.end()) {
+ bool located = false;
+ if (current_shard->first == prefix) {
+ r1 = current_shard->second->upper_bound(after);
+ if (r1 != 0)
+ return r1;
+ if (current_shard->second->valid()) {
+ located = true;
+ }
+ }
+ if (!located) {
+ while (current_shard != shards.end()) {
+ r1 = current_shard->second->seek_to_first();
+ if (r1 != 0)
+ return r1;
+ if (current_shard->second->valid())
+ break;
+ ++current_shard;
+ }
+ }
+ }
+ smaller = is_main_smaller() ? on_main : on_shard;
+ return 0;
+ }
+
+ int lower_bound(const std::string &prefix, const std::string &to) override {
+ int r0 = main->lower_bound(prefix, to);
+ int r1 = 0;
+ if (r0 != 0)
+ return r0;
+ current_shard = shards.lower_bound(prefix);
+ if (current_shard != shards.end()) {
+ bool located = false;
+ if (current_shard->first == prefix) {
+ r1 = current_shard->second->lower_bound(to);
+ if (r1 != 0)
+ return r1;
+ if (current_shard->second->valid()) {
+ located = true;
+ }
+ }
+ if (!located) {
+ while (current_shard != shards.end()) {
+ r1 = current_shard->second->seek_to_first();
+ if (r1 != 0)
+ return r1;
+ if (current_shard->second->valid())
+ break;
+ ++current_shard;
+ }
+ }
+ }
+ smaller = is_main_smaller() ? on_main : on_shard;
+ return 0;
+ }
+
+ bool valid() override {
+ if (smaller == on_main) {
+ return main->valid();
+ } else {
+ if (current_shard == shards.end())
+ return false;
+ return current_shard->second->valid();
+ }
+ };
+
+ int next() override {
+ int r;
+ if (smaller == on_main) {
+ r = main->next();
+ } else {
+ r = shards_next();
+ }
+ if (r != 0)
+ return r;
+ smaller = is_main_smaller() ? on_main : on_shard;
+ return 0;
+ }
+
+ int prev() override {
+ int r;
+ bool main_was_valid = false;
+ if (main->valid()) {
+ main_was_valid = true;
+ r = main->prev();
+ } else {
+ r = main->seek_to_last();
+ }
+ if (r != 0)
+ return r;
+
+ bool shards_was_valid = false;
+ if (shards_valid()) {
+ shards_was_valid = true;
+ r = shards_prev();
+ } else {
+ r = shards_seek_to_last();
+ }
+ if (r != 0)
+ return r;
+
+ if (!main->valid() && !shards_valid()) {
+ //end, no previous. set marker so valid() can work
+ smaller = on_main;
+ return 0;
+ }
+
+ //if 1 is valid, select it
+ //if 2 are valid select larger and advance the other
+ if (main->valid()) {
+ if (shards_valid()) {
+ if (is_main_smaller()) {
+ smaller = on_shard;
+ if (main_was_valid) {
+ if (main->valid()) {
+ r = main->next();
+ } else {
+ r = main->seek_to_first();
+ }
+ } else {
+ //if we have resurrected main, kill it
+ if (main->valid()) {
+ main->next();
+ }
+ }
+ } else {
+ smaller = on_main;
+ if (shards_was_valid) {
+ if (shards_valid()) {
+ r = shards_next();
+ } else {
+ r = shards_seek_to_first();
+ }
+ } else {
+ //if we have resurected shards, kill it
+ if (shards_valid()) {
+ shards_next();
+ }
+ }
+ }
+ } else {
+ smaller = on_main;
+ r = shards_seek_to_first();
+ }
+ } else {
+ smaller = on_shard;
+ r = main->seek_to_first();
+ }
+ return r;
+ }
+
+ std::string key() override
+ {
+ if (smaller == on_main) {
+ return main->key();
+ } else {
+ return current_shard->second->key();
+ }
+ }
+
+ std::pair<std::string,std::string> raw_key() override
+ {
+ if (smaller == on_main) {
+ return main->raw_key();
+ } else {
+ return { current_shard->first, current_shard->second->key() };
+ }
+ }
+
+ bool raw_key_is_prefixed(const std::string &prefix) override
+ {
+ if (smaller == on_main) {
+ return main->raw_key_is_prefixed(prefix);
+ } else {
+ return current_shard->first == prefix;
+ }
+ }
+
+ ceph::buffer::list value() override
+ {
+ if (smaller == on_main) {
+ return main->value();
+ } else {
+ return current_shard->second->value();
+ }
+ }
+
+ int status() override
+ {
+ //because we already had to inspect key, it must be ok
+ return 0;
+ }
+
+ size_t key_size() override
+ {
+ if (smaller == on_main) {
+ return main->key_size();
+ } else {
+ return current_shard->second->key().size();
+ }
+ }
+ size_t value_size() override
+ {
+ if (smaller == on_main) {
+ return main->value_size();
+ } else {
+ return current_shard->second->value().length();
+ }
+ }
+
+ int shards_valid() {
+ if (current_shard == shards.end())
+ return false;
+ return current_shard->second->valid();
+ }
+
+ int shards_next() {
+ if (current_shard == shards.end()) {
+ //illegal to next() on !valid()
+ return -1;
+ }
+ int r = 0;
+ r = current_shard->second->next();
+ if (r != 0)
+ return r;
+ if (current_shard->second->valid())
+ return 0;
+ //current shard exhaused, search for key
+ ++current_shard;
+ while (current_shard != shards.end()) {
+ r = current_shard->second->seek_to_first();
+ if (r != 0)
+ return r;
+ if (current_shard->second->valid())
+ break;
+ ++current_shard;
+ }
+ //either we found key or not, but it is success
+ return 0;
+ }
+
+ int shards_prev() {
+ if (current_shard == shards.end()) {
+ //illegal to prev() on !valid()
+ return -1;
+ }
+ int r = current_shard->second->prev();
+ while (r == 0) {
+ if (current_shard->second->valid()) {
+ break;
+ }
+ if (current_shard == shards.begin()) {
+ //we have reached pre-first element
+ //this makes it !valid(), but guarantees next() moves to first element
+ break;
+ }
+ --current_shard;
+ r = current_shard->second->seek_to_last();
+ }
+ return r;
+ }
+
+ int shards_seek_to_last() {
+ int r = 0;
+ current_shard = shards.end();
+ if (current_shard == shards.begin()) {
+ //no shards at all
+ return 0;
+ }
+ while (current_shard != shards.begin()) {
+ --current_shard;
+ r = current_shard->second->seek_to_last();
+ if (r != 0)
+ return r;
+ if (current_shard->second->valid()) {
+ return 0;
+ }
+ }
+ //no keys at all
+ current_shard = shards.end();
+ return r;
+ }
+
+ int shards_seek_to_first() {
+ int r = 0;
+ current_shard = shards.begin();
+ while (current_shard != shards.end()) {
+ r = current_shard->second->seek_to_first();
+ if (r != 0)
+ break;
+ if (current_shard->second->valid()) {
+ //this is the first shard that will yield some keys
+ break;
+ }
+ ++current_shard;
+ }
+ return r;
+ }
+};
+
+class ShardMergeIteratorImpl : public KeyValueDB::IteratorImpl {
+private:
+ struct KeyLess {
+ private:
+ const rocksdb::Comparator* comparator;
+ public:
+ KeyLess(const rocksdb::Comparator* comparator) : comparator(comparator) { };
+
+ bool operator()(rocksdb::Iterator* a, rocksdb::Iterator* b) const
+ {
+ if (a->Valid()) {
+ if (b->Valid()) {
+ return comparator->Compare(a->key(), b->key()) < 0;
+ } else {
+ return true;
+ }
+ } else {
+ if (b->Valid()) {
+ return false;
+ } else {
+ return false;
+ }
+ }
+ }
+ };
+
+ const RocksDBStore* db;
+ KeyLess keyless;
+ string prefix;
+ std::vector<rocksdb::Iterator*> iters;
+public:
+ explicit ShardMergeIteratorImpl(const RocksDBStore* db,
+ const std::string& prefix,
+ const std::vector<rocksdb::ColumnFamilyHandle*>& shards)
+ : db(db), keyless(db->comparator), prefix(prefix)
+ {
+ iters.reserve(shards.size());
+ for (auto& s : shards) {
+ iters.push_back(db->db->NewIterator(rocksdb::ReadOptions(), s));
+ }
+ }
+ ~ShardMergeIteratorImpl() {
+ for (auto& it : iters) {
+ delete it;
+ }
+ }
+ int seek_to_first() override {
+ for (auto& it : iters) {
+ it->SeekToFirst();
+ if (!it->status().ok()) {
+ return -1;
+ }
+ }
+ //all iterators seeked, sort
+ std::sort(iters.begin(), iters.end(), keyless);
+ return 0;
+ }
+ int seek_to_last() override {
+ for (auto& it : iters) {
+ it->SeekToLast();
+ if (!it->status().ok()) {
+ return -1;
+ }
+ }
+ for (size_t i = 1; i < iters.size(); i++) {
+ if (iters[0]->Valid()) {
+ if (iters[i]->Valid()) {
+ if (keyless(iters[0], iters[i])) {
+ swap(iters[0], iters[i]);
+ }
+ } else {
+ //iters[i] empty
+ }
+ } else {
+ if (iters[i]->Valid()) {
+ swap(iters[0], iters[i]);
+ }
+ }
+ //it might happen that cf was empty
+ if (iters[i]->Valid()) {
+ iters[i]->Next();
+ }
+ }
+ //no need to sort, as at most 1 iterator is valid now
+ return 0;
+ }
+ int upper_bound(const string &after) override {
+ rocksdb::Slice slice_bound(after);
+ for (auto& it : iters) {
+ it->Seek(slice_bound);
+ if (it->Valid() && it->key() == after) {
+ it->Next();
+ }
+ if (!it->status().ok()) {
+ return -1;
+ }
+ }
+ std::sort(iters.begin(), iters.end(), keyless);
+ return 0;
+ }
+ int lower_bound(const string &to) override {
+ rocksdb::Slice slice_bound(to);
+ for (auto& it : iters) {
+ it->Seek(slice_bound);
+ if (!it->status().ok()) {
+ return -1;
+ }
+ }
+ std::sort(iters.begin(), iters.end(), keyless);
+ return 0;
+ }
+ int next() override {
+ int r = -1;
+ if (iters[0]->Valid()) {
+ iters[0]->Next();
+ if (iters[0]->status().ok()) {
+ r = 0;
+ //bubble up
+ for (size_t i = 0; i < iters.size() - 1; i++) {
+ if (keyless(iters[i], iters[i + 1])) {
+ //matches, fixed
+ break;
+ }
+ std::swap(iters[i], iters[i + 1]);
+ }
+ }
+ }
+ return r;
+ }
+ // iters are sorted, so
+ // a[0] < b[0] < c[0] < d[0]
+ // a[0] > a[-1], a[0] > b[-1], a[0] > c[-1], a[0] > d[-1]
+ // so, prev() will be one of:
+ // a[-1], b[-1], c[-1], d[-1]
+ // prev() will be the one that is *largest* of them
+ //
+ // alg:
+ // 1. go prev() on each iterator we can
+ // 2. select largest key from those iterators
+ // 3. go next() on all iterators except (2)
+ // 4. sort
+ int prev() override {
+ std::vector<rocksdb::Iterator*> prev_done;
+ //1
+ for (auto it: iters) {
+ if (it->Valid()) {
+ it->Prev();
+ if (it->Valid()) {
+ prev_done.push_back(it);
+ } else {
+ it->SeekToFirst();
+ }
+ } else {
+ it->SeekToLast();
+ if (it->Valid()) {
+ prev_done.push_back(it);
+ }
+ }
+ }
+ if (prev_done.size() == 0) {
+ /* there is no previous element */
+ if (iters[0]->Valid()) {
+ iters[0]->Prev();
+ ceph_assert(!iters[0]->Valid());
+ }
+ return 0;
+ }
+ //2,3
+ rocksdb::Iterator* highest = prev_done[0];
+ for (size_t i = 1; i < prev_done.size(); i++) {
+ if (keyless(highest, prev_done[i])) {
+ highest->Next();
+ highest = prev_done[i];
+ } else {
+ prev_done[i]->Next();
+ }
+ }
+ //4
+ //insert highest in the beginning, and shift values until we pick highest
+ //untouched rest is sorted - we just prev()/next() them
+ rocksdb::Iterator* hold = highest;
+ for (size_t i = 0; i < iters.size(); i++) {
+ std::swap(hold, iters[i]);
+ if (hold == highest) break;
+ }
+ ceph_assert(hold == highest);
+ return 0;
+ }
+ bool valid() override {
+ return iters[0]->Valid();
+ }
+ string key() override {
+ return iters[0]->key().ToString();
+ }
+ std::pair<std::string, std::string> raw_key() override {
+ return make_pair(prefix, key());
+ }
+ bufferlist value() override {
+ return to_bufferlist(iters[0]->value());
+ }
+ bufferptr value_as_ptr() override {
+ rocksdb::Slice val = iters[0]->value();
+ return bufferptr(val.data(), val.size());
+ }
+ int status() override {
+ return iters[0]->status().ok() ? 0 : -1;
+ }
+};
+
+KeyValueDB::Iterator RocksDBStore::get_iterator(const std::string& prefix, IteratorOpts opts)
+{
+ auto cf_it = cf_handles.find(prefix);
+ if (cf_it != cf_handles.end()) {
+ if (cf_it->second.handles.size() == 1) {
+ return std::make_shared<CFIteratorImpl>(
+ prefix,
+ db->NewIterator(rocksdb::ReadOptions(), cf_it->second.handles[0]));
+ } else {
+ return std::make_shared<ShardMergeIteratorImpl>(
+ this,
+ prefix,
+ cf_it->second.handles);
+ }
+ } else {
+ return KeyValueDB::get_iterator(prefix, opts);
+ }
+}
+
+rocksdb::Iterator* RocksDBStore::new_shard_iterator(rocksdb::ColumnFamilyHandle* cf)
+{
+ return db->NewIterator(rocksdb::ReadOptions(), cf);
+}
+
+RocksDBStore::WholeSpaceIterator RocksDBStore::get_wholespace_iterator(IteratorOpts opts)
{
- rocksdb::ColumnFamilyHandle *cf_handle =
- static_cast<rocksdb::ColumnFamilyHandle*>(get_cf_handle(prefix));
- if (cf_handle) {
- return std::make_shared<CFIteratorImpl>(
- prefix,
- db->NewIterator(rocksdb::ReadOptions(), cf_handle));
+ if (cf_handles.size() == 0) {
+ rocksdb::ReadOptions opt = rocksdb::ReadOptions();
+ if (opts & ITERATOR_NOCACHE)
+ opt.fill_cache=false;
+ return std::make_shared<RocksDBWholeSpaceIteratorImpl>(
+ db->NewIterator(opt, default_cf));
} else {
- return KeyValueDB::get_iterator(prefix);
+ return std::make_shared<WholeMergeIteratorImpl>(this);
+ }
+}
+
+RocksDBStore::WholeSpaceIterator RocksDBStore::get_default_cf_iterator()
+{
+ return std::make_shared<RocksDBWholeSpaceIteratorImpl>(
+ db->NewIterator(rocksdb::ReadOptions(), default_cf));
+}
+
+int RocksDBStore::prepare_for_reshard(const std::string& new_sharding,
+ RocksDBStore::columns_t& to_process_columns)
+{
+ //0. lock db from opening
+ //1. list existing columns
+ //2. apply merge operator to (main + columns) opts
+ //3. prepare std::vector<rocksdb::ColumnFamilyDescriptor> existing_cfs
+ //4. open db, acquire existing column handles
+ //5. calculate missing columns
+ //6. create missing columns
+ //7. construct cf_handles according to new sharding
+ //8. check is all cf_handles are filled
+
+ bool b;
+ std::vector<ColumnFamily> new_sharding_def;
+ char const* error_position;
+ std::string error_msg;
+ b = parse_sharding_def(new_sharding, new_sharding_def, &error_position, &error_msg);
+ if (!b) {
+ dout(1) << __func__ << " bad sharding: " << dendl;
+ dout(1) << __func__ << new_sharding << dendl;
+ dout(1) << __func__ << std::string(error_position - &new_sharding[0], ' ') << "^" << error_msg << dendl;
+ return -EINVAL;
+ }
+
+ //0. lock db from opening
+ std::string stored_sharding_text;
+ rocksdb::ReadFileToString(env,
+ sharding_def_file,
+ &stored_sharding_text);
+ if (stored_sharding_text.find(resharding_column_lock) == string::npos) {
+ rocksdb::Status status;
+ if (stored_sharding_text.size() != 0)
+ stored_sharding_text += " ";
+ stored_sharding_text += resharding_column_lock;
+ env->CreateDir(sharding_def_dir);
+ status = rocksdb::WriteStringToFile(env, stored_sharding_text,
+ sharding_def_file, true);
+ if (!status.ok()) {
+ derr << __func__ << " cannot write to " << sharding_def_file << dendl;
+ return -EIO;
+ }
+ }
+
+ //1. list existing columns
+
+ rocksdb::Status status;
+ std::vector<std::string> existing_columns;
+ rocksdb::Options opt;
+ int r = load_rocksdb_options(false, opt);
+ if (r) {
+ dout(1) << __func__ << " load rocksdb options failed" << dendl;
+ return r;
}
+ status = rocksdb::DB::ListColumnFamilies(rocksdb::DBOptions(opt), path, &existing_columns);
+ if (!status.ok()) {
+ derr << "Unable to list column families: " << status.ToString() << dendl;
+ return -EINVAL;
+ }
+ dout(5) << "existing columns = " << existing_columns << dendl;
+
+ //2. apply merge operator to (main + columns) opts
+ //3. prepare std::vector<rocksdb::ColumnFamilyDescriptor> cfs_to_open
+
+ std::vector<rocksdb::ColumnFamilyDescriptor> cfs_to_open;
+ for (const auto& full_name : existing_columns) {
+ //split col_name to <prefix>-<number>
+ std::string base_name;
+ size_t pos = full_name.find('-');
+ if (std::string::npos == pos)
+ base_name = full_name;
+ else
+ base_name = full_name.substr(0,pos);
+
+ rocksdb::ColumnFamilyOptions cf_opt(opt);
+ // search if we have options for this column
+ std::string options;
+ for (const auto& nsd : new_sharding_def) {
+ if (nsd.name == base_name) {
+ options = nsd.options;
+ break;
+ }
+ }
+ status = rocksdb::GetColumnFamilyOptionsFromString(cf_opt, options, &cf_opt);
+ if (!status.ok()) {
+ derr << __func__ << " failure parsing column options: " << options << dendl;
+ return -EINVAL;
+ }
+ if (base_name != rocksdb::kDefaultColumnFamilyName)
+ install_cf_mergeop(base_name, &cf_opt);
+ cfs_to_open.emplace_back(full_name, cf_opt);
+ }
+
+ //4. open db, acquire existing column handles
+ std::vector<rocksdb::ColumnFamilyHandle*> handles;
+ status = rocksdb::DB::Open(rocksdb::DBOptions(opt),
+ path, cfs_to_open, &handles, &db);
+ if (!status.ok()) {
+ derr << status.ToString() << dendl;
+ return -EINVAL;
+ }
+ for (size_t i = 0; i < cfs_to_open.size(); i++) {
+ dout(10) << "column " << cfs_to_open[i].name << " handle " << (void*)handles[i] << dendl;
+ }
+
+ //5. calculate missing columns
+ std::vector<std::string> new_sharding_columns;
+ std::vector<std::string> missing_columns;
+ sharding_def_to_columns(new_sharding_def,
+ new_sharding_columns);
+ dout(5) << "target columns = " << new_sharding_columns << dendl;
+ for (const auto& n : new_sharding_columns) {
+ bool found = false;
+ for (const auto& e : existing_columns) {
+ if (n == e) {
+ found = true;
+ break;
+ }
+ }
+ if (!found) {
+ missing_columns.push_back(n);
+ }
+ }
+ dout(5) << "missing columns = " << missing_columns << dendl;
+
+ //6. create missing columns
+ for (const auto& full_name : missing_columns) {
+ std::string base_name;
+ size_t pos = full_name.find('-');
+ if (std::string::npos == pos)
+ base_name = full_name;
+ else
+ base_name = full_name.substr(0,pos);
+
+ rocksdb::ColumnFamilyOptions cf_opt(opt);
+ // search if we have options for this column
+ std::string options;
+ for (const auto& nsd : new_sharding_def) {
+ if (nsd.name == base_name) {
+ options = nsd.options;
+ break;
+ }
+ }
+ status = rocksdb::GetColumnFamilyOptionsFromString(cf_opt, options, &cf_opt);
+ if (!status.ok()) {
+ derr << __func__ << " failure parsing column options: " << options << dendl;
+ return -EINVAL;
+ }
+ install_cf_mergeop(base_name, &cf_opt);
+ rocksdb::ColumnFamilyHandle *cf;
+ status = db->CreateColumnFamily(cf_opt, full_name, &cf);
+ if (!status.ok()) {
+ derr << __func__ << " Failed to create rocksdb column family: "
+ << full_name << dendl;
+ return -EINVAL;
+ }
+ dout(10) << "created column " << full_name << " handle = " << (void*)cf << dendl;
+ existing_columns.push_back(full_name);
+ handles.push_back(cf);
+ }
+
+ //7. construct cf_handles according to new sharding
+ for (size_t i = 0; i < existing_columns.size(); i++) {
+ std::string full_name = existing_columns[i];
+ rocksdb::ColumnFamilyHandle *cf = handles[i];
+ std::string base_name;
+ size_t shard_idx = 0;
+ size_t pos = full_name.find('-');
+ dout(10) << "processing column " << full_name << dendl;
+ if (std::string::npos == pos) {
+ base_name = full_name;
+ } else {
+ base_name = full_name.substr(0,pos);
+ shard_idx = atoi(full_name.substr(pos+1).c_str());
+ }
+ if (rocksdb::kDefaultColumnFamilyName == base_name) {
+ default_cf = handles[i];
+ must_close_default_cf = true;
+ std::unique_ptr<rocksdb::ColumnFamilyHandle, cf_deleter_t> ptr{
+ cf, [](rocksdb::ColumnFamilyHandle*) {}};
+ to_process_columns.emplace(full_name, std::move(ptr));
+ } else {
+ for (const auto& nsd : new_sharding_def) {
+ if (nsd.name == base_name) {
+ if (shard_idx < nsd.shard_cnt) {
+ add_column_family(base_name, nsd.hash_l, nsd.hash_h, shard_idx, cf);
+ } else {
+ //ignore columns with index larger then shard count
+ }
+ break;
+ }
+ }
+ std::unique_ptr<rocksdb::ColumnFamilyHandle, cf_deleter_t> ptr{
+ cf, [this](rocksdb::ColumnFamilyHandle* handle) {
+ db->DestroyColumnFamilyHandle(handle);
+ }};
+ to_process_columns.emplace(full_name, std::move(ptr));
+ }
+ }
+
+ //8. check if all cf_handles are filled
+ for (const auto& col : cf_handles) {
+ for (size_t i = 0; i < col.second.handles.size(); i++) {
+ if (col.second.handles[i] == nullptr) {
+ derr << "missing handle for column " << col.first << " shard " << i << dendl;
+ return -EIO;
+ }
+ }
+ }
+ return 0;
+}
+
+int RocksDBStore::reshard_cleanup(const RocksDBStore::columns_t& current_columns)
+{
+ std::vector<std::string> new_sharding_columns;
+ for (const auto& [name, handle] : cf_handles) {
+ if (handle.handles.size() == 1) {
+ new_sharding_columns.push_back(name);
+ } else {
+ for (size_t i = 0; i < handle.handles.size(); i++) {
+ new_sharding_columns.push_back(name + "-" + to_string(i));
+ }
+ }
+ }
+
+ for (auto& [name, handle] : current_columns) {
+ auto found = std::find(new_sharding_columns.begin(),
+ new_sharding_columns.end(),
+ name) != new_sharding_columns.end();
+ if (found || name == rocksdb::kDefaultColumnFamilyName) {
+ dout(5) << "Column " << name << " is part of new sharding." << dendl;
+ continue;
+ }
+ dout(5) << "Column " << name << " not part of new sharding. Deleting." << dendl;
+
+ // verify that column is empty
+ std::unique_ptr<rocksdb::Iterator> it{
+ db->NewIterator(rocksdb::ReadOptions(), handle.get())};
+ ceph_assert(it);
+ it->SeekToFirst();
+ ceph_assert(!it->Valid());
+
+ if (rocksdb::Status status = db->DropColumnFamily(handle.get()); !status.ok()) {
+ derr << __func__ << " Failed to drop column: " << name << dendl;
+ return -EINVAL;
+ }
+ }
+ return 0;
+}
+
+int RocksDBStore::reshard(const std::string& new_sharding, const RocksDBStore::resharding_ctrl* ctrl_in)
+{
+
+ resharding_ctrl ctrl = ctrl_in ? *ctrl_in : resharding_ctrl();
+ size_t bytes_in_batch = 0;
+ size_t keys_in_batch = 0;
+ size_t bytes_per_iterator = 0;
+ size_t keys_per_iterator = 0;
+ size_t keys_processed = 0;
+ size_t keys_moved = 0;
+
+ auto flush_batch = [&](rocksdb::WriteBatch* batch) {
+ dout(10) << "flushing batch, " << keys_in_batch << " keys, for "
+ << bytes_in_batch << " bytes" << dendl;
+ rocksdb::WriteOptions woptions;
+ woptions.sync = true;
+ rocksdb::Status s = db->Write(woptions, batch);
+ ceph_assert(s.ok());
+ bytes_in_batch = 0;
+ keys_in_batch = 0;
+ batch->Clear();
+ };
+
+ auto process_column = [&](rocksdb::ColumnFamilyHandle* handle,
+ const std::string& fixed_prefix)
+ {
+ dout(5) << " column=" << (void*)handle << " prefix=" << fixed_prefix << dendl;
+ std::unique_ptr<rocksdb::Iterator> it{
+ db->NewIterator(rocksdb::ReadOptions(), handle)};
+ ceph_assert(it);
+
+ rocksdb::WriteBatch bat;
+ for (it->SeekToFirst(); it->Valid(); it->Next()) {
+ rocksdb::Slice raw_key = it->key();
+ dout(30) << "key=" << pretty_binary_string(raw_key.ToString()) << dendl;
+ //check if need to refresh iterator
+ if (bytes_per_iterator >= ctrl.bytes_per_iterator ||
+ keys_per_iterator >= ctrl.keys_per_iterator) {
+ dout(8) << "refreshing iterator" << dendl;
+ bytes_per_iterator = 0;
+ keys_per_iterator = 0;
+ std::string raw_key_str = raw_key.ToString();
+ it.reset(db->NewIterator(rocksdb::ReadOptions(), handle));
+ ceph_assert(it);
+ it->Seek(raw_key_str);
+ ceph_assert(it->Valid());
+ raw_key = it->key();
+ }
+ rocksdb::Slice value = it->value();
+ std::string prefix, key;
+ if (fixed_prefix.size() == 0) {
+ split_key(raw_key, &prefix, &key);
+ } else {
+ prefix = fixed_prefix;
+ key = raw_key.ToString();
+ }
+ keys_processed++;
+ if ((keys_processed % 10000) == 0) {
+ dout(10) << "processed " << keys_processed << " keys, moved " << keys_moved << dendl;
+ }
+ rocksdb::ColumnFamilyHandle* new_handle = get_cf_handle(prefix, key);
+ if (new_handle == nullptr) {
+ new_handle = default_cf;
+ }
+ if (handle == new_handle) {
+ continue;
+ }
+ std::string new_raw_key;
+ if (new_handle == default_cf) {
+ new_raw_key = combine_strings(prefix, key);
+ } else {
+ new_raw_key = key;
+ }
+ bat.Delete(handle, raw_key);
+ bat.Put(new_handle, new_raw_key, value);
+ dout(25) << "moving " << (void*)handle << "/" << pretty_binary_string(raw_key.ToString()) <<
+ " to " << (void*)new_handle << "/" << pretty_binary_string(new_raw_key) <<
+ " size " << value.size() << dendl;
+ keys_moved++;
+ bytes_in_batch += new_raw_key.size() * 2 + value.size();
+ keys_in_batch++;
+ bytes_per_iterator += new_raw_key.size() * 2 + value.size();
+ keys_per_iterator++;
+
+ //check if need to write batch
+ if (bytes_in_batch >= ctrl.bytes_per_batch ||
+ keys_in_batch >= ctrl.keys_per_batch) {
+ flush_batch(&bat);
+ if (ctrl.unittest_fail_after_first_batch) {
+ return -1000;
+ }
+ }
+ }
+ if (bat.Count() > 0) {
+ flush_batch(&bat);
+ }
+ return 0;
+ };
+
+ auto close_column_handles = make_scope_guard([this] {
+ cf_handles.clear();
+ close();
+ });
+ columns_t to_process_columns;
+ int r = prepare_for_reshard(new_sharding, to_process_columns);
+ if (r != 0) {
+ dout(1) << "failed to prepare db for reshard" << dendl;
+ return r;
+ }
+
+ for (auto& [name, handle] : to_process_columns) {
+ dout(5) << "Processing column=" << name
+ << " handle=" << handle.get() << dendl;
+ if (name == rocksdb::kDefaultColumnFamilyName) {
+ ceph_assert(handle.get() == default_cf);
+ r = process_column(default_cf, std::string());
+ } else {
+ std::string fixed_prefix = name.substr(0, name.find('-'));
+ dout(10) << "Prefix: " << fixed_prefix << dendl;
+ r = process_column(handle.get(), fixed_prefix);
+ }
+ if (r != 0) {
+ derr << "Error processing column " << name << dendl;
+ return r;
+ }
+ if (ctrl.unittest_fail_after_processing_column) {
+ return -1001;
+ }
+ }
+
+ r = reshard_cleanup(to_process_columns);
+ if (r != 0) {
+ dout(5) << "failed to cleanup after reshard" << dendl;
+ return r;
+ }
+
+ if (ctrl.unittest_fail_after_successful_processing) {
+ return -1002;
+ }
+ env->CreateDir(sharding_def_dir);
+ if (auto status = rocksdb::WriteStringToFile(env, new_sharding,
+ sharding_def_file, true);
+ !status.ok()) {
+ derr << __func__ << " cannot write to " << sharding_def_file << dendl;
+ return -EIO;
+ }
+
+ return r;
+}
+
+bool RocksDBStore::get_sharding(std::string& sharding) {
+ rocksdb::Status status;
+ std::string stored_sharding_text;
+ bool result = false;
+ sharding.clear();
+
+ status = env->FileExists(sharding_def_file);
+ if (status.ok()) {
+ status = rocksdb::ReadFileToString(env,
+ sharding_def_file,
+ &stored_sharding_text);
+ if(status.ok()) {
+ result = true;
+ sharding = stored_sharding_text;
+ }
+ }
+ return result;
}
projects / ceph.git / blobdiff