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 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 resyncing_(initial_offset > 0) {}
30 31 Reader::~Reader() { delete[] backing_store_; }
32 33 bool Reader::SkipToInitialBlock() {
34 const size_t offset_in_block = initial_offset_ % kBlockSize;
35 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 if (offset_in_block > kBlockSize - 6) {
39 block_start_location += kBlockSize;
40 }
41 42 end_of_buffer_offset_ = block_start_location;
43 44 // Skip to start of first block that can contain the initial record
45 if (block_start_location > 0) {
46 Status skip_status = file_->Skip(block_start_location);
47 if (!skip_status.ok()) {
48 ReportDrop(block_start_location, skip_status);
49 return false;
50 }
51 }
52 53 return true;
54 }
55 56 bool Reader::ReadRecord(Slice* record, std::string* scratch) {
57 if (last_record_offset_ < initial_offset_) {
58 if (!SkipToInitialBlock()) {
59 return false;
60 }
61 }
62 63 scratch->clear();
64 record->clear();
65 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 uint64_t prospective_record_offset = 0;
69 70 Slice fragment;
71 while (true) {
72 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 uint64_t physical_record_offset =
78 end_of_buffer_offset_ - buffer_.size() - kHeaderSize - fragment.size();
79 80 if (resyncing_) {
81 if (record_type == kMiddleType) {
82 continue;
83 } else if (record_type == kLastType) {
84 resyncing_ = false;
85 continue;
86 } else {
87 resyncing_ = false;
88 }
89 }
90 91 switch (record_type) {
92 case kFullType:
93 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 if (!scratch->empty()) {
99 ReportCorruption(scratch->size(), "partial record without end(1)");
100 }
101 }
102 prospective_record_offset = physical_record_offset;
103 scratch->clear();
104 *record = fragment;
105 last_record_offset_ = prospective_record_offset;
106 return true;
107 108 case kFirstType:
109 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 if (!scratch->empty()) {
115 ReportCorruption(scratch->size(), "partial record without end(2)");
116 }
117 }
118 prospective_record_offset = physical_record_offset;
119 scratch->assign(fragment.data(), fragment.size());
120 in_fragmented_record = true;
121 break;
122 123 case kMiddleType:
124 if (!in_fragmented_record) {
125 ReportCorruption(fragment.size(),
126 "missing start of fragmented record(1)");
127 } else {
128 scratch->append(fragment.data(), fragment.size());
129 }
130 break;
131 132 case kLastType:
133 if (!in_fragmented_record) {
134 ReportCorruption(fragment.size(),
135 "missing start of fragmented record(2)");
136 } else {
137 scratch->append(fragment.data(), fragment.size());
138 *record = Slice(*scratch);
139 last_record_offset_ = prospective_record_offset;
140 return true;
141 }
142 break;
143 144 case kEof:
145 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 scratch->clear();
150 }
151 return false;
152 153 case kBadRecord:
154 if (in_fragmented_record) {
155 ReportCorruption(scratch->size(), "error in middle of record");
156 in_fragmented_record = false;
157 scratch->clear();
158 }
159 break;
160 161 default: {
162 char buf[40];
163 snprintf(buf, sizeof(buf), "unknown record type %u", record_type);
164 ReportCorruption(
165 (fragment.size() + (in_fragmented_record ? scratch->size() : 0)),
166 buf);
167 in_fragmented_record = false;
168 scratch->clear();
169 break;
170 }
171 }
172 }
173 return false;
174 }
175 176 uint64_t Reader::LastRecordOffset() { return last_record_offset_; }
177 178 void Reader::ReportCorruption(uint64_t bytes, const char* reason) {
179 ReportDrop(bytes, Status::Corruption(reason, file_->GetName()));
180 }
181 182 void Reader::ReportDrop(uint64_t bytes, const Status& reason) {
183 if (reporter_ != nullptr &&
184 end_of_buffer_offset_ - buffer_.size() - bytes >= initial_offset_) {
185 reporter_->Corruption(static_cast<size_t>(bytes), reason);
186 }
187 }
188 189 unsigned int Reader::ReadPhysicalRecord(Slice* result) {
190 while (true) {
191 if (buffer_.size() < kHeaderSize) {
192 if (!eof_) {
193 // Last read was a full read, so this is a trailer to skip
194 buffer_.clear();
195 Status status = file_->Read(kBlockSize, &buffer_, backing_store_);
196 end_of_buffer_offset_ += buffer_.size();
197 if (!status.ok()) {
198 buffer_.clear();
199 ReportDrop(kBlockSize, status);
200 eof_ = true;
201 return kEof;
202 } else if (buffer_.size() < kBlockSize) {
203 eof_ = true;
204 }
205 continue;
206 } 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 buffer_.clear();
212 return kEof;
213 }
214 }
215 216 // Parse the header
217 const char* header = buffer_.data();
218 const uint32_t a = static_cast<uint32_t>(header[4]) & 0xff;
219 const uint32_t b = static_cast<uint32_t>(header[5]) & 0xff;
220 const unsigned int type = header[6];
221 const uint32_t length = a | (b << 8);
222 if (kHeaderSize + length > buffer_.size()) {
223 size_t drop_size = buffer_.size();
224 buffer_.clear();
225 if (!eof_) {
226 ReportCorruption(drop_size, "bad record length");
227 return kBadRecord;
228 }
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 return kEof;
233 }
234 235 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 buffer_.clear();
240 return kBadRecord;
241 }
242 243 // Check crc
244 if (checksum_) {
245 uint32_t expected_crc = crc32c::Unmask(DecodeFixed32(header));
246 uint32_t actual_crc = crc32c::Value(header + 6, 1 + length);
247 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 size_t drop_size = buffer_.size();
253 buffer_.clear();
254 ReportCorruption(drop_size, "checksum mismatch");
255 return kBadRecord;
256 }
257 }
258 259 buffer_.remove_prefix(kHeaderSize + length);
260 261 // Skip physical record that started before initial_offset_
262 if (end_of_buffer_offset_ - buffer_.size() - kHeaderSize - length <
263 initial_offset_) {
264 result->clear();
265 return kBadRecord;
266 }
267 268 *result = Slice(header + kHeaderSize, length);
269 return type;
270 }
271 }
272 273 } // namespace log
274 } // namespace leveldb
275