## Pair Generator
+`pair_generator()` reads a range through a transaction supplied by the caller.
+Use it when the query is expected to fit within one transaction and/or you want
+control over the transaction's lifetime and options.
+
```cpp
-/* Use pair_generator() for ordinary range scans where processing one key/value
- * pair at a time is natural. */
-std::map<std::string, std::string> people;
+auto txn = lfdb::make_transaction(dbh);
+
+for (const auto& [key, value] : lfdb::pair_generator(txn, lfdb::select { "person/" })) {
+ fmt::println("{}: {}", key, value);
+}
+```
+
+To get results in reverse order, set the reverse_order property in the selector:
+
+```cpp
+auto people = lfdb::select { "person/" };
+people.options.reverse_order = true;
+auto txn = lfdb::make_transaction(dbh);
+
+for (const auto& [key, value] : lfdb::pair_generator(txn, people)) {
+ /* process results from high keys to low keys */
+}
+```
+
+It may be useful to group pair_generator()'s output into discrete groups of N items. One way to do that is
+with a chunk_view:
+
+```cpp
+// Stream groups of 100:
+auto txn = lfdb::make_transaction(dbh);
+auto keys = lfdb::pair_generator(txn, lfdb::select { "key_" });
-std::ranges::copy(lfdb::pair_generator(dbh, lfdb::select { "person/" }),
- std::inserter(people, std::end(people)));
+for (const auto& chunk : keys | std::views::chunk(100)) {
+ for (const auto& [key, value] : chunk) {
+ // ...
+ }
+}
```
To get results in reverse order, set the reverse_order property in the selector:
## Block Generator
-For very large prefix scans, use the same one-argument selector with
-`block_generator()`. This lets libfdb plan and read the range in blocks while
-the call site still names the logical key family directly.
+`block_generator()` is useful for reads that may become very large. Given a database
+handle, it internally manages transactions for each planned block/window. Use it for very
+large scans where a single transaction may get too old or where block-at-a-time
+processing is preferable.
```cpp
/* Use block_generator() for large range scans where split planning and
* block-at-a-time processing are useful. */
for (auto&& block : lfdb::block_generator(dbh, lfdb::select { "object/metadata/" })) {
- /* process block */
+ for (const auto& [key, value] : block) {
+ fmt::println("{}: {}", key, value);
+ }
}
```
std::span<const std::uint8_t> k,
std::span<const std::uint8_t> v);
-inline future_value await_ready_key_range_future(transaction_handle txn, const ceph::libfdb::select&, int);
-inline future_value await_ready_keyvalue_range_future(transaction_handle txn, const ceph::libfdb::select& key_range, int& iteration);
+inline future_value block_until_ready(future_value&& fv);
+inline fdb_error_t get_future_error(const future_value& fv);
+inline future_value wait_for_on_error(FDBTransaction* txn, fdb_error_t original_error);
+inline future_value get_range_future_from_transaction(ceph::libfdb::transaction& txn, const ceph::libfdb::select& selection, int iteration);
// A generator that produces successive spans for a range:
inline std::generator<std::span<const FDBKeyValue>> generate_FDB_pairs(ceph::libfdb::transaction& txn, ceph::libfdb::select key_range);
friend class ceph::libfdb::transaction;
};
-constexpr auto as_fdb_span(const char *s)
-{
- return std::span<const std::uint8_t>((const std::uint8_t *)s, std::strlen(s));
-}
-
-constexpr auto as_fdb_span(std::string_view sv)
+inline auto as_fdb_span(std::string_view sv)
{
return std::span<const std::uint8_t>((const std::uint8_t *)sv.data(), sv.size());
}
} // namespace ceph::libfdb::detail
+struct range_endpoint final
+{
+ std::string key;
+ bool inclusive;
+
+ explicit range_endpoint(std::string_view key_, bool inclusive_)
+ : key(key_), inclusive(inclusive_)
+ {}
+};
+
+inline range_endpoint inclusive(std::string_view key)
+{
+ return range_endpoint(key, true);
+}
+
+inline range_endpoint exclusive(std::string_view key)
+{
+ return range_endpoint(key, false);
+}
+
/* lfdb::select is for handling batch queries-- there are two constructors to consider:
- single-key select creates the range containing all keys with that prefix;
- - passing a start and end key creates a half-open lexicographic key range: [begin_key, end_key). */
+ - passing a start and end key creates a half-open lexicographic key range: [begin_key, end_key);
+ - wrap endpoints with inclusive() or exclusive() when you need a different shape. */
struct select final
{
std::string begin_key, end_key;
+ bool begin_inclusive = true;
+ bool end_inclusive = false;
- mutable struct {
+ struct {
int stride = 0; // "unlimited"
+ int target_bytes = 0; // "unlimited"
bool reverse_order = false; // should we return results in reverse order?
} options;
public:
+ select(range_endpoint begin, range_endpoint end)
+ : begin_key(std::move(begin.key)),
+ end_key(std::move(end.key)),
+ begin_inclusive(begin.inclusive),
+ end_inclusive(end.inclusive)
+ {}
+
select(std::string_view begin_key_, std::string_view end_key_)
- : begin_key(begin_key_), end_key(end_key_)
+ : select(inclusive(begin_key_), exclusive(end_key_))
+ {}
+
+ select(std::string_view begin_key_, range_endpoint end)
+ : select(inclusive(begin_key_), std::move(end))
+ {}
+
+ select(range_endpoint begin, std::string_view end_key_)
+ : select(std::move(begin), exclusive(end_key_))
{}
select(std::string_view prefix)
};
-// Flag-only options are indicated with an explicit option_flag, as they have no actual value:
+// Helpers for selectors:
+namespace detail {
+
+inline std::string key_after(std::string_view key)
+{
+ return std::string(key) + '\0';
+}
+
+inline select as_half_open_select(const select& selection)
+{
+ auto half_open = select {
+ selection.begin_inclusive ? selection.begin_key : key_after(selection.begin_key),
+ selection.end_inclusive ? key_after(selection.end_key) : selection.end_key
+ };
+
+ half_open.options = selection.options;
+
+ return half_open;
+}
+
+} // namespace detail
+
+// Flag-only options are indicated with an explicit option_flag, as they have
+// no actual value:
struct option_flag_t final {};
inline constexpr option_flag_t option_flag;
namespace detail {
// Note that these are specific to FDB's needs (see return type, casts):
-constexpr const std::uint8_t *data_of(const std::vector<std::uint8_t>& xs) { return (const std::uint8_t *)xs.data(); }
-constexpr const std::uint8_t *data_of(const std::string& xs) { return (const std::uint8_t *)xs.data(); }
-constexpr const std::uint8_t *data_of(const std::int64_t& x) { return reinterpret_cast<const std::uint8_t *>(&x); }
-constexpr const std::uint8_t *data_of(const option_flag_t&) { return nullptr; }
+inline const std::uint8_t *data_of(const std::vector<std::uint8_t>& xs) { return (const std::uint8_t *)xs.data(); }
+inline const std::uint8_t *data_of(const std::string& xs) { return (const std::uint8_t *)xs.data(); }
+inline const std::uint8_t *data_of(const std::int64_t& x) { return reinterpret_cast<const std::uint8_t *>(&x); }
+inline const std::uint8_t *data_of(const option_flag_t&) { return nullptr; }
// Note that these are specific to FDB's needs (see return type, casts):
constexpr int size_of(const std::vector<std::uint8_t>& xs) { return static_cast<int>(xs.size()); }
constexpr int size_of(const std::string& xs) { return static_cast<int>(xs.size()); }
-constexpr int size_of(const std::int64_t) { return 8; } // 8b = size required by FDB
+constexpr int size_of(const std::int64_t) { return sizeof(std::int64_t); }
constexpr int size_of(const option_flag_t&) { return 0; }
// ...also used:
-constexpr std::uint8_t *data_of(std::string_view xs) { return (std::uint8_t *)xs.data(); }
+inline const std::uint8_t *data_of(std::string_view xs) { return (const std::uint8_t *)xs.data(); }
constexpr int size_of(std::string_view xs) { return static_cast<int>(xs.size()); }
inline auto as_fdb_option_args(const auto& option)
// The global DB state and management thread:
// JFW: more user hooks that go into FDB system possible here
-class database_system final
+namespace database_system
{
- database_system() = delete;
-
- private:
- static inline bool was_initialized = false;
+inline bool was_initialized = false;
+inline bool was_shutdown = false;
- static inline std::once_flag fdb_was_initialized;
- static inline std::jthread fdb_network_thread;
+inline std::once_flag fdb_was_initialized;
+inline std::jthread fdb_network_thread;
- static inline void initialize_fdb(const network_options& options)
- {
- // This must be called before ANY other API function:
- if (fdb_error_t r = fdb_select_api_version(FDB_API_VERSION); 0 != r)
- throw libfdb_exception(r);
+inline void initialize_fdb(const network_options& options)
+{
+ // This must be called before ANY other API function-- the structure
+ // of fdb_database_system accomplishes this:
+ if (fdb_error_t r = fdb_select_api_version(FDB_API_VERSION); 0 != r)
+ throw libfdb_exception(r);
- // Zero or more calls to this may now be made:
- apply_options(options, fdb_network_set_option);
+ // Zero or more calls to this may now be made:
+ apply_options(options, fdb_network_set_option);
- // This must be called before any other API function (besides >= 0 calls to fdb_network_set_option()):
- if (fdb_error_t r = fdb_setup_network(); 0 != r)
- throw libfdb_exception(r);
+ // This must be called before any other API function (besides >= 0 calls to fdb_network_set_option()):
+ if (fdb_error_t r = fdb_setup_network(); 0 != r)
+ throw libfdb_exception(r);
- // Launch network thread:
- fdb_network_thread = std::jthread { &fdb_run_network };
+ // Launch network thread:
+ fdb_network_thread = std::jthread { &fdb_run_network };
- // Okie-dokie, we're all set (distinct from fdb_was_initialized):
- was_initialized = true;
- }
-
- private:
- database_system(const network_options& network_opts)
- {
- std::call_once(ceph::libfdb::detail::database_system::fdb_was_initialized,
- ceph::libfdb::detail::database_system::initialize_fdb,
- network_opts);
- }
+ // Okie-dokie, we're all set (distinct from fdb_was_initialized):
+ was_initialized = true;
+ was_shutdown = false;
+}
- public:
- static bool& initialized() noexcept { return was_initialized; }
+inline bool initialized() noexcept { return was_initialized; }
- public:
- static inline void shutdown_fdb()
- {
- using namespace std::chrono_literals;
+inline void shutdown_fdb()
+{
+ using namespace std::chrono_literals;
- if (not initialized()) {
- return;
+ if (not initialized() or was_shutdown) {
+ return;
+ }
+
+ // shut down network and database:
+ if (int r = fdb_stop_network(); 0 != r)
+ {
+ // In this case, we likely don't want to throw from our dtor, but we may
+ // have something to log. As there's no higher-level hook for that, we
+ // have nothing to do right now.
+ // fmt::println("database::shutdown_fdb() error {}", r);
}
- // shut down network and database:
- if (int r = fdb_stop_network(); 0 != r)
- {
- // In this case, we likely don't want to throw from our dtor, but we may
- // have something to log. As there's no higher-level hook for that, we
- // have nothing to do right now.
- // fmt::println("database::shutdown_fdb() error {}", r);
- }
-
- // This may not actually be needed, but it's a traditional courtesy:
- std::this_thread::sleep_for(7ms);
+ // This may not actually be needed, but it's a traditional courtesy:
+ std::this_thread::sleep_for(7ms);
- if (fdb_network_thread.joinable()) {
- fdb_network_thread.join();
- }
+ if (fdb_network_thread.joinable()) {
+ fdb_network_thread.join();
}
- private:
- friend struct ceph::libfdb::database;
-};
+ was_shutdown = true;
+ was_initialized = false;
+}
+
+} // namespace database_system
} // namespace ceph::libfdb::detail
class database final
{
- detail::database_system db_system;
-
private:
- std::shared_ptr<FDBDatabase> db_handle;
+ struct database_deleter final
+ {
+ void operator()(FDBDatabase *db) const noexcept
+ {
+ fdb_database_destroy(db);
+ }
+ };
+
+ std::unique_ptr<FDBDatabase, database_deleter> db_handle;
+
+ static FDBDatabase *create_database_ptr(const std::filesystem::path& cluster_file_path,
+ const network_options& network_opts) {
+ std::call_once(detail::database_system::fdb_was_initialized,
+ detail::database_system::initialize_fdb,
+ network_opts);
+
+ if (detail::database_system::was_shutdown) {
+ throw libfdb_exception("FoundationDB already shut down");
+ }
- FDBDatabase *create_database_ptr(const std::filesystem::path cluster_file_path) {
FDBDatabase *fdbp = nullptr;
if (fdb_error_t r = fdb_create_database(cluster_file_path.c_str(), &fdbp); 0 != r) {
public:
database(const std::filesystem::path cluster_file_path, const ceph::libfdb::database_options& db_opts, const network_options& network_opts)
- : db_system(network_opts),
- db_handle(create_database_ptr(cluster_file_path), &fdb_database_destroy)
+ : db_handle(create_database_ptr(cluster_file_path, network_opts))
{
detail::apply_options(db_opts,
[handle = raw_handle()](auto option_code, auto data, auto size) {
std::unique_ptr<FDBTransaction, decltype(&fdb_transaction_destroy)> txn_handle;
private:
- bool get_single_value_from_transaction(const std::span<const std::uint8_t>& key, std::invocable<std::span<const std::uint8_t>> auto&& write_output);
+ bool get_single_value_from_transaction(const std::span<const std::uint8_t>& key,
+ std::invocable<std::span<const std::uint8_t>> auto&& write_output_fn);
public:
transaction(database_handle dbh_)
: transaction(dbh_)
{
detail::apply_options(opts,
- [handle = raw_handle()](auto code, auto data, auto size) {
- return fdb_transaction_set_option(handle, code, data, size);
+ [handle = raw_handle()](auto opt, auto val, auto sz) {
+ return fdb_transaction_set_option(handle, opt, val, sz);
});
}
}
void erase(const ceph::libfdb::concepts::selector auto& key_range) {
+ const auto half_open_range = detail::as_half_open_select(key_range);
+
fdb_transaction_clear_range(raw_handle(),
- (const uint8_t *)key_range.begin_key.data(), key_range.begin_key.size(),
- (const uint8_t *)key_range.end_key.data(), key_range.end_key.size());
+ (const uint8_t *)half_open_range.begin_key.data(), half_open_range.begin_key.size(),
+ (const uint8_t *)half_open_range.end_key.data(), half_open_range.end_key.size());
}
bool key_exists(std::string_view k) {
return get_single_value_from_transaction(detail::as_fdb_span(k), [](auto) {});
}
- bool commit(fdb_error_t *replay_error = nullptr);
+ bool commit();
+ bool commit(fdb_error_t *replay_error);
void destroy() noexcept { txn_handle.reset(); }
// As you can see, friends are used extensively to implement the public interface; it would be nice to come up
{
const fdb_bool_t is_snapshot = false;
- detail::future_value fv(fdb_transaction_get(
- raw_handle(),
- (const uint8_t *)key.data(), key.size(),
- is_snapshot));
-
- if (fdb_error_t r = fdb_future_block_until_ready(fv.raw_handle()); 0 != r) {
- throw libfdb_exception(r);
- }
+ auto fv = detail::block_until_ready(detail::future_value(fdb_transaction_get(
+ raw_handle(),
+ (const uint8_t *)key.data(), key.size(),
+ is_snapshot)));
fdb_bool_t key_was_found = false;
const uint8_t *out_buffer = nullptr;
return true;
}
+[[nodiscard]] inline bool ceph::libfdb::transaction::commit()
+{
+ return commit(nullptr);
+}
+
[[nodiscard]] inline bool ceph::libfdb::transaction::commit(fdb_error_t *replay_error)
{
if (nullptr != replay_error) {
}
if (fdb_error_t r = fdb_future_get_error(fv.raw_handle()); 0 != r) {
- detail::future_value ferror_result(fdb_transaction_on_error(raw_handle(), r));
+ detail::future_value ferror_result = detail::wait_for_on_error(raw_handle(), r);
- if (0 != fdb_future_block_until_ready(ferror_result.raw_handle())) {
- // In their example, they use the first error as the message source, so I will do that also:
- throw libfdb_exception(r);
- }
-
- if (0 != fdb_future_get_error(ferror_result.raw_handle())) {
+ if (0 != detail::get_future_error(ferror_result)) {
// Destroy the futures AND be sure to invalidate the transaction-- according to the documentation,
// this must happen in the specified order (which /should/ currently match the dtor order, but I am
// not going to touch this any more right now):
inline future_value block_until_ready(future_value&& fv)
{
if (fdb_error_t r = fdb_future_block_until_ready(fv.raw_handle()); 0 != r) {
- throw libfdb_exception(r);
+ throw libfdb_exception(r);
}
// Note that fdb_future_block_until_ready() does not by itself check for errors
// with the value; so, we need to do this separately:
fdb_error_t r = fdb_future_get_error(fv.raw_handle());
- if (r) {
+ if (0 != r) {
throw libfdb_exception(r);
}
inline future_value wait_until_ready(future_value&& fv)
{
if (fdb_error_t r = fdb_future_block_until_ready(fv.raw_handle()); 0 != r) {
- throw libfdb_exception(r);
+ throw libfdb_exception(r);
}
return fv;
}
+inline fdb_error_t get_future_error(const future_value& fv)
+{
+ return fdb_future_get_error(fv.raw_handle());
+}
+
+inline future_value wait_for_on_error(FDBTransaction* txn, const fdb_error_t original_error)
+{
+ future_value on_error_future(fdb_transaction_on_error(txn, original_error));
+
+ if (0 != fdb_future_block_until_ready(on_error_future.raw_handle())) {
+ throw libfdb_exception(original_error);
+ }
+
+ return on_error_future;
+}
+
template <typename FnT, typename... XS>
requires std::invocable<FnT, XS...>
inline future_value await_future_of(FnT&& fn, XS&& ...params)
{
return block_until_ready(
std::invoke(std::forward<FnT>(fn), std::forward<XS>(params)...));
-}
+}
+
+// Tracks a set of results and their corresponding Future:
+struct query_window final
+{
+ future_value result_owner;
+ std::span<const FDBKeyValue> result_pairs;
+ bool more_available = false;
+};
+
+struct split_point_result final
+{
+ future_value result_owner;
+ std::span<const FDBKey> result_keys;
+ fdb_error_t error = 0;
+};
+
+inline query_window extract_result_pairs(future_value result_owner)
+{
+ int more_available = 0;
+ int out_count = 0;
+ const FDBKeyValue *out_kvs = nullptr;
+
+ if (fdb_error_t r = fdb_future_get_keyvalue_array(result_owner.raw_handle(),
+ &out_kvs,
+ &out_count,
+ &more_available); 0 != r) {
+ throw libfdb_exception(r);
+ }
+
+ return query_window {
+ .result_owner = std::move(result_owner),
+ .result_pairs = std::span<const FDBKeyValue>(out_kvs, out_count),
+ .more_available = more_available
+ };
+}
+
+inline split_point_result extract_split_points(future_value result_owner)
+{
+ const FDBKey *result_keys = nullptr;
+ int result_count = 0;
+
+ const auto error = fdb_future_get_key_array(result_owner.raw_handle(), &result_keys, &result_count);
+
+ return split_point_result {
+ .result_owner = std::move(result_owner),
+ .result_keys = 0 == error ? std::span<const FDBKey>(result_keys, result_count)
+ : std::span<const FDBKey>(),
+ .error = error
+ };
+}
+
+inline query_window read_query_window(transaction& txn, const select& key_range, const int iteration)
+{
+ return extract_result_pairs(await_future_of([&]() {
+ return get_range_future_from_transaction(txn, key_range, iteration);
+ }));
+}
+
+inline std::optional<select> next_range_after(select key_range, const query_window& window)
+{
+ if (not window.more_available or window.result_pairs.empty()) {
+ return std::nullopt;
+ }
+
+ const auto& last_key = window.result_pairs.back();
+ auto cursor = std::string_view((const char *)last_key.key, last_key.key_length);
+
+ if (key_range.options.reverse_order) {
+ key_range.end_key = cursor;
+ key_range.end_inclusive = false;
+ return key_range;
+ }
+
+ key_range.begin_key = cursor;
+ key_range.begin_inclusive = false;
+
+ return key_range;
+}
+
+template <typename ValueT = std::string>
+inline auto decode_pairs(std::span<const FDBKeyValue> pairs)
+{
+ return pairs | std::views::transform(to_decoded_kv_pair<ValueT>);
+}
+
+template <typename ValueT, typename AssocT>
+inline AssocT collect_pairs(std::span<const FDBKeyValue> pairs)
+{
+ AssocT out;
+ std::ranges::copy(decode_pairs<ValueT>(pairs), std::inserter(out, std::end(out)));
+ return out;
+}
+
+template <typename AssocT>
+struct query_window_result final
+{
+ AssocT result_block;
+ std::optional<select> next_range;
+};
+
+template <typename ValueT, typename AssocT>
+inline query_window_result<AssocT> materialize_query_window(transaction& txn, select key_range, const int iteration = 1)
+{
+ auto window = read_query_window(txn, key_range, iteration);
+
+ return {
+ .result_block = collect_pairs<ValueT, AssocT>(window.result_pairs),
+ .next_range = next_range_after(std::move(key_range), window)
+ };
+}
template <typename OutIterT>
requires std::output_iterator<OutIterT, std::pair<std::string, std::string>>
return true;
}
-inline fdb_error_t value_from_db(future_value& fv,
- std::invocable<future_value&> auto extract_fn)
-{
- return std::invoke(extract_fn, fv);
-}
-
inline bool retry_after_error(ceph::libfdb::transaction_handle& txn, const fdb_error_t r)
{
if (0 == r) {
- return false;
+ return false;
}
if (not fdb_error_predicate(FDB_ERROR_PREDICATE_RETRYABLE, r)) {
- // Non-retryable errors cannot be repaired by on_error():
- throw ceph::libfdb::libfdb_exception(r);
+ // Non-retryable errors cannot be repaired by on_error():
+ throw ceph::libfdb::libfdb_exception(r);
}
- future_value on_error_future(fdb_transaction_on_error(txn->raw_handle(), r));
+ auto on_error_future = wait_for_on_error(txn->raw_handle(), r);
- if (fdb_error_t on_error_r = fdb_future_block_until_ready(on_error_future.raw_handle()); 0 != on_error_r) {
- throw ceph::libfdb::libfdb_exception(r);
- }
-
- if (fdb_error_t on_error_r = fdb_future_get_error(on_error_future.raw_handle()); 0 != on_error_r) {
- throw ceph::libfdb::libfdb_exception(r);
+ if (fdb_error_t on_error_r = get_future_error(on_error_future); 0 != on_error_r) {
+ throw ceph::libfdb::libfdb_exception(r);
}
return true;
}
// Convert FDBKey array into something useful:
-inline std::vector<ceph::libfdb::select> as_select_seq(const FDBKey* const xs,
- const int n,
+inline std::vector<ceph::libfdb::select> as_select_seq(std::span<const FDBKey> xs,
const ceph::libfdb::select& parent)
{
- std::vector<ceph::libfdb::select> out;
-
- if (1 < n) {
- out.reserve(static_cast<std::size_t>(n - 1));
+ if (2 > xs.size()) {
+ return {};
}
// Gather the flattened list into *overlapping* libfdb::select pairs:
- for (const auto& dyad : std::span(xs, n) | std::views::slide(2)) {
- const auto& fst = std::ranges::begin(dyad)[0];
- const auto& snd = std::ranges::begin(dyad)[1];
-
- ceph::libfdb::select split {
- { (const char *)fst.key, static_cast<std::string::size_type>(fst.key_length) },
- { (const char *)snd.key, static_cast<std::string::size_type>(snd.key_length) }
- };
-
- split.options = parent.options;
- out.push_back(std::move(split));
- }
-
- return out;
+ return std::views::iota(std::size_t{0}, xs.size() - 1)
+ | std::views::transform([&parent, xs](const auto i) {
+ const auto& fst = xs[i];
+ const auto& snd = xs[i + 1];
+ const auto first_key = std::string_view((const char *)fst.key,
+ static_cast<std::string::size_type>(fst.key_length));
+ const auto second_key = std::string_view((const char *)snd.key,
+ static_cast<std::string::size_type>(snd.key_length));
+
+ ceph::libfdb::select split(first_key, second_key);
+
+ split.options = parent.options;
+ split.begin_inclusive = (0 == i) ? parent.begin_inclusive : true;
+ split.end_inclusive = (i + 2 == xs.size()) ? parent.end_inclusive : false;
+ return split;
+ })
+ | std::ranges::to<std::vector<ceph::libfdb::select>>();
}
// Finding a clear example both in the samples and in the documentation is not very easy. The
-// statelessness of FDB requests bleeds into here with basically no hand-holding, so it's fairly subtle and
-// not easy to see what to do (I'm still not sure I have it right), but note for instance that the call
-// parameters have to change for subsequent reads.
+// statelessness of FDB requests bleeds into here with basically no hand-holding, but note for instance
+// that the call parameters have to change for subsequent reads.
inline future_value get_range_future_from_transaction(ceph::libfdb::transaction& txn, const ceph::libfdb::select& selection, int iteration)
{
const auto& begin_key = selection.begin_key;
// The documentation makes this stuff about as clear as mud... read VERY carefully
// when you fiddle with these:
- const int begin_or_eq = (not options.reverse_order and 1 < iteration) ? 1 : 0;
+ const bool continuing_forward = not options.reverse_order and 1 < iteration;
+ const bool continuing_reverse = options.reverse_order and 1 < iteration;
+
+ const int begin_or_eq = (continuing_forward or not selection.begin_inclusive) ? 1 : 0;
const int begin_offset = 1;
- const int end_or_eq = 0;
+ const int end_or_eq = (not continuing_reverse and selection.end_inclusive) ? 1 : 0;
const int end_offset = 1;
- // See validate_and_update_parameters() in fdb_c.cpp (FDB source) if greater clarity is needed on the meaning of some of these, it can be
- // hard to deduce from the documentation:
- // Returns an FDBKeyValueArray in the future:
-
+ // See validate_and_update_parameters() in fdb_c.cpp (FDB source) if greater clarity is needed on
+ // the meaning of some of these, it can be hard to deduce from the documentation; Returns an
+ // FDBKeyValueArray in the future:
return future_value(fdb_transaction_get_range(txn.raw_handle(),
(const uint8_t *)begin_key.data(), begin_key.size(), // the reference-point key
begin_or_eq, // begin or eq
// How should results be grouped/chunked:
options.stride, // limit (0 == unlimited)
- 0, // target bytes (0 == unlimited)
+ options.target_bytes, // target bytes (0 == unlimited)
options.streaming_mode, // streaming mode (e.g.: FDB_STREAMING_MODE_WANT_ALL)
iteration, // iteration # (produced side effect)
using ceph::libfdb::detail::wait_until_ready;
auto txn = ceph::libfdb::make_transaction(dbh);
-
- // We can do this without side-effects by combining some lambdas, but for now I'm satisfied if it works:
- FDBKey const *keys = nullptr;
- int nkeys = 0;
+ auto split_selector = ceph::libfdb::detail::as_half_open_select(selector);
for (bool should_retry = true; should_retry;) {
+ auto result_owner = wait_until_ready(future_value(fdb_transaction_get_range_split_points(
+ txn->raw_handle(),
+ (const std::uint8_t *)split_selector.begin_key.data(), static_cast<int>(split_selector.begin_key.length()),
+ (const std::uint8_t *)split_selector.end_key.data(), static_cast<int>(split_selector.end_key.length()),
+ remote_chunk_size)));
- auto fv = wait_until_ready(future_value(fdb_transaction_get_range_split_points(
- txn->raw_handle(),
- (const std::uint8_t *)selector.begin_key.data(), static_cast<int>(selector.begin_key.length()),
- (const std::uint8_t *)selector.end_key.data(), static_cast<int>(selector.end_key.length()),
- remote_chunk_size)));
+ auto split_points = extract_split_points(std::move(result_owner));
- should_retry = retry_after_error(txn, value_from_db(fv,
- [&keys, &nkeys](future_value& fv) {
- return fdb_future_get_key_array(fv.raw_handle(), &keys, &nkeys);
- }));
+ should_retry = retry_after_error(txn, split_points.error);
- if (not should_retry) {
- return as_select_seq(keys, nkeys, selector);
- }
+ if (not should_retry) {
+ return as_select_seq(split_points.result_keys, split_selector);
+ }
}
return {};
}
-// Generators:
+// Generators (internal guts):
// The returned memory should be copied immediately as its lifetime will end once the Future is destroyed:
-// (This is pretty much why this is in the "detail" namespace for the moment.)
-inline std::generator<std::span<const FDBKeyValue>> generate_FDB_pairs(transaction& txn, select key_range)
+// (This is pretty much why this is in the "detail" namespace-- we use this to implement the user-facing
+// stuff, it shouldn't really be touched outside.)
+inline std::generator<std::span<const FDBKeyValue>> generate_FDB_pairs(transaction& txn, select key_range)
{
- // As FDB is stateless, we need to track state somewhere:
- int more_available = 1;
- int out_count = 0;
- int iteration = 1; // expected to be 1 when FDB is called in iterator mode
-
- const FDBKeyValue *out_kvs = nullptr;
+ // FDB uses iteration for FDB_STREAMING_MODE_ITERATOR (but, other streaming modes
+ // just ignore it, per fdb_c's documentation. We do have to keep this state somewhere,
+ // though. See: "https://apple.github.io/foundationdb/api-c.html".
+ int iteration = 1;
- for (; more_available; iteration++) {
+ for (auto more_available = true; more_available; iteration++) {
+ auto window = read_query_window(txn, key_range, iteration);
+ auto next_range = next_range_after(key_range, window);
- auto fv = await_future_of([&]() { return get_range_future_from_transaction(txn, key_range, iteration); });
+ more_available = next_range.has_value();
- if (fdb_error_t r = fdb_future_get_keyvalue_array(fv.raw_handle(), &out_kvs, &out_count, &more_available); 0 != r) {
- throw libfdb_exception(r);
- }
+ co_yield window.result_pairs;
- auto result = std::span<const FDBKeyValue>(out_kvs, out_count);
-
- if (more_available) {
- // Make the first part of the range for the new search equal to the last one from the old search:
- const auto& last_key = result.back();
- auto& cursor = key_range.options.reverse_order ? key_range.end_key : key_range.begin_key;
-
- cursor = std::string_view((const char *)last_key.key, last_key.key_length);
- }
-
- co_yield result;
+ if (next_range) {
+ key_range = std::move(*next_range);
}
+ }
}
} // namespace ceph::libfdb::detail
#include <type_traits>
#include <system_error>
-/* This module is for converting internal types "owned" by FoundationDB. They've initially been implemented in
-a conversion namespace with overloads, which is the same way that user conversions work, but especially with Concepts
-this technique doesn't have to be used-- it is likely possible to avoid default construction and possible extra copies.
-Since I'm building a prototype right now, it's more important to me on this pass to make things understandable and
-clear, because past experience informs me that it's better to have a clear understanding of what the goals of "to"
-and "from" versions actually are in relationship to user-level types than to have every nanosecond of performance
-be available on day one.
-
-The target of "to" conversions is not a USER type, but rather the FUNCTIONS provided inside of libfdb-- users
-should NOT see the output of these or have to handle them outside of tests or edge-cases (and even then, I doubt it's
-needed, though I won't work hard to stop it).
-
-I'm hoping that later down the line I can sit and spend more time with this-- it would be nice, for example, if we could
-use memory provided by the caller.
-
-Additionally, this mechanism is mostly obviated by forwarding the work to zpp_bits, but keeping it here provides an additional
-hook "ahead" of that library, and indeed eliminates any actual dependency on it-- we use it to smooth out a few areas where
-zpp_bits is designed for serialization ONLY rather than also playing nice with some conversions, especially those where the
-input and output sizes of an array may not match.
-*/
+/* This module is the conversion boundary between C++ values and the byte buffers used by FoundationDB. Most of
+the serialization work is delegated to zpp_bits, but this layer has two jobs: it gives libfdb's gadgets a place
+to live (abstracting array/span behavior, callback outputs, error translation, etc.), and it provides a clean point
+for future features-- a good example would be caller-owned memory, which we currently don't support but certainly
+could. It also provides a fixed point where another serialization library could be swapped in. */
+
namespace ceph::libfdb::to {
inline auto convert(const auto& from, std::vector<std::uint8_t>& out_data) -> std::span<const std::uint8_t>
}
template <std::invocable<const char *, size_t> OutputFunction>
-inline void convert(const std::span<const std::uint8_t>& in, OutputFunction& fn)
+inline void convert(const std::span<const std::uint8_t>& in, OutputFunction& write_output_fn)
{
- fn((const char *)in.data(), in.size());
+ write_output_fn((const char *)in.data(), in.size());
}
} // namespace ceph::libfdb::from
#ifndef CEPH_RGW_FDB_H
#define CEPH_RGW_FDB_H
-#include "base.h"
#include "interface.h"
-#include "conversion.h"
#endif
#include "base.h"
#include "conversion.h"
-#include <span>
-#include <ranges>
-#include <cstdint>
-#include <utility>
-#include <optional>
-#include <concepts>
-#include <iterator>
-#include <generator>
-#include <exception>
-#include <algorithm>
-#include <functional>
-#include <filesystem>
-#include <type_traits>
-
namespace ceph::libfdb {
-/* This needs to be called once and only once during application lifetime,
-when no more FoundationDB calls will be made: */
+/* This should be called when the application is all done with FoundationDB: */
inline void shutdown_libfdb()
{
ceph::libfdb::detail::database_system::shutdown_fdb();
auto commit_noreplay(transaction_handle txn, const commit_after_op commit_after, FnT&& fn)
-> operation_result_t<FnT>;
+template <supported_transaction_invocation FnT>
+auto commit_in_new_transaction(database_handle dbh, FnT&& fn)
+ -> operation_result_t<FnT>;
+
} // namespace ceph::libfdb::detail
namespace ceph::libfdb {
inline void set(database_handle dbh,
std::string_view k, const auto& v)
{
- return detail::maybe_commit(make_transaction(dbh), commit_after_op::commit,
+ return detail::commit_in_new_transaction(dbh,
[key = detail::as_fdb_span(k), &v](const transaction_handle& txn) {
return txn->set(key, ceph::libfdb::to::convert(v));
});
concepts::key_value_iterator IteratorT>
inline void set(database_handle dbh, IteratorT b, IteratorT e)
{
- return detail::maybe_commit(make_transaction(dbh), commit_after_op::commit,
+ return detail::commit_in_new_transaction(dbh,
[b, e](const transaction_handle& txn) {
return set(txn, b, e, commit_after_op::no_commit);
});
inline void set(database_handle dbh,
std::string_view k, const ceph::libfdb::concepts::stringview_convertible auto& v)
{
- return detail::maybe_commit(make_transaction(dbh), commit_after_op::commit,
+ return detail::commit_in_new_transaction(dbh,
[key = detail::as_fdb_span(k), value = std::string_view(v)](const transaction_handle& txn) {
return txn->set(key, ceph::libfdb::to::convert(value));
});
inline void erase(ceph::libfdb::database_handle dbh,
const ceph::libfdb::select& key_range)
{
- return detail::maybe_commit(ceph::libfdb::make_transaction(dbh), commit_after_op::commit,
+ return detail::commit_in_new_transaction(dbh,
[&key_range](const transaction_handle& txn) {
return txn->erase(key_range);
});
inline void erase(ceph::libfdb::database_handle dbh, std::string_view k)
{
- return detail::maybe_commit(make_transaction(dbh), commit_after_op::commit,
+ return detail::commit_in_new_transaction(dbh,
[key = detail::as_fdb_span(k)](const transaction_handle& txn) {
return txn->erase(key);
});
// Basic Generators:
namespace ceph::libfdb {
-// For ordinary range scans, pair_generator() is usually the right default:
+// For ordinary range scans inside one explicit transaction, pair_generator() is
+// usually the right default:
template <typename ValueT = std::string>
inline auto pair_generator(ceph::libfdb::transaction_handle txn, ceph::libfdb::select key_range)
-> std::generator<std::pair<std::string, ValueT>>
{
- for (const std::span<const FDBKeyValue>& kvp_block : ceph::libfdb::detail::generate_FDB_pairs(*txn, key_range)) {
- for (const auto& kvp : kvp_block) {
- co_yield ceph::libfdb::detail::to_decoded_kv_pair<ValueT>(kvp);
- }
- }
-}
+ auto decoded_pairs = ceph::libfdb::detail::generate_FDB_pairs(*txn, key_range)
+ | std::views::join
+ | std::views::transform(ceph::libfdb::detail::to_decoded_kv_pair<ValueT>);
-template <typename ValueT = std::string>
-inline auto pair_generator(const ceph::libfdb::database_handle& dbh, ceph::libfdb::select key_range)
-{
- return pair_generator<ValueT>(ceph::libfdb::make_transaction(dbh), key_range);
+ co_yield std::ranges::elements_of(decoded_pairs);
}
-// Note: block_generator() uses split planning to tackle large sets; use pair_generator() for
-// direct scans.
+// Note: block_generator() uses split planning to tackle large sets; use pair_generator() for
+// direct scans.
//
// What block_generator() gives you:
-// - avoids one huge transaction getting too old
-// - gives caller block-at-a-time processing
-// - can bound memory and transaction duration better than a monolithic scan
+// - avoids one huge transaction getting too old
+// - gives caller block-at-a-time processing
+// - can bound memory and transaction duration better than a monolithic scan
//
// Note: block_generator() was originally parallel, and could be again, but preliminary benchmarking
// showed it to be a significant performance impediment. The database must be truly large to see benefits.
//
// Note: This is meant to be straightforward and easy-to-understand-- hence, there's not
-// a recover strategy or other things (you can replay the entire query)-- as new needs arise, this
+// a recovery strategy or other things (you can replay the entire query)-- as new needs arise, this
// can be made more flexible via selector options, dynamic range-splitting, etc., but so far there
// has been no need:
-template <typename AssocT = flat_map<std::string, std::string>>
+template <typename ValueT = std::string,
+ typename AssocT = std::vector<std::pair<std::string, ValueT>>>
auto block_generator(ceph::libfdb::database_handle dbh, ceph::libfdb::select selector)
-> std::generator<AssocT>
{
+ if (0 == selector.options.stride) {
+ selector.options.stride = 4096;
+ }
+
// JFW: Although this is tunable, in my measurement it isn't a large factor so far-- likely
// these can move into the select instance itself (this is hard to measure in tests-- N
// has to be pretty big; I've adjusted chunk_size to where I guesstimate it needs to be,
// we need real-world experience to tweak this further):
- const auto chunk_size = 4*(1024*1024);
+ const auto chunk_size = 4 * 1024 * 1024;
auto split_ranges = detail::plan_split_ranges(dbh, selector, chunk_size);
- auto read_block = [txr = make_transactor(dbh)](const auto& range) {
- return txr([&range](auto& txn) {
- return pair_generator(txn, range) | std::ranges::to<AssocT>();
+ auto read_blocks = [txr = make_transactor(dbh)](this auto& read_blocks, ceph::libfdb::select range, const int iteration)
+ -> std::generator<AssocT> {
+ auto read_result = txr([range, iteration](auto& txn) {
+ return detail::materialize_query_window<ValueT, AssocT>(*txn, range, iteration);
});
+
+ auto next_range = std::move(read_result.next_range);
+
+ if (read_result.result_block.empty()) {
+ co_return;
+ }
+
+ co_yield std::move(read_result.result_block);
+
+ if (next_range) {
+ co_yield std::ranges::elements_of(read_blocks(std::move(*next_range), 2));
+ }
};
- co_yield std::ranges::elements_of(split_ranges | std::views::transform(read_block));
+ auto expand_range = [&read_blocks](ceph::libfdb::select range) {
+ return read_blocks(std::move(range), 1);
+ };
+
+ co_yield std::ranges::elements_of(split_ranges
+ | std::views::transform(expand_range)
+ | std::views::join);
}
} // namespace ceph::libfdb
return 0;
}
+inline bool commit_or_throw(transaction_handle& txn)
+{
+ if (fdb_error_t r = do_commit(txn); 0 != r) {
+ throw ceph::libfdb::libfdb_exception(r);
+ }
+
+ return true;
+}
+
template <typename OutValuesT>
void value_collector_t<OutValuesT>::operator()(std::span<const std::uint8_t> out_data) const
{
txn, std::forward<FnT>(fn),
[](transaction_handle& txn) {
return ceph::libfdb::commit(txn);
- });
+ });
+}
+
+template <supported_transaction_invocation FnT>
+auto commit_in_new_transaction(database_handle dbh, FnT&& fn)
+ -> operation_result_t<FnT>
+{
+ return maybe_commit(make_transaction(dbh),
+ commit_after_op::commit,
+ std::forward<FnT>(fn));
}
// Commit only once; the caller is responsible for transaction replay:
auto commit_noreplay(transaction_handle txn, const commit_after_op commit_after, FnT&& fn)
-> operation_result_t<FnT>
{
- using result_t = transaction_invocation_result_t<FnT>;
-
- auto commit_fn = [](transaction_handle& txn) {
- if (fdb_error_t r = do_commit(txn); 0 != r) {
- throw ceph::libfdb::libfdb_exception(r);
- }
-
- return true;
- };
-
if (commit_after_op::commit != commit_after) {
return std::invoke(fn, txn);
}
- if constexpr (std::is_void_v<result_t>) {
- attempt_invocation<invocation_failure_policy::no_retry>(txn, fn, commit_fn);
- return;
- }
-
- if constexpr (concepts::storable_invocation_result<result_t>) {
- auto result = attempt_invocation<invocation_failure_policy::no_retry>(
- txn, fn, commit_fn);
- return std::remove_cvref_t<result_t>(std::move(*result));
- }
+ return invoke_with_retry<invocation_failure_policy::no_retry>(
+ txn, std::forward<FnT>(fn), commit_or_throw);
}
} // namespace ceph::libfdb::detail
{
auto kvs = make_monotonic_kvs(N, prefix);
- for(const auto& [k, v] : kvs)
+ for (const auto& [k, v] : kvs)
lfdb::set(lfdb::make_transaction(dbh), k, v, lfdb::commit_after_op::commit);
return kvs;
}
+inline auto decode_raw_fdb_pairs(std::span<const FDBKeyValue> pairs)
+{
+ return pairs
+ | std::views::transform(ceph::libfdb::detail::to_decoded_kv_pair<std::string>)
+ | std::ranges::to<std::vector<string_pair>>();
+}
+
+inline auto first_keys(const auto& kvs, const std::size_t n)
+{
+ return kvs
+ | std::views::take(n)
+ | std::ranges::to<std::vector<string_pair>>();
+}
+
+inline auto last_keys_reversed(const auto& kvs, const std::size_t n)
+{
+ return kvs
+ | std::views::reverse
+ | std::views::take(n)
+ | std::ranges::to<std::vector<string_pair>>();
+}
+
// Basically, make sure we're actually linking with the library:
TEST_CASE()
{
TEST_CASE("fdb simple", "[rgw][fdb]") {
janitor j;
- auto dbh = j.dbh();
-
- const string_view k = "key";
+ const string k = test_key("key");
const string v = fmt::format("value-{:%c}", std::chrono::system_clock::now());
SECTION("read missing key") {
- const string_view missing_key = "missing_key";
+ const string missing_key = test_key("missing_key");
SECTION("with transaction") {
std::string out_value;
- auto txn_handle = lfdb::make_transaction(dbh);
+ auto txn_handle = lfdb::make_transaction(j);
REQUIRE(nullptr != txn_handle);
CAPTURE(missing_key);
std::string out_value;
// The key initially either exists, or we'll write it anew, either is fine:
- CHECK_NOTHROW(lfdb::set(lfdb::make_transaction(dbh), k, v, lfdb::commit_after_op::commit));
+ CHECK_NOTHROW(lfdb::set(lfdb::make_transaction(j), k, v, lfdb::commit_after_op::commit));
// Make sure that it DOES exist:
- CHECK(lfdb::get(lfdb::make_transaction(dbh), k, out_value, lfdb::commit_after_op::no_commit));
+ CHECK(lfdb::get(lfdb::make_transaction(j), k, out_value, lfdb::commit_after_op::no_commit));
CHECK(v == out_value);
// "erase()" is known as "clear" in FDB parlance, deleting a record:
- REQUIRE_NOTHROW(lfdb::erase(lfdb::make_transaction(dbh), k, lfdb::commit_after_op::commit));
+ REQUIRE_NOTHROW(lfdb::erase(lfdb::make_transaction(j), k, lfdb::commit_after_op::commit));
// ...as this shouldn't be updated again, make sure there isn't an accidental match:
out_value.clear();
// ...and, POOF!-- it should be gone:
- CHECK_FALSE(lfdb::get(lfdb::make_transaction(dbh), k, out_value, lfdb::commit_after_op::no_commit));
+ CHECK_FALSE(lfdb::get(lfdb::make_transaction(j), k, out_value, lfdb::commit_after_op::no_commit));
CHECK(v != out_value);
}
SECTION("read/write single key") {
- REQUIRE(nullptr != dbh);
+ REQUIRE(nullptr != j.dbh());
// First, be sure we have a valid value written to the database:
- REQUIRE_NOTHROW(lfdb::set(lfdb::make_transaction(dbh), k, v, lfdb::commit_after_op::commit));
+ REQUIRE_NOTHROW(lfdb::set(lfdb::make_transaction(j), k, v, lfdb::commit_after_op::commit));
SECTION("read transaction") {
std::string out_value;
- CHECK(lfdb::get(lfdb::make_transaction(dbh), k, out_value, lfdb::commit_after_op::no_commit));
+ CHECK(lfdb::get(lfdb::make_transaction(j), k, out_value, lfdb::commit_after_op::no_commit));
CHECK(v == out_value);
}
}
SECTION("check for existence of key") {
- REQUIRE(nullptr != dbh);
+ REQUIRE(nullptr != j.dbh());
// Erase the key if it's already there:
- lfdb::erase(lfdb::make_transaction(dbh), k, lfdb::commit_after_op::commit);
+ lfdb::erase(lfdb::make_transaction(j), k, lfdb::commit_after_op::commit);
// Now, we shouldn't find anything:
- CHECK_FALSE(lfdb::key_exists(lfdb::make_transaction(dbh), k, lfdb::commit_after_op::no_commit));
+ CHECK_FALSE(lfdb::key_exists(lfdb::make_transaction(j), k, lfdb::commit_after_op::no_commit));
// Write the key:
- lfdb::set(lfdb::make_transaction(dbh), k, v, lfdb::commit_after_op::commit);
+ lfdb::set(lfdb::make_transaction(j), k, v, lfdb::commit_after_op::commit);
// ...it should magically be there!
- CHECK(lfdb::key_exists(lfdb::make_transaction(dbh), k, lfdb::commit_after_op::no_commit));
+ CHECK(lfdb::key_exists(lfdb::make_transaction(j), k, lfdb::commit_after_op::no_commit));
// ...and now it should be gone again:
- lfdb::erase(lfdb::make_transaction(dbh), k, lfdb::commit_after_op::commit);
- CHECK_FALSE(lfdb::key_exists(lfdb::make_transaction(dbh), k, lfdb::commit_after_op::no_commit));
+ lfdb::erase(lfdb::make_transaction(j), k, lfdb::commit_after_op::commit);
+ CHECK_FALSE(lfdb::key_exists(lfdb::make_transaction(j), k, lfdb::commit_after_op::no_commit));
}
}
// Erase the entire range:
lfdb::erase(dbh, selector);
CHECK(0 == key_count(dbh, selector));
+
+ const auto bounded_selector = lfdb::select { make_key(0, "bounded"), make_key(20, "bounded") };
+ write_monotonic_kvs(dbh, 20, "bounded");
+ lfdb::erase(dbh, lfdb::select { lfdb::exclusive(make_key(5, "bounded")), lfdb::inclusive(make_key(7, "bounded")) });
+
+ CHECK(18 == key_count(dbh, bounded_selector));
+ CHECK(lfdb::key_exists(dbh, make_key(5, "bounded")));
+ CHECK_FALSE(lfdb::key_exists(dbh, make_key(6, "bounded")));
+ CHECK_FALSE(lfdb::key_exists(dbh, make_key(7, "bounded")));
+ CHECK(lfdb::key_exists(dbh, make_key(8, "bounded")));
}
TEMPLATE_PRODUCT_TEST_CASE("multi-key ops", "[rgw][fdb]",
{
janitor j;
- auto dbh = j.dbh();
-
// Write a sequence of keys so we have some data to work with:
- const auto kvs = write_monotonic_kvs(dbh, 100);
+ const auto kvs = write_monotonic_kvs(j, 100);
SECTION("check multiple key write", "[fdb]") {
- auto txn = lfdb::make_transaction(dbh);
+ auto txn = lfdb::make_transaction(j);
std::string out_value;
SECTION("check multiple key selection", "[fdb]") {
TestType out_values;
- auto txn = lfdb::make_transaction(dbh);
+ auto txn = lfdb::make_transaction(j);
lfdb::get(txn, lfdb::select { make_key(0), make_key(100) }, std::back_inserter(out_values), lfdb::commit_after_op::no_commit);
CHECK(100 == out_values.size());
// Maybe not the world's most creative test, but the idea is to try getting some random keys:
- for(auto i = ceph::util::generate_random_number(out_values.size() - 1); i; --i) {
+ for (auto i = ceph::util::generate_random_number(out_values.size() - 1); i; --i) {
CHECK(std::end(out_values) != std::ranges::find(out_values, string_pair { make_key(i), make_value(i) }));
}
}
CHECK("abd" == lfdb::select { std::string("abc\xFF", 4) }.end_key);
// Make sure that there's nothing in our test range:
- dbh.drop_all();
+ dbh.drop_test_namespace();
REQUIRE(0 == key_count(dbh, select_all));
const auto kvs = write_monotonic_kvs(dbh, nentries);
CHECK(make_key(0) == out.front().first);
CHECK(make_key(nentries - 1) == out.back().first);
+ auto keys_in = [&dbh](lfdb::select selector) {
+ std::vector<std::pair<std::string, std::string>> entries;
+ lfdb::get(dbh, selector, std::back_inserter(entries));
+
+ return entries
+ | std::views::transform([](const auto& kv) { return kv.first; })
+ | std::ranges::to<std::vector<std::string>>();
+ };
+
+ CHECK_THAT(keys_in(lfdb::select { make_key(3), make_key(6) }),
+ Catch::Matchers::RangeEquals(std::vector { make_key(3), make_key(4), make_key(5) }));
+ CHECK_THAT(keys_in(lfdb::select { lfdb::exclusive(make_key(3)), make_key(6) }),
+ Catch::Matchers::RangeEquals(std::vector { make_key(4), make_key(5) }));
+ CHECK_THAT(keys_in(lfdb::select { make_key(3), lfdb::inclusive(make_key(6)) }),
+ Catch::Matchers::RangeEquals(std::vector { make_key(3), make_key(4), make_key(5), make_key(6) }));
+ CHECK_THAT(keys_in(lfdb::select { lfdb::exclusive(make_key(3)), lfdb::inclusive(make_key(6)) }),
+ Catch::Matchers::RangeEquals(std::vector { make_key(4), make_key(5), make_key(6) }));
+
+ auto reverse_open_closed = lfdb::select { lfdb::exclusive(make_key(3)), lfdb::inclusive(make_key(6)) };
+ reverse_open_closed.options.reverse_order = true;
+ reverse_open_closed.options.stride = 2;
+ CHECK_THAT(keys_in(reverse_open_closed),
+ Catch::Matchers::RangeEquals(std::vector { make_key(6), make_key(5), make_key(4) }));
+
SECTION("reverse order") {
auto reverse_all = select_all;
reverse_all.options.reverse_order = true;
&std::pair<std::string, std::string>::first));
}
- lfdb::set(dbh, "keyx", "outside");
+ lfdb::set(dbh, test_key("keyx"), "outside");
out.clear();
- lfdb::get(dbh, lfdb::select { "key_" }, std::back_inserter(out));
+ lfdb::get(dbh, lfdb::select { make_key_prefix() }, std::back_inserter(out));
CHECK(nentries == out.size());
// Get exactly no entries:
TEST_CASE("fdb conversions (round-trip)", "[fdb][rgw]") {
janitor j;
- auto dbh = j.dbh();
-
// string_view -> string
{
const std::string_view n = "Hello, World!";
std::string o;
- lfdb::set(lfdb::make_transaction(dbh), "key", n, lfdb::commit_after_op::commit);
- lfdb::get(lfdb::make_transaction(dbh), "key", o, lfdb::commit_after_op::no_commit);
+ const auto key = test_key("key");
+ lfdb::set(lfdb::make_transaction(j), key, n, lfdb::commit_after_op::commit);
+ lfdb::get(lfdb::make_transaction(j), key, o, lfdb::commit_after_op::no_commit);
REQUIRE_THAT(n, Catch::Matchers::RangeEquals(o));
}
const std::vector<uint8_t> n = { 1, 2, 3, 4, 5 };
std::vector<uint8_t> o;
- lfdb::set(lfdb::make_transaction(dbh), "key", n, lfdb::commit_after_op::commit);
- lfdb::get(lfdb::make_transaction(dbh), "key", o, lfdb::commit_after_op::no_commit);
+ const auto key = test_key("key");
+ lfdb::set(lfdb::make_transaction(j), key, n, lfdb::commit_after_op::commit);
+ lfdb::get(lfdb::make_transaction(j), key, o, lfdb::commit_after_op::no_commit);
REQUIRE_THAT(n, Catch::Matchers::RangeEquals(o));
}
SECTION("get with a raw value callback")
{
janitor j;
- auto dbh = j.dbh();
std::string_view n { pearl_msg };
std::string o;
- lfdb::set(dbh, "key", n);
+ const auto key = test_key("key");
+ lfdb::set(j, key, n);
- REQUIRE(lfdb::get(dbh, "key", [&o](std::span<const std::uint8_t> in) {
+ REQUIRE(lfdb::get(j, key, [&o](std::span<const std::uint8_t> in) {
ceph::libfdb::from::convert(in, o);
}));
}
}
+TEST_CASE("read_query_window", "[fdb]")
+{
+ janitor j;
+
+ const std::size_t stride = 5;
+ const std::size_t nkeys = 12;
+
+ const auto kvs_in = write_monotonic_kvs(j, nkeys);
+
+ SECTION("reads one bounded forward window") {
+ auto selector = lfdb::select { make_key(0), make_key(nkeys) };
+ selector.options.stride = stride;
+
+ auto txn = lfdb::make_transaction(j);
+ auto window = ceph::libfdb::detail::read_query_window(*txn, selector, 1);
+ const auto out = decode_raw_fdb_pairs(window.result_pairs);
+
+ REQUIRE_FALSE(out.empty());
+ CHECK(out.size() <= stride);
+ CHECK(window.more_available);
+ CHECK_THAT(out, Catch::Matchers::RangeEquals(first_keys(kvs_in, out.size())));
+ }
+
+ SECTION("reads one bounded reverse window") {
+ auto selector = lfdb::select { make_key(0), make_key(nkeys) };
+ selector.options.stride = stride;
+ selector.options.reverse_order = true;
+
+ auto txn = lfdb::make_transaction(j);
+ auto window = ceph::libfdb::detail::read_query_window(*txn, selector, 1);
+ const auto out = decode_raw_fdb_pairs(window.result_pairs);
+
+ REQUIRE_FALSE(out.empty());
+ CHECK(out.size() <= stride);
+ CHECK(window.more_available);
+ CHECK_THAT(out, Catch::Matchers::RangeEquals(last_keys_reversed(kvs_in, out.size())));
+ }
+
+ SECTION("reports terminal windows") {
+ auto selector = lfdb::select { make_key(0), make_key(1) };
+ selector.options.stride = stride;
+
+ auto txn = lfdb::make_transaction(j);
+ auto window = ceph::libfdb::detail::read_query_window(*txn, selector, 1);
+ const auto out = decode_raw_fdb_pairs(window.result_pairs);
+
+ REQUIRE(1 == out.size());
+ CHECK_FALSE(window.more_available);
+ CHECK(make_key(0) == out.front().first);
+ }
+}
+
+TEST_CASE("generate_FDB_pairs", "[fdb]")
+{
+ janitor j;
+
+ const std::size_t stride = 5;
+ const std::size_t nkeys = 12;
+
+ const auto kvs_in = write_monotonic_kvs(j, nkeys);
+
+ auto collect_pages = [&j](auto selector) {
+ auto txn = lfdb::make_transaction(j);
+
+ return ceph::libfdb::detail::generate_FDB_pairs(*txn, selector)
+ | std::views::transform(decode_raw_fdb_pairs)
+ | std::ranges::to<std::vector<std::vector<string_pair>>>();
+ };
+
+ SECTION("drains paged forward windows") {
+ auto selector = lfdb::select { make_key(0), make_key(nkeys) };
+ selector.options.stride = stride;
+
+ const auto pages = collect_pages(selector);
+ const auto out = pages | std::views::join | std::ranges::to<std::vector<string_pair>>();
+
+ CAPTURE(pages.size());
+ CAPTURE(out.size());
+ CHECK(1 < pages.size());
+ CHECK(std::ranges::all_of(pages, [stride](const auto& page) {
+ return page.size() <= stride;
+ }));
+ CHECK_THAT(out, Catch::Matchers::RangeEquals(first_keys(kvs_in, nkeys)));
+ }
+
+ SECTION("drains paged reverse windows") {
+ auto selector = lfdb::select { make_key(0), make_key(nkeys) };
+ selector.options.stride = stride;
+ selector.options.reverse_order = true;
+
+ const auto pages = collect_pages(selector);
+ const auto out = pages | std::views::join | std::ranges::to<std::vector<string_pair>>();
+
+ CAPTURE(pages.size());
+ CAPTURE(out.size());
+ CHECK(1 < pages.size());
+ CHECK(std::ranges::all_of(pages, [stride](const auto& page) {
+ return page.size() <= stride;
+ }));
+ CHECK_THAT(out, Catch::Matchers::RangeEquals(last_keys_reversed(kvs_in, nkeys)));
+ }
+}
+
TEST_CASE("basic generators", "[fdb]") {
janitor j;
const unsigned nkeys = GENERATE(0, 1, 2, 3, 10, 100, 1'000);
- auto dbh = j.dbh();
-
- const auto kvs_in = write_monotonic_kvs(dbh, nkeys);
+ const auto kvs_in = write_monotonic_kvs(j, nkeys);
REQUIRE(nkeys == kvs_in.size());
SECTION("pair_generator forward") {
std::vector<std::pair<std::string, std::string>> out;
+ auto txn = lfdb::make_transaction(j);
// pair_generator returns key-value pairs:
- for(auto&& kvp : lfdb::pair_generator(dbh, lfdb::select { make_key(0), make_key(nkeys) }))
+ for (auto&& kvp : lfdb::pair_generator(txn, lfdb::select { make_key(0), make_key(nkeys) }))
out.emplace_back(std::move(kvp));
CAPTURE(nkeys);
REQUIRE(nkeys == out.size());
// Be sure we captured the head and the tail:
- if(0 < nkeys) {
+ if (0 < nkeys) {
CAPTURE(out.front().first);
CAPTURE(out.back().first);
CHECK(make_key(0) == out.front().first);
selector.options.reverse_order = true;
std::vector<std::pair<std::string, std::string>> out;
- std::ranges::copy(lfdb::pair_generator(dbh, selector), std::back_inserter(out));
+ auto txn = lfdb::make_transaction(j);
+ std::ranges::copy(lfdb::pair_generator(txn, selector), std::back_inserter(out));
CAPTURE(nkeys);
CAPTURE(out.size());
REQUIRE(nkeys == out.size());
- if(0 < nkeys) {
+ if (0 < nkeys) {
CAPTURE(out.front().first);
CAPTURE(out.back().first);
CHECK(make_key(nkeys - 1) == out.front().first);
selector.options.stride = 5; // one of the most prime of prime numbers
std::vector<std::pair<std::string, std::string>> out;
- std::ranges::copy(lfdb::pair_generator(dbh, selector), std::back_inserter(out));
+ auto txn = lfdb::make_transaction(j);
+ std::ranges::copy(lfdb::pair_generator(txn, selector), std::back_inserter(out));
CAPTURE(nkeys);
CAPTURE(out.size());
REQUIRE(nkeys == out.size());
- if(0 < nkeys) {
+ if (0 < nkeys) {
CAPTURE(out.front().first);
CAPTURE(out.back().first);
CHECK(make_key(0) == out.front().first);
selector.options.stride = 5; // one of the most prime of prime numbers
std::vector<std::pair<std::string, std::string>> out;
- std::ranges::copy(lfdb::pair_generator(dbh, selector), std::back_inserter(out));
+ auto txn = lfdb::make_transaction(j);
+ std::ranges::copy(lfdb::pair_generator(txn, selector), std::back_inserter(out));
CAPTURE(nkeys);
CAPTURE(out.size());
REQUIRE(nkeys == out.size());
- if(0 < nkeys) {
+ if (0 < nkeys) {
CAPTURE(out.front().first);
CAPTURE(out.back().first);
CHECK(make_key(nkeys - 1) == out.front().first);
// The transaction handle local to this lambda must remain alive inside the
// generator's coroutine frame after the lambda returns:
- auto gen = [&dbh, selector] {
- auto txn = lfdb::make_transaction(dbh);
+ auto gen = [&j, selector] {
+ auto txn = lfdb::make_transaction(j);
return lfdb::pair_generator(txn, selector);
}();
CAPTURE(out.size());
REQUIRE(nkeys == out.size());
- if(0 < nkeys) {
+ if (0 < nkeys) {
CHECK(out.contains(make_key(0)));
CHECK(out.contains(make_key(nkeys - 1)));
}
}
}
+TEST_CASE("generators honor selector endpoints", "[fdb]") {
+ janitor j;
+
+ constexpr auto prefix = "generator-selector";
+ write_monotonic_kvs(j, 10, prefix);
+
+ auto collect_pair_keys = [&j](lfdb::select selector) {
+ std::vector<std::string> out;
+ auto txn = lfdb::make_transaction(j);
+ std::ranges::copy(lfdb::pair_generator(txn, selector)
+ | std::views::transform([](const auto& kv) { return kv.first; }),
+ std::back_inserter(out));
+ return out;
+ };
+
+ auto collect_block_keys = [&j](lfdb::select selector) {
+ std::vector<std::string> out;
+ std::ranges::for_each(lfdb::block_generator(j, selector), [&out](const auto& block) {
+ std::ranges::copy(block | std::views::transform([](const auto& kv) { return kv.first; }),
+ std::back_inserter(out));
+ });
+ return out;
+ };
+
+ auto selector = lfdb::select { lfdb::exclusive(make_key(3, prefix)), lfdb::inclusive(make_key(6, prefix)) };
+ selector.options.stride = 2;
+
+ const auto forward_keys = std::vector { make_key(4, prefix), make_key(5, prefix), make_key(6, prefix) };
+ CHECK_THAT(collect_pair_keys(selector), Catch::Matchers::RangeEquals(forward_keys));
+ CHECK_THAT(collect_block_keys(selector), Catch::Matchers::RangeEquals(forward_keys));
+
+ selector.options.reverse_order = true;
+
+ const auto reverse_keys = std::vector { make_key(6, prefix), make_key(5, prefix), make_key(4, prefix) };
+ CHECK_THAT(collect_pair_keys(selector), Catch::Matchers::RangeEquals(reverse_keys));
+ CHECK_THAT(collect_block_keys(selector), Catch::Matchers::RangeEquals(reverse_keys));
+}
+
TEMPLATE_PRODUCT_TEST_CASE("associative data", "[fdb][rgw]",
(std::map, std::unordered_map, boost::container::flat_map), ((std::string, std::string)))
{
janitor j;
- auto dbh = j.dbh();
-
TestType kvs{
{ "hello", "world" },
{ "lorem", "ipsum" },
// From the "database" point of view, the structure is now that we have a single
// key pointing (p) to an associative array, e.g. map<p, map<k, v>>:
- lfdb::set(lfdb::make_transaction(dbh), "key", kvs, lfdb::commit_after_op::commit);
+ const auto key = test_key("key");
+ lfdb::set(lfdb::make_transaction(j), key, kvs, lfdb::commit_after_op::commit);
TestType out_kvs;
- lfdb::get(lfdb::make_transaction(dbh), "key", out_kvs, lfdb::commit_after_op::no_commit);
+ lfdb::get(lfdb::make_transaction(j), key, out_kvs, lfdb::commit_after_op::no_commit);
CHECK(pearl_msg == out_kvs["pearl"]);
}
{
janitor j;
- auto dbh = j.dbh();
-
- std::string_view k = "hi", v = "there";
+ const auto k = test_key("hi");
+ std::string_view v = "there";
CAPTURE(k);
CAPTURE(v);
SECTION("implicitly create and complete transactions") {
- REQUIRE_FALSE(lfdb::key_exists(dbh, k));
- CHECK_NOTHROW(lfdb::set(dbh, k, v));
- CHECK(lfdb::key_exists(dbh, k));
+ REQUIRE_FALSE(lfdb::key_exists(j, k));
+ CHECK_NOTHROW(lfdb::set(j, k, v));
+ CHECK(lfdb::key_exists(j, k));
std::string ov;
- CHECK(lfdb::get(dbh, k, ov));
+ CHECK(lfdb::get(j, k, ov));
CAPTURE(ov);
REQUIRE(v == ov);
- CHECK_NOTHROW(lfdb::erase(dbh, k));
- REQUIRE_FALSE(lfdb::key_exists(dbh, k));
+ CHECK_NOTHROW(lfdb::erase(j, k));
+ REQUIRE_FALSE(lfdb::key_exists(j, k));
- REQUIRE_FALSE(lfdb::get(dbh, k, ov));
+ REQUIRE_FALSE(lfdb::get(j, k, ov));
}
SECTION("implicitly create and complete transactions-- selection operations") {
// With an implicit transaction, mutating transactions should commit by default:
const auto selector = lfdb::select { make_key(0), make_key(20) };
- const auto kvs = write_monotonic_kvs(dbh, 20);
+ const auto kvs = write_monotonic_kvs(j, 20);
- lfdb::erase(dbh, lfdb::select { make_key(1), make_key(6) });
+ lfdb::erase(j, lfdb::select { make_key(1), make_key(6) });
- CHECK(15 == key_count(dbh, selector));
+ CHECK(15 == key_count(j, selector));
// Let's look around the edge cases of the selection:
- CHECK_FALSE(lfdb::key_exists(dbh, make_key(1)));
- CHECK_FALSE(lfdb::key_exists(dbh, make_key(5)));
+ CHECK_FALSE(lfdb::key_exists(j, make_key(1)));
+ CHECK_FALSE(lfdb::key_exists(j, make_key(5)));
- CHECK(lfdb::key_exists(dbh, make_key(0)));
- CHECK(lfdb::key_exists(dbh, make_key(6)));
+ CHECK(lfdb::key_exists(j, make_key(0)));
+ CHECK(lfdb::key_exists(j, make_key(6)));
}
SECTION("test behavior with shared transaction") {
SECTION("write in uncommitted transaction") {
using lfdb::commit_after_op;
- auto txn = lfdb::make_transaction(dbh);
+ auto txn = lfdb::make_transaction(j);
- lfdb::set(txn, "Herman", "Hollerith", commit_after_op::no_commit);
+ const auto herman = test_key("Herman");
+ const auto john = test_key("John");
+
+ lfdb::set(txn, herman, "Hollerith", commit_after_op::no_commit);
// Key exists with respect to this transaction:
- CHECK(lfdb::key_exists(txn, "Herman"));
+ CHECK(lfdb::key_exists(txn, herman));
- lfdb::set(txn, "John", "Backus", commit_after_op::no_commit);
+ lfdb::set(txn, john, "Backus", commit_after_op::no_commit);
// Key exists with respect to this transaction:
- CHECK(lfdb::key_exists(txn, "John", commit_after_op::no_commit));
+ CHECK(lfdb::key_exists(txn, john, commit_after_op::no_commit));
// transaction is abandoned
}
// These were only set in the abandoned transaction:
- CHECK_FALSE(lfdb::key_exists(dbh, "Herman"));
- CHECK_FALSE(lfdb::key_exists(dbh, "John"));
+ CHECK_FALSE(lfdb::key_exists(j, test_key("Herman")));
+ CHECK_FALSE(lfdb::key_exists(j, test_key("John")));
}
SECTION("round trip") {
- janitor j(dbh);
+ janitor scoped_j(j);
using namespace ceph::libfdb;
- set(dbh, "key_0000", "value");
+ const auto key = test_key("key_0000");
+ set(j, key, "value");
std::string out;
- get(dbh, "key_0000", out);
+ get(j, key, out);
CHECK("value" == out);
}
// works around this so that the "right" thing to do is what gets done, with
// performance-maximzation left as an available, but explicit operation.
- janitor j(dbh);
+ janitor scoped_j(j);
using namespace ceph::libfdb;
// Notice the raw literal going in here:
- set(dbh, "key_0000", "value");
+ const auto key = test_key("key_0000");
+ set(j, key, "value");
std::string out;
- CHECK_NOTHROW(get(dbh, "key_0000", out));
+ CHECK_NOTHROW(get(j, key, out));
CHECK(std::string_view("value") == std::string_view(out));
// Explicit raw buffers:
char out_buffer[9] = {};
- CHECK_NOTHROW(get(dbh, "key_0000", out_buffer));
+ CHECK_NOTHROW(get(j, key, out_buffer));
CHECK(std::string_view("value") == std::string_view(out));
}
{
janitor j;
- auto dbh = j.dbh();
-
SECTION("transaction function returns nothing") {
- auto txr = lfdb::make_transactor(dbh);
+ auto txr = lfdb::make_transactor(j);
+ const auto key = test_key("key");
- txr([](auto txn) {
- lfdb::set(txn, "key", "value");
+ txr([&key](auto txn) {
+ lfdb::set(txn, key, "value");
});
std::string out;
- CHECK(lfdb::get(dbh, "key", out));
+ CHECK(lfdb::get(j, key, out));
CHECK("value" == out);
}
SECTION("transaction function returns value") {
- auto txr = lfdb::make_transactor(dbh);
+ auto txr = lfdb::make_transactor(j);
+ const auto key = test_key("key");
- auto [found, out] = txr([](auto txn) {
- lfdb::set(txn, "key", "value");
+ auto [found, out] = txr([&key](auto txn) {
+ lfdb::set(txn, key, "value");
std::string out;
- auto found = lfdb::get(txn, "key", out);
+ auto found = lfdb::get(txn, key, out);
return std::pair(found, out);
});
{ FDB_TR_OPTION_READ_YOUR_WRITES_DISABLE, lfdb::option_flag }
};
- auto txr = lfdb::make_transactor(dbh, opts);
+ auto txr = lfdb::make_transactor(j, opts);
+ const auto key = test_key("key");
- txr([](auto txn) {
- lfdb::set(txn, "key", "value");
+ txr([&key](auto txn) {
+ lfdb::set(txn, key, "value");
});
std::string out;
- CHECK(lfdb::get(dbh, "key", out));
+ CHECK(lfdb::get(j, key, out));
CHECK("value" == out);
}
SECTION("transactor replays after conflict") {
- auto txr = lfdb::make_transactor(dbh);
+ auto txr = lfdb::make_transactor(j);
+ const auto key = test_key("key");
- lfdb::set(dbh, "key", "initial");
+ lfdb::set(j, key, "initial");
- txr([dbh](auto txn) {
+ txr([&j, &key](auto txn) {
std::string out;
- if(not lfdb::get(txn, "key", out)) {
+ if (not lfdb::get(txn, key, out)) {
throw std::runtime_error("expected key does not exist");
}
// Force a conflict, making the transactor replay the body:
- if("initial" == out) {
- lfdb::set(dbh, "key", "conflict");
+ if ("initial" == out) {
+ lfdb::set(j, key, "conflict");
}
- lfdb::set(txn, "key", "final");
+ lfdb::set(txn, key, "final");
});
std::string out;
- CHECK(lfdb::get(dbh, "key", out));
+ CHECK(lfdb::get(j, key, out));
CHECK("final" == out);
}
SECTION("transactor propagates transaction body exceptions") {
- auto txr = lfdb::make_transactor(dbh);
+ auto txr = lfdb::make_transactor(j);
CHECK_THROWS_WITH(txr([](auto) {
throw std::runtime_error("transaction body failed");
{ "displayName", "John Doe" }
};
- // In a "real" program, it would be "auto dbh = lfdb::database_handle(...)":
- auto dbh = j.dbh();
-
// This write will make and commit its own transaction:
- lfdb::set(dbh, "bucket_obj", bucket_entries);
+ const auto key = test_key("bucket_obj");
+ lfdb::set(j, key, bucket_entries);
map<string, string> out;
- lfdb::get(dbh, "bucket_obj", out);
+ lfdb::get(j, key, out);
// For "demo" purposes, you can ignore everything below here:
CAPTURE(out["userId"]);
REQUIRE(bucket_entries == out);
- j.drop_all();
+ j.drop_test_namespace();
}
+TEST_CASE("block_generator should correctly handle value types") {
+
+ SECTION("wrangle the unwrangled!") {
+ janitor dbh;
+
+ using person = std::map<std::string, std::string>; // person => things about a person
+
+ auto check_generators = [&dbh]<typename ValueT>(std::string_view prefix,
+ const std::map<std::string, ValueT>& values) {
+ auto key_for = [prefix](std::string_view name) {
+ return test_key(fmt::format("{}/{}", prefix, name));
+ };
+
+ lfdb::make_transactor(dbh)([&values, &key_for](auto txn) {
+ std::ranges::for_each(values, [&txn, &key_for](const auto& individual) {
+ lfdb::set(txn, key_for(individual.first), individual.second);
+ });
+ });
+
+ const std::map<std::string, ValueT> expected {
+ { key_for("Alice"), values.at("Alice") },
+ { key_for("Bob"), values.at("Bob") }
+ };
+
+ auto selector = lfdb::select { key_for("A"), key_for("C") };
+
+ std::map<std::string, ValueT> from_pairs;
+ auto txn = lfdb::make_transaction(dbh);
+ std::ranges::copy(lfdb::pair_generator<ValueT>(txn, selector),
+ std::inserter(from_pairs, std::end(from_pairs)));
+
+ REQUIRE(from_pairs == expected);
+
+ std::map<std::string, ValueT> from_blocks;
+ for (auto&& block : lfdb::block_generator<ValueT>(dbh, selector)) {
+ from_blocks.insert(std::begin(block), std::end(block));
+ }
+
+ REQUIRE(from_blocks == expected);
+ REQUIRE(from_blocks == from_pairs);
+ };
+
+ check_generators("generator-values/string",
+ std::map<std::string, std::string> {
+ { "Alice", "boysenberry" },
+ { "Bob", "coconut" },
+ { "X", "coconut" },
+ { "Y", "coconut" }
+ });
+
+ check_generators("generator-values/bytes",
+ std::map<std::string, std::vector<std::uint8_t>> {
+ { "Alice", { 1, 2, 3 } },
+ { "Bob", { 4, 5, 6 } },
+ { "X", { 7, 8, 9 } },
+ { "Y", { 10, 11, 12 } }
+ });
+
+ check_generators("generator-values/person",
+ std::map<std::string, person> {
+ { "Alice", {
+ { "name", "Alice" },
+ { "ice_cream", "boysenberry" }
+ } },
+ { "Bob", {
+ { "name", "Bob" },
+ { "ice_cream", "coconut" }
+ } },
+ { "X", {
+ { "name", "X" },
+ { "ice_cream", "coconut" }
+ } },
+ { "Y", {
+ { "name", "Y" },
+ { "ice_cream", "coconut" }
+ } }
+ });
+ }
+}
+
+
// Adapted from Catch2 documentation:
#include <catch2/catch_session.hpp>
SECTION("check write and cleanup") {
janitor j;
- j.drop_all_keys();
+ j.drop_key_prefix("VEAL");
// Check that there are no keys:
std::string s;
lfdb::get(j, make_key(2, "VEAL"), s);
REQUIRE(make_value(2) == s);
- j.drop_all_keys();
+ j.drop_key_prefix("VEAL");
}
}
const char *msg = "Hello, World!";
- auto dbh = lfdb::create_database();
+ janitor j;
+ auto dbh = j.dbh();
// ceph::buffer::list -> span<uint8_t> -> std::string
{
ceph::buffer::list o;
ceph::libfdb::from::convert(x, o);
- lfdb::set(lfdb::make_transaction(dbh), "key", n, lfdb::commit_after_op::commit);
- lfdb::get(lfdb::make_transaction(dbh), "key", o);
+ const auto key = test_key("key");
+ lfdb::set(lfdb::make_transaction(dbh), key, n, lfdb::commit_after_op::commit);
+ lfdb::get(lfdb::make_transaction(dbh), key, o);
REQUIRE_THAT(n, Catch::Matchers::RangeEquals(o));
- lfdb::erase(dbh, "key");
+ lfdb::erase(dbh, key);
}
SECTION("buffer::list direct and const conversion paths match")
{
ceph::buffer::list in;
- lfdb::set(dbh, "empty-buffer-list", in);
+ const auto key = test_key("empty-buffer-list");
+ lfdb::set(dbh, key, in);
ceph::buffer::list out;
- REQUIRE(lfdb::get(dbh, "empty-buffer-list", out));
+ REQUIRE(lfdb::get(dbh, key, out));
CHECK(0 == out.length());
- lfdb::erase(dbh, "empty-buffer-list");
+ lfdb::erase(dbh, key);
}
SECTION("buffer::list with embedded nulls round trips through public set/get")
ceph::buffer::list in;
in.append(data, sizeof(data));
- lfdb::set(dbh, "embedded-null-buffer-list", in);
+ const auto key = test_key("embedded-null-buffer-list");
+ lfdb::set(dbh, key, in);
ceph::buffer::list out;
- REQUIRE(lfdb::get(dbh, "embedded-null-buffer-list", out));
+ REQUIRE(lfdb::get(dbh, key, out));
REQUIRE_THAT(in, Catch::Matchers::RangeEquals(out));
- lfdb::erase(dbh, "embedded-null-buffer-list");
+ lfdb::erase(dbh, key);
}
SECTION("buffer::list get replaces existing output")
ceph::buffer::list out;
out.append("old-value");
- lfdb::set(dbh, "replace-buffer-list-output", in);
+ const auto key = test_key("replace-buffer-list-output");
+ lfdb::set(dbh, key, in);
- REQUIRE(lfdb::get(dbh, "replace-buffer-list-output", out));
+ REQUIRE(lfdb::get(dbh, key, out));
REQUIRE_THAT(in, Catch::Matchers::RangeEquals(out));
- lfdb::erase(dbh, "replace-buffer-list-output");
+ lfdb::erase(dbh, key);
}
SECTION("buffer::list (and buffer::list key) -> buffer::list")
ceph::buffer::list o;
o.append(n);
- lfdb::set(lfdb::make_transaction(dbh), "key", n, lfdb::commit_after_op::commit);
- lfdb::get(lfdb::make_transaction(dbh), "key", o);
+ const auto key = test_key("key");
+ lfdb::set(lfdb::make_transaction(dbh), key, n, lfdb::commit_after_op::commit);
+ lfdb::get(lfdb::make_transaction(dbh), key, o);
REQUIRE_THAT(n, Catch::Matchers::RangeEquals(o));
- lfdb::erase(dbh, "key");
+ lfdb::erase(dbh, key);
}
SECTION("buffer::ptr")
ceph::buffer::ptr p(ceph::buffer::claim_char(in.length(), const_cast<char *>(in.data())));
CHECK(in == string_view(p.c_str(), p.length()));
- lfdb::set(dbh, "key", p);
+ const auto key = test_key("key");
+ lfdb::set(dbh, key, p);
std::string o;
- lfdb::get(dbh, "key", o);
+ lfdb::get(dbh, key, o);
CHECK(in == o);
}
[dbh](const ceph::libfdb::select& selector) mutable {
AssocT out;
+ auto txn = ceph::libfdb::make_transaction(dbh);
// ...my libstdc++ lacks std::from_range overloads; the Hard Way(TM) it is:
- for(auto&& kvp : ceph::libfdb::pair_generator(dbh, selector)) {
+ for(auto&& kvp : ceph::libfdb::pair_generator(txn, selector)) {
out.emplace(kvp);
}
populate_monotonic(j, nkeys);
- SECTION("forward") {
- auto selector = lfdb::select { "key_" };
-
+ struct collected_blocks
+ {
+ std::vector<std::pair<std::string, std::string>> entries;
+ size_t nblocks = 0;
size_t total = 0;
- for(const auto& block : lfdb::block_generator(j, selector)) {
- total += block.size();
- }
+ };
- CAPTURE(nkeys);
- CAPTURE(total);
- CHECK(nkeys == total);
- }
+ auto collect_blocks = [&j](lfdb::select selector) {
+ collected_blocks out;
+ const auto requested_stride = selector.options.stride;
- SECTION("reverse") {
- auto selector = lfdb::select { "key_" };
- selector.options.reverse_order = true;
+ std::ranges::for_each(lfdb::block_generator<std::string, ordered_block>(j, selector), [&](const auto& block) {
+ ++out.nblocks;
+ out.total += block.size();
+
+ if (0 < requested_stride) {
+ CHECK(block.size() <= static_cast<size_t>(requested_stride));
+ }
- std::vector<std::pair<std::string, std::string>> out;
+ std::ranges::copy(block, std::back_inserter(out.entries));
+ });
- for(const auto& block : lfdb::block_generator<ordered_block>(j, selector)) {
- std::ranges::copy(block, std::back_inserter(out));
+ return out;
+ };
+
+ auto check_paged = [nkeys](const collected_blocks& blocks, const lfdb::select& selector) {
+ if (selector.options.stride < static_cast<int>(nkeys)) {
+ CHECK(1 < blocks.nblocks);
}
+ };
+ auto check_reverse = [nkeys](const auto& out) {
CAPTURE(nkeys);
CAPTURE(out.size());
REQUIRE(nkeys == out.size());
- if(0 < nkeys) {
+ if (0 < nkeys) {
CAPTURE(out.front().first);
CAPTURE(out.back().first);
CHECK(make_key(nkeys - 1) == out.front().first);
CHECK(std::ranges::is_sorted(out, std::ranges::greater {},
&std::pair<std::string, std::string>::first));
}
+ };
+
+ SECTION("forward") {
+ auto selector = lfdb::select { make_key_prefix() };
+ auto blocks = collect_blocks(selector);
+
+ CAPTURE(nkeys);
+ CAPTURE(blocks.total);
+ CHECK(nkeys == blocks.total);
+ }
+
+ SECTION("forward, paged") {
+ auto selector = lfdb::select { make_key_prefix() };
+ selector.options.stride = 5;
+ auto blocks = collect_blocks(selector);
+
+ CAPTURE(nkeys);
+ CAPTURE(blocks.total);
+ CAPTURE(blocks.nblocks);
+ CHECK(nkeys == blocks.total);
+ check_paged(blocks, selector);
+ }
+
+ SECTION("reverse") {
+ auto selector = lfdb::select { make_key_prefix() };
+ selector.options.reverse_order = true;
+ auto blocks = collect_blocks(selector);
+
+ check_reverse(blocks.entries);
+ }
+
+ SECTION("reverse, paged") {
+ auto selector = lfdb::select { make_key_prefix() };
+ selector.options.reverse_order = true;
+ selector.options.stride = 5;
+ auto blocks = collect_blocks(selector);
+
+ CAPTURE(blocks.nblocks);
+ check_reverse(blocks.entries);
+ check_paged(blocks, selector);
}
}
}
populate_monotonic(j, nkeys);
- auto selector = lfdb::select { "key_" };
+ auto selector = lfdb::select { make_key_prefix() };
size_t total = 0;
meter.measure([&j, &selector, &total] {
total = 0;
- for(const auto& block : lfdb::block_generator(j, selector)) {
+ for (const auto& block : lfdb::block_generator(j, selector)) {
total += block.size();
}
});
janitor dbh;
populate_monotonic(dbh, nkeys);
- const auto selector = lfdb::select { "key_" };
+ const auto selector = lfdb::select { make_key_prefix() };
// Track enough work to prevent the benchmarked reads from being optimized away:
struct {
bytes += value.size();
}
+ void add_raw(const FDBKeyValue& kv)
+ {
+ ++total;
+ bytes += kv.value_length;
+ }
+
} read_tally;
auto mark_failure = [&read_tally](const std::exception& e) {
read_tally.add(kv.second);
};
+ auto add_raw_pairs = [&read_tally](std::span<const FDBKeyValue> raw_pairs) {
+ std::ranges::for_each(raw_pairs, [&read_tally](const FDBKeyValue& kv) {
+ read_tally.add_raw(kv);
+ });
+ };
+
auto key_indices = [nkeys] {
return std::views::iota(0u, static_cast<unsigned>(nkeys));
};
- auto expected_bytes = [&] {
+ auto expected_decoded_bytes = [&] {
size_t bytes = 0;
std::ranges::for_each(key_indices(), [&bytes](const auto n) {
return bytes;
}();
- auto require_expected = [nkeys, expected_bytes](const auto& tally) {
- if(not tally.ok) {
+ auto expected_encoded_bytes = [&] {
+ size_t bytes = 0;
+
+ std::ranges::for_each(key_indices(), [&bytes](const auto n) {
+ bytes += ceph::libfdb::to::convert(make_value(n)).size();
+ });
+
+ return bytes;
+ }();
+
+ auto require_expected = [nkeys](const auto& tally, const size_t expected_bytes) {
+ if (not tally.ok) {
WARN(tally.error);
return;
}
try {
auto txn = lfdb::make_transaction(dbh);
- for(const auto n : key_indices()) {
+ for (const auto n : key_indices()) {
std::string out;
if (lfdb::get(txn, make_key(n), out, lfdb::commit_after_op::no_commit)) {
read_tally.add(out);
}
}
- } catch(const ceph::libfdb::libfdb_exception& e) {
+ } catch (const ceph::libfdb::libfdb_exception& e) {
mark_failure(e);
}
});
- require_expected(read_tally);
+ require_expected(read_tally, expected_decoded_bytes);
};
BENCHMARK_ADVANCED("single-key get, implicit transaction per key")(Catch::Benchmark::Chronometer meter) {
read_tally.reset();
try {
- for(const auto n : key_indices()) {
+ for (const auto n : key_indices()) {
std::string out;
if (lfdb::get(dbh, make_key(n), out)) {
read_tally.add(out);
}
}
- } catch(const ceph::libfdb::libfdb_exception& e) {
+ } catch (const ceph::libfdb::libfdb_exception& e) {
+ mark_failure(e);
+ }
+ });
+
+ require_expected(read_tally, expected_decoded_bytes);
+ };
+
+ BENCHMARK_ADVANCED("raw FDB pair windows read all")(Catch::Benchmark::Chronometer meter) {
+ meter.measure([&] {
+ read_tally.reset();
+
+ try {
+ auto txn = lfdb::make_transaction(dbh);
+ std::ranges::for_each(ceph::libfdb::detail::generate_FDB_pairs(*txn, selector),
+ add_raw_pairs);
+ } catch (const ceph::libfdb::libfdb_exception& e) {
mark_failure(e);
}
});
- require_expected(read_tally);
+ require_expected(read_tally, expected_encoded_bytes);
};
BENCHMARK_ADVANCED("pair generator read all")(Catch::Benchmark::Chronometer meter) {
read_tally.reset();
try {
- std::ranges::for_each(lfdb::pair_generator(dbh, selector), add_kv);
- } catch(const ceph::libfdb::libfdb_exception& e) {
+ auto txn = lfdb::make_transaction(dbh);
+ std::ranges::for_each(lfdb::pair_generator(txn, selector), add_kv);
+ } catch (const ceph::libfdb::libfdb_exception& e) {
mark_failure(e);
}
});
- require_expected(read_tally);
+ require_expected(read_tally, expected_decoded_bytes);
};
BENCHMARK_ADVANCED("block generator read all")(Catch::Benchmark::Chronometer meter) {
std::ranges::for_each(lfdb::block_generator(dbh, selector), [&](const auto& block) {
std::ranges::for_each(block, add_kv);
});
- } catch(const ceph::libfdb::libfdb_exception& e) {
+ } catch (const ceph::libfdb::libfdb_exception& e) {
mark_failure(e);
}
});
- require_expected(read_tally);
+ require_expected(read_tally, expected_decoded_bytes);
};
}
inputs.reserve(N);
// Sadly, my version of libstdc++ lacks std::from_range_t:
-for(auto n : views::iota(0, N)) {
+for (auto n : views::iota(0, N)) {
inputs.emplace_back(std::make_pair<std::string, std::string>(make_key(n), make_value(n)));
}
-if(run_slow_baselines) {
+if (run_slow_baselines) {
BENCHMARK_ADVANCED("write simple records-- implicit transaction per key")(Catch::Benchmark::Chronometer meter) {
janitor j;
meter.measure([&j, &inputs, stride] {
// Notice that here we are passing around the database handle, winding up making a new one per-operation:
- for(auto block : inputs | views::chunk(stride)) {
+ for (auto block : inputs | views::chunk(stride)) {
std::for_each(std::begin(block), std::end(block), [&j](const auto& kv) mutable {
lfdb::set(j, kv.first, kv.second);
});
meter.measure([&j, &inputs, stride] {
// Shared-transaction:
- for(auto block : inputs | views::chunk(stride)) {
+ for (auto block : inputs | views::chunk(stride)) {
auto txn = lfdb::make_transaction(j);
std::for_each(std::begin(block), std::end(block), [&txn](const auto& kv) mutable {
lfdb::set(txn, kv.first, kv.second);
});
- if(false == lfdb::commit(txn)) {
+ if (false == lfdb::commit(txn)) {
throw std::runtime_error("unable to commit transaction");
}
}
janitor j;
meter.measure([&j, &inputs, stride] {
- for(auto block : inputs | views::chunk(stride)) {
+ for (auto block : inputs | views::chunk(stride)) {
lfdb::set(j, std::begin(block), std::end(block));
}
});
lfdb::set(txn, kv.first, kv.second);
});
- if(false == lfdb::commit(txn)) {
+ if (false == lfdb::commit(txn)) {
throw std::runtime_error("unable to commit transaction");
}
});
#include "rgw/fdb/fdb.h"
#include <algorithm>
+#include <chrono>
#include <map>
#include <ranges>
#include <stdexcept>
#include <string>
#include <string_view>
+#include <unistd.h>
+
namespace lfdb = ceph::libfdb;
constexpr const char* const msg = "Hello, World!";
Ne proved I never her precios pere.
)";
-// It's a little moribund, but for tests we use this highly-creative (*cough!*) prefix
-// to make it easy to not only remove stale keys but diagnose them when there's a problem:
+// Tests use a unique process-local namespace so different test binaries can share an FDB
+// backing store without deleting or reading each other's keys.
+inline std::string test_namespace_prefix()
+{
+ static const auto prefix = fmt::format("ceph-libfdb-test/{}/{}/",
+ ::getpid(),
+ std::chrono::steady_clock::now().time_since_epoch().count());
+ return prefix;
+}
+
+inline std::string test_key(std::string_view key)
+{
+ return fmt::format("{}{}", test_namespace_prefix(), key);
+}
+
+inline std::string make_key_prefix(std::string_view prefix = "key")
+{
+ return fmt::format("{}{}_", test_namespace_prefix(), prefix);
+}
+
inline std::string make_key(const int n, std::string_view prefix = "key")
{
- return fmt::format("{}_{:010d}", prefix, n);
+ return fmt::format("{}{:010d}", make_key_prefix(prefix), n);
}
inline std::string make_value(const int n)
{
std::map<std::string, std::string> kvs;
- for(const auto i : std::ranges::iota_view(0u, N)) {
+ for (const auto i : std::ranges::iota_view(0u, N)) {
kvs.insert({ make_key(i, prefix), make_value(i) });
}
return kvs;
}
-// Gadget to help clean up test keys-- both removing stale ones when we enter and also tidying
-// up after ourselves when we leave scope:
+// Gadget to help clean up this test process's key namespace when we enter and leave scope:
struct janitor final
{
ceph::libfdb::database_handle dbh_;
// flip this off if you need artifacts after debugging:
bool drop_after_scope = true;
- janitor(ceph::libfdb::database_handle dbh_)
- : dbh_(dbh_)
+ janitor(ceph::libfdb::database_handle dbh)
+ : dbh_(dbh)
{
- drop_all();
+ drop_test_namespace();
}
janitor()
~janitor()
{
- if(drop_after_scope)
- drop_all();
+ if (drop_after_scope) {
+ drop_test_namespace();
+ }
}
public:
ceph::libfdb::database_handle dbh() { return dbh_; }
public:
- // This type coersion turns out to be very useful:
+ // This type coercion turns out to be very useful:
operator ceph::libfdb::database_handle() { return dbh(); }
public:
lfdb::commit_after_op::commit);
}
- // Drop everything in the database (except system keys):
- void drop_all()
+ // Drop everything made by this test process:
+ void drop_test_namespace()
{
- // Note: technically, [0x00, 0xFF) is needed to include the system keys (if the transaction's allowed to
- // access these). However, special permissions are needed to access these magical "system keys" and we
- // probably don't actually want to delete them erroneously. So, we stick with the normal user-key range...
- return drop(lfdb::select { "" });
+ drop(lfdb::select { test_namespace_prefix() });
}
// Drop keys made with our key convention:
- void drop_all_keys()
+ void drop_key_prefix(std::string_view prefix = "key")
{
- return drop(lfdb::select { "key_" });
+ drop(lfdb::select { make_key_prefix(prefix) });
}
};
const auto total = static_cast<unsigned>(N);
const auto nblocks = (total + stride - 1) / stride;
- for(auto block_index : views::iota(0u, nblocks)) {
+ for (auto block_index : views::iota(0u, nblocks)) {
const auto first = block_index * stride;
const auto last = std::min(first + stride, total);
// We make a new transaction each block to avoid session timeout:
auto txn = make_transaction(dbh);
- for(auto n : views::iota(first, last)) {
- lfdb::set(txn, make_key(n, prefix), make_value(n));
+ for (auto n : views::iota(first, last)) {
+ lfdb::set(txn, make_key(n, prefix), make_value(n));
}
- if(false == lfdb::commit(txn)) {
- throw std::runtime_error(fmt::format("unable to commit transaction; {} entries, {} stride", total, stride));
+ if (false == lfdb::commit(txn)) {
+ throw std::runtime_error(fmt::format("unable to commit transaction; {} entries, {} stride", total, stride));
}
}
}
projects / ceph.git / commitdiff