Merge pull request ceph#62530 from liu-chunmei/omap_iterate

crimson: implement crimson Omap iterate interface

Reviewed-by: Yingxin Cheng <[email protected]>

Author: Matan-B

src/crimson/os/alienstore/alien_store.cc

@@ -424,42 +424,30 @@ auto AlienStore::omap_get_values(CollectionRef ch,
   });
 }
 
-auto AlienStore::omap_get_values(CollectionRef ch,
-                                 const ghobject_t &oid,
-                                 const std::optional<string> &start,
-				 uint32_t op_flags)
-  -> read_errorator::future<std::tuple<bool, omap_values_t>>
+AlienStore::read_errorator::future<ObjectStore::omap_iter_ret_t>
+AlienStore::omap_iterate(CollectionRef ch,
+                         const ghobject_t &oid,
+                         ObjectStore::omap_iter_seek_t start_from,
+                         omap_iterate_cb_t callback,
+                         uint32_t op_flags)
 {
   logger().debug("{} with_start", __func__);
   assert(tp);
-  return do_with_op_gate(omap_values_t{}, [=, this] (auto &values) {
-    return tp->submit(ch->get_cid().hash_to_shard(tp->size()), [=, this, &values] {
+  return do_with_op_gate(oid, [ch, start_from, callback, this] (auto& oid) {
+    return tp->submit(ch->get_cid().hash_to_shard(tp->size()), [ch, oid, start_from, callback, this] {
       auto c = static_cast<AlienCollection*>(ch.get());
       return store->omap_iterate(
-        c->collection, oid,
-        ObjectStore::omap_iter_seek_t{
-          .seek_position = start.value_or(std::string{}),
-          // FIXME: classical OSDs begins iteration from LOWER_BOUND
-          // (or UPPER_BOUND if filter_prefix > start). However, these
-          // bits are not implemented yet
-          .seek_type = ObjectStore::omap_iter_seek_t::UPPER_BOUND
-        },
-        [&values]
-        (std::string_view key, std::string_view value) mutable {
-          values[std::string{key}].append(value);
-          // FIXME: there is limit on number of entries yet
-          return ObjectStore::omap_iter_ret_t::NEXT;
-        });
-    }).then([&values] (int r)
-      -> read_errorator::future<std::tuple<bool, omap_values_t>> {
+        c->collection, oid, start_from, callback);
+    }).then([] (int r)
+      -> read_errorator::future<ObjectStore::omap_iter_ret_t> {
       if (r == -ENOENT) {
         return crimson::ct_error::enoent::make();
-      } else if (r < 0){
-        logger().error("omap_get_values(start): {}", r);
-        return crimson::ct_error::input_output_error::make();
       } else {
-        return read_errorator::make_ready_future<std::tuple<bool, omap_values_t>>(
-          true, std::move(values));
+        if (r == 1) {
+          return read_errorator::make_ready_future<ObjectStore::omap_iter_ret_t>(ObjectStore::omap_iter_ret_t::STOP);
+        } else {
+          return read_errorator::make_ready_future<ObjectStore::omap_iter_ret_t>(ObjectStore::omap_iter_ret_t::NEXT);
+        }
       }
     });
   });

src/crimson/os/alienstore/alien_store.h

@@ -66,21 +66,20 @@ class AlienStore final : public FuturizedStore,
     const omap_keys_t& keys,
     uint32_t op_flags = 0) final;
 
-  /// Retrieves paged set of values > start (if present)
-  read_errorator::future<std::tuple<bool, omap_values_t>> omap_get_values(
-    CollectionRef c,           ///< [in] collection
-    const ghobject_t &oid,     ///< [in] oid
-    const std::optional<std::string> &start, ///< [in] start, empty for begin
-    uint32_t op_flags = 0
-    ) final; ///< @return <done, values> values.empty() iff done
-
   seastar::future<std::tuple<std::vector<ghobject_t>, ghobject_t>> list_objects(
     CollectionRef c,
     const ghobject_t& start,
     const ghobject_t& end,
     uint64_t limit,
     uint32_t op_flags = 0) const final;
 
+  read_errorator::future<ObjectStore::omap_iter_ret_t> omap_iterate(
+    CollectionRef c,
+    const ghobject_t &oid,
+    ObjectStore::omap_iter_seek_t start_from,
+    omap_iterate_cb_t callback,
+    uint32_t op_flags = 0) final;
+
   seastar::future<CollectionRef> create_new_collection(const coll_t& cid) final;
   seastar::future<CollectionRef> open_collection(const coll_t& cid) final;
   seastar::future<std::vector<coll_core_t>> list_collections() final;

src/crimson/os/cyanstore/cyan_store.cc

@@ -390,29 +390,33 @@ auto CyanStore::Shard::omap_get_values(
   return seastar::make_ready_future<omap_values_t>(std::move(values));
 }
 
-auto CyanStore::Shard::omap_get_values(
+auto CyanStore::Shard::omap_iterate(
   CollectionRef ch,
   const ghobject_t &oid,
-  const std::optional<string> &start,
+  ObjectStore::omap_iter_seek_t start_from,
+  omap_iterate_cb_t callback,
   uint32_t op_flags)
-  -> CyanStore::Shard::read_errorator::future<std::tuple<bool, omap_values_t>>
+  -> CyanStore::Shard::read_errorator::future<ObjectStore::omap_iter_ret_t>
 {
   auto c = static_cast<Collection*>(ch.get());
   logger().debug("{} {} {}", __func__, c->get_cid(), oid);
   auto o = c->get_object(oid);
   if (!o) {
     return crimson::ct_error::enoent::make();
   }
-  omap_values_t values;
-  for (auto i = start ? o->omap.upper_bound(*start) : o->omap.begin();
-       i != o->omap.end();
-       ++i) {
-    values.insert(*i);
+  auto ret = ObjectStore::omap_iter_ret_t::NEXT;
+  auto i = (start_from.seek_type == ObjectStore::omap_iter_seek_t::LOWER_BOUND) ?
+            o->omap.lower_bound(start_from.seek_position) :
+            o->omap.upper_bound(start_from.seek_position);
+  for (; i != o->omap.end(); ++i) {
+    ceph::bufferlist bl = i->second;
+    std::string result(bl.c_str(), bl.length());
+    ret = callback(i->first, result);
+    if (ret == ObjectStore::omap_iter_ret_t::STOP)
+      break;
   }
-  return seastar::make_ready_future<std::tuple<bool, omap_values_t>>(
-    std::make_tuple(true, std::move(values)));
+  return read_errorator::make_ready_future<ObjectStore::omap_iter_ret_t>(ret);
 }
-
 auto CyanStore::Shard::omap_get_header(
   CollectionRef ch,
   const ghobject_t& oid,

src/crimson/os/cyanstore/cyan_store.h

@@ -72,12 +72,13 @@ class CyanStore final : public FuturizedStore {
       const omap_keys_t& keys,
       uint32_t op_flags = 0) final;
 
-    read_errorator::future<std::tuple<bool, omap_values_t>> omap_get_values(
-      CollectionRef c,           ///< [in] collection
-      const ghobject_t &oid,     ///< [in] oid
-      const std::optional<std::string> &start, ///< [in] start, empty for begin
+    read_errorator::future<ObjectStore::omap_iter_ret_t> omap_iterate(
+      CollectionRef c,
+      const ghobject_t &oid,
+      ObjectStore::omap_iter_seek_t start_from,
+      omap_iterate_cb_t callback,
       uint32_t op_flags = 0
-      ) final;
+    ) final;
 
     get_attr_errorator::future<ceph::bufferlist> omap_get_header(
       CollectionRef c,

src/crimson/os/futurized_store.h

@@ -17,6 +17,7 @@
 #include "include/buffer_fwd.h"
 #include "include/uuid.h"
 #include "osd/osd_types.h"
+#include "os/ObjectStore.h"
 
 namespace ceph::os {
 class Transaction;
@@ -87,13 +88,36 @@ class FuturizedStore {
       const omap_keys_t& keys,
       uint32_t op_flags = 0) = 0;
 
-    using omap_values_paged_t = std::tuple<bool, omap_values_t>;
-    virtual read_errorator::future<omap_values_paged_t> omap_get_values(
-      CollectionRef c,           ///< [in] collection
-      const ghobject_t &oid,     ///< [in] oid
-      const std::optional<std::string> &start, ///< [in] start, empty for begin
+    /**
+     * Iterate over object map with user-provided callable
+     *
+     * Warning! f cannot block or perform IO and must not wait on a future.
+     *
+     * @param c collection
+     * @param oid object
+     * @param start_from where the iterator should point to at
+     *                   the beginning
+     * @param f callable that takes OMAP key and corresponding
+     *          value as string_views and controls iteration
+     *          by the return. It is executed for every object's
+     *          OMAP entry from `start_from` till end of the
+     *          object's OMAP or till the iteration is stopped
+     *          by `STOP`. Please note that if there is no such
+     *          entry, `visitor` will be called 0 times.
+     * @return omap_iter_ret_t on success
+     *         omap_iter_ret_t::STOP means omap_iterate() is stopped by f,
+     *         omap_iter_ret_t::NEXT means omap_iterate() reaches the end of omap tree
+     */
+    using omap_iterate_cb_t = std::function<ObjectStore::omap_iter_ret_t(std::string_view, std::string_view)>;
+    virtual read_errorator::future<ObjectStore::omap_iter_ret_t> omap_iterate(
+      CollectionRef c,   ///< [in] collection
+      const ghobject_t &oid, ///< [in] object
+      ObjectStore::omap_iter_seek_t start_from, ///< [in] where the iterator should point to at the beginning
+      omap_iterate_cb_t callback,
+      ///< [in] the callback function for each OMAP entry after start_from till end of the OMAP or
+      /// till the iteration is stopped by `STOP`.
       uint32_t op_flags = 0
-      ) = 0; ///< @return <done, values> values.empty() only if done
+      ) = 0;
 
     virtual get_attr_errorator::future<bufferlist> omap_get_header(
       CollectionRef c,

src/crimson/os/seastore/omap_manager.h

@@ -4,6 +4,7 @@
 #pragma once
 
 #include <iostream>
+#include <fmt/format.h>
 
 #include <boost/intrusive_ptr.hpp>
 #include <boost/smart_ptr/intrusive_ref_counter.hpp>
@@ -13,13 +14,31 @@
 #include "crimson/osd/exceptions.h"
 #include "crimson/os/seastore/seastore_types.h"
 #include "crimson/os/seastore/transaction_manager.h"
+#include "os/ObjectStore.h" // for ObjectStore::omap_iter_seek_t
+
 
 //TODO: calculate the max key and value sizes the current layout supports,
 //	and return errors during insert if the max is exceeded.
 #define OMAP_INNER_BLOCK_SIZE 8192
 #define OMAP_LEAF_BLOCK_SIZE 65536
 #define LOG_LEAF_BLOCK_SIZE 16384
 
+template <>
+struct fmt::formatter<ObjectStore::omap_iter_seek_t> {
+  constexpr auto parse(format_parse_context& ctx) const {
+    return ctx.begin();
+  }
+
+  auto format(const ObjectStore::omap_iter_seek_t& seek, format_context& ctx) const {
+    return fmt::format_to(
+      ctx.out(),
+      "omap_iter_seek_t{{seek_position='{}', seek_type={}}}",
+      seek.seek_position,
+      seek.seek_type == ObjectStore::omap_iter_seek_t::LOWER_BOUND ? "LOWER_BOUND" : "UPPER_BOUND"
+    );
+  }
+};
+
 namespace crimson::os::seastore {
 
 std::ostream &operator<<(std::ostream &out, const std::list<std::string> &rhs);
@@ -98,10 +117,35 @@ class OMapManager {
     Transaction &t,
     const std::string &key) = 0;
 
+  /**
+   * omap_iterate
+   *
+   * scan key/value pairs and give key/value from start_from.seek_poistion to function f
+   *
+   * @param omap_root: omap btree root information
+   * @param t: current transaction
+   * @param start_from: seek start key and seek type
+   *                    Upon transaction conflict, start_from will be updated to the
+   *                    upper bound of the last key of the last visited omap leaf node.
+   * @param f: function is called for each seeked key/value pair
+   *
+   */
+   using omap_iterate_cb_t = std::function<ObjectStore::omap_iter_ret_t(std::string_view, std::string_view)>;
+   using omap_iterate_iertr = base_iertr;
+   using omap_iterate_ret = omap_iterate_iertr::future<ObjectStore::omap_iter_ret_t>;
+   virtual omap_iterate_ret omap_iterate(
+     const omap_root_t &omap_root,
+     Transaction &t,
+     ObjectStore::omap_iter_seek_t &start_from,
+     omap_iterate_cb_t callback
+   ) = 0;
+
   /**
    * omap_list
    *
    * Scans key/value pairs in order.
+   * // TODO: omap_list() is deprecated in favor of omap_iterate()
+   * // TODO: omap_rm_key_range() that omap_list_config_t needs to be dropped from that interface
    *
    * @param omap_root: omap btree root information
    * @param t: current transaction

src/crimson/os/seastore/omap_manager/btree/btree_omap_manager.cc

@@ -268,6 +268,26 @@ BtreeOMapManager::omap_rm_key_range(
   });
 }
 
+BtreeOMapManager::omap_iterate_ret
+BtreeOMapManager::omap_iterate(
+  const omap_root_t &omap_root,
+  Transaction &t,
+  ObjectStore::omap_iter_seek_t &start_from,
+  omap_iterate_cb_t callback)
+{
+  LOG_PREFIX(BtreeOMapManager::omap_iterate);
+  DEBUGT("{}, {}", t, omap_root, start_from);
+  return get_omap_root(
+    get_omap_context(t, omap_root),
+    omap_root
+  ).si_then([this, &t, &start_from, callback, &omap_root](auto extent) {
+    return extent->iterate(
+      get_omap_context(t, omap_root),
+      start_from,
+      callback);
+  });
+}
+
 BtreeOMapManager::omap_list_ret
 BtreeOMapManager::omap_list(
   const omap_root_t &omap_root,

src/crimson/os/seastore/omap_manager/btree/btree_omap_manager.h

@@ -96,6 +96,12 @@ class BtreeOMapManager : public OMapManager {
     const std::string &last,
     omap_list_config_t config) final;
 
+  omap_iterate_ret omap_iterate(
+    const omap_root_t &omap_root,
+    Transaction &t,
+    ObjectStore::omap_iter_seek_t &start_from,
+    omap_iterate_cb_t callback) final;
+
   omap_list_ret omap_list(
     const omap_root_t &omap_root,
     Transaction &t,

src/crimson/os/seastore/omap_manager/btree/omap_btree_node.h

@@ -80,6 +80,14 @@ struct OMapNode : LogicalChildNode {
     omap_context_t oc,
     const std::string &key) = 0;
 
+  using iterate_iertr = base_iertr;
+  using iterate_ret = OMapManager::omap_iterate_ret;
+  using omap_iterate_cb_t = OMapManager::omap_iterate_cb_t;
+  virtual iterate_ret iterate(
+    omap_context_t oc,
+    ObjectStore::omap_iter_seek_t &start_from,
+    omap_iterate_cb_t callback) = 0;
+
   using omap_list_config_t = OMapManager::omap_list_config_t;
   using list_iertr = base_iertr;
   using list_bare_ret = OMapManager::omap_list_bare_ret;