From: Matan Breizman Date: Tue, 14 Jul 2026 14:14:55 +0000 (+0000) Subject: crimson/seastore: batch same-collection transactions X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=548f65f03edc78b0ea549f9503a400159effd24e;p=ceph.git 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 --- diff --git a/src/crimson/os/seastore/seastore.cc b/src/crimson/os/seastore/seastore.cc index 0fcfd13efd1a..1102bfbe768a 100644 --- 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(*_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>& 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(*ch); + while (!coll.pending_txns.empty()) { + std::vector> 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(*(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, 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(*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 e4f65be42b30..7ad4a20d1dd5 100644 --- a/src/crimson/os/seastore/seastore.h +++ b/src/crimson/os/seastore/seastore.h @@ -3,6 +3,7 @@ #pragma once +#include #include #include #include @@ -71,6 +72,13 @@ public: FuturizedCollection(std::forward(args)...) {} seastar::shared_mutex ordering_lock; + + struct batch_entry_t { + ceph::os::Transaction txn; + seastar::promise<> pr; + }; + std::deque 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>& pending_txns_promises); + seastar::future<> run_one_batch(CollectionRef ch, ceph::os::Transaction&& t); + TransactionManager::read_extent_iertr::future> get_coll_bits(CollectionRef ch, Transaction &t) const;