You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 
rocksdb/db/log_reader.cc

178 lines
5.0 KiB

// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file. See the AUTHORS file for names of contributors.
#include "db/log_reader.h"
#include <stdint.h>
#include "include/env.h"
#include "util/coding.h"
#include "util/crc32c.h"
namespace leveldb {
namespace log {
Reader::Reporter::~Reporter() {
}
Reader::Reader(SequentialFile* file, Reporter* reporter, bool checksum)
: file_(file),
reporter_(reporter),
checksum_(checksum),
backing_store_(new char[kBlockSize]),
buffer_(),
eof_(false) {
}
Reader::~Reader() {
delete[] backing_store_;
}
bool Reader::ReadRecord(Slice* record, std::string* scratch) {
scratch->clear();
record->clear();
bool in_fragmented_record = false;
Slice fragment;
while (true) {
switch (ReadPhysicalRecord(&fragment)) {
case kFullType:
if (in_fragmented_record) {
ReportDrop(scratch->size(), "partial record without end");
}
scratch->clear();
*record = fragment;
return true;
case kFirstType:
if (in_fragmented_record) {
ReportDrop(scratch->size(), "partial record without end");
}
scratch->assign(fragment.data(), fragment.size());
in_fragmented_record = true;
break;
case kMiddleType:
if (!in_fragmented_record) {
ReportDrop(fragment.size(), "missing start of fragmented record");
} else {
scratch->append(fragment.data(), fragment.size());
}
break;
case kLastType:
if (!in_fragmented_record) {
ReportDrop(fragment.size(), "missing start of fragmented record");
} else {
scratch->append(fragment.data(), fragment.size());
*record = Slice(*scratch);
return true;
}
break;
case kEof:
if (in_fragmented_record) {
ReportDrop(scratch->size(), "partial record without end");
scratch->clear();
}
return false;
case kBadRecord:
if (in_fragmented_record) {
ReportDrop(scratch->size(), "error in middle of record");
in_fragmented_record = false;
scratch->clear();
}
break;
default:
ReportDrop(
(fragment.size() + (in_fragmented_record ? scratch->size() : 0)),
"unknown record type");
in_fragmented_record = false;
scratch->clear();
break;
}
}
return false;
}
void Reader::ReportDrop(size_t bytes, const char* reason) {
if (reporter_ != NULL) {
reporter_->Corruption(bytes, Status::Corruption(reason));
}
}
unsigned int Reader::ReadPhysicalRecord(Slice* result) {
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_);
if (!status.ok()) {
if (reporter_ != NULL) {
reporter_->Corruption(kBlockSize, status);
}
buffer_.clear();
eof_ = true;
return kEof;
} else if (buffer_.size() < kBlockSize) {
eof_ = true;
}
continue;
} else if (buffer_.size() == 0) {
// End of file
return kEof;
} else if (buffer_.size() < kHeaderSize) {
ReportDrop(buffer_.size(), "truncated record at end of file");
buffer_.clear();
return kEof;
} else {
// We have a trailing zero-length record. Fall through and check it.
}
}
// 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()) {
ReportDrop(buffer_.size(), "bad record length");
buffer_.clear();
return kBadRecord;
}
// Check crc
if (checksum_) {
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;
}
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.
ReportDrop(buffer_.size(), "checksum mismatch");
buffer_.clear();
return kBadRecord;
}
}
buffer_.remove_prefix(kHeaderSize + length);
*result = Slice(header + kHeaderSize, length);
return type;
}
}
}
}