From: Xuehan Xu Date: Fri, 5 Jun 2026 09:01:25 +0000 (+0800) Subject: crimson/os/seastore/journal: change CircularBoundedJournal::replay() to X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=aa9247a36cfc26b791ce10efab0e1eae7624faae;p=ceph.git crimson/os/seastore/journal: change CircularBoundedJournal::replay() to use coroutine Signed-off-by: Xuehan Xu --- diff --git a/src/crimson/os/seastore/journal/circular_bounded_journal.cc b/src/crimson/os/seastore/journal/circular_bounded_journal.cc index 9ab7d629f59..62e6c175aed 100644 --- 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(), - std::map>(), - [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(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(applied); - }); - } - return replay_ertr::make_ready_future(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 map; + std::map> 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(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(applied); + }); + } + return replay_ertr::make_ready_future(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()