crimson/os/seastore/journal: change CircularBoundedJournal::replay() to

use coroutine

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

diff --git a/src/crimson/os/seastore/journal/circular_bounded_journal.cc b/src/crimson/os/seastore/journal/circular_bounded_journal.cc
index 9ab7d629f591513782ca1ff2b67f5fcf70775318..62e6c175aed2d18c7720c011382c54ff0f310064 100644 (file)
--- a/src/crimson/os/seastore/journal/circular_bounded_journal.cc
+++ b/src/crimson/os/seastore/journal/circular_bounded_journal.cc
@@ -332,94 +332,80 @@ Journal::replay_ret CircularBoundedJournal::replay(
   /*
    * read records from last applied record prior to written_to, and replay
    */
+  auto d_handler = std::move(delta_handler);
   LOG_PREFIX(CircularBoundedJournal::replay);
-  return cjs.read_header(
-  ).handle_error(
+  auto p = co_await cjs.read_header().handle_error(
     open_for_mount_ertr::pass_further{},
-    crimson::ct_error::assert_all(
-      "Invalid error read_header"
-  )).safe_then([this, FNAME, delta_handler=std::move(delta_handler)](auto p)
-    mutable {
-    auto &[head, bl] = *p;
-    cjs.set_cbj_header(head);
-    DEBUG("header : {}", cjs.get_cbj_header());
-    cjs.set_initialized(true);
-    return seastar::do_with(
-      std::move(delta_handler),
-      std::map<paddr_t, journal_seq_t>(),
-      std::map<paddr_t, std::pair<CachedExtentRef, uint32_t>>(),
-      [this](auto &d_handler, auto &map, auto &crc_info) {
-      auto build_paddr_seq_map = [&map](
-        const auto &offsets,
-        const auto &e,
-       sea_time_point modify_time)
-      {
-       if (e.type == extent_types_t::ALLOC_INFO) {
-         alloc_delta_t alloc_delta;
-         decode(alloc_delta, e.bl);
-         if (alloc_delta.op == alloc_delta_t::op_types_t::CLEAR) {
-           for (auto &alloc_blk : alloc_delta.alloc_blk_ranges) {
-             map[alloc_blk.paddr] = offsets.write_result.start_seq;
-           }
-         }
-       }
-       return replay_ertr::make_ready_future<bool>(true);
-      };
-      auto tail = get_dirty_tail() <= get_alloc_tail() ?
-       get_dirty_tail() : get_alloc_tail();
-      set_written_to(tail);
-      // The first pass to build the paddr->journal_seq_t map 
-      // from extent allocations
-      return scan_valid_record_delta(std::move(build_paddr_seq_map), tail
-      ).safe_then([this, &map, &d_handler, tail, &crc_info]() {
-       auto call_d_handler_if_valid = [this, &map, &d_handler, &crc_info](
-         const auto &offsets,
-         const auto &e,
-         sea_time_point modify_time)
-       {
-         if (map.find(e.paddr) == map.end() ||
-             map[e.paddr] <= offsets.write_result.start_seq) {
-           return d_handler(
-             offsets,
-             e,
-             get_dirty_tail(),
-             get_alloc_tail(),
-             modify_time
-           ).safe_then([&e, &crc_info](auto ret) {
-             auto [applied, ext] = ret;
-             if (applied && ext && can_inplace_rewrite(
-                 ext->get_type())) {
-               crc_info[ext->get_paddr()] =
-                 std::make_pair(ext, e.final_crc);
-             }
-             return replay_ertr::make_ready_future<bool>(applied);
-           });
-         }
-         return replay_ertr::make_ready_future<bool>(true);
-       };
-       // The second pass to replay deltas
-       return scan_valid_record_delta(std::move(call_d_handler_if_valid), tail
-       ).safe_then([&crc_info]() {
-         for (auto p : crc_info) {
-           ceph_assert_always(p.second.first->get_last_committed_crc() == p.second.second);    
-         }
-         crc_info.clear();
-         return replay_ertr::now();
-       });
-      });
-    }).safe_then([this]() {
-      // make sure that committed_to is JOURNAL_SEQ_NULL if jounal is the initial state
-      if (get_written_to() != 
-         journal_seq_t{0,
-           convert_abs_addr_to_paddr(get_records_start(),
-           get_device_id())}) {
-       record_submitter.update_committed_to(get_written_to());
+    crimson::ct_error::assert_all("Invalid error read_header"));
+  auto &[head, bl] = *p;
+  cjs.set_cbj_header(head);
+  DEBUG("header : {}", cjs.get_cbj_header());
+  cjs.set_initialized(true);
+  std::map<paddr_t, journal_seq_t> map;
+  std::map<paddr_t, std::pair<CachedExtentRef, uint32_t>> crc_info;
+  auto build_paddr_seq_map = [&map](
+    const auto &offsets,
+    const auto &e,
+    sea_time_point modify_time)
+  {
+    if (e.type == extent_types_t::ALLOC_INFO) {
+      alloc_delta_t alloc_delta;
+      decode(alloc_delta, e.bl);
+      if (alloc_delta.op == alloc_delta_t::op_types_t::CLEAR) {
+        for (auto &alloc_blk : alloc_delta.alloc_blk_ranges) {
+          map[alloc_blk.paddr] = offsets.write_result.start_seq;
+        }
       }
-      trimmer.update_journal_tails(
-       get_dirty_tail(),
-       get_alloc_tail());
-    });
-  });
+    }
+    return replay_ertr::make_ready_future<bool>(true);
+  };
+  auto tail = get_dirty_tail() <= get_alloc_tail() ?
+    get_dirty_tail() : get_alloc_tail();
+  set_written_to(tail);
+  // The first pass to build the paddr->journal_seq_t map 
+  // from extent allocations
+  co_await scan_valid_record_delta(std::move(build_paddr_seq_map), tail);
+  auto call_d_handler_if_valid = [this, &map, &d_handler, &crc_info](
+    const auto &offsets,
+    const auto &e,
+    sea_time_point modify_time)
+  {
+    if (map.find(e.paddr) == map.end() ||
+        map[e.paddr] <= offsets.write_result.start_seq) {
+      return d_handler(
+        offsets,
+        e,
+        get_dirty_tail(),
+        get_alloc_tail(),
+        modify_time
+      ).safe_then([&e, &crc_info](auto ret) {
+        auto [applied, ext] = ret;
+        if (applied && ext && can_inplace_rewrite(
+            ext->get_type())) {
+          crc_info[ext->get_paddr()] =
+            std::make_pair(ext, e.final_crc);
+        }
+        return replay_ertr::make_ready_future<bool>(applied);
+      });
+    }
+    return replay_ertr::make_ready_future<bool>(true);
+  };
+  // The second pass to replay deltas
+  co_await scan_valid_record_delta(std::move(call_d_handler_if_valid), tail);
+  for (auto p : crc_info) {
+    ceph_assert_always(p.second.first->get_last_committed_crc() == p.second.second);   
+  }
+  crc_info.clear();
+  // make sure that committed_to is JOURNAL_SEQ_NULL if jounal is the initial state
+  if (get_written_to() != 
+      journal_seq_t{0,
+        convert_abs_addr_to_paddr(get_records_start(),
+        get_device_id())}) {
+    record_submitter.update_committed_to(get_written_to());
+  }
+  trimmer.update_journal_tails(
+    get_dirty_tail(),
+    get_alloc_tail());
 }
 
 void CircularBoundedJournal::register_metrics()