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