crimson/seastore: batch same-collection transactions

Previously, the per-collection ordering_lock provided both ordering and
single-holder exclusion, causing same-collection transactions to be serialized
one at a time.

Replace that with a per-collection queue and a single dispatch loop. When a
collection has a batch in flight, newly arriving transactions are queued.
The dispatch loop later drains the queue and merges transactions into one
SeaStore transaction (batch). Resulting in a one build, one ool_write and one journal record for the entire batch.

Only one batch is in flight per collection, so same-collection concurrency is
still prevented and ordering is preserved by the queue.

Callbacks are unchanged: FuturizedStore drains each caller's on_commit before
enqueueing, so only ops are merged. Each caller's promise is fulfilled when the
merged batch commits.

Signed-off-by: Matan Breizman <mbreizma@redhat.com>

diff --git a/src/crimson/os/seastore/seastore.cc b/src/crimson/os/seastore/seastore.cc
index 0fcfd13efd1a85e266c4f3443d53cb0e5d96ab42..1102bfbe768a8aadc2f653a4cb4dbfcafc0ca160 100644 (file)
--- a/src/crimson/os/seastore/seastore.cc
+++ b/src/crimson/os/seastore/seastore.cc
@@ -1733,9 +1733,69 @@ 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);
+  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;
+  while (!coll.pending_txns.empty()) {
+    auto e = std::move(coll.pending_txns.front());
+    coll.pending_txns.pop_front();
+    if (first) {
+      merged = std::move(e.txn);
+      first = false;
+    } else {
+      merged.append(e.txn);
+    }
+    pending_txns_promises.push_back(std::move(e.pr));
+  }
+  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 +1810,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 +1889,8 @@ 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::COLLOCK_WAIT,          seastar::lowres_clock::duration::zero()},
+        {txn_stage_t::COLLOCK_HOLD,          seastar::lowres_clock::duration::zero()},
         {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 +1932,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);
   });

diff --git a/src/crimson/os/seastore/seastore.h b/src/crimson/os/seastore/seastore.h
index e4f65be42b30f74ea84c9089799bcebf484c94ed..7ad4a20d1dd5a7ed49c0ffca30684732af0ed6c5 100644 (file)
--- a/src/crimson/os/seastore/seastore.h
+++ b/src/crimson/os/seastore/seastore.h
@@ -3,6 +3,7 @@
 
 #pragma once
 
+#include <deque>
 #include <map>
 #include <optional>
 #include <string>
@@ -71,6 +72,13 @@ public:
     FuturizedCollection(std::forward<T>(args)...) {}
 
   seastar::shared_mutex ordering_lock;
+
+  struct batch_entry_t {
+    ceph::os::Transaction txn;
+    seastar::promise<> pr;
+  };
+  std::deque<batch_entry_t> pending_txns;
+  bool collection_in_flight = false;
 };
 
 /**
@@ -273,6 +281,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;