Merge pull request ceph#58913 from cyx1231st/wip-seastore-unlock-ool

crimson/os/seastore: move ool writes from collection lock to concurrent DeviceSubmission phase

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

Author: cyx1231st

src/crimson/os/seastore/cache.cc

@@ -1322,7 +1322,7 @@ record_t Cache::prepare_record(
     }
   }
 
-  for (auto &i: t.written_ool_block_list) {
+  for (auto &i: t.ool_block_list) {
     TRACET("fresh ool extent -- {}", t, *i);
     ceph_assert(i->is_valid());
     assert(!i->is_inline());
@@ -1340,7 +1340,7 @@ record_t Cache::prepare_record(
     }
   }
 
-  for (auto &i: t.written_inplace_ool_block_list) {
+  for (auto &i: t.inplace_ool_block_list) {
     if (!i->is_valid()) {
       continue;
     }
@@ -1422,13 +1422,13 @@ record_t Cache::prepare_record(
 
   ceph_assert(t.get_fresh_block_stats().num ==
               t.inline_block_list.size() +
-              t.written_ool_block_list.size() +
+              t.ool_block_list.size() +
               t.num_delayed_invalid_extents +
 	      t.num_allocated_invalid_extents);
 
   auto& ool_stats = t.get_ool_write_stats();
-  ceph_assert(ool_stats.extents.num == t.written_ool_block_list.size() +
-    t.written_inplace_ool_block_list.size());
+  ceph_assert(ool_stats.extents.num == t.ool_block_list.size() +
+    t.inplace_ool_block_list.size());
 
   if (record.is_empty()) {
     SUBINFOT(seastore_t,

src/crimson/os/seastore/extent_placement_manager.cc

@@ -28,8 +28,7 @@ SegmentedOolWriter::SegmentedOolWriter(
 {
 }
 
-SegmentedOolWriter::alloc_write_ertr::future<>
-SegmentedOolWriter::write_record(
+void SegmentedOolWriter::write_record(
   Transaction& t,
   record_t&& record,
   std::list<LogicalCachedExtentRef>&& extents,
@@ -47,24 +46,36 @@ SegmentedOolWriter::write_record(
   stats.md_bytes += record.size.get_raw_mdlength();
   stats.num_records += 1;
 
-  return record_submitter.submit(
+  auto ret = record_submitter.submit(
     std::move(record),
-    with_atomic_roll_segment
-  ).safe_then([this, FNAME, &t, extents=std::move(extents)
-              ](record_locator_t ret) mutable {
-    DEBUGT("{} finish with {} and {} extents",
+    with_atomic_roll_segment);
+  DEBUGT("{} start at {} with {} extents ...",
+         t, segment_allocator.get_name(),
+         ret.record_base_regardless_md,
+         extents.size());
+  paddr_t extent_addr = ret.record_base_regardless_md.offset;
+  for (auto& extent : extents) {
+    TRACET("{} extent will be written at {} -- {}",
            t, segment_allocator.get_name(),
-           ret, extents.size());
-    paddr_t extent_addr = ret.record_block_base;
-    for (auto& extent : extents) {
-      TRACET("{} ool extent written at {} -- {}",
-             t, segment_allocator.get_name(),
-             extent_addr, *extent);
-      t.update_delayed_ool_extent_addr(extent, extent_addr);
-      extent_addr = extent_addr.as_seg_paddr().add_offset(
-          extent->get_length());
-    }
+           extent_addr, *extent);
+    t.update_delayed_ool_extent_addr(extent, extent_addr);
+    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);
+    });
   });
+  t.get_handle().add_write_future(std::move(write_future));
 }
 
 SegmentedOolWriter::alloc_write_iertr::future<>
@@ -101,19 +112,15 @@ 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);
-        fut_write = write_record(
+        write_record(
             t, std::move(record), std::move(pending_extents),
             true/* with_atomic_roll_segment */);
       }
       return trans_intr::make_interruptible(
-        record_submitter.roll_segment(
-        ).safe_then([fut_write=std::move(fut_write)]() mutable {
-          return std::move(fut_write);
-        })
+        record_submitter.roll_segment()
       ).si_then([this, &t, &extents] {
         return do_write(t, extents);
       });
@@ -144,15 +151,12 @@ SegmentedOolWriter::do_write(
       DEBUGT("{} extents={} submit {} extents ...",
              t, segment_allocator.get_name(),
              extents.size(), pending_extents.size());
-      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();
-        }
-      });
+      write_record(t, std::move(record), std::move(pending_extents));
+      if (!extents.empty()) {
+        return do_write(t, extents);
+      } else {
+        return alloc_write_iertr::now();
+      }
     }
     // SUBMIT_NOT_FULL: evaluate the next extent
   }
@@ -162,8 +166,8 @@ SegmentedOolWriter::do_write(
          t, segment_allocator.get_name(),
          num_extents);
   assert(num_extents > 0);
-  return trans_intr::make_interruptible(
-    write_record(t, std::move(record), std::move(pending_extents)));
+  write_record(t, std::move(record), std::move(pending_extents));
+  return alloc_write_iertr::now();
 }
 
 SegmentedOolWriter::alloc_write_iertr::future<>
@@ -498,6 +502,7 @@ ExtentPlacementManager::write_delayed_ool_extents(
       assert(extent->is_valid());
     });
 #endif
+    assert(writer->get_type() == backend_type_t::SEGMENTED);
     return writer->alloc_write_ool_extents(t, extents);
   });
 }
@@ -524,6 +529,7 @@ ExtentPlacementManager::write_preallocated_ool_extents(
     return trans_intr::do_for_each(alloc_map, [&t](auto& p) {
       auto writer = p.first;
       auto& extents = p.second;
+      assert(writer->get_type() == backend_type_t::RANDOM_BLOCK);
       return writer->alloc_write_ool_extents(t, extents);
     });
   });
@@ -985,27 +991,27 @@ RandomBlockOolWriter::alloc_write_ool_extents(
   if (extents.empty()) {
     return alloc_write_iertr::now();
   }
-  return seastar::with_gate(write_guard, [this, &t, &extents] {
-    return do_write(t, extents);
-  });
+  do_write(t, extents);
+  return alloc_write_iertr::now();
 }
 
-RandomBlockOolWriter::alloc_write_iertr::future<>
-RandomBlockOolWriter::do_write(
+void RandomBlockOolWriter::do_write(
   Transaction& t,
   std::list<CachedExtentRef>& extents)
 {
   LOG_PREFIX(RandomBlockOolWriter::do_write);
   assert(!extents.empty());
   DEBUGT("start with {} allocated extents",
          t, extents.size());
-  return trans_intr::do_for_each(extents,
-    [this, &t, FNAME](auto& ex) {
+  std::vector<write_info_t> writes;
+  writes.reserve(extents.size());
+  for (auto& ex : extents) {
     auto paddr = ex->get_paddr();
     assert(paddr.is_absolute());
     RandomBlockManager * rbm = rb_cleaner->get_rbm(paddr); 
     assert(rbm);
-    TRACE("extent {}, allocated addr {}", fmt::ptr(ex.get()), paddr);
+    TRACE("write extent {}, paddr {} ...",
+          fmt::ptr(ex.get()), paddr);
     auto& stats = t.get_ool_write_stats();
     stats.extents.num += 1;
     stats.extents.bytes += ex->get_length();
@@ -1029,29 +1035,36 @@ RandomBlockOolWriter::do_write(
       trans_stats.data_bytes += ex->get_length();
       w_stats.data_bytes += ex->get_length();
     }
-    return trans_intr::make_interruptible(
-      rbm->write(paddr + offset,
-	bp
-      ).handle_error(
-	alloc_write_iertr::pass_further{},
-	crimson::ct_error::assert_all{
-	  "Invalid error when writing record"}
-      )
-    ).si_then([this, &t, &ex, paddr, FNAME] {
-      TRACET("ool extent written at {} -- {}",
-	     t, paddr, *ex);
-      if (ex->is_initial_pending()) {
-	t.mark_allocated_extent_ool(ex);
-      } else if (can_inplace_rewrite(t, ex)) {
-        assert(ex->is_logical());
-	t.mark_inplace_rewrite_extent_ool(
-          ex->template cast<LogicalCachedExtent>());
-      } else {
-	ceph_assert("impossible");
-      }
-      return alloc_write_iertr::now();
+    writes.push_back(write_info_t{paddr + offset, std::move(bp), rbm});
+
+    if (ex->is_initial_pending()) {
+      t.mark_allocated_extent_ool(ex);
+    } else if (can_inplace_rewrite(t, ex)) {
+      assert(ex->is_logical());
+      t.mark_inplace_rewrite_extent_ool(
+        ex->template cast<LogicalCachedExtent>());
+    } else {
+      ceph_assert("impossible");
+    }
+  }
+
+  // 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),
+      [](auto& writes) {
+      return crimson::do_for_each(writes,
+        [](auto& info) {
+        return info.rbm->write(info.offset, info.bp
+        ).handle_error(
+          alloc_write_ertr::pass_further{},
+          crimson::ct_error::assert_all{
+            "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);
 
-  alloc_write_ertr::future<> write_record(
+  void write_record(
     Transaction& t,
     record_t&& record,
     std::list<LogicalCachedExtentRef> &&extents,
@@ -190,7 +190,12 @@ class RandomBlockOolWriter : public ExtentOolWriter {
   }
 #endif
 private:
-  alloc_write_iertr::future<> do_write(
+  struct write_info_t {
+    paddr_t offset;
+    ceph::bufferptr bp;
+    RandomBlockManager* rbm;
+  };
+  void do_write(
     Transaction& t,
     std::list<CachedExtentRef> &extent);
 

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

@@ -94,9 +94,12 @@ CircularBoundedJournal::do_submit_record(
 	(void*)&handle,
 	action == RecordSubmitter::action_t::SUBMIT_FULL ?
 	"FULL" : "NOT_FULL");
-  auto submit_fut = record_submitter.submit(std::move(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([submit_fut=std::move(submit_fut)]() mutable {
+  ).then([&handle] {
+    return handle.take_write_future();
+  }).safe_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/circular_journal_space.cc

@@ -49,7 +49,7 @@ CircularJournalSpace::roll_ertr::future<> CircularJournalSpace::roll() {
   return roll_ertr::now();
 }
 
-CircularJournalSpace::write_ret
+CircularJournalSpace::write_ertr::future<>
 CircularJournalSpace::write(ceph::bufferlist&& to_write) {
   LOG_PREFIX(CircularJournalSpace::write);
   assert(get_written_to().segment_seq != NULL_SEG_SEQ);
@@ -60,7 +60,6 @@ CircularJournalSpace::write(ceph::bufferlist&& to_write) {
   assert(encoded_size + get_rbm_addr(get_written_to())
 	 < get_journal_end());
 
-  journal_seq_t j_seq = get_written_to();
   auto target = get_rbm_addr(get_written_to());
   auto new_written_to = target + encoded_size;
   assert(new_written_to < get_journal_end());
@@ -69,22 +68,12 @@ CircularJournalSpace::write(ceph::bufferlist&& to_write) {
     get_device_id());
   set_written_to(
     journal_seq_t{get_written_to().segment_seq, paddr});
-  DEBUG("{}, target {}", to_write.length(), target);
+  DEBUG("length {}, commit target {}, used_size {}",
+        encoded_size, target, get_records_used_size());
 
-  auto write_result = write_result_t{
-    j_seq,
-    encoded_size
-  };
   return device_write_bl(target, to_write
-  ).safe_then([this, target,
-    length=encoded_size,
-    write_result,
-    FNAME] {
-    DEBUG("commit target {} used_size {} written length {}",
-          target, get_records_used_size(), length);
-    return write_result;
-  }).handle_error(
-    base_ertr::pass_further{},
+  ).handle_error(
+    write_ertr::pass_further{},
     crimson::ct_error::assert_all{ "Invalid error" }
   );
 }
@@ -167,7 +156,8 @@ ceph::bufferlist CircularJournalSpace::encode_header()
   return bl;
 }
 
-CircularJournalSpace::write_ertr::future<> CircularJournalSpace::device_write_bl(
+CircularJournalSpace::submit_ertr::future<>
+CircularJournalSpace::device_write_bl(
     rbm_abs_addr offset, bufferlist &bl)
 {
   LOG_PREFIX(CircularJournalSpace::device_write_bl);
@@ -181,7 +171,7 @@ CircularJournalSpace::write_ertr::future<> CircularJournalSpace::device_write_bl
     length);
   return device->writev(offset, bl
   ).handle_error(
-    write_ertr::pass_further{},
+    submit_ertr::pass_further{},
     crimson::ct_error::assert_all{ "Invalid error device->write" }
   );
 }
@@ -229,7 +219,7 @@ CircularJournalSpace::read_header()
   });
 }
 
-CircularJournalSpace::write_ertr::future<>
+CircularJournalSpace::submit_ertr::future<>
 CircularJournalSpace::write_header()
 {
   LOG_PREFIX(CircularJournalSpace::write_header);
@@ -245,7 +235,7 @@ CircularJournalSpace::write_header()
   iter.copy(bl.length(), bp.c_str());
   return device->write(device->get_shard_journal_start(), std::move(bp)
   ).handle_error(
-    write_ertr::pass_further{},
+    submit_ertr::pass_further{},
     crimson::ct_error::assert_all{ "Invalid error device->write" }
   );
 }