summaryrefslogtreecommitdiffstats
path: root/src/test/crimson/seastore/test_seastore_journal.cc
blob: 0bed505ffa1d1c7ed10b6e2146bedfbd3e84feba (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
// -*- 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/journal.h"
#include "crimson/os/seastore/segment_manager/ephemeral.h"

using namespace crimson;
using namespace crimson::os;
using namespace crimson::os::seastore;

namespace {
  [[maybe_unused]] seastar::logger& logger() {
    return crimson::get_logger(ceph_subsys_test);
  }
}

struct record_validator_t {
  record_t record;
  paddr_t record_final_offset;

  template <typename... T>
  record_validator_t(T&&... record) : record(std::forward<T>(record)...) {}

  void validate(SegmentManager &manager) {
    paddr_t addr = make_record_relative_paddr(0);
    for (auto &&block : record.extents) {
      auto test = manager.read(
	record_final_offset.add_relative(addr),
	block.bl.length()).unsafe_get0();
      addr.offset += block.bl.length();
      bufferlist bl;
      bl.push_back(test);
      ASSERT_EQ(
	bl.length(),
	block.bl.length());
      ASSERT_EQ(
	bl.begin().crc32c(bl.length(), 1),
	block.bl.begin().crc32c(block.bl.length(), 1));
    }
  }

  auto get_replay_handler() {
    auto checker = [this, iter=record.deltas.begin()] (
      paddr_t base,
      const delta_info_t &di) mutable {
      EXPECT_EQ(base, record_final_offset);
      ceph_assert(iter != record.deltas.end());
      EXPECT_EQ(di, *iter++);
      EXPECT_EQ(base, record_final_offset);
      return iter != record.deltas.end();
    };
    if (record.deltas.size()) {
      return std::make_optional(std::move(checker));
    } else {
      return std::optional<decltype(checker)>();
    }
  }
};

struct journal_test_t : seastar_test_suite_t, JournalSegmentProvider {
  segment_manager::EphemeralSegmentManagerRef segment_manager;
  std::unique_ptr<Journal> journal;

  std::vector<record_validator_t> records;

  std::default_random_engine generator;

  const segment_off_t block_size;

  journal_test_t()
    : segment_manager(segment_manager::create_test_ephemeral()),
      block_size(segment_manager->get_block_size())
  {
  }

  segment_id_t next = 0;
  get_segment_ret get_segment() final {
    return get_segment_ret(
      get_segment_ertr::ready_future_marker{},
      next++);
  }

  journal_seq_t get_journal_tail_target() const final { return journal_seq_t{}; }
  void update_journal_tail_committed(journal_seq_t paddr) final {}

  seastar::future<> set_up_fut() final {
    journal.reset(new Journal(*segment_manager));
    journal->set_segment_provider(this);
    return segment_manager->init(
    ).safe_then([this] {
      return journal->open_for_write();
    }).safe_then(
      [](auto){},
      crimson::ct_error::all_same_way([] {
	ASSERT_FALSE("Unable to mount");
      }));
  }

  template <typename T>
  auto replay(T &&f) {
    return journal->close(
    ).safe_then([this, f=std::move(f)]() mutable {
      journal.reset(new Journal(*segment_manager));
      journal->set_segment_provider(this);
      return journal->replay(std::forward<T>(std::move(f)));
    }).safe_then([this] {
      return journal->open_for_write();
    });
  }

  auto replay_and_check() {
    auto record_iter = records.begin();
    decltype(record_iter->get_replay_handler()) delta_checker = std::nullopt;
    auto advance = [this, &record_iter, &delta_checker] {
      ceph_assert(!delta_checker);
      while (record_iter != records.end()) {
	auto checker = record_iter->get_replay_handler();
	record_iter++;
	if (checker) {
	  delta_checker.emplace(std::move(*checker));
	  break;
	}
      }
    };
    advance();
    replay(
      [&advance,
       &delta_checker]
      (auto seq, auto base, const auto &di) mutable {
	if (!delta_checker) {
	  EXPECT_FALSE("No Deltas Left");
	}
	if (!(*delta_checker)(base, di)) {
	  delta_checker = std::nullopt;
	  advance();
	}
	return Journal::replay_ertr::now();
      }).unsafe_get0();
    ASSERT_EQ(record_iter, records.end());
    for (auto &i : records) {
      i.validate(*segment_manager);
    }
  }

  template <typename... T>
  auto submit_record(T&&... _record) {
    auto record{std::forward<T>(_record)...};
    records.push_back(record);
    auto [addr, _] = journal->submit_record(std::move(record)).unsafe_get0();
    records.back().record_final_offset = addr;
    return addr;
  }

  seastar::future<> tear_down_fut() final {
    return seastar::now();
  }

  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,
      block_size,
      1,
      bl
    };
  }
};

TEST_F(journal_test_t, replay_one_journal_segment)
{
 run_async([this] {
   submit_record(record_t{
     { generate_extent(1), generate_extent(2) },
     { generate_delta(23), generate_delta(30) }
     });
   replay_and_check();
 });
}

TEST_F(journal_test_t, replay_two_records)
{
 run_async([this] {
   submit_record(record_t{
     { generate_extent(1), generate_extent(2) },
     { generate_delta(23), generate_delta(30) }
     });
   submit_record(record_t{
     { generate_extent(4), generate_extent(1) },
     { generate_delta(23), generate_delta(400) }
     });
   replay_and_check();
 });
}

TEST_F(journal_test_t, replay_twice)
{
 run_async([this] {
   submit_record(record_t{
     { generate_extent(1), generate_extent(2) },
     { generate_delta(23), generate_delta(30) }
     });
   submit_record(record_t{
     { generate_extent(4), generate_extent(1) },
     { generate_delta(23), generate_delta(400) }
     });
   replay_and_check();
   submit_record(record_t{
     { generate_extent(2), generate_extent(5) },
     { generate_delta(230), generate_delta(40) }
     });
   replay_and_check();
 });
}

TEST_F(journal_test_t, roll_journal_and_replay)
{
 run_async([this] {
   paddr_t current = submit_record(
     record_t{
       { generate_extent(1), generate_extent(2) },
       { generate_delta(23), generate_delta(30) }
     });
   auto starting_segment = current.segment;
   unsigned so_far = 0;
   while (current.segment == starting_segment) {
     current = submit_record(record_t{
	 { generate_extent(512), generate_extent(512) },
	 { generate_delta(23), generate_delta(400) }
       });
     ++so_far;
     ASSERT_FALSE(so_far > 10);
   }
   replay_and_check();
 });
}