recovery_test.cc raw
1 // Copyright (c) 2014 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/db_impl.h"
6 #include "db/filename.h"
7 #include "db/version_set.h"
8 #include "db/write_batch_internal.h"
9 #include "leveldb/db.h"
10 #include "leveldb/env.h"
11 #include "leveldb/write_batch.h"
12 #include "util/logging.h"
13 #include "util/testharness.h"
14 #include "util/testutil.h"
15
16 namespace leveldb {
17
18 class RecoveryTest {
19 public:
20 RecoveryTest() : env_(Env::Default()), db_(nullptr) {
21 dbname_ = test::TmpDir() + "/recovery_test";
22 DestroyDB(dbname_, Options());
23 Open();
24 }
25
26 ~RecoveryTest() {
27 Close();
28 DestroyDB(dbname_, Options());
29 }
30
31 DBImpl* dbfull() const { return reinterpret_cast<DBImpl*>(db_); }
32 Env* env() const { return env_; }
33
34 bool CanAppend() {
35 WritableFile* tmp;
36 Status s = env_->NewAppendableFile(CurrentFileName(dbname_), &tmp);
37 delete tmp;
38 if (s.IsNotSupportedError()) {
39 return false;
40 } else {
41 return true;
42 }
43 }
44
45 void Close() {
46 delete db_;
47 db_ = nullptr;
48 }
49
50 Status OpenWithStatus(Options* options = nullptr) {
51 Close();
52 Options opts;
53 if (options != nullptr) {
54 opts = *options;
55 } else {
56 opts.reuse_logs = true; // TODO(sanjay): test both ways
57 opts.create_if_missing = true;
58 }
59 if (opts.env == nullptr) {
60 opts.env = env_;
61 }
62 return DB::Open(opts, dbname_, &db_);
63 }
64
65 void Open(Options* options = nullptr) {
66 ASSERT_OK(OpenWithStatus(options));
67 ASSERT_EQ(1, NumLogs());
68 }
69
70 Status Put(const std::string& k, const std::string& v) {
71 return db_->Put(WriteOptions(), k, v);
72 }
73
74 std::string Get(const std::string& k, const Snapshot* snapshot = nullptr) {
75 std::string result;
76 Status s = db_->Get(ReadOptions(), k, &result);
77 if (s.IsNotFound()) {
78 result = "NOT_FOUND";
79 } else if (!s.ok()) {
80 result = s.ToString();
81 }
82 return result;
83 }
84
85 std::string ManifestFileName() {
86 std::string current;
87 ASSERT_OK(ReadFileToString(env_, CurrentFileName(dbname_), ¤t));
88 size_t len = current.size();
89 if (len > 0 && current[len - 1] == '\n') {
90 current.resize(len - 1);
91 }
92 return dbname_ + "/" + current;
93 }
94
95 std::string LogName(uint64_t number) { return LogFileName(dbname_, number); }
96
97 size_t DeleteLogFiles() {
98 // Linux allows unlinking open files, but Windows does not.
99 // Closing the db allows for file deletion.
100 Close();
101 std::vector<uint64_t> logs = GetFiles(kLogFile);
102 for (size_t i = 0; i < logs.size(); i++) {
103 ASSERT_OK(env_->DeleteFile(LogName(logs[i]))) << LogName(logs[i]);
104 }
105 return logs.size();
106 }
107
108 void DeleteManifestFile() { ASSERT_OK(env_->DeleteFile(ManifestFileName())); }
109
110 uint64_t FirstLogFile() { return GetFiles(kLogFile)[0]; }
111
112 std::vector<uint64_t> GetFiles(FileType t) {
113 std::vector<std::string> filenames;
114 ASSERT_OK(env_->GetChildren(dbname_, &filenames));
115 std::vector<uint64_t> result;
116 for (size_t i = 0; i < filenames.size(); i++) {
117 uint64_t number;
118 FileType type;
119 if (ParseFileName(filenames[i], &number, &type) && type == t) {
120 result.push_back(number);
121 }
122 }
123 return result;
124 }
125
126 int NumLogs() { return GetFiles(kLogFile).size(); }
127
128 int NumTables() { return GetFiles(kTableFile).size(); }
129
130 uint64_t FileSize(const std::string& fname) {
131 uint64_t result;
132 ASSERT_OK(env_->GetFileSize(fname, &result)) << fname;
133 return result;
134 }
135
136 void CompactMemTable() { dbfull()->TEST_CompactMemTable(); }
137
138 // Directly construct a log file that sets key to val.
139 void MakeLogFile(uint64_t lognum, SequenceNumber seq, Slice key, Slice val) {
140 std::string fname = LogFileName(dbname_, lognum);
141 WritableFile* file;
142 ASSERT_OK(env_->NewWritableFile(fname, &file));
143 log::Writer writer(file);
144 WriteBatch batch;
145 batch.Put(key, val);
146 WriteBatchInternal::SetSequence(&batch, seq);
147 ASSERT_OK(writer.AddRecord(WriteBatchInternal::Contents(&batch)));
148 ASSERT_OK(file->Flush());
149 delete file;
150 }
151
152 private:
153 std::string dbname_;
154 Env* env_;
155 DB* db_;
156 };
157
158 TEST(RecoveryTest, ManifestReused) {
159 if (!CanAppend()) {
160 fprintf(stderr, "skipping test because env does not support appending\n");
161 return;
162 }
163 ASSERT_OK(Put("foo", "bar"));
164 Close();
165 std::string old_manifest = ManifestFileName();
166 Open();
167 ASSERT_EQ(old_manifest, ManifestFileName());
168 ASSERT_EQ("bar", Get("foo"));
169 Open();
170 ASSERT_EQ(old_manifest, ManifestFileName());
171 ASSERT_EQ("bar", Get("foo"));
172 }
173
174 TEST(RecoveryTest, LargeManifestCompacted) {
175 if (!CanAppend()) {
176 fprintf(stderr, "skipping test because env does not support appending\n");
177 return;
178 }
179 ASSERT_OK(Put("foo", "bar"));
180 Close();
181 std::string old_manifest = ManifestFileName();
182
183 // Pad with zeroes to make manifest file very big.
184 {
185 uint64_t len = FileSize(old_manifest);
186 WritableFile* file;
187 ASSERT_OK(env()->NewAppendableFile(old_manifest, &file));
188 std::string zeroes(3 * 1048576 - static_cast<size_t>(len), 0);
189 ASSERT_OK(file->Append(zeroes));
190 ASSERT_OK(file->Flush());
191 delete file;
192 }
193
194 Open();
195 std::string new_manifest = ManifestFileName();
196 ASSERT_NE(old_manifest, new_manifest);
197 ASSERT_GT(10000, FileSize(new_manifest));
198 ASSERT_EQ("bar", Get("foo"));
199
200 Open();
201 ASSERT_EQ(new_manifest, ManifestFileName());
202 ASSERT_EQ("bar", Get("foo"));
203 }
204
205 TEST(RecoveryTest, NoLogFiles) {
206 ASSERT_OK(Put("foo", "bar"));
207 ASSERT_EQ(1, DeleteLogFiles());
208 Open();
209 ASSERT_EQ("NOT_FOUND", Get("foo"));
210 Open();
211 ASSERT_EQ("NOT_FOUND", Get("foo"));
212 }
213
214 TEST(RecoveryTest, LogFileReuse) {
215 if (!CanAppend()) {
216 fprintf(stderr, "skipping test because env does not support appending\n");
217 return;
218 }
219 for (int i = 0; i < 2; i++) {
220 ASSERT_OK(Put("foo", "bar"));
221 if (i == 0) {
222 // Compact to ensure current log is empty
223 CompactMemTable();
224 }
225 Close();
226 ASSERT_EQ(1, NumLogs());
227 uint64_t number = FirstLogFile();
228 if (i == 0) {
229 ASSERT_EQ(0, FileSize(LogName(number)));
230 } else {
231 ASSERT_LT(0, FileSize(LogName(number)));
232 }
233 Open();
234 ASSERT_EQ(1, NumLogs());
235 ASSERT_EQ(number, FirstLogFile()) << "did not reuse log file";
236 ASSERT_EQ("bar", Get("foo"));
237 Open();
238 ASSERT_EQ(1, NumLogs());
239 ASSERT_EQ(number, FirstLogFile()) << "did not reuse log file";
240 ASSERT_EQ("bar", Get("foo"));
241 }
242 }
243
244 TEST(RecoveryTest, MultipleMemTables) {
245 // Make a large log.
246 const int kNum = 1000;
247 for (int i = 0; i < kNum; i++) {
248 char buf[100];
249 snprintf(buf, sizeof(buf), "%050d", i);
250 ASSERT_OK(Put(buf, buf));
251 }
252 ASSERT_EQ(0, NumTables());
253 Close();
254 ASSERT_EQ(0, NumTables());
255 ASSERT_EQ(1, NumLogs());
256 uint64_t old_log_file = FirstLogFile();
257
258 // Force creation of multiple memtables by reducing the write buffer size.
259 Options opt;
260 opt.reuse_logs = true;
261 opt.write_buffer_size = (kNum * 100) / 2;
262 Open(&opt);
263 ASSERT_LE(2, NumTables());
264 ASSERT_EQ(1, NumLogs());
265 ASSERT_NE(old_log_file, FirstLogFile()) << "must not reuse log";
266 for (int i = 0; i < kNum; i++) {
267 char buf[100];
268 snprintf(buf, sizeof(buf), "%050d", i);
269 ASSERT_EQ(buf, Get(buf));
270 }
271 }
272
273 TEST(RecoveryTest, MultipleLogFiles) {
274 ASSERT_OK(Put("foo", "bar"));
275 Close();
276 ASSERT_EQ(1, NumLogs());
277
278 // Make a bunch of uncompacted log files.
279 uint64_t old_log = FirstLogFile();
280 MakeLogFile(old_log + 1, 1000, "hello", "world");
281 MakeLogFile(old_log + 2, 1001, "hi", "there");
282 MakeLogFile(old_log + 3, 1002, "foo", "bar2");
283
284 // Recover and check that all log files were processed.
285 Open();
286 ASSERT_LE(1, NumTables());
287 ASSERT_EQ(1, NumLogs());
288 uint64_t new_log = FirstLogFile();
289 ASSERT_LE(old_log + 3, new_log);
290 ASSERT_EQ("bar2", Get("foo"));
291 ASSERT_EQ("world", Get("hello"));
292 ASSERT_EQ("there", Get("hi"));
293
294 // Test that previous recovery produced recoverable state.
295 Open();
296 ASSERT_LE(1, NumTables());
297 ASSERT_EQ(1, NumLogs());
298 if (CanAppend()) {
299 ASSERT_EQ(new_log, FirstLogFile());
300 }
301 ASSERT_EQ("bar2", Get("foo"));
302 ASSERT_EQ("world", Get("hello"));
303 ASSERT_EQ("there", Get("hi"));
304
305 // Check that introducing an older log file does not cause it to be re-read.
306 Close();
307 MakeLogFile(old_log + 1, 2000, "hello", "stale write");
308 Open();
309 ASSERT_LE(1, NumTables());
310 ASSERT_EQ(1, NumLogs());
311 if (CanAppend()) {
312 ASSERT_EQ(new_log, FirstLogFile());
313 }
314 ASSERT_EQ("bar2", Get("foo"));
315 ASSERT_EQ("world", Get("hello"));
316 ASSERT_EQ("there", Get("hi"));
317 }
318
319 TEST(RecoveryTest, ManifestMissing) {
320 ASSERT_OK(Put("foo", "bar"));
321 Close();
322 DeleteManifestFile();
323
324 Status status = OpenWithStatus();
325 ASSERT_TRUE(status.IsCorruption());
326 }
327
328 } // namespace leveldb
329
330 int main(int argc, char** argv) { return leveldb::test::RunAllTests(); }
331