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 <cstdio> |
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 | std::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)); |
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* = 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 | |