Home | History | Annotate | Download | only in db 
      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 
      7 #include <stdio.h>
      8 #include "leveldb/env.h"
      9 #include "util/coding.h"
     10 #include "util/crc32c.h"
     11 
     12 namespace leveldb {
     13 namespace log {
     14 
     15 Reader::Reporter::~Reporter() {
     16 }
     17 
     18 Reader::Reader(SequentialFile* file, Reporter* reporter, bool checksum,
     19                uint64_t initial_offset)
     20     : file_(file),
     21       reporter_(reporter),
     22       checksum_(checksum),
     23       backing_store_(new char[kBlockSize]),
     24       buffer_(),
     25       eof_(false),
     26       last_record_offset_(0),
     27       end_of_buffer_offset_(0),
     28       initial_offset_(initial_offset) {
     29 }
     30 
     31 Reader::~Reader() {
     32   delete[] backing_store_;
     33 }
     34 
     35 bool Reader::SkipToInitialBlock() {
     36   size_t offset_in_block = initial_offset_ % kBlockSize;
     37   uint64_t block_start_location = initial_offset_ - offset_in_block;
     38 
     39   // Don't search a block if we'd be in the trailer
     40   if (offset_in_block > kBlockSize - 6) {
     41     offset_in_block = 0;
     42     block_start_location += kBlockSize;
     43   }
     44 
     45   end_of_buffer_offset_ = block_start_location;
     46 
     47   // Skip to start of first block that can contain the initial record
     48   if (block_start_location > 0) {
     49     Status skip_status = file_->Skip(block_start_location);
     50     if (!skip_status.ok()) {
     51       ReportDrop(block_start_location, skip_status);
     52       return false;
     53     }
     54   }
     55 
     56   return true;
     57 }
     58 
     59 bool Reader::ReadRecord(Slice* record, std::string* scratch) {
     60   if (last_record_offset_ < initial_offset_) {
     61     if (!SkipToInitialBlock()) {
     62       return false;
     63     }
     64   }
     65 
     66   scratch->clear();
     67   record->clear();
     68   bool in_fragmented_record = false;
     69   // Record offset of the logical record that we're reading
     70   // 0 is a dummy value to make compilers happy
     71   uint64_t prospective_record_offset = 0;
     72 
     73   Slice fragment;
     74   while (true) {
     75     uint64_t physical_record_offset = end_of_buffer_offset_ - buffer_.size();
     76     const unsigned int record_type = ReadPhysicalRecord(&fragment);
     77     switch (record_type) {
     78       case kFullType:
     79         if (in_fragmented_record) {
     80           // Handle bug in earlier versions of log::Writer where
     81           // it could emit an empty kFirstType record at the tail end
     82           // of a block followed by a kFullType or kFirstType record
     83           // at the beginning of the next block.
     84           if (scratch->empty()) {
     85             in_fragmented_record = false;
     86           } else {
     87             ReportCorruption(scratch->size(), "partial record without end(1)");
     88           }
     89         }
     90         prospective_record_offset = physical_record_offset;
     91         scratch->clear();
     92         *record = fragment;
     93         last_record_offset_ = prospective_record_offset;
     94         return true;
     95 
     96       case kFirstType:
     97         if (in_fragmented_record) {
     98           // Handle bug in earlier versions of log::Writer where
     99           // it could emit an empty kFirstType record at the tail end
    100           // of a block followed by a kFullType or kFirstType record
    101           // at the beginning of the next block.
    102           if (scratch->empty()) {
    103             in_fragmented_record = false;
    104           } else {
    105             ReportCorruption(scratch->size(), "partial record without end(2)");
    106           }
    107         }
    108         prospective_record_offset = physical_record_offset;
    109         scratch->assign(fragment.data(), fragment.size());
    110         in_fragmented_record = true;
    111         break;
    112 
    113       case kMiddleType:
    114         if (!in_fragmented_record) {
    115           ReportCorruption(fragment.size(),
    116                            "missing start of fragmented record(1)");
    117         } else {
    118           scratch->append(fragment.data(), fragment.size());
    119         }
    120         break;
    121 
    122       case kLastType:
    123         if (!in_fragmented_record) {
    124           ReportCorruption(fragment.size(),
    125                            "missing start of fragmented record(2)");
    126         } else {
    127           scratch->append(fragment.data(), fragment.size());
    128           *record = Slice(*scratch);
    129           last_record_offset_ = prospective_record_offset;
    130           return true;
    131         }
    132         break;
    133 
    134       case kEof:
    135         if (in_fragmented_record) {
    136           // This can be caused by the writer dying immediately after
    137           // writing a physical record but before completing the next; don't
    138           // treat it as a corruption, just ignore the entire logical record.
    139           scratch->clear();
    140         }
    141         return false;
    142 
    143       case kBadRecord:
    144         if (in_fragmented_record) {
    145           ReportCorruption(scratch->size(), "error in middle of record");
    146           in_fragmented_record = false;
    147           scratch->clear();
    148         }
    149         break;
    150 
    151       default: {
    152         char buf[40];
    153         snprintf(buf, sizeof(buf), "unknown record type %u", record_type);
    154         ReportCorruption(
    155             (fragment.size() + (in_fragmented_record ? scratch->size() : 0)),
    156             buf);
    157         in_fragmented_record = false;
    158         scratch->clear();
    159         break;
    160       }
    161     }
    162   }
    163   return false;
    164 }
    165 
    166 uint64_t Reader::LastRecordOffset() {
    167   return last_record_offset_;
    168 }
    169 
    170 void Reader::ReportCorruption(size_t bytes, const char* reason) {
    171   ReportDrop(bytes, Status::Corruption(reason));
    172 }
    173 
    174 void Reader::ReportDrop(size_t bytes, const Status& reason) {
    175   if (reporter_ != NULL &&
    176       end_of_buffer_offset_ - buffer_.size() - bytes >= initial_offset_) {
    177     reporter_->Corruption(bytes, reason);
    178   }
    179 }
    180 
    181 unsigned int Reader::ReadPhysicalRecord(Slice* result) {
    182   while (true) {
    183     if (buffer_.size() < kHeaderSize) {
    184       if (!eof_) {
    185         // Last read was a full read, so this is a trailer to skip
    186         buffer_.clear();
    187         Status status = file_->Read(kBlockSize, &buffer_, backing_store_);
    188         end_of_buffer_offset_ += buffer_.size();
    189         if (!status.ok()) {
    190           buffer_.clear();
    191           ReportDrop(kBlockSize, status);
    192           eof_ = true;
    193           return kEof;
    194         } else if (buffer_.size() < kBlockSize) {
    195           eof_ = true;
    196         }
    197         continue;
    198       } else {
    199         // Note that if buffer_ is non-empty, we have a truncated header at the
    200         // end of the file, which can be caused by the writer crashing in the
    201         // middle of writing the header. Instead of considering this an error,
    202         // just report EOF.
    203         buffer_.clear();
    204         return kEof;
    205       }
    206     }
    207 
    208     // Parse the header
    209     const char* header = buffer_.data();
    210     const uint32_t a = static_cast<uint32_t>(header[4]) & 0xff;
    211     const uint32_t b = static_cast<uint32_t>(header[5]) & 0xff;
    212     const unsigned int type = header[6];
    213     const uint32_t length = a | (b << 8);
    214     if (kHeaderSize + length > buffer_.size()) {
    215       size_t drop_size = buffer_.size();
    216       buffer_.clear();
    217       if (!eof_) {
    218         ReportCorruption(drop_size, "bad record length");
    219         return kBadRecord;
    220       }
    221       // If the end of the file has been reached without reading |length| bytes
    222       // of payload, assume the writer died in the middle of writing the record.
    223       // Don't report a corruption.
    224       return kEof;
    225     }
    226 
    227     if (type == kZeroType && length == 0) {
    228       // Skip zero length record without reporting any drops since
    229       // such records are produced by the mmap based writing code in
    230       // env_posix.cc that preallocates file regions.
    231       buffer_.clear();
    232       return kBadRecord;
    233     }
    234 
    235     // Check crc
    236     if (checksum_) {
    237       uint32_t expected_crc = crc32c::Unmask(DecodeFixed32(header));
    238       uint32_t actual_crc = crc32c::Value(header + 6, 1 + length);
    239       if (actual_crc != expected_crc) {
    240         // Drop the rest of the buffer since "length" itself may have
    241         // been corrupted and if we trust it, we could find some
    242         // fragment of a real log record that just happens to look
    243         // like a valid log record.
    244         size_t drop_size = buffer_.size();
    245         buffer_.clear();
    246         ReportCorruption(drop_size, "checksum mismatch");
    247         return kBadRecord;
    248       }
    249     }
    250 
    251     buffer_.remove_prefix(kHeaderSize + length);
    252 
    253     // Skip physical record that started before initial_offset_
    254     if (end_of_buffer_offset_ - buffer_.size() - kHeaderSize - length <
    255         initial_offset_) {
    256       result->clear();
    257       return kBadRecord;
    258     }
    259 
    260     *result = Slice(header + kHeaderSize, length);
    261     return type;
    262   }
    263 }
    264 
    265 }  // namespace log
    266 }  // namespace leveldb
    267