Merge pull request #70180 from Matan-B/wip-matanb-seastore-coll-batch

crimson/seastore: batch same-collection transactions

Reviewed-by: Josh Durgin <jdurgin@redhat.com>
Reviewed-by: Xuehan Xu <xuxuehan@qianxin.com>

src/crimson/os/seastore/omap_manager/log/log_manager.cc

@@ -153,7 +153,8 @@ LogManager::omap_set_keys(
 	  continue;
 	}
 	if (cur->expect_overflow(p.first, p.second.length(),
-	    !is_ow_key(p.first) ? cur->can_ow() : false)) {
+	    (!is_ow_key(p.first) && !cur->is_initial_pending())
+	      ? cur->can_ow() : false)) {
 	  // This means the first entry of the new LogNode is not _fastinfo
 	  if (!is_ow_key(p.first)) {
 	    // remove _fastinfo in old LogNode

src/crimson/os/seastore/omap_manager/log/log_node.cc

@@ -323,15 +323,10 @@ bool LogNode::expect_overflow(const std::string &key,
   size_t vsize, bool can_ow) {
   size_t ksize = key.size();
   if (can_ow) { 
-    int gap = ow_gap_from_last_entry(key.size(), vsize);
-    uint64_t remain = capacity() - get_last_pos() - reserved_len;
-    if (gap >= 0) {
-      gap += static_cast<uint64_t>(gap);
-    } else {
-      uint64_t d = static_cast<uint64_t>(-gap);
-      gap -= d;
-    }
-    return remain < get_entry_size(ksize, vsize);
+    // Reserve only the additional space required by the overwrite.
+    int gap = ow_gap_from_last_entry(ksize, vsize);
+    return gap > 0 &&
+      free_space() < static_cast<size_t>(gap) + reserved_len;
   } else if (get_size() + reserved_size + 1 > d_bitmap_t::MAX_ENTRY) {
     return true;
   } else if (is_ow_key(key) && !can_ow) {
@@ -351,8 +346,10 @@ int LogNode::ow_gap_from_last_entry(const size_t key, const size_t val) {
   if (p) {
     auto ret = p->get_latest_write_delta();
     if (ret && (*ret).key == get_ow_key()) {
-      if ((*ret).val.length() < val) {
-	gap = val - (*ret).val.length();
+      auto old_size = get_entry_size((*ret).key.size(), (*ret).val.length());
+      auto new_size = get_entry_size(key, val);
+      if (new_size > old_size) {
+        gap = new_size - old_size;
       }
     } else {
       gap = _ow_gap_from_last_entry(key, val);

src/crimson/os/seastore/omap_manager/log/log_node.h

@@ -664,8 +664,10 @@ public:
   void _append(const std::string &key, const ceph::bufferlist &val) {
     iterator prev_iter(this, get_last_pos());
     auto last = prev_iter->get_node_key();
-    iterator next_iter(this, get_size() == 0 ? get_last_pos() :
-      get_last_pos() + get_entry_size(last.key_len, last.val_len));
+    uint32_t pos = get_size() == 0 ? get_last_pos() :
+      get_last_pos() + get_entry_size(last.key_len, last.val_len);
+    assert(pos + get_entry_size(key.size(), val.length()) <= capacity());
+    iterator next_iter(this, pos);
     next_iter.set_node_key(log_key_t(key.size(), val.length()));
     next_iter.set_node_val(key, val);
     if (get_size() >= 1) {
@@ -687,6 +689,8 @@ public:
   }
 
   void _overwrite(const std::string &key, const ceph::bufferlist &val) {
+    assert(get_last_pos() + get_entry_size(key.size(), val.length())
+      <= capacity());
     iterator iter(this, get_last_pos());
     iter.set_node_key(log_key_t(key.size(), val.length()));
     iter.set_node_val(key, val);

src/crimson/os/seastore/ordering_handle.h

@@ -123,45 +123,12 @@ struct OrderingHandle {
   // we can easily optimize this dynalloc out as all concretes are
   // supposed to have exactly the same size.
   std::unique_ptr<OperationProxy> op;
-  seastar::shared_mutex *collection_ordering_lock = nullptr;
-
-  seastar::lowres_clock::time_point lock_acquire_time{};
-  seastar::lowres_clock::duration lock_hold_time{0};
 
   // in the future we might add further constructors / template to type
   // erasure while extracting the location of tracking events.
   OrderingHandle(std::unique_ptr<OperationProxy> op) : op(std::move(op)) {}
   OrderingHandle(OrderingHandle &&other)
-    : op(std::move(other.op)),
-      collection_ordering_lock(other.collection_ordering_lock),
-      lock_acquire_time(other.lock_acquire_time),
-      lock_hold_time(other.lock_hold_time) {
-    other.collection_ordering_lock = nullptr;
-  }
-
-  seastar::future<> take_collection_lock(seastar::shared_mutex &mutex) {
-    ceph_assert(!collection_ordering_lock);
-    collection_ordering_lock = &mutex;
-    return collection_ordering_lock->lock();
-  }
-
-  void set_lock_acquire_time(seastar::lowres_clock::time_point tp) {
-    lock_acquire_time = tp;
-  }
-
-  seastar::lowres_clock::duration get_lock_hold_time() const {
-    return lock_hold_time;
-  }
-
-  void maybe_release_collection_lock() {
-    if (collection_ordering_lock) {
-      if (lock_acquire_time != seastar::lowres_clock::time_point{}) {
-        lock_hold_time = seastar::lowres_clock::now() - lock_acquire_time;
-      }
-      collection_ordering_lock->unlock();
-      collection_ordering_lock = nullptr;
-    }
-  }
+    : op(std::move(other.op)) {}
 
   template <typename T>
   seastar::future<> enter(T &t) {
@@ -175,10 +142,6 @@ struct OrderingHandle {
   seastar::future<> complete() {
     return op->complete();
   }
-
-  ~OrderingHandle() {
-    maybe_release_collection_lock();
-  }
 };
 
 inline OrderingHandle get_dummy_ordering_handle() {

src/crimson/os/seastore/seastore.cc

@@ -215,8 +215,6 @@ void SeaStore::Shard::register_metrics(store_index_t store_index)
   );
 
   std::pair<txn_stage_t, sm::label_instance> labels_by_stage[] = {
-    {txn_stage_t::COLLOCK_WAIT,          sm::label_instance("stage", "collock_wait")},
-    {txn_stage_t::COLLOCK_HOLD,          sm::label_instance("stage", "collock_hold")},
     {txn_stage_t::THROTTLER_WAIT,        sm::label_instance("stage", "throttler_wait")},
     {txn_stage_t::BUILD,                 sm::label_instance("stage", "build")},
     {txn_stage_t::BUILD_GET_ONODE,       sm::label_instance("stage", "build_get_onode")},
@@ -921,14 +919,13 @@ seastar::future<> SeaStore::report_stats()
          calc_conflicts(io_total.read_num, io_total.repeat_read_num),
          calc_conflicts(io_total.get_bg_num(), io_total.get_repeat_bg_num()));
     INFO("trans outstanding: {},{},{},{} "
-         "per-shard: {:.2f}({:.2f},{:.2f},{:.2f},{:.2f},{:.2f}),{:.2f},{:.2f},{:.2f}",
+         "per-shard: {:.2f}({:.2f},{:.2f},{:.2f},{:.2f}),{:.2f},{:.2f},{:.2f}",
          io_total.pending_io_num,
          io_total.pending_read_num,
          io_total.pending_bg_num,
          io_total.pending_flush_num,
          (double)io_total.pending_io_num/seastar::smp::count,
          (double)io_total.starting_io_num/seastar::smp::count,
-         (double)io_total.waiting_collock_io_num/seastar::smp::count,
          (double)io_total.waiting_throttler_io_num/seastar::smp::count,
          (double)io_total.processing_inlock_io_num/seastar::smp::count,
          (double)io_total.processing_postlock_io_num/seastar::smp::count,
@@ -940,7 +937,6 @@ seastar::future<> SeaStore::report_stats()
     for (const auto &s : shard_io_stats) {
       oss_pending << s.pending_io_num
                  << "(" << s.starting_io_num
-                 << "," << s.waiting_collock_io_num
                  << "," << s.waiting_throttler_io_num
                  << "," << s.processing_inlock_io_num
                  << "," << s.processing_postlock_io_num
@@ -1729,13 +1725,104 @@ void SeaStore::Shard::transaction_dump(ceph::os::Transaction &t) {
   ERROR("{}", str.str());
 }
 
+// Whether a client txn may be merged with others into one seastore batch
+// (build_next_batch). Default is yes, specific op patterns opt out and run solo.
+
+// Bathced pg-log trim (OP_OMAP_RMKEYS / OP_OMAP_RMKEYRANGE)
+static bool txn_is_batchable(ceph::os::Transaction& t)
+{
+  using ceph::os::Transaction;
+  auto i = t.begin();
+  while (i.have_op()) {
+    auto op = i.decode_op();
+    if (op->op == Transaction::OP_OMAP_RMKEYS ||
+        op->op == Transaction::OP_OMAP_RMKEYRANGE) {
+      return false;
+    }
+  }
+  return true;
+}
+
 seastar::future<> SeaStore::Shard::do_transaction_no_callbacks(
   CollectionRef _ch,
   ceph::os::Transaction&& _t)
 {
-  assert(store_active);
   LOG_PREFIX(SeaStoreS::do_transaction_no_callbacks);
+  assert(store_active);
   ++(shard_stats.io_num);
+
+  auto& coll = static_cast<SeastoreCollection&>(*_ch);
+  auto& entry = coll.pending_txns.emplace_back();
+  entry.txn = std::move(_t);
+  entry.batchable = txn_is_batchable(entry.txn);
+  auto fut = entry.pr.get_future();
+  DEBUG("enqueue cid={} queue_depth={} in_flight={}",
+        coll.get_cid(), coll.pending_txns.size(), coll.collection_in_flight);
+  if (!coll.collection_in_flight) {
+    coll.collection_in_flight = true;
+    DEBUG("cid={} gate closed, starting dispatch", coll.get_cid());
+    std::ignore = dispatch_collection(_ch);
+  }
+  return fut;
+}
+
+ceph::os::Transaction SeaStore::Shard::build_next_batch(
+  SeastoreCollection& coll,
+  std::vector<seastar::promise<>>& pending_txns_promises)
+{
+  ceph::os::Transaction merged;
+  bool first = true;
+  uint64_t batch_features = 0;
+  while (!coll.pending_txns.empty()) {
+    const bool no_batch = !coll.pending_txns.front().batchable;
+    if (!first &&
+        (no_batch || coll.pending_txns.front().txn.get_data_features() != batch_features)) {
+      // Batch boundary: seal what we have so far in the batch
+      break;
+    }
+    auto e = std::move(coll.pending_txns.front());
+    coll.pending_txns.pop_front();
+    if (first) {
+      batch_features = e.txn.get_data_features();
+      merged = std::move(e.txn);
+      first = false;
+    } else {
+      merged.append(e.txn);
+    }
+    pending_txns_promises.push_back(std::move(e.pr));
+    if (no_batch) {
+      // no_batch runs solo (never is appended)
+      break;
+    }
+  }
+  return merged;
+}
+
+seastar::future<> SeaStore::Shard::dispatch_collection(CollectionRef ch)
+{
+  LOG_PREFIX(SeaStoreS::dispatch_collection);
+  auto& coll = static_cast<SeastoreCollection&>(*ch);
+  while (!coll.pending_txns.empty()) {
+    std::vector<seastar::promise<>> pending_txns_promises;
+    auto merged = build_next_batch(coll, pending_txns_promises);
+    DEBUG("draining {} txns from cid={}, committing batch ({} ops)",
+          pending_txns_promises.size(), coll.get_cid(), merged.get_num_ops());
+    co_await run_one_batch(ch, std::move(merged));
+    DEBUG("committed batch of {} txns for cid={}",
+          pending_txns_promises.size(), coll.get_cid());
+    for (auto& p : pending_txns_promises) {
+      p.set_value();
+    }
+  }
+  DEBUG("cid={} drained, gate open", coll.get_cid());
+  coll.collection_in_flight = false;
+}
+
+seastar::future<> SeaStore::Shard::run_one_batch(
+  CollectionRef _ch,
+  ceph::os::Transaction&& _t)
+{
+  LOG_PREFIX(SeaStoreS::run_one_batch);
   ++(shard_stats.pending_io_num);
   ++(shard_stats.starting_io_num);
 
@@ -1750,18 +1837,6 @@ seastar::future<> SeaStore::Shard::do_transaction_no_callbacks(
 
   assert(shard_stats.starting_io_num);
   --(shard_stats.starting_io_num);
-  ++(shard_stats.waiting_collock_io_num);
-
-  auto t_pre_collock = seastar::lowres_clock::now();
-  co_await ctx.transaction->get_handle().take_collection_lock(
-    static_cast<SeastoreCollection&>(*(ctx.ch)).ordering_lock
-  );
-  auto t_post_collock = seastar::lowres_clock::now();
-  auto collock_wait = t_post_collock - t_pre_collock;
-  ctx.transaction->get_handle().set_lock_acquire_time(t_post_collock);
-
-  assert(shard_stats.waiting_collock_io_num);
-  --(shard_stats.waiting_collock_io_num);
   ++(shard_stats.waiting_throttler_io_num);
 
   auto t_pre_throttler = seastar::lowres_clock::now();
@@ -1841,8 +1916,6 @@ seastar::future<> SeaStore::Shard::do_transaction_no_callbacks(
     const std::array<
       std::pair<txn_stage_t, seastar::lowres_clock::duration>, STAGE_MAX>
       stage_samples = {{
-        {txn_stage_t::COLLOCK_WAIT,          collock_wait},
-        {txn_stage_t::COLLOCK_HOLD,          ctx.transaction->get_handle().get_lock_hold_time()},
         {txn_stage_t::THROTTLER_WAIT,        throttler_wait},
         {txn_stage_t::BUILD,                 ctx.build_time},
         {txn_stage_t::BUILD_GET_ONODE,       ctx.get_onode_time},
@@ -1884,16 +1957,15 @@ seastar::future<> SeaStore::Shard::flush(CollectionRef ch)
   ++(shard_stats.flush_num);
   ++(shard_stats.pending_flush_num);
 
-  return seastar::do_with(
-    get_dummy_ordering_handle(),
-    [this, ch](auto &handle) {
-      return handle.take_collection_lock(
-	static_cast<SeastoreCollection&>(*ch).ordering_lock
-      ).then([this, &handle] {
+  return do_transaction_no_callbacks(
+    ch, ceph::os::Transaction{}
+  ).then([this] {
+    return seastar::do_with(
+      get_dummy_ordering_handle(),
+      [this](auto &handle) {
 	return transaction_manager->flush(handle);
       });
-    }
-  ).finally([this] {
+  }).finally([this] {
     assert(shard_stats.pending_flush_num);
     --(shard_stats.pending_flush_num);
   });
@@ -3004,7 +3076,7 @@ shard_stats_t SeaStore::Shard::get_io_stats(
     };
     INFO("iops={:.2f},{:.2f},{:.2f}({:.2f},{:.2f},{:.2f},{:.2f}),{:.2f} "
          "conflicts={:.2f},{:.2f},{:.2f}({:.2f},{:.2f},{:.2f},{:.2f}) "
-         "outstanding={}({},{},{},{},{}),{},{},{}",
+         "outstanding={}({},{},{},{}),{},{},{}",
          // iops
          ret.io_num/seconds,
          ret.read_num/seconds,
@@ -3025,7 +3097,6 @@ shard_stats_t SeaStore::Shard::get_io_stats(
          // outstanding
          ret.pending_io_num,
          ret.starting_io_num,
-         ret.waiting_collock_io_num,
          ret.waiting_throttler_io_num,
          ret.processing_inlock_io_num,
          ret.processing_postlock_io_num,

src/crimson/os/seastore/seastore.h

@@ -3,6 +3,7 @@
 
 #pragma once
 
+#include <deque>
 #include <map>
 #include <optional>
 #include <string>
@@ -48,9 +49,7 @@ enum class op_type_t : uint8_t {
 };
 
 enum class txn_stage_t : uint8_t {
-    COLLOCK_WAIT = 0,  // waiting on the collection ordering_lock
-    COLLOCK_HOLD,      // collection ordering_lock held (acquire -> release at prepare_record)
-    THROTTLER_WAIT,    // waiting for a throttler slot
+    THROTTLER_WAIT = 0, // waiting for a throttler slot
     BUILD,             // building the transaction (_do_transaction_step loop)
     BUILD_GET_ONODE,   // onode_manager get/get_or_create calls within BUILD
     SUBMIT_TOTAL,      // the whole submit_transaction (pipeline + journal write)
@@ -70,7 +69,13 @@ public:
   SeastoreCollection(T&&... args) :
     FuturizedCollection(std::forward<T>(args)...) {}
 
-  seastar::shared_mutex ordering_lock;
+  struct batch_entry_t {
+    ceph::os::Transaction txn;
+    seastar::promise<> pr;
+    bool batchable = true;
+  };
+  std::deque<batch_entry_t> pending_txns;
+  bool collection_in_flight = false;
 };
 
 /**
@@ -273,6 +278,12 @@ public:
       }
     };
 
+    seastar::future<> dispatch_collection(CollectionRef ch);
+    ceph::os::Transaction build_next_batch(
+      SeastoreCollection& coll,
+      std::vector<seastar::promise<>>& pending_txns_promises);
+    seastar::future<> run_one_batch(CollectionRef ch, ceph::os::Transaction&& t);
+
     TransactionManager::read_extent_iertr::future<std::optional<unsigned>>
     get_coll_bits(CollectionRef ch, Transaction &t) const;
 

src/crimson/os/seastore/seastore_types.h

@@ -3293,7 +3293,6 @@ struct shard_stats_t {
   uint64_t repeat_io_num = 0;
   uint64_t pending_io_num = 0;
   uint64_t starting_io_num = 0;
-  uint64_t waiting_collock_io_num = 0;
   uint64_t waiting_throttler_io_num = 0;
   uint64_t processing_inlock_io_num = 0;
   uint64_t processing_postlock_io_num = 0;
@@ -3336,7 +3335,6 @@ struct shard_stats_t {
     repeat_io_num += o.repeat_io_num;
     pending_io_num += o.pending_io_num;
     starting_io_num += o.starting_io_num;
-    waiting_collock_io_num += o.waiting_collock_io_num;
     waiting_throttler_io_num += o.waiting_throttler_io_num;
     processing_inlock_io_num += o.processing_inlock_io_num;
     processing_postlock_io_num += o.processing_postlock_io_num;

src/crimson/os/seastore/transaction_manager.cc

@@ -703,7 +703,6 @@ TransactionManager::do_submit_transaction(
   tref.get_phase_durations().prepare_record +=
     std::chrono::steady_clock::now() - prepare_record_start;
 
-  tref.get_handle().maybe_release_collection_lock();
   if (tref.get_src() == Transaction::src_t::MUTATE) {
     --(shard_stats.processing_inlock_io_num);
     ++(shard_stats.processing_postlock_io_num);
@@ -749,7 +748,6 @@ seastar::future<> TransactionManager::flush(OrderingHandle &handle)
   }).then([this, &handle] {
     return handle.enter(write_pipeline.prepare);
   }).then([this, &handle] {
-    handle.maybe_release_collection_lock();
     return journal->flush(handle);
   }).then([FNAME, &handle] {
     SUBDEBUG(seastore_t, "H{} completed", (void*)&handle);

src/os/Transaction.h

@@ -393,6 +393,8 @@ public:
   }
   uint32_t get_fadvise_flags() { return data.fadvise_flags; }
 
+  uint64_t get_data_features() const { return data_features; }
+
   void swap(Transaction& other) noexcept {
     std::swap(data, other.data);
     std::swap(on_applied, other.on_applied);