1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
|
// 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 "leveldb/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 {
ReportDrop(buffer_.size(), "truncated record at end of file");
buffer_.clear();
return kEof;
}
}
// 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;
}
}
}
}
|
