diff --git a/src/crimson/os/seastore/omap_manager/log/log_manager.cc b/src/crimson/os/seastore/omap_manager/log/log_manager.cc index 03f16875c5c..96b7f5e0726 100644 --- a/src/crimson/os/seastore/omap_manager/log/log_manager.cc +++ b/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 diff --git a/src/crimson/os/seastore/omap_manager/log/log_node.cc b/src/crimson/os/seastore/omap_manager/log/log_node.cc index e87a3997b3f..e0d6f173972 100644 --- a/src/crimson/os/seastore/omap_manager/log/log_node.cc +++ b/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(gap); - } else { - uint64_t d = static_cast(-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(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); diff --git a/src/crimson/os/seastore/omap_manager/log/log_node.h b/src/crimson/os/seastore/omap_manager/log/log_node.h index 5ce65be0c51..6a499df0780 100644 --- a/src/crimson/os/seastore/omap_manager/log/log_node.h +++ b/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); diff --git a/src/crimson/os/seastore/ordering_handle.h b/src/crimson/os/seastore/ordering_handle.h index e4b375416e8..957b39f6243 100644 --- a/src/crimson/os/seastore/ordering_handle.h +++ b/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 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 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 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() { diff --git a/src/crimson/os/seastore/seastore.cc b/src/crimson/os/seastore/seastore.cc index 0fcfd13efd1..15c9a3fcb1f 100644 --- a/src/crimson/os/seastore/seastore.cc +++ b/src/crimson/os/seastore/seastore.cc @@ -215,8 +215,6 @@ void SeaStore::Shard::register_metrics(store_index_t store_index) ); std::pair 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(*_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>& 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(*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 +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(*(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, 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(*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, diff --git a/src/crimson/os/seastore/seastore.h b/src/crimson/os/seastore/seastore.h index e4f65be42b3..ca7ffa02712 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 @@ -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(args)...) {} - seastar::shared_mutex ordering_lock; + struct batch_entry_t { + ceph::os::Transaction txn; + seastar::promise<> pr; + bool batchable = true; + }; + std::deque 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>& 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; diff --git a/src/crimson/os/seastore/seastore_types.h b/src/crimson/os/seastore/seastore_types.h index 082baeacea3..58b23da3814 100644 --- a/src/crimson/os/seastore/seastore_types.h +++ b/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; diff --git a/src/crimson/os/seastore/transaction_manager.cc b/src/crimson/os/seastore/transaction_manager.cc index a8d28e276f3..b4bfd50a2a5 100644 --- a/src/crimson/os/seastore/transaction_manager.cc +++ b/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); diff --git a/src/os/Transaction.h b/src/os/Transaction.h index 220e701f45d..c226e0399a3 100644 --- a/src/os/Transaction.h +++ b/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);