crimson/os/seastore/transaction_manager: block client transactions if

they conflict with rewriting transactions until the rewriting
transactions finishes

Signed-off-by: Xuehan Xu <xuxuehan@qianxin.com>

diff --git a/src/crimson/os/seastore/cache.cc b/src/crimson/os/seastore/cache.cc
index bb0f63f024426efd16123daf745ae323ba7fca62..34ab33e4f700f417901dc03c217a405f23c81347 100644 (file)
--- a/src/crimson/os/seastore/cache.cc
+++ b/src/crimson/os/seastore/cache.cc
@@ -1301,6 +1301,7 @@ record_t Cache::prepare_record(
     if (is_rewrite_transaction(t.get_src()) &&
         !is_root_type(i->get_type())) {
       i->new_committer(t);
+      i->committer->block_trans(t);
     }
     assert(i->is_exist_mutation_pending() ||
           i->prior_instance);
@@ -1573,6 +1574,7 @@ record_t Cache::prepare_record(
       if (is_lba_backref_node(i->get_type())) {
         committer.sync_checksum();
       }
+      committer.block_trans(t);
       i->get_prior_instance()->set_io_wait(
         CachedExtent::extent_state_t::CLEAN, true);
     }
@@ -1939,6 +1941,7 @@ void Cache::complete_commit(
         committer.share_prior_data_to_pending();
       }
       touch_extent_fully(prior, &t_src, t.get_cache_hint());
+      committer.unblock_trans(t);
       prior.complete_io();
       i->committer.reset();
       prior.committer.reset();
@@ -2015,6 +2018,7 @@ void Cache::complete_commit(
         t, *i, *i->prior_instance);
       assert(i->committer);
       auto &committer = *i->committer;
+      committer.unblock_trans(t);
       auto &prior = *i->prior_instance;
       prior.pending_for_transaction = TRANS_ID_NULL;
       ceph_assert(prior.is_valid());

diff --git a/src/crimson/os/seastore/cached_extent.cc b/src/crimson/os/seastore/cached_extent.cc
index 72fe6840993dbfa719255fa4d042057b11b7133c..3063c7cb01b24594e9eb643e61317e81e8b10ea8 100644 (file)
--- a/src/crimson/os/seastore/cached_extent.cc
+++ b/src/crimson/os/seastore/cached_extent.cc
@@ -18,6 +18,8 @@ namespace {
   }
 }
 
+SET_SUBSYS(seastore_cache);
+
 namespace crimson::os::seastore {
 
 #ifdef DEBUG_CACHED_EXTENT_REF
@@ -466,4 +468,24 @@ void CachedExtent::new_committer(Transaction &t) {
   prior_instance->committer = committer;
 }
 
+void ExtentCommitter::block_trans(Transaction &t) {
+  LOG_PREFIX(ExtentCommitter::block_trans);
+  auto &prior = *extent.prior_instance;
+  for (auto &item : prior.read_transactions) {
+    TRACET("blocking trans {} for rewriting {}",
+      t, item.t->get_trans_id(), *item.ref);
+    item.t->need_wait_rewrite = true;
+  }
+}
+
+void ExtentCommitter::unblock_trans(Transaction &t) {
+  LOG_PREFIX(ExtentCommitter::unblock_trans);
+  auto &prior = *extent.prior_instance;
+  for (auto &item : prior.read_transactions) {
+    TRACET("unblocking trans {} for rewriting {}",
+      t, item.t->get_trans_id(), *item.ref);
+    item.t->need_wait_rewrite = false;
+  }
+}
+
 }

diff --git a/src/crimson/os/seastore/cached_extent.h b/src/crimson/os/seastore/cached_extent.h
index 5c4ad192a67e566821f5144c5f9d676d97bd3534..2cb3ebe4321a85a3eafb1e610067d57b8a15fe10 100644 (file)
--- a/src/crimson/os/seastore/cached_extent.h
+++ b/src/crimson/os/seastore/cached_extent.h
@@ -279,6 +279,8 @@ public:
   ExtentCommitter(CachedExtent &extent, Transaction &t)
     : extent(extent), t(t) {}
 
+  void block_trans(Transaction &);
+  void unblock_trans(Transaction &);
   // commit all extent states to the prior instance,
   // except poffset and extent content
   void commit_state();

diff --git a/src/crimson/os/seastore/transaction.h b/src/crimson/os/seastore/transaction.h
index 0ca723cc44400b2d83e423e801d5b35c8eccf952..723c87eb66ddb166b79cb20d32a64fd5a5f6b734 100644 (file)
--- a/src/crimson/os/seastore/transaction.h
+++ b/src/crimson/os/seastore/transaction.h
@@ -550,6 +550,7 @@ public:
     ool_write_stats = {};
     rewrite_stats = {};
     conflicted = false;
+    need_wait_rewrite = false;
     assert(backref_entries.empty());
     if (!has_reset) {
       has_reset = true;
@@ -667,6 +668,8 @@ public:
   }
 
   btree_cursor_stats_t cursor_stats;
+  bool need_wait_rewrite = false;
+
 private:
   friend class Cache;
   friend Ref make_test_transaction();

diff --git a/src/crimson/os/seastore/transaction_manager.cc b/src/crimson/os/seastore/transaction_manager.cc
index ca1a45af9a77431ae68831c2d138406413c31414..a44748d27727d18a442a24770a83173a2f12b57b 100644 (file)
--- a/src/crimson/os/seastore/transaction_manager.cc
+++ b/src/crimson/os/seastore/transaction_manager.cc
@@ -583,6 +583,9 @@ TransactionManager::do_submit_transaction(
     tref.get_handle().enter(write_pipeline.prepare)
   );
 
+  while (tref.need_wait_rewrite) {
+    co_await trans_intr::make_interruptible(seastar::yield());
+  }
   if (trim_alloc_to && *trim_alloc_to != JOURNAL_SEQ_NULL) {
     SUBTRACET(seastore_t, "trim backref_bufs to {}", tref, *trim_alloc_to);
     cache->trim_backref_bufs(*trim_alloc_to);