leveldb::log::Reader Class Reference

#include <log_reader.h>

Collaboration diagram for leveldb::log::Reader:

Classes
class	Reporter

Public Member Functions
	Reader (SequentialFile *file, Reporter *reporter, bool checksum, uint64_t initial_offset)
	~Reader ()
bool	ReadRecord (Slice *record, std::string *scratch)
uint64_t	LastRecordOffset ()

Private Types
enum	{ kEof = kMaxRecordType + 1, kBadRecord = kMaxRecordType + 2 }

Private Member Functions
bool	SkipToInitialBlock ()
unsigned int	ReadPhysicalRecord (Slice *result)
void	ReportCorruption (uint64_t bytes, const char *reason)
void	ReportDrop (uint64_t bytes, const Status &reason)
	Reader (const Reader &)
void	operator= (const Reader &)

Private Attributes
SequentialFile *const	file_
Reporter *const	reporter_
bool const	checksum_
char *const	backing_store_
Slice	buffer_
bool	eof_
uint64_t	last_record_offset_
uint64_t	end_of_buffer_offset_
uint64_t const	initial_offset_
bool	resyncing_

Detailed Description

Definition at line 20 of file log_reader.h.

Member Enumeration Documentation

anonymous enum

anonymous enum

private

Enumerator
kEof
kBadRecord

Definition at line 82 of file log_reader.h.

       {
    kEof = kMaxRecordType + 1,
    // Returned whenever we find an invalid physical record.
    // Currently there are three situations in which this happens:
    // * The record has an invalid CRC (ReadPhysicalRecord reports a drop)
    // * The record is a 0-length record (No drop is reported)
    // * The record is below constructor's initial_offset (No drop is reported)
    kBadRecord = kMaxRecordType + 2
  };

Constructor & Destructor Documentation

Reader() [1/2]

leveldb::log::Reader::Reader	(	SequentialFile *	file,
		Reporter *	reporter,
		bool	checksum,
		uint64_t	initial_offset
	)

Definition at line 18 of file log_reader.cc.

    : file_(file),
      reporter_(reporter),
      checksum_(checksum),
      backing_store_(new char[kBlockSize]),
      buffer_(),
      eof_(false),
      last_record_offset_(0),
      end_of_buffer_offset_(0),
      initial_offset_(initial_offset),
      resyncing_(initial_offset > 0) {
}

~Reader()

leveldb::log::Reader::~Reader

(

)

Definition at line 32 of file log_reader.cc.

                {
  delete[] backing_store_;
}

Reader() [2/2]

leveldb::log::Reader::Reader ( const Reader &  )

private

Member Function Documentation

LastRecordOffset()

uint64_t leveldb::log::Reader::LastRecordOffset

(

)

Definition at line 184 of file log_reader.cc.

                                  {
  return last_record_offset_;
}

Here is the caller graph for this function:

operator=()

void leveldb::log::Reader::operator= ( const Reader &  )

private

ReadPhysicalRecord()

unsigned int leveldb::log::Reader::ReadPhysicalRecord ( Slice *  result )

private

Definition at line 199 of file log_reader.cc.

                                                     {
  while (true) {
    if (buffer_.size() < kHeaderSize) {
      if (!eof_) {
        // Last read was a full read, so this is a trailer to skip
        buffer_.clear();
        Status status = file_->Read(kBlockSize, &buffer_, backing_store_);
        end_of_buffer_offset_ += buffer_.size();
        if (!status.ok()) {
          buffer_.clear();
          ReportDrop(kBlockSize, status);
          eof_ = true;
          return kEof;
        } else if (buffer_.size() < kBlockSize) {
          eof_ = true;
        }
        continue;
      } else {
        // Note that if buffer_ is non-empty, we have a truncated header at the
        // end of the file, which can be caused by the writer crashing in the
        // middle of writing the header. Instead of considering this an error,
        // just report EOF.
        buffer_.clear();
        return kEof;
      }
    }

    // Parse the header
    const char* header = buffer_.data();
    const uint32_t a = static_cast<uint32_t>(header[4]) & 0xff;
    const uint32_t b = static_cast<uint32_t>(header[5]) & 0xff;
    const unsigned int type = header[6];
    const uint32_t length = a | (b << 8);
    if (kHeaderSize + length > buffer_.size()) {
      size_t drop_size = buffer_.size();
      buffer_.clear();
      if (!eof_) {
        ReportCorruption(drop_size, "bad record length");
        return kBadRecord;
      }
      // If the end of the file has been reached without reading |length| bytes
      // of payload, assume the writer died in the middle of writing the record.
      // Don't report a corruption.
      return kEof;
    }

    if (type == kZeroType && length == 0) {
      // Skip zero length record without reporting any drops since
      // such records are produced by the mmap based writing code in
      // env_posix.cc that preallocates file regions.
      buffer_.clear();
      return kBadRecord;
    }

    // Check crc
    if (checksum_) {
      uint32_t expected_crc = crc32c::Unmask(DecodeFixed32(header));
      uint32_t actual_crc = crc32c::Value(header + 6, 1 + length);
      if (actual_crc != expected_crc) {
        // Drop the rest of the buffer since "length" itself may have
        // been corrupted and if we trust it, we could find some
        // fragment of a real log record that just happens to look
        // like a valid log record.
        size_t drop_size = buffer_.size();
        buffer_.clear();
        ReportCorruption(drop_size, "checksum mismatch");
        return kBadRecord;
      }
    }

    buffer_.remove_prefix(kHeaderSize + length);

    // Skip physical record that started before initial_offset_
    if (end_of_buffer_offset_ - buffer_.size() - kHeaderSize - length <
        initial_offset_) {
      result->clear();
      return kBadRecord;
    }

    *result = Slice(header + kHeaderSize, length);
    return type;
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

ReadRecord()

bool leveldb::log::Reader::ReadRecord	(	Slice *	record,
		std::string *	scratch
	)

Definition at line 60 of file log_reader.cc.

                                                         {
  if (last_record_offset_ < initial_offset_) {
    if (!SkipToInitialBlock()) {
      return false;
    }
  }

  scratch->clear();
  record->clear();
  bool in_fragmented_record = false;
  // Record offset of the logical record that we're reading
  // 0 is a dummy value to make compilers happy
  uint64_t prospective_record_offset = 0;

  Slice fragment;
  while (true) {
    const unsigned int record_type = ReadPhysicalRecord(&fragment);

    // ReadPhysicalRecord may have only had an empty trailer remaining in its
    // internal buffer. Calculate the offset of the next physical record now
    // that it has returned, properly accounting for its header size.
    uint64_t physical_record_offset =
        end_of_buffer_offset_ - buffer_.size() - kHeaderSize - fragment.size();

    if (resyncing_) {
      if (record_type == kMiddleType) {
        continue;
      } else if (record_type == kLastType) {
        resyncing_ = false;
        continue;
      } else {
        resyncing_ = false;
      }
    }

    switch (record_type) {
      case kFullType:
        if (in_fragmented_record) {
          // Handle bug in earlier versions of log::Writer where
          // it could emit an empty kFirstType record at the tail end
          // of a block followed by a kFullType or kFirstType record
          // at the beginning of the next block.
          if (scratch->empty()) {
            in_fragmented_record = false;
          } else {
            ReportCorruption(scratch->size(), "partial record without end(1)");
          }
        }
        prospective_record_offset = physical_record_offset;
        scratch->clear();
        *record = fragment;
        last_record_offset_ = prospective_record_offset;
        return true;

      case kFirstType:
        if (in_fragmented_record) {
          // Handle bug in earlier versions of log::Writer where
          // it could emit an empty kFirstType record at the tail end
          // of a block followed by a kFullType or kFirstType record
          // at the beginning of the next block.
          if (scratch->empty()) {
            in_fragmented_record = false;
          } else {
            ReportCorruption(scratch->size(), "partial record without end(2)");
          }
        }
        prospective_record_offset = physical_record_offset;
        scratch->assign(fragment.data(), fragment.size());
        in_fragmented_record = true;
        break;

      case kMiddleType:
        if (!in_fragmented_record) {
          ReportCorruption(fragment.size(),
                           "missing start of fragmented record(1)");
        } else {
          scratch->append(fragment.data(), fragment.size());
        }
        break;

      case kLastType:
        if (!in_fragmented_record) {
          ReportCorruption(fragment.size(),
                           "missing start of fragmented record(2)");
        } else {
          scratch->append(fragment.data(), fragment.size());
          *record = Slice(*scratch);
          last_record_offset_ = prospective_record_offset;
          return true;
        }
        break;

      case kEof:
        if (in_fragmented_record) {
          // This can be caused by the writer dying immediately after
          // writing a physical record but before completing the next; don't
          // treat it as a corruption, just ignore the entire logical record.
          scratch->clear();
        }
        return false;

      case kBadRecord:
        if (in_fragmented_record) {
          ReportCorruption(scratch->size(), "error in middle of record");
          in_fragmented_record = false;
          scratch->clear();
        }
        break;

      default: {
        char buf[40];
        snprintf(buf, sizeof(buf), "unknown record type %u", record_type);
        ReportCorruption(
            (fragment.size() + (in_fragmented_record ? scratch->size() : 0)),
            buf);
        in_fragmented_record = false;
        scratch->clear();
        break;
      }
    }
  }
  return false;
}

Here is the call graph for this function:

Here is the caller graph for this function:

ReportCorruption()

void leveldb::log::Reader::ReportCorruption ( uint64_t  bytes, const char *  reason  )

private

Definition at line 188 of file log_reader.cc.

                                                                {
  ReportDrop(bytes, Status::Corruption(reason));
}

Here is the call graph for this function:

Here is the caller graph for this function:

ReportDrop()

void leveldb::log::Reader::ReportDrop ( uint64_t  bytes, const Status &  reason  )

private

Definition at line 192 of file log_reader.cc.

                                                            {
  if (reporter_ != NULL &&
      end_of_buffer_offset_ - buffer_.size() - bytes >= initial_offset_) {
    reporter_->Corruption(static_cast<size_t>(bytes), reason);
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

SkipToInitialBlock()

bool leveldb::log::Reader::SkipToInitialBlock ( )

private

Definition at line 36 of file log_reader.cc.

                                {
  size_t offset_in_block = initial_offset_ % kBlockSize;
  uint64_t block_start_location = initial_offset_ - offset_in_block;

  // Don't search a block if we'd be in the trailer
  if (offset_in_block > kBlockSize - 6) {
    offset_in_block = 0;
    block_start_location += kBlockSize;
  }

  end_of_buffer_offset_ = block_start_location;

  // Skip to start of first block that can contain the initial record
  if (block_start_location > 0) {
    Status skip_status = file_->Skip(block_start_location);
    if (!skip_status.ok()) {
      ReportDrop(block_start_location, skip_status);
      return false;
    }
  }

  return true;
}

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

backing_store_

char* const leveldb::log::Reader::backing_store_

private

Definition at line 64 of file log_reader.h.

buffer_

Slice leveldb::log::Reader::buffer_

private

Definition at line 65 of file log_reader.h.

checksum_

bool const leveldb::log::Reader::checksum_

private

Definition at line 63 of file log_reader.h.

end_of_buffer_offset_

uint64_t leveldb::log::Reader::end_of_buffer_offset_

private

Definition at line 71 of file log_reader.h.

eof_

bool leveldb::log::Reader::eof_

private

Definition at line 66 of file log_reader.h.

file_

SequentialFile* const leveldb::log::Reader::file_

private

Definition at line 61 of file log_reader.h.

initial_offset_

uint64_t const leveldb::log::Reader::initial_offset_

private

Definition at line 74 of file log_reader.h.

last_record_offset_

uint64_t leveldb::log::Reader::last_record_offset_

private

Definition at line 69 of file log_reader.h.

reporter_

Reporter* const leveldb::log::Reader::reporter_

private

Definition at line 62 of file log_reader.h.

resyncing_

bool leveldb::log::Reader::resyncing_

private

Definition at line 79 of file log_reader.h.

---

The documentation for this class was generated from the following files:

• db/log_reader.h
• db/log_reader.cc