src/leveldb/db/log_test.cc

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