Merge pull request #44591 from athanatos/sjust/wip-seastore-flush

crimson/os/seastore: avoid empty Transactions by adding explicit flush() call

Reviewed-by: Yingxin Cheng <yingxin.cheng@intel.com>

8 files changed, 86 insertions, 13 deletions

diff --git a/src/crimson/os/futurized_store.h b/src/crimson/os/futurized_store.h
index d25c0948450..c35cba9b05d 100644
--- a/src/crimson/os/futurized_store.h
+++ b/src/crimson/os/futurized_store.h
@@ -10,6 +10,7 @@
 
 #include <seastar/core/future.hh>
 
+#include "os/Transaction.h"
 #include "crimson/osd/exceptions.h"
 #include "include/buffer_fwd.h"
 #include "include/uuid.h"
@@ -134,6 +135,18 @@ public:
 
   virtual seastar::future<> do_transaction(CollectionRef ch,
 					   ceph::os::Transaction&& txn) = 0;
+  /**
+   * flush
+   *
+   * Flushes outstanding transactions on ch, returned future resolves
+   * after any previously submitted transactions on ch have committed.
+   *
+   * @param ch [in] collection on which to flush
+   */
+  virtual seastar::future<> flush(CollectionRef ch) {
+    return do_transaction(ch, ceph::os::Transaction{});
+  }
+
   // error injection
   virtual seastar::future<> inject_data_error(const ghobject_t& o) {
     return seastar::now();
diff --git a/src/crimson/os/seastore/journal.cc b/src/crimson/os/seastore/journal.cc
index ebcc366c3e9..39a9063d62a 100644
--- a/src/crimson/os/seastore/journal.cc
+++ b/src/crimson/os/seastore/journal.cc
@@ -646,6 +646,14 @@ Journal::RecordSubmitter::submit(
   return do_submit(std::move(record), handle);
 }
 
+seastar::future<> Journal::RecordSubmitter::flush(OrderingHandle &handle)
+{
+  return handle.enter(write_pipeline->device_submission
+  ).then([this, &handle] {
+    return handle.enter(write_pipeline->finalize);
+  });
+}
+
 void Journal::RecordSubmitter::update_state()
 {
   if (num_outstanding_io == 0) {
diff --git a/src/crimson/os/seastore/journal.h b/src/crimson/os/seastore/journal.h
index c7be7cbab2b..4a68a9a9308 100644
--- a/src/crimson/os/seastore/journal.h
+++ b/src/crimson/os/seastore/journal.h
@@ -94,6 +94,17 @@ public:
   }
 
   /**
+   * flush
+   *
+   * Wait for all outstanding IOs on handle to commit.
+   * Note, flush() machinery must go through the same pipeline
+   * stages and locks as submit_record.
+   */
+  seastar::future<> flush(OrderingHandle &handle) {
+    return record_submitter.flush(handle);
+  }
+
+  /**
    * Read deltas and pass to delta_handler
    *
    * record_block_start (argument to delta_handler) is the start of the
@@ -387,6 +398,7 @@ private:
 
     using submit_ret = Journal::submit_record_ret;
     submit_ret submit(record_t&&, OrderingHandle&);
+    seastar::future<> flush(OrderingHandle &handle);
 
   private:
     void update_state();
diff --git a/src/crimson/os/seastore/seastore.cc b/src/crimson/os/seastore/seastore.cc
index eb166b1ad32..1a89fa79a44 100644
--- a/src/crimson/os/seastore/seastore.cc
+++ b/src/crimson/os/seastore/seastore.cc
@@ -925,6 +925,20 @@ seastar::future<> SeaStore::do_transaction(
     });
 }
 
+
+seastar::future<> SeaStore::flush(CollectionRef ch)
+{
+  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 transaction_manager->flush(handle);
+      });
+    });
+}
+
 SeaStore::tm_ret SeaStore::_do_transaction_step(
   internal_context_t &ctx,
   CollectionRef &col,
diff --git a/src/crimson/os/seastore/seastore.h b/src/crimson/os/seastore/seastore.h
index d1c325432d7..93e34c28801 100644
--- a/src/crimson/os/seastore/seastore.h
+++ b/src/crimson/os/seastore/seastore.h
@@ -139,6 +139,10 @@ public:
     CollectionRef ch,
     ceph::os::Transaction&& txn) final;
 
+  /* Note, flush() machinery must go through the same pipeline
+   * stages and locks as do_transaction. */
+  seastar::future<> flush(CollectionRef ch) final;
+
   seastar::future<OmapIteratorRef> get_omap_iterator(
     CollectionRef ch,
     const ghobject_t& oid) final;
diff --git a/src/crimson/os/seastore/transaction_manager.cc b/src/crimson/os/seastore/transaction_manager.cc
index 624ae4c0d5f..c9dc00e4ead 100644
--- a/src/crimson/os/seastore/transaction_manager.cc
+++ b/src/crimson/os/seastore/transaction_manager.cc
@@ -321,6 +321,19 @@ TransactionManager::submit_transaction_direct(
   });
 }
 
+seastar::future<> TransactionManager::flush(OrderingHandle &handle)
+{
+  return handle.enter(write_pipeline.reserve_projected_usage
+  ).then([this, &handle] {
+    return handle.enter(write_pipeline.ool_writes);
+  }).then([this, &handle] {
+    return handle.enter(write_pipeline.prepare);
+  }).then([this, &handle] {
+    handle.maybe_release_collection_lock();
+    return journal->flush(handle);
+  });
+}
+
 TransactionManager::get_next_dirty_extents_ret
 TransactionManager::get_next_dirty_extents(
   Transaction &t,
diff --git a/src/crimson/os/seastore/transaction_manager.h b/src/crimson/os/seastore/transaction_manager.h
index cb3cdec8d19..82c825c284c 100644
--- a/src/crimson/os/seastore/transaction_manager.h
+++ b/src/crimson/os/seastore/transaction_manager.h
@@ -374,6 +374,15 @@ public:
   submit_transaction_direct_ret submit_transaction_direct(
     Transaction &t) final;
 
+  /**
+   * flush
+   *
+   * Block until all outstanding IOs on handle are committed.
+   * Note, flush() machinery must go through the same pipeline
+   * stages and locks as submit_transaction.
+   */
+  seastar::future<> flush(OrderingHandle &handle);
+
   using SegmentCleaner::ExtentCallbackInterface::get_next_dirty_extents_ret;
   get_next_dirty_extents_ret get_next_dirty_extents(
     Transaction &t,
diff --git a/src/crimson/osd/pg.cc b/src/crimson/osd/pg.cc
index 7fd940f1e76..45ae1875b2b 100644
--- a/src/crimson/osd/pg.cc
+++ b/src/crimson/osd/pg.cc
@@ -131,19 +131,19 @@ PG::~PG() {}
 
 bool PG::try_flush_or_schedule_async() {
   logger().debug("PG::try_flush_or_schedule_async: do_transaction...");
-  (void)shard_services.get_store().do_transaction(
-    coll_ref,
-    ObjectStore::Transaction()).then(
-      [this, epoch=get_osdmap_epoch()]() {
-	return shard_services.start_operation<LocalPeeringEvent>(
-	  this,
-	  shard_services,
-	  pg_whoami,
-	  pgid,
-	  epoch,
-	  epoch,
-	  PeeringState::IntervalFlush());
-      });
+  (void)shard_services.get_store().flush(
+    coll_ref
+  ).then(
+    [this, epoch=get_osdmap_epoch()]() {
+      return shard_services.start_operation<LocalPeeringEvent>(
+	this,
+	shard_services,
+	pg_whoami,
+	pgid,
+	epoch,
+	epoch,
+	PeeringState::IntervalFlush());
+    });
   return false;
 }