From: Jesse Williamson Date: Wed, 8 Jul 2026 23:31:52 +0000 (-0700) Subject: Fixup for (EXPERIMENTAL) FoundationDB support X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=d21a73039ad0bd2c7418242ed57a7a09a367d0fd;p=ceph.git Fixup for (EXPERIMENTAL) FoundationDB support Better-defined generator semantics; many internal improvements, a couple found bug fixes, minor feature enhancements. Considerable simplification that streamlines and improves the baseline performance of library internals. Updates tests and example documentation. - Add inclusive()/exclusive() selector endpoints. - default is still: select { a, b } == [a, b). - endpoint semantics work for reads, pagination, range-erase, split planning. - tests: selector endpoint combinations; endpoint handling; multiple value types from generators.. - runtime byte-pointer helpers now actually just runtime Assisted-by: Codex:GPT-5 Signed-off-by: Jesse Williamson --- diff --git a/src/rgw/fdb/EXAMPLES.md b/src/rgw/fdb/EXAMPLES.md index 9f1521813a8..e23422f1eea 100644 --- a/src/rgw/fdb/EXAMPLES.md +++ b/src/rgw/fdb/EXAMPLES.md @@ -175,13 +175,43 @@ auto medieval_people = lfdb::select { "person/charlemagne", "person/saladin/" }; ## 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 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: @@ -212,15 +242,18 @@ for (const auto& chunk : keys | std::views::chunk(100)) { ## 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); + } } ``` diff --git a/src/rgw/fdb/base.h b/src/rgw/fdb/base.h index 877fce14778..c6780431889 100644 --- a/src/rgw/fdb/base.h +++ b/src/rgw/fdb/base.h @@ -87,8 +87,10 @@ inline void transaction_set_kv_bytes(const transaction_handle& txn, std::span k, std::span 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> generate_FDB_pairs(ceph::libfdb::transaction& txn, ceph::libfdb::select key_range); @@ -205,12 +207,7 @@ struct future_value final friend class ceph::libfdb::transaction; }; -constexpr auto as_fdb_span(const char *s) -{ - return std::span((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 *)sv.data(), sv.size()); } @@ -239,15 +236,39 @@ inline std::string make_range_end_key_for_prefix(std::string_view prefix) } // 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? @@ -261,8 +282,23 @@ struct select final } 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) @@ -272,7 +308,30 @@ struct select final }; -// 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; @@ -289,19 +348,19 @@ using transaction_options = option_map; 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& 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(&x); } -constexpr const std::uint8_t *data_of(const option_flag_t&) { return nullptr; } +inline const std::uint8_t *data_of(const std::vector& 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(&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& xs) { return static_cast(xs.size()); } constexpr int size_of(const std::string& xs) { return static_cast(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(xs.size()); } inline auto as_fdb_option_args(const auto& option) @@ -329,87 +388,93 @@ inline void apply_options(const auto& option_map, auto&& set_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 db_handle; + struct database_deleter final + { + void operator()(FDBDatabase *db) const noexcept + { + fdb_database_destroy(db); + } + }; + + std::unique_ptr 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) { @@ -431,8 +496,7 @@ class database final 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) { @@ -478,7 +542,8 @@ class transaction final std::unique_ptr txn_handle; private: - bool get_single_value_from_transaction(const std::span& key, std::invocable> auto&& write_output); + bool get_single_value_from_transaction(const std::span& key, + std::invocable> auto&& write_output_fn); public: transaction(database_handle dbh_) @@ -490,8 +555,8 @@ class transaction final : 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); }); } @@ -526,16 +591,19 @@ class transaction final } 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 @@ -588,14 +656,10 @@ inline bool ceph::libfdb::transaction::get_single_value_from_transaction(const s { 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; @@ -616,6 +680,11 @@ inline bool ceph::libfdb::transaction::get_single_value_from_transaction(const s 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) { @@ -633,14 +702,9 @@ inline bool ceph::libfdb::transaction::get_single_value_from_transaction(const s } 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): @@ -670,14 +734,14 @@ namespace ceph::libfdb::detail { 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); } @@ -687,19 +751,145 @@ inline future_value block_until_ready(future_value&& fv) 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 requires std::invocable inline future_value await_future_of(FnT&& fn, XS&& ...params) { return block_until_ready( std::invoke(std::forward(fn), std::forward(params)...)); -} +} + +// Tracks a set of results and their corresponding Future: +struct query_window final +{ + future_value result_owner; + std::span result_pairs; + bool more_available = false; +}; + +struct split_point_result final +{ + future_value result_owner; + std::span 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(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(result_keys, result_count) + : std::span(), + .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 next_range; +}; + +template +inline query_window_result 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(window.result_pairs), + .next_range = next_range_after(std::move(key_range), window) + }; +} template requires std::output_iterator> @@ -711,67 +901,56 @@ inline bool get_value_range_from_transaction(transaction& txn, const select& key return true; } -inline fdb_error_t value_from_db(future_value& fv, - std::invocable 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 as_select_seq(const FDBKey* const xs, - const int n, +inline std::vector as_select_seq(std::span xs, const ceph::libfdb::select& parent) { - std::vector out; - - if (1 < n) { - out.reserve(static_cast(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(fst.key_length) }, - { (const char *)snd.key, static_cast(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(fst.key_length)); + const auto second_key = std::string_view((const char *)snd.key, + static_cast(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>(); } // 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; @@ -781,15 +960,17 @@ inline future_value get_range_future_from_transaction(ceph::libfdb::transaction& // 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 @@ -801,7 +982,7 @@ inline future_value get_range_future_from_transaction(ceph::libfdb::transaction& // 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) @@ -819,64 +1000,50 @@ inline std::vector plan_split_ranges( 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(split_selector.begin_key.length()), + (const std::uint8_t *)split_selector.end_key.data(), static_cast(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(selector.begin_key.length()), - (const std::uint8_t *)selector.end_key.data(), static_cast(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> 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> 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(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 diff --git a/src/rgw/fdb/conversion.h b/src/rgw/fdb/conversion.h index eea6882fe22..3ec66fb37d3 100644 --- a/src/rgw/fdb/conversion.h +++ b/src/rgw/fdb/conversion.h @@ -30,26 +30,12 @@ #include #include -/* 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& out_data) -> std::span @@ -93,9 +79,9 @@ inline void convert(const std::span& from, auto& to) } template OutputFunction> -inline void convert(const std::span& in, OutputFunction& fn) +inline void convert(const std::span& 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 diff --git a/src/rgw/fdb/fdb.h b/src/rgw/fdb/fdb.h index 0165a6c96d1..5042765fb21 100644 --- a/src/rgw/fdb/fdb.h +++ b/src/rgw/fdb/fdb.h @@ -16,8 +16,6 @@ #ifndef CEPH_RGW_FDB_H #define CEPH_RGW_FDB_H -#include "base.h" #include "interface.h" -#include "conversion.h" #endif diff --git a/src/rgw/fdb/interface.h b/src/rgw/fdb/interface.h index eb1059da87a..7ad4183b2b6 100644 --- a/src/rgw/fdb/interface.h +++ b/src/rgw/fdb/interface.h @@ -17,24 +17,9 @@ #include "base.h" #include "conversion.h" -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - 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(); @@ -137,6 +122,10 @@ template auto commit_noreplay(transaction_handle txn, const commit_after_op commit_after, FnT&& fn) -> operation_result_t; +template +auto commit_in_new_transaction(database_handle dbh, FnT&& fn) + -> operation_result_t; + } // namespace ceph::libfdb::detail namespace ceph::libfdb { @@ -162,7 +151,7 @@ inline void set(transaction_handle txn, std::string_view k, const auto& v) 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)); }); @@ -191,7 +180,7 @@ template