log_test.cc raw
1 // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
2 // Use of this source code is governed by a BSD-style license that can be
3 // found in the LICENSE file. See the AUTHORS file for names of contributors.
4
5 #include "db/log_reader.h"
6 #include "db/log_writer.h"
7 #include "leveldb/env.h"
8 #include "util/coding.h"
9 #include "util/crc32c.h"
10 #include "util/random.h"
11 #include "util/testharness.h"
12
13 namespace leveldb {
14 namespace log {
15
16 // Construct a string of the specified length made out of the supplied
17 // partial string.
18 static std::string BigString(const std::string& partial_string, size_t n) {
19 std::string result;
20 while (result.size() < n) {
21 result.append(partial_string);
22 }
23 result.resize(n);
24 return result;
25 }
26
27 // Construct a string from a number
28 static std::string NumberString(int n) {
29 char buf[50];
30 snprintf(buf, sizeof(buf), "%d.", n);
31 return std::string(buf);
32 }
33
34 // Return a skewed potentially long string
35 static std::string RandomSkewedString(int i, Random* rnd) {
36 return BigString(NumberString(i), rnd->Skewed(17));
37 }
38
39 class LogTest {
40 public:
41 LogTest()
42 : reading_(false),
43 writer_(new Writer(&dest_)),
44 reader_(new Reader(&source_, &report_, true /*checksum*/,
45 0 /*initial_offset*/)) {}
46
47 ~LogTest() {
48 delete writer_;
49 delete reader_;
50 }
51
52 void ReopenForAppend() {
53 delete writer_;
54 writer_ = new Writer(&dest_, dest_.contents_.size());
55 }
56
57 void Write(const std::string& msg) {
58 ASSERT_TRUE(!reading_) << "Write() after starting to read";
59 writer_->AddRecord(Slice(msg));
60 }
61
62 size_t WrittenBytes() const { return dest_.contents_.size(); }
63
64 std::string Read() {
65 if (!reading_) {
66 reading_ = true;
67 source_.contents_ = Slice(dest_.contents_);
68 }
69 std::string scratch;
70 Slice record;
71 if (reader_->ReadRecord(&record, &scratch)) {
72 return record.ToString();
73 } else {
74 return "EOF";
75 }
76 }
77
78 void IncrementByte(int offset, int delta) {
79 dest_.contents_[offset] += delta;
80 }
81
82 void SetByte(int offset, char new_byte) {
83 dest_.contents_[offset] = new_byte;
84 }
85
86 void ShrinkSize(int bytes) {
87 dest_.contents_.resize(dest_.contents_.size() - bytes);
88 }
89
90 void FixChecksum(int header_offset, int len) {
91 // Compute crc of type/len/data
92 uint32_t crc = crc32c::Value(&dest_.contents_[header_offset + 6], 1 + len);
93 crc = crc32c::Mask(crc);
94 EncodeFixed32(&dest_.contents_[header_offset], crc);
95 }
96
97 void ForceError() { source_.force_error_ = true; }
98
99 size_t DroppedBytes() const { return report_.dropped_bytes_; }
100
101 std::string ReportMessage() const { return report_.message_; }
102
103 // Returns OK iff recorded error message contains "msg"
104 std::string MatchError(const std::string& msg) const {
105 if (report_.message_.find(msg) == std::string::npos) {
106 return report_.message_;
107 } else {
108 return "OK";
109 }
110 }
111
112 void WriteInitialOffsetLog() {
113 for (int i = 0; i < num_initial_offset_records_; i++) {
114 std::string record(initial_offset_record_sizes_[i],
115 static_cast<char>('a' + i));
116 Write(record);
117 }
118 }
119
120 void StartReadingAt(uint64_t initial_offset) {
121 delete reader_;
122 reader_ = new Reader(&source_, &report_, true /*checksum*/, initial_offset);
123 }
124
125 void CheckOffsetPastEndReturnsNoRecords(uint64_t offset_past_end) {
126 WriteInitialOffsetLog();
127 reading_ = true;
128 source_.contents_ = Slice(dest_.contents_);
129 Reader* offset_reader = new Reader(&source_, &report_, true /*checksum*/,
130 WrittenBytes() + offset_past_end);
131 Slice record;
132 std::string scratch;
133 ASSERT_TRUE(!offset_reader->ReadRecord(&record, &scratch));
134 delete offset_reader;
135 }
136
137 void CheckInitialOffsetRecord(uint64_t initial_offset,
138 int expected_record_offset) {
139 WriteInitialOffsetLog();
140 reading_ = true;
141 source_.contents_ = Slice(dest_.contents_);
142 Reader* offset_reader =
143 new Reader(&source_, &report_, true /*checksum*/, initial_offset);
144
145 // Read all records from expected_record_offset through the last one.
146 ASSERT_LT(expected_record_offset, num_initial_offset_records_);
147 for (; expected_record_offset < num_initial_offset_records_;
148 ++expected_record_offset) {
149 Slice record;
150 std::string scratch;
151 ASSERT_TRUE(offset_reader->ReadRecord(&record, &scratch));
152 ASSERT_EQ(initial_offset_record_sizes_[expected_record_offset],
153 record.size());
154 ASSERT_EQ(initial_offset_last_record_offsets_[expected_record_offset],
155 offset_reader->LastRecordOffset());
156 ASSERT_EQ((char)('a' + expected_record_offset), record.data()[0]);
157 }
158 delete offset_reader;
159 }
160
161 private:
162 class StringDest : public WritableFile {
163 public:
164 Status Close() override { return Status::OK(); }
165 Status Flush() override { return Status::OK(); }
166 Status Sync() override { return Status::OK(); }
167 Status Append(const Slice& slice) override {
168 contents_.append(slice.data(), slice.size());
169 return Status::OK();
170 }
171 std::string GetName() const override { return ""; }
172
173 std::string contents_;
174 };
175
176 class StringSource : public SequentialFile {
177 public:
178 StringSource() : force_error_(false), returned_partial_(false) {}
179
180 Status Read(size_t n, Slice* result, char* scratch) override {
181 ASSERT_TRUE(!returned_partial_) << "must not Read() after eof/error";
182
183 if (force_error_) {
184 force_error_ = false;
185 returned_partial_ = true;
186 return Status::Corruption("read error");
187 }
188
189 if (contents_.size() < n) {
190 n = contents_.size();
191 returned_partial_ = true;
192 }
193 *result = Slice(contents_.data(), n);
194 contents_.remove_prefix(n);
195 return Status::OK();
196 }
197
198 Status Skip(uint64_t n) override {
199 if (n > contents_.size()) {
200 contents_.clear();
201 return Status::NotFound("in-memory file skipped past end");
202 }
203
204 contents_.remove_prefix(n);
205
206 return Status::OK();
207 }
208 std::string GetName() const override { return ""; }
209
210 Slice contents_;
211 bool force_error_;
212 bool returned_partial_;
213 };
214
215 class ReportCollector : public Reader::Reporter {
216 public:
217 ReportCollector() : dropped_bytes_(0) {}
218 void Corruption(size_t bytes, const Status& status) override {
219 dropped_bytes_ += bytes;
220 message_.append(status.ToString());
221 }
222
223 size_t dropped_bytes_;
224 std::string message_;
225 };
226
227 // Record metadata for testing initial offset functionality
228 static size_t initial_offset_record_sizes_[];
229 static uint64_t initial_offset_last_record_offsets_[];
230 static int num_initial_offset_records_;
231
232 StringDest dest_;
233 StringSource source_;
234 ReportCollector report_;
235 bool reading_;
236 Writer* writer_;
237 Reader* reader_;
238 };
239
240 size_t LogTest::initial_offset_record_sizes_[] = {
241 10000, // Two sizable records in first block
242 10000,
243 2 * log::kBlockSize - 1000, // Span three blocks
244 1,
245 13716, // Consume all but two bytes of block 3.
246 log::kBlockSize - kHeaderSize, // Consume the entirety of block 4.
247 };
248
249 uint64_t LogTest::initial_offset_last_record_offsets_[] = {
250 0,
251 kHeaderSize + 10000,
252 2 * (kHeaderSize + 10000),
253 2 * (kHeaderSize + 10000) + (2 * log::kBlockSize - 1000) + 3 * kHeaderSize,
254 2 * (kHeaderSize + 10000) + (2 * log::kBlockSize - 1000) + 3 * kHeaderSize +
255 kHeaderSize + 1,
256 3 * log::kBlockSize,
257 };
258
259 // LogTest::initial_offset_last_record_offsets_ must be defined before this.
260 int LogTest::num_initial_offset_records_ =
261 sizeof(LogTest::initial_offset_last_record_offsets_) / sizeof(uint64_t);
262
263 TEST(LogTest, Empty) { ASSERT_EQ("EOF", Read()); }
264
265 TEST(LogTest, ReadWrite) {
266 Write("foo");
267 Write("bar");
268 Write("");
269 Write("xxxx");
270 ASSERT_EQ("foo", Read());
271 ASSERT_EQ("bar", Read());
272 ASSERT_EQ("", Read());
273 ASSERT_EQ("xxxx", Read());
274 ASSERT_EQ("EOF", Read());
275 ASSERT_EQ("EOF", Read()); // Make sure reads at eof work
276 }
277
278 TEST(LogTest, ManyBlocks) {
279 for (int i = 0; i < 100000; i++) {
280 Write(NumberString(i));
281 }
282 for (int i = 0; i < 100000; i++) {
283 ASSERT_EQ(NumberString(i), Read());
284 }
285 ASSERT_EQ("EOF", Read());
286 }
287
288 TEST(LogTest, Fragmentation) {
289 Write("small");
290 Write(BigString("medium", 50000));
291 Write(BigString("large", 100000));
292 ASSERT_EQ("small", Read());
293 ASSERT_EQ(BigString("medium", 50000), Read());
294 ASSERT_EQ(BigString("large", 100000), Read());
295 ASSERT_EQ("EOF", Read());
296 }
297
298 TEST(LogTest, MarginalTrailer) {
299 // Make a trailer that is exactly the same length as an empty record.
300 const int n = kBlockSize - 2 * kHeaderSize;
301 Write(BigString("foo", n));
302 ASSERT_EQ(kBlockSize - kHeaderSize, WrittenBytes());
303 Write("");
304 Write("bar");
305 ASSERT_EQ(BigString("foo", n), Read());
306 ASSERT_EQ("", Read());
307 ASSERT_EQ("bar", Read());
308 ASSERT_EQ("EOF", Read());
309 }
310
311 TEST(LogTest, MarginalTrailer2) {
312 // Make a trailer that is exactly the same length as an empty record.
313 const int n = kBlockSize - 2 * kHeaderSize;
314 Write(BigString("foo", n));
315 ASSERT_EQ(kBlockSize - kHeaderSize, WrittenBytes());
316 Write("bar");
317 ASSERT_EQ(BigString("foo", n), Read());
318 ASSERT_EQ("bar", Read());
319 ASSERT_EQ("EOF", Read());
320 ASSERT_EQ(0, DroppedBytes());
321 ASSERT_EQ("", ReportMessage());
322 }
323
324 TEST(LogTest, ShortTrailer) {
325 const int n = kBlockSize - 2 * kHeaderSize + 4;
326 Write(BigString("foo", n));
327 ASSERT_EQ(kBlockSize - kHeaderSize + 4, WrittenBytes());
328 Write("");
329 Write("bar");
330 ASSERT_EQ(BigString("foo", n), Read());
331 ASSERT_EQ("", Read());
332 ASSERT_EQ("bar", Read());
333 ASSERT_EQ("EOF", Read());
334 }
335
336 TEST(LogTest, AlignedEof) {
337 const int n = kBlockSize - 2 * kHeaderSize + 4;
338 Write(BigString("foo", n));
339 ASSERT_EQ(kBlockSize - kHeaderSize + 4, WrittenBytes());
340 ASSERT_EQ(BigString("foo", n), Read());
341 ASSERT_EQ("EOF", Read());
342 }
343
344 TEST(LogTest, OpenForAppend) {
345 Write("hello");
346 ReopenForAppend();
347 Write("world");
348 ASSERT_EQ("hello", Read());
349 ASSERT_EQ("world", Read());
350 ASSERT_EQ("EOF", Read());
351 }
352
353 TEST(LogTest, RandomRead) {
354 const int N = 500;
355 Random write_rnd(301);
356 for (int i = 0; i < N; i++) {
357 Write(RandomSkewedString(i, &write_rnd));
358 }
359 Random read_rnd(301);
360 for (int i = 0; i < N; i++) {
361 ASSERT_EQ(RandomSkewedString(i, &read_rnd), Read());
362 }
363 ASSERT_EQ("EOF", Read());
364 }
365
366 // Tests of all the error paths in log_reader.cc follow:
367
368 TEST(LogTest, ReadError) {
369 Write("foo");
370 ForceError();
371 ASSERT_EQ("EOF", Read());
372 ASSERT_EQ(kBlockSize, DroppedBytes());
373 ASSERT_EQ("OK", MatchError("read error"));
374 }
375
376 TEST(LogTest, BadRecordType) {
377 Write("foo");
378 // Type is stored in header[6]
379 IncrementByte(6, 100);
380 FixChecksum(0, 3);
381 ASSERT_EQ("EOF", Read());
382 ASSERT_EQ(3, DroppedBytes());
383 ASSERT_EQ("OK", MatchError("unknown record type"));
384 }
385
386 TEST(LogTest, TruncatedTrailingRecordIsIgnored) {
387 Write("foo");
388 ShrinkSize(4); // Drop all payload as well as a header byte
389 ASSERT_EQ("EOF", Read());
390 // Truncated last record is ignored, not treated as an error.
391 ASSERT_EQ(0, DroppedBytes());
392 ASSERT_EQ("", ReportMessage());
393 }
394
395 TEST(LogTest, BadLength) {
396 const int kPayloadSize = kBlockSize - kHeaderSize;
397 Write(BigString("bar", kPayloadSize));
398 Write("foo");
399 // Least significant size byte is stored in header[4].
400 IncrementByte(4, 1);
401 ASSERT_EQ("foo", Read());
402 ASSERT_EQ(kBlockSize, DroppedBytes());
403 ASSERT_EQ("OK", MatchError("bad record length"));
404 }
405
406 TEST(LogTest, BadLengthAtEndIsIgnored) {
407 Write("foo");
408 ShrinkSize(1);
409 ASSERT_EQ("EOF", Read());
410 ASSERT_EQ(0, DroppedBytes());
411 ASSERT_EQ("", ReportMessage());
412 }
413
414 TEST(LogTest, ChecksumMismatch) {
415 Write("foo");
416 IncrementByte(0, 10);
417 ASSERT_EQ("EOF", Read());
418 ASSERT_EQ(10, DroppedBytes());
419 ASSERT_EQ("OK", MatchError("checksum mismatch"));
420 }
421
422 TEST(LogTest, UnexpectedMiddleType) {
423 Write("foo");
424 SetByte(6, kMiddleType);
425 FixChecksum(0, 3);
426 ASSERT_EQ("EOF", Read());
427 ASSERT_EQ(3, DroppedBytes());
428 ASSERT_EQ("OK", MatchError("missing start"));
429 }
430
431 TEST(LogTest, UnexpectedLastType) {
432 Write("foo");
433 SetByte(6, kLastType);
434 FixChecksum(0, 3);
435 ASSERT_EQ("EOF", Read());
436 ASSERT_EQ(3, DroppedBytes());
437 ASSERT_EQ("OK", MatchError("missing start"));
438 }
439
440 TEST(LogTest, UnexpectedFullType) {
441 Write("foo");
442 Write("bar");
443 SetByte(6, kFirstType);
444 FixChecksum(0, 3);
445 ASSERT_EQ("bar", Read());
446 ASSERT_EQ("EOF", Read());
447 ASSERT_EQ(3, DroppedBytes());
448 ASSERT_EQ("OK", MatchError("partial record without end"));
449 }
450
451 TEST(LogTest, UnexpectedFirstType) {
452 Write("foo");
453 Write(BigString("bar", 100000));
454 SetByte(6, kFirstType);
455 FixChecksum(0, 3);
456 ASSERT_EQ(BigString("bar", 100000), Read());
457 ASSERT_EQ("EOF", Read());
458 ASSERT_EQ(3, DroppedBytes());
459 ASSERT_EQ("OK", MatchError("partial record without end"));
460 }
461
462 TEST(LogTest, MissingLastIsIgnored) {
463 Write(BigString("bar", kBlockSize));
464 // Remove the LAST block, including header.
465 ShrinkSize(14);
466 ASSERT_EQ("EOF", Read());
467 ASSERT_EQ("", ReportMessage());
468 ASSERT_EQ(0, DroppedBytes());
469 }
470
471 TEST(LogTest, PartialLastIsIgnored) {
472 Write(BigString("bar", kBlockSize));
473 // Cause a bad record length in the LAST block.
474 ShrinkSize(1);
475 ASSERT_EQ("EOF", Read());
476 ASSERT_EQ("", ReportMessage());
477 ASSERT_EQ(0, DroppedBytes());
478 }
479
480 TEST(LogTest, SkipIntoMultiRecord) {
481 // Consider a fragmented record:
482 // first(R1), middle(R1), last(R1), first(R2)
483 // If initial_offset points to a record after first(R1) but before first(R2)
484 // incomplete fragment errors are not actual errors, and must be suppressed
485 // until a new first or full record is encountered.
486 Write(BigString("foo", 3 * kBlockSize));
487 Write("correct");
488 StartReadingAt(kBlockSize);
489
490 ASSERT_EQ("correct", Read());
491 ASSERT_EQ("", ReportMessage());
492 ASSERT_EQ(0, DroppedBytes());
493 ASSERT_EQ("EOF", Read());
494 }
495
496 TEST(LogTest, ErrorJoinsRecords) {
497 // Consider two fragmented records:
498 // first(R1) last(R1) first(R2) last(R2)
499 // where the middle two fragments disappear. We do not want
500 // first(R1),last(R2) to get joined and returned as a valid record.
501
502 // Write records that span two blocks
503 Write(BigString("foo", kBlockSize));
504 Write(BigString("bar", kBlockSize));
505 Write("correct");
506
507 // Wipe the middle block
508 for (int offset = kBlockSize; offset < 2 * kBlockSize; offset++) {
509 SetByte(offset, 'x');
510 }
511
512 ASSERT_EQ("correct", Read());
513 ASSERT_EQ("EOF", Read());
514 const size_t dropped = DroppedBytes();
515 ASSERT_LE(dropped, 2 * kBlockSize + 100);
516 ASSERT_GE(dropped, 2 * kBlockSize);
517 }
518
519 TEST(LogTest, ReadStart) { CheckInitialOffsetRecord(0, 0); }
520
521 TEST(LogTest, ReadSecondOneOff) { CheckInitialOffsetRecord(1, 1); }
522
523 TEST(LogTest, ReadSecondTenThousand) { CheckInitialOffsetRecord(10000, 1); }
524
525 TEST(LogTest, ReadSecondStart) { CheckInitialOffsetRecord(10007, 1); }
526
527 TEST(LogTest, ReadThirdOneOff) { CheckInitialOffsetRecord(10008, 2); }
528
529 TEST(LogTest, ReadThirdStart) { CheckInitialOffsetRecord(20014, 2); }
530
531 TEST(LogTest, ReadFourthOneOff) { CheckInitialOffsetRecord(20015, 3); }
532
533 TEST(LogTest, ReadFourthFirstBlockTrailer) {
534 CheckInitialOffsetRecord(log::kBlockSize - 4, 3);
535 }
536
537 TEST(LogTest, ReadFourthMiddleBlock) {
538 CheckInitialOffsetRecord(log::kBlockSize + 1, 3);
539 }
540
541 TEST(LogTest, ReadFourthLastBlock) {
542 CheckInitialOffsetRecord(2 * log::kBlockSize + 1, 3);
543 }
544
545 TEST(LogTest, ReadFourthStart) {
546 CheckInitialOffsetRecord(
547 2 * (kHeaderSize + 1000) + (2 * log::kBlockSize - 1000) + 3 * kHeaderSize,
548 3);
549 }
550
551 TEST(LogTest, ReadInitialOffsetIntoBlockPadding) {
552 CheckInitialOffsetRecord(3 * log::kBlockSize - 3, 5);
553 }
554
555 TEST(LogTest, ReadEnd) { CheckOffsetPastEndReturnsNoRecords(0); }
556
557 TEST(LogTest, ReadPastEnd) { CheckOffsetPastEndReturnsNoRecords(5); }
558
559 } // namespace log
560 } // namespace leveldb
561
562 int main(int argc, char** argv) { return leveldb::test::RunAllTests(); }
563