--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "test/crimson/gtest_seastar.h"
+
+#include <random>
+
+#include "crimson/common/log.h"
+#include "crimson/os/seastore/async_cleaner.h"
+#include "crimson/os/seastore/journal.h"
+#include "crimson/os/seastore/journal/circular_bounded_journal.h"
+#include "crimson/os/seastore/random_block_manager.h"
+#include "crimson/os/seastore/random_block_manager/rbm_device.h"
+#include "crimson/os/seastore/seastore_types.h"
+#include "test/crimson/seastore/transaction_manager_test_state.h"
+#include "crimson/os/seastore/random_block_manager/block_rb_manager.h"
+
+using namespace crimson;
+using namespace crimson::os;
+using namespace crimson::os::seastore;
+using namespace crimson::os::seastore::journal;
+
+namespace {
+ [[maybe_unused]] seastar::logger& logger() {
+ return crimson::get_logger(ceph_subsys_test);
+ }
+}
+
+std::optional<record_t> decode_record(
+ bufferlist& bl)
+{
+ record_t record;
+ record_group_header_t r_header;
+ auto bliter = bl.cbegin();
+ decode(r_header, bliter);
+ logger().debug(" decode_record mdlength {} records {}",
+ r_header.mdlength, r_header.records);
+ device_id_t d_id = 1 << (std::numeric_limits<device_id_t>::digits - 1);
+
+ auto del_infos = try_decode_deltas(r_header, bl,
+ paddr_t::make_blk_paddr(d_id, 0));
+ for (auto &iter : *del_infos) {
+ for (auto r : iter.deltas) {
+ record.deltas.push_back(r.second);
+ }
+ }
+ auto ex_infos = try_decode_extent_infos(r_header, bl);
+ auto bliter_ex = bl.cbegin();
+ bliter_ex += r_header.mdlength;
+ for (auto &iter: *ex_infos) {
+ for (auto e : iter.extent_infos) {
+ extent_t ex;
+ auto bptr = bufferptr(ceph::buffer::create_page_aligned(e.len));
+ logger().debug(" exten len {} remaining {} ", e.len, bliter_ex.get_remaining());
+ bliter_ex.copy(e.len, bptr.c_str());
+ ex.bl.append(bptr);
+ record.extents.push_back(ex);
+ }
+ }
+ return record;
+}
+
+struct entry_validator_t {
+ bufferlist bl;
+ int entries;
+ journal_seq_t last_seq;
+ record_t record;
+ rbm_abs_addr addr = 0;
+
+ template <typename... T>
+ entry_validator_t(T&&... entry) : record(std::forward<T>(entry)...) {}
+
+ void validate(record_t read) {
+ auto iter = read.extents.begin();
+ for (auto &&block : record.extents) {
+ ASSERT_EQ(
+ iter->bl.length(),
+ block.bl.length());
+ ASSERT_EQ(
+ iter->bl.begin().crc32c(iter->bl.length(), 1),
+ block.bl.begin().crc32c(block.bl.length(), 1));
+ ++iter;
+ }
+ auto iter_delta = read.deltas.begin();
+ for (auto &&block : record.deltas) {
+ ASSERT_EQ(
+ iter_delta->bl.length(),
+ block.bl.length());
+ ASSERT_EQ(
+ iter_delta->bl.begin().crc32c(iter_delta->bl.length(), 1),
+ block.bl.begin().crc32c(block.bl.length(), 1));
+ ++iter_delta;
+ }
+ }
+
+ void validate(CircularBoundedJournal &cbj) {
+ rbm_abs_addr offset = 0;
+ for (int i = 0; i < entries; i++) {
+ paddr_t paddr = convert_abs_addr_to_paddr(
+ addr + offset,
+ cbj.get_device_id());
+ auto [header, buf] = *(cbj.read_record(paddr, NULL_SEG_SEQ).unsafe_get0());
+ auto record = decode_record(buf);
+ validate(*record);
+ offset += header.mdlength + header.dlength;
+ }
+ }
+
+ bool validate_delta(bufferlist bl) {
+ for (auto &&block : record.deltas) {
+ if (bl.begin().crc32c(bl.length(), 1) ==
+ block.bl.begin().crc32c(block.bl.length(), 1)) {
+ return true;
+ }
+ }
+ return false;
+ }
+};
+
+struct cbjournal_test_t : public seastar_test_suite_t, JournalTrimmer
+{
+ std::vector<entry_validator_t> entries;
+ std::unique_ptr<CircularBoundedJournal> cbj;
+ random_block_device::EphemeralRBMDeviceRef device;
+
+ std::default_random_engine generator;
+ uint64_t block_size;
+ WritePipeline pipeline;
+
+ cbjournal_test_t() = default;
+
+ /*
+ * JournalTrimmer interfaces
+ */
+ journal_seq_t get_journal_head() const {
+ return JOURNAL_SEQ_NULL;
+ }
+
+ journal_seq_t get_dirty_tail() const final {
+ return JOURNAL_SEQ_NULL;
+ }
+
+ journal_seq_t get_alloc_tail() const final {
+ return JOURNAL_SEQ_NULL;
+ }
+
+ void set_journal_head(journal_seq_t head) final {}
+
+ void update_journal_tails(
+ journal_seq_t dirty_tail,
+ journal_seq_t alloc_tail) final {}
+
+ bool try_reserve_inline_usage(std::size_t) final { return true; }
+
+ void release_inline_usage(std::size_t) final {}
+
+ std::size_t get_trim_size_per_cycle() const final {
+ return 0;
+ }
+
+ auto submit_record(record_t&& record) {
+ entries.push_back(record);
+ OrderingHandle handle = get_dummy_ordering_handle();
+ auto [addr, w_result] = cbj->submit_record(
+ std::move(record),
+ handle).unsafe_get0();
+ entries.back().addr =
+ convert_paddr_to_abs_addr(w_result.start_seq.offset);
+ entries.back().entries = 1;
+ logger().debug("submit entry to addr {}", entries.back().addr);
+ return entries.back().addr;
+ }
+
+ seastar::future<> tear_down_fut() final {
+ return close();
+ }
+
+ extent_t generate_extent(size_t blocks) {
+ std::uniform_int_distribution<char> distribution(
+ std::numeric_limits<char>::min(),
+ std::numeric_limits<char>::max()
+ );
+ char contents = distribution(generator);
+ bufferlist bl;
+ bl.append(buffer::ptr(buffer::create(blocks * block_size, contents)));
+ return extent_t{extent_types_t::TEST_BLOCK, L_ADDR_NULL, bl};
+ }
+
+ delta_info_t generate_delta(size_t bytes) {
+ std::uniform_int_distribution<char> distribution(
+ std::numeric_limits<char>::min(),
+ std::numeric_limits<char>::max()
+ );
+ char contents = distribution(generator);
+ bufferlist bl;
+ bl.append(buffer::ptr(buffer::create(bytes, contents)));
+ return delta_info_t{
+ extent_types_t::TEST_BLOCK,
+ paddr_t{},
+ L_ADDR_NULL,
+ 0, 0,
+ device->get_block_size(),
+ 1,
+ 0,
+ segment_type_t::JOURNAL,
+ bl
+ };
+ }
+
+ auto replay_and_check() {
+ for (auto &i : entries) {
+ i.validate(*(cbj.get()));
+ }
+ }
+
+ auto replay() {
+ return cbj->replay(
+ [this](const auto &offsets,
+ const auto &e,
+ auto &dirty_seq,
+ auto &alloc_seq,
+ auto last_modified) {
+ bool found = false;
+ for (auto &i : entries) {
+ paddr_t base = offsets.write_result.start_seq.offset;
+ rbm_abs_addr addr = convert_paddr_to_abs_addr(base);
+ if (addr == i.addr) {
+ logger().debug(" compare addr: {} and i.addr {} ", base, i.addr);
+ found = i.validate_delta(e.bl);
+ break;
+ }
+ }
+ assert(found == true);
+ return Journal::replay_ertr::make_ready_future<bool>(true);
+ });
+ }
+
+ auto mkfs() {
+ device_config_t config = get_rbm_ephemeral_device_config(0, 1);
+ return device->mkfs(config
+ ).safe_then([this]() {
+ return device->mount(
+ ).safe_then([this]() {
+ return cbj->open_for_mkfs(
+ ).safe_then([](auto q) {
+ return seastar::now();
+ });
+ });
+ }).safe_then([this] {
+ return cbj->close();
+ });
+ }
+ auto open() {
+ return cbj->open_for_mount(
+ ).safe_then([](auto q) {
+ return seastar::now();
+ });
+ }
+ seastar::future<> close() {
+ return cbj->close().handle_error(crimson::ct_error::assert_all{});
+ }
+ auto get_records_available_size() {
+ return cbj->get_cjs().get_records_available_size();
+ }
+ auto get_records_total_size() {
+ return cbj->get_cjs().get_records_total_size();
+ }
+ auto get_block_size() {
+ return device->get_block_size();
+ }
+ auto get_written_to_rbm_addr() {
+ return cbj->get_rbm_addr(cbj->get_cjs().get_written_to());
+ }
+ auto get_written_to() {
+ return cbj->get_cjs().get_written_to();
+ }
+ auto get_journal_tail() {
+ return cbj->get_dirty_tail();
+ }
+ auto get_records_used_size() {
+ return cbj->get_cjs().get_records_used_size();
+ }
+ bool is_available_size(uint64_t size) {
+ return cbj->get_cjs().is_available_size(size);
+ }
+ void update_journal_tail(rbm_abs_addr addr, uint32_t len) {
+ paddr_t paddr =
+ convert_abs_addr_to_paddr(
+ addr + len,
+ cbj->get_device_id());
+ journal_seq_t seq = {0, paddr};
+ cbj->update_journal_tail(
+ seq,
+ seq
+ ).get0();
+ }
+ void set_written_to(journal_seq_t seq) {
+ cbj->set_written_to(seq);
+ }
+
+ seastar::future<> set_up_fut() final {
+ device = random_block_device::create_test_ephemeral(
+ random_block_device::DEFAULT_TEST_CBJOURNAL_SIZE, 0);
+ cbj.reset(new CircularBoundedJournal(*this, device.get(), std::string()));
+ block_size = device->get_block_size();
+ cbj->set_write_pipeline(&pipeline);
+ return mkfs(
+ ).safe_then([this] {
+ return replay(
+ ).safe_then([this] {
+ return open();
+ });
+ }).handle_error(crimson::ct_error::assert_all{});
+ }
+};
+
+TEST_F(cbjournal_test_t, submit_one_record)
+{
+ run_async([this] {
+ submit_record(
+ record_t{
+ { generate_extent(1), generate_extent(2) },
+ { generate_delta(3), generate_delta(4) }
+ });
+ replay_and_check();
+ });
+}
+
+TEST_F(cbjournal_test_t, submit_three_records)
+{
+ run_async([this] {
+ submit_record(
+ record_t{
+ { generate_extent(1), generate_extent(2) },
+ { generate_delta(3), generate_delta(4) }
+ });
+ submit_record(
+ record_t{
+ { generate_extent(8), generate_extent(9) },
+ { generate_delta(20), generate_delta(21) }
+ });
+ submit_record(
+ record_t{
+ { generate_extent(5), generate_extent(6) },
+ { generate_delta(200), generate_delta(210) }
+ });
+ replay_and_check();
+ });
+}
+
+TEST_F(cbjournal_test_t, submit_full_records)
+{
+ run_async([this] {
+ record_t rec {
+ { generate_extent(1), generate_extent(2) },
+ { generate_delta(20), generate_delta(21) }
+ };
+ auto r_size = record_group_size_t(rec.size, block_size);
+ auto record_total_size = r_size.get_encoded_length();
+
+ submit_record(std::move(rec));
+ while (is_available_size(record_total_size)) {
+ submit_record(
+ record_t {
+ { generate_extent(1), generate_extent(2) },
+ { generate_delta(20), generate_delta(21) }
+ });
+ }
+
+ update_journal_tail(entries.back().addr, record_total_size);
+ ASSERT_EQ(get_records_total_size(),
+ get_records_available_size());
+
+ // will be appended at the begining of log
+ submit_record(
+ record_t {
+ { generate_extent(1), generate_extent(2) },
+ { generate_delta(20), generate_delta(21) }
+ });
+
+ while (is_available_size(record_total_size)) {
+ submit_record(
+ record_t {
+ { generate_extent(1), generate_extent(2) },
+ { generate_delta(20), generate_delta(21) }
+ });
+ }
+ ASSERT_TRUE(record_total_size > get_records_available_size());
+ });
+}
+
+TEST_F(cbjournal_test_t, boudary_check_verify)
+{
+ run_async([this] {
+ record_t rec {
+ { generate_extent(1), generate_extent(2) },
+ { generate_delta(20), generate_delta(21) }
+ };
+ auto r_size = record_group_size_t(rec.size, block_size);
+ auto record_total_size = r_size.get_encoded_length();
+ submit_record(std::move(rec));
+ while (is_available_size(record_total_size)) {
+ submit_record(
+ record_t {
+ { generate_extent(1), generate_extent(2) },
+ { generate_delta(20), generate_delta(21) }
+ });
+ }
+
+ uint64_t avail = get_records_available_size();
+ // forward 2 recod size here because 1 block is reserved between head and tail
+ update_journal_tail(entries.front().addr, record_total_size * 2);
+ entries.erase(entries.begin());
+ entries.erase(entries.begin());
+ ASSERT_EQ(avail + (record_total_size * 2), get_records_available_size());
+ avail = get_records_available_size();
+ // will be appended at the begining of WAL
+ submit_record(
+ record_t {
+ { generate_extent(1), generate_extent(2) },
+ { generate_delta(20), generate_delta(21) }
+ });
+ ASSERT_TRUE(avail - record_total_size >= get_records_available_size());
+ replay_and_check();
+ });
+}
+
+TEST_F(cbjournal_test_t, update_header)
+{
+ run_async([this] {
+ auto [header, _buf] = *(cbj->get_cjs().read_header().unsafe_get0());
+ record_t rec {
+ { generate_extent(1), generate_extent(2) },
+ { generate_delta(20), generate_delta(21) }
+ };
+ auto r_size = record_group_size_t(rec.size, block_size);
+ auto record_total_size = r_size.get_encoded_length();
+ submit_record(std::move(rec));
+
+ update_journal_tail(entries.front().addr, record_total_size);
+ cbj->get_cjs().write_header().unsafe_get0();
+ auto [update_header, update_buf2] = *(cbj->get_cjs().read_header().unsafe_get0());
+ cbj->close().unsafe_get0();
+ replay().unsafe_get0();
+
+ ASSERT_EQ(update_header.dirty_tail.offset, update_header.dirty_tail.offset);
+ });
+}
+
+TEST_F(cbjournal_test_t, replay)
+{
+ run_async([this] {
+ record_t rec {
+ { generate_extent(1), generate_extent(2) },
+ { generate_delta(20), generate_delta(21) }
+ };
+ auto r_size = record_group_size_t(rec.size, block_size);
+ auto record_total_size = r_size.get_encoded_length();
+ submit_record(std::move(rec));
+ while (is_available_size(record_total_size)) {
+ submit_record(
+ record_t {
+ { generate_extent(1), generate_extent(2) },
+ { generate_delta(20), generate_delta(21) }
+ });
+ }
+ // will be appended at the begining of WAL
+ uint64_t avail = get_records_available_size();
+ update_journal_tail(entries.front().addr, record_total_size * 2);
+ entries.erase(entries.begin());
+ entries.erase(entries.begin());
+ ASSERT_EQ(avail + (record_total_size * 2), get_records_available_size());
+ avail = get_records_available_size();
+ submit_record(
+ record_t {
+ { generate_extent(1), generate_extent(2) },
+ { generate_delta(20), generate_delta(21) }
+ });
+ ASSERT_TRUE(avail - record_total_size >= get_records_available_size());
+ cbj->close().unsafe_get0();
+ replay().unsafe_get0();
+ });
+}
+
+TEST_F(cbjournal_test_t, replay_after_reset)
+{
+ run_async([this] {
+ record_t rec {
+ { generate_extent(1), generate_extent(2) },
+ { generate_delta(20), generate_delta(21) }
+ };
+ auto r_size = record_group_size_t(rec.size, block_size);
+ auto record_total_size = r_size.get_encoded_length();
+ submit_record(std::move(rec));
+ while (is_available_size(record_total_size)) {
+ submit_record(
+ record_t {
+ { generate_extent(1), generate_extent(2) },
+ { generate_delta(20), generate_delta(21) }
+ });
+ }
+ auto old_written_to = get_written_to();
+ auto old_used_size = get_records_used_size();
+ set_written_to(
+ journal_seq_t{0,
+ convert_abs_addr_to_paddr(
+ cbj->get_records_start(),
+ cbj->get_device_id())});
+ cbj->close().unsafe_get0();
+ replay().unsafe_get0();
+ ASSERT_EQ(old_written_to, get_written_to());
+ ASSERT_EQ(old_used_size,
+ get_records_used_size());
+ });
+}
projects / ceph.git / blobdiff