// Copyright (c) 2011-present, Facebook, Inc. All rights reserved. // This source code is licensed under both the GPLv2 (found in the // COPYING file in the root directory) and Apache 2.0 License // (found in the LICENSE.Apache file in the root directory). // // 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 <stdio.h> #include "rocksdb/env.h" #include "util/coding.h" #include "util/crc32c.h" #include "util/file_reader_writer.h" namespace rocksdb { namespace log { Reader::Reporter::~Reporter() { } Reader::Reader(std::shared_ptr<Logger> info_log, unique_ptr<SequentialFileReader>&& _file, Reporter* reporter, bool checksum, uint64_t initial_offset, uint64_t log_num) : info_log_(info_log), file_(std::move(_file)), reporter_(reporter), checksum_(checksum), backing_store_(new char[kBlockSize]), buffer_(), eof_(false), read_error_(false), eof_offset_(0), last_record_offset_(0), end_of_buffer_offset_(0), initial_offset_(initial_offset), log_number_(log_num), recycled_(false) {} Reader::~Reader() { delete[] backing_store_; } bool Reader::SkipToInitialBlock() { size_t initial_offset_in_block = initial_offset_ % kBlockSize; uint64_t block_start_location = initial_offset_ - initial_offset_in_block; // Don't search a block if we'd be in the trailer if (initial_offset_in_block > kBlockSize - 6) { 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(static_cast<size_t>(block_start_location), skip_status); return false; } } return true; } // For kAbsoluteConsistency, on clean shutdown we don't expect any error // in the log files. For other modes, we can ignore only incomplete records // in the last log file, which are presumably due to a write in progress // during restart (or from log recycling). // // TODO krad: Evaluate if we need to move to a more strict mode where we // restrict the inconsistency to only the last log bool Reader::ReadRecord(Slice* record, std::string* scratch, WALRecoveryMode wal_recovery_mode) { 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) { uint64_t physical_record_offset = end_of_buffer_offset_ - buffer_.size(); size_t drop_size = 0; const unsigned int record_type = ReadPhysicalRecord(&fragment, &drop_size); switch (record_type) { case kFullType: case kRecyclableFullType: if (in_fragmented_record && !scratch->empty()) { // 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. 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: case kRecyclableFirstType: if (in_fragmented_record && !scratch->empty()) { // 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. 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: case kRecyclableMiddleType: if (!in_fragmented_record) { ReportCorruption(fragment.size(), "missing start of fragmented record(1)"); } else { scratch->append(fragment.data(), fragment.size()); } break; case kLastType: case kRecyclableLastType: 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 kBadHeader: if (wal_recovery_mode == WALRecoveryMode::kAbsoluteConsistency) { // in clean shutdown we don't expect any error in the log files ReportCorruption(drop_size, "truncated header"); } // fall-thru case kEof: if (in_fragmented_record) { if (wal_recovery_mode == WALRecoveryMode::kAbsoluteConsistency) { // in clean shutdown we don't expect any error in the log files ReportCorruption(scratch->size(), "error reading trailing data"); } // 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 kOldRecord: if (wal_recovery_mode != WALRecoveryMode::kSkipAnyCorruptedRecords) { // Treat a record from a previous instance of the log as EOF. if (in_fragmented_record) { if (wal_recovery_mode == WALRecoveryMode::kAbsoluteConsistency) { // in clean shutdown we don't expect any error in the log files ReportCorruption(scratch->size(), "error reading trailing data"); } // 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; } // fall-thru case kBadRecord: if (in_fragmented_record) { ReportCorruption(scratch->size(), "error in middle of record"); in_fragmented_record = false; scratch->clear(); } break; case kBadRecordLen: case kBadRecordChecksum: if (recycled_ && wal_recovery_mode == WALRecoveryMode::kTolerateCorruptedTailRecords) { scratch->clear(); return false; } if (record_type == kBadRecordLen) { ReportCorruption(drop_size, "bad record length"); } else { ReportCorruption(drop_size, "checksum mismatch"); } 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; } uint64_t Reader::LastRecordOffset() { return last_record_offset_; } void Reader::UnmarkEOF() { if (read_error_) { return; } eof_ = false; if (eof_offset_ == 0) { return; } // If the EOF was in the middle of a block (a partial block was read) we have // to read the rest of the block as ReadPhysicalRecord can only read full // blocks and expects the file position indicator to be aligned to the start // of a block. // // consumed_bytes + buffer_size() + remaining == kBlockSize size_t consumed_bytes = eof_offset_ - buffer_.size(); size_t remaining = kBlockSize - eof_offset_; // backing_store_ is used to concatenate what is left in buffer_ and // the remainder of the block. If buffer_ already uses backing_store_, // we just append the new data. if (buffer_.data() != backing_store_ + consumed_bytes) { // Buffer_ does not use backing_store_ for storage. // Copy what is left in buffer_ to backing_store. memmove(backing_store_ + consumed_bytes, buffer_.data(), buffer_.size()); } Slice read_buffer; Status status = file_->Read(remaining, &read_buffer, backing_store_ + eof_offset_); size_t added = read_buffer.size(); end_of_buffer_offset_ += added; if (!status.ok()) { if (added > 0) { ReportDrop(added, status); } read_error_ = true; return; } if (read_buffer.data() != backing_store_ + eof_offset_) { // Read did not write to backing_store_ memmove(backing_store_ + eof_offset_, read_buffer.data(), read_buffer.size()); } buffer_ = Slice(backing_store_ + consumed_bytes, eof_offset_ + added - consumed_bytes); if (added < remaining) { eof_ = true; eof_offset_ += added; } else { eof_offset_ = 0; } } void Reader::ReportCorruption(size_t bytes, const char* reason) { ReportDrop(bytes, Status::Corruption(reason)); } void Reader::ReportDrop(size_t bytes, const Status& reason) { if (reporter_ != nullptr && end_of_buffer_offset_ - buffer_.size() - bytes >= initial_offset_) { reporter_->Corruption(bytes, reason); } } bool Reader::ReadMore(size_t* drop_size, int *error) { if (!eof_ && !read_error_) { // 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); read_error_ = true; *error = kEof; return false; } else if (buffer_.size() < (size_t)kBlockSize) { eof_ = true; eof_offset_ = buffer_.size(); } return true; } 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. Unless explicitly requested we don't // considering this an error, just report EOF. if (buffer_.size()) { *drop_size = buffer_.size(); buffer_.clear(); *error = kBadHeader; return false; } buffer_.clear(); *error = kEof; return false; } } unsigned int Reader::ReadPhysicalRecord(Slice* result, size_t* drop_size) { while (true) { // We need at least the minimum header size if (buffer_.size() < (size_t)kHeaderSize) { int r; if (!ReadMore(drop_size, &r)) { return r; } continue; } // 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); int header_size = kHeaderSize; if (type >= kRecyclableFullType && type <= kRecyclableLastType) { if (end_of_buffer_offset_ - buffer_.size() == 0) { recycled_ = true; } header_size = kRecyclableHeaderSize; // We need enough for the larger header if (buffer_.size() < (size_t)kRecyclableHeaderSize) { int r; if (!ReadMore(drop_size, &r)) { return r; } continue; } const uint32_t log_num = DecodeFixed32(header + 7); if (log_num != log_number_) { return kOldRecord; } } if (header_size + length > buffer_.size()) { *drop_size = buffer_.size(); buffer_.clear(); if (!eof_) { return kBadRecordLen; } // 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 unless requested. if (*drop_size) { return kBadHeader; } 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. // NOTE: this should never happen in DB written by new RocksDB versions, // since we turn off mmap writes to manifest and log files buffer_.clear(); return kBadRecord; } // Check crc if (checksum_) { uint32_t expected_crc = crc32c::Unmask(DecodeFixed32(header)); uint32_t actual_crc = crc32c::Value(header + 6, length + header_size - 6); 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. *drop_size = buffer_.size(); buffer_.clear(); return kBadRecordChecksum; } } buffer_.remove_prefix(header_size + length); // Skip physical record that started before initial_offset_ if (end_of_buffer_offset_ - buffer_.size() - header_size - length < initial_offset_) { result->clear(); return kBadRecord; } *result = Slice(header + header_size, length); return type; } } } // namespace log } // namespace rocksdb