Merge pull request ceph#59292 from cyx1231st/wip-seastore-revert-decouple-ool-writes

Revert "crimson/os/seastore: wait ool writes in DeviceSubmission phase"

Reviewed-by: Xuehan Xu <[email protected]>
Reviewed-by: Myoungwon Oh <[email protected]>

Author: cyx1231st

src/crimson/os/seastore/extent_placement_manager.cc

@@ -28,7 +28,8 @@ SegmentedOolWriter::SegmentedOolWriter(
 {
 }
 
-void SegmentedOolWriter::write_record(
+SegmentedOolWriter::alloc_write_ertr::future<>
+SegmentedOolWriter::write_record(
   Transaction& t,
   record_t&& record,
   std::list<LogicalCachedExtentRef>&& extents,
@@ -62,20 +63,15 @@ void SegmentedOolWriter::write_record(
     extent_addr = extent_addr.as_seg_paddr().add_offset(
         extent->get_length());
   }
-  // t might be destructed inside write_future
-  auto write_future = seastar::with_gate(write_guard,
-    [this, FNAME, tid=t.get_trans_id(),
-     record_base=ret.record_base_regardless_md,
-     submit_fut=std::move(ret.future)]() mutable {
-    return std::move(submit_fut
-    ).safe_then([this, FNAME, tid, record_base](record_locator_t ret) {
-      TRACE("trans.{} {} finish {}=={}",
-            tid, segment_allocator.get_name(), ret, record_base);
-      // ool won't write metadata, so the paddrs must be equal
-      assert(ret.record_block_base == record_base.offset);
-    });
+  return std::move(ret.future
+  ).safe_then([this, FNAME, &t,
+               record_base=ret.record_base_regardless_md
+              ](record_locator_t ret) {
+    TRACET("{} finish {}=={}",
+           t, segment_allocator.get_name(), ret, record_base);
+    // ool won't write metadata, so the paddrs must be equal
+    assert(ret.record_block_base == record_base.offset);
   });
-  t.get_handle().add_write_future(std::move(write_future));
 }
 
 SegmentedOolWriter::alloc_write_iertr::future<>
@@ -112,15 +108,19 @@ SegmentedOolWriter::do_write(
       DEBUGT("{} extents={} submit {} extents and roll, unavailable ...",
              t, segment_allocator.get_name(),
              extents.size(), num_extents);
+      auto fut_write = alloc_write_ertr::now();
       if (num_extents > 0) {
         assert(record_submitter.check_action(record.size) !=
                action_t::ROLL);
-        write_record(
+        fut_write = write_record(
             t, std::move(record), std::move(pending_extents),
             true/* with_atomic_roll_segment */);
       }
       return trans_intr::make_interruptible(
-        record_submitter.roll_segment()
+        record_submitter.roll_segment(
+        ).safe_then([fut_write=std::move(fut_write)]() mutable {
+          return std::move(fut_write);
+        })
       ).si_then([this, &t, &extents] {
         return do_write(t, extents);
       });
@@ -151,12 +151,15 @@ SegmentedOolWriter::do_write(
       DEBUGT("{} extents={} submit {} extents ...",
              t, segment_allocator.get_name(),
              extents.size(), pending_extents.size());
-      write_record(t, std::move(record), std::move(pending_extents));
-      if (!extents.empty()) {
-        return do_write(t, extents);
-      } else {
-        return alloc_write_iertr::now();
-      }
+      return trans_intr::make_interruptible(
+        write_record(t, std::move(record), std::move(pending_extents))
+      ).si_then([this, &t, &extents] {
+        if (!extents.empty()) {
+          return do_write(t, extents);
+        } else {
+          return alloc_write_iertr::now();
+        }
+      });
     }
     // SUBMIT_NOT_FULL: evaluate the next extent
   }
@@ -166,8 +169,8 @@ SegmentedOolWriter::do_write(
          t, segment_allocator.get_name(),
          num_extents);
   assert(num_extents > 0);
-  write_record(t, std::move(record), std::move(pending_extents));
-  return alloc_write_iertr::now();
+  return trans_intr::make_interruptible(
+    write_record(t, std::move(record), std::move(pending_extents)));
 }
 
 SegmentedOolWriter::alloc_write_iertr::future<>
@@ -983,11 +986,13 @@ RandomBlockOolWriter::alloc_write_ool_extents(
   if (extents.empty()) {
     return alloc_write_iertr::now();
   }
-  do_write(t, extents);
-  return alloc_write_iertr::now();
+  return seastar::with_gate(write_guard, [this, &t, &extents] {
+    return do_write(t, extents);
+  });
 }
 
-void RandomBlockOolWriter::do_write(
+RandomBlockOolWriter::alloc_write_iertr::future<>
+RandomBlockOolWriter::do_write(
   Transaction& t,
   std::list<CachedExtentRef>& extents)
 {
@@ -1040,10 +1045,8 @@ void RandomBlockOolWriter::do_write(
     }
   }
 
-  // t might be destructed inside write_future
-  auto write_future = seastar::with_gate(write_guard,
-    [writes=std::move(writes)]() mutable {
-    return seastar::do_with(std::move(writes),
+  return trans_intr::make_interruptible(
+    seastar::do_with(std::move(writes),
       [](auto& writes) {
       return crimson::do_for_each(writes,
         [](auto& info) {
@@ -1054,9 +1057,8 @@ void RandomBlockOolWriter::do_write(
             "Invalid error when writing record"}
         );
       });
-    });
-  });
-  t.get_handle().add_write_future(std::move(write_future));
+    })
+  );
 }
 
 }

src/crimson/os/seastore/extent_placement_manager.h

@@ -115,7 +115,7 @@ class SegmentedOolWriter : public ExtentOolWriter {
     Transaction& t,
     std::list<CachedExtentRef> &extent);
 
-  void write_record(
+  alloc_write_ertr::future<> write_record(
     Transaction& t,
     record_t&& record,
     std::list<LogicalCachedExtentRef> &&extents,
@@ -195,7 +195,7 @@ class RandomBlockOolWriter : public ExtentOolWriter {
     ceph::bufferptr bp;
     RandomBlockManager* rbm;
   };
-  void do_write(
+  alloc_write_iertr::future<> do_write(
     Transaction& t,
     std::list<CachedExtentRef> &extent);
 

src/crimson/os/seastore/journal/circular_bounded_journal.cc

@@ -97,9 +97,7 @@ CircularBoundedJournal::do_submit_record(
   auto submit_ret = record_submitter.submit(std::move(record));
   // submit_ret.record_base_regardless_md is wrong for journaling
   return handle.enter(write_pipeline->device_submission
-  ).then([&handle] {
-    return handle.take_write_future();
-  }).safe_then([submit_fut=std::move(submit_ret.future)]() mutable {
+  ).then([submit_fut=std::move(submit_ret.future)]() mutable {
     return std::move(submit_fut);
   }).safe_then([FNAME, this, &handle](record_locator_t result) {
     return handle.enter(write_pipeline->finalize

src/crimson/os/seastore/journal/segmented_journal.cc

@@ -396,9 +396,7 @@ SegmentedJournal::do_submit_record(
     auto submit_ret = record_submitter.submit(std::move(record));
     // submit_ret.record_base_regardless_md is wrong for journaling
     return handle.enter(write_pipeline->device_submission
-    ).then([&handle] {
-      return handle.take_write_future();
-    }).safe_then([submit_fut=std::move(submit_ret.future)]() mutable {
+    ).then([submit_fut=std::move(submit_ret.future)]() mutable {
       return std::move(submit_fut);
     }).safe_then([FNAME, this, &handle](record_locator_t result) {
       return handle.enter(write_pipeline->finalize

src/crimson/os/seastore/ordering_handle.h

@@ -122,11 +122,6 @@ struct OrderingHandle {
   std::unique_ptr<OperationProxy> op;
   seastar::shared_mutex *collection_ordering_lock = nullptr;
 
-  using write_ertr = crimson::errorator<
-      crimson::ct_error::input_output_error>;
-  // the pending writes that should complete at DeviceSubmission phase
-  write_ertr::future<> write_future = write_ertr::now();
-
   // in the future we might add further constructors / template to type
   // erasure while extracting the location of tracking events.
   OrderingHandle(std::unique_ptr<OperationProxy> op) : op(std::move(op)) {}
@@ -149,31 +144,13 @@ struct OrderingHandle {
     }
   }
 
-  void add_write_future(write_ertr::future<>&& fut) {
-    auto appended = std::move(write_future
-    ).safe_then([fut=std::move(fut)]() mutable {
-      return std::move(fut);
-    });
-    write_future = std::move(appended);
-  }
-
-  write_ertr::future<> take_write_future() {
-    auto ret = std::move(write_future);
-    write_future = write_ertr::now();
-    return ret;
-  }
-
   template <typename T>
   seastar::future<> enter(T &t) {
     return op->enter(t);
   }
 
   void exit() {
     op->exit();
-
-    auto ignore_writes = std::move(write_future);
-    std::ignore = ignore_writes;
-    write_future = write_ertr::now();
   }
 
   seastar::future<> complete() {
@@ -182,9 +159,6 @@ struct OrderingHandle {
 
   ~OrderingHandle() {
     maybe_release_collection_lock();
-
-    assert(write_future.available());
-    assert(!write_future.failed());
   }
 };