Coverage Report

Created: 2026-07-14 18:13

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
/bitcoin/src/leveldb/db/db_impl.cc
Line
Count
Source
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/db_impl.h"
6
7
#include <stdint.h>
8
#include <stdio.h>
9
10
#include <algorithm>
11
#include <atomic>
12
#include <set>
13
#include <string>
14
#include <vector>
15
16
#include "db/builder.h"
17
#include "db/db_iter.h"
18
#include "db/dbformat.h"
19
#include "db/filename.h"
20
#include "db/log_reader.h"
21
#include "db/log_writer.h"
22
#include "db/memtable.h"
23
#include "db/table_cache.h"
24
#include "db/version_set.h"
25
#include "db/write_batch_internal.h"
26
#include "leveldb/db.h"
27
#include "leveldb/env.h"
28
#include "leveldb/status.h"
29
#include "leveldb/table.h"
30
#include "leveldb/table_builder.h"
31
#include "port/port.h"
32
#include "table/block.h"
33
#include "table/merger.h"
34
#include "table/two_level_iterator.h"
35
#include "util/coding.h"
36
#include "util/logging.h"
37
#include "util/mutexlock.h"
38
39
namespace leveldb {
40
41
const int kNumNonTableCacheFiles = 10;
42
43
// Information kept for every waiting writer
44
struct DBImpl::Writer {
45
  explicit Writer(port::Mutex* mu)
46
831k
      : batch(nullptr), sync(false), done(false), cv(mu) {}
47
48
  Status status;
49
  WriteBatch* batch;
50
  bool sync;
51
  bool done;
52
  port::CondVar cv;
53
};
54
55
struct DBImpl::CompactionState {
56
  // Files produced by compaction
57
  struct Output {
58
    uint64_t number;
59
    uint64_t file_size;
60
    InternalKey smallest, largest;
61
  };
62
63
0
  Output* current_output() { return &outputs[outputs.size() - 1]; }
64
65
  explicit CompactionState(Compaction* c)
66
0
      : compaction(c),
67
0
        smallest_snapshot(0),
68
0
        outfile(nullptr),
69
0
        builder(nullptr),
70
0
        total_bytes(0) {}
71
72
  Compaction* const compaction;
73
74
  // Sequence numbers < smallest_snapshot are not significant since we
75
  // will never have to service a snapshot below smallest_snapshot.
76
  // Therefore if we have seen a sequence number S <= smallest_snapshot,
77
  // we can drop all entries for the same key with sequence numbers < S.
78
  SequenceNumber smallest_snapshot;
79
80
  std::vector<Output> outputs;
81
82
  // State kept for output being generated
83
  WritableFile* outfile;
84
  TableBuilder* builder;
85
86
  uint64_t total_bytes;
87
};
88
89
// Fix user-supplied options to be reasonable
90
template <class T, class V>
91
324
static void ClipToRange(T* ptr, V minvalue, V maxvalue) {
92
324
  if (static_cast<V>(*ptr) > maxvalue) *ptr = maxvalue;
  Branch (92:7): [True: 0, False: 81]
  Branch (92:7): [True: 0, False: 243]
93
324
  if (static_cast<V>(*ptr) < minvalue) *ptr = minvalue;
  Branch (93:7): [True: 0, False: 81]
  Branch (93:7): [True: 27, False: 216]
94
324
}
db_impl.cc:void leveldb::ClipToRange<int, int>(int*, int, int)
Line
Count
Source
91
81
static void ClipToRange(T* ptr, V minvalue, V maxvalue) {
92
81
  if (static_cast<V>(*ptr) > maxvalue) *ptr = maxvalue;
  Branch (92:7): [True: 0, False: 81]
93
81
  if (static_cast<V>(*ptr) < minvalue) *ptr = minvalue;
  Branch (93:7): [True: 0, False: 81]
94
81
}
db_impl.cc:void leveldb::ClipToRange<unsigned long, int>(unsigned long*, int, int)
Line
Count
Source
91
243
static void ClipToRange(T* ptr, V minvalue, V maxvalue) {
92
243
  if (static_cast<V>(*ptr) > maxvalue) *ptr = maxvalue;
  Branch (92:7): [True: 0, False: 243]
93
243
  if (static_cast<V>(*ptr) < minvalue) *ptr = minvalue;
  Branch (93:7): [True: 27, False: 216]
94
243
}
95
Options SanitizeOptions(const std::string& dbname,
96
                        const InternalKeyComparator* icmp,
97
                        const InternalFilterPolicy* ipolicy,
98
81
                        const Options& src) {
99
81
  Options result = src;
100
81
  result.comparator = icmp;
101
81
  result.filter_policy = (src.filter_policy != nullptr) ? ipolicy : nullptr;
  Branch (101:26): [True: 81, False: 0]
102
81
  ClipToRange(&result.max_open_files, 64 + kNumNonTableCacheFiles, 50000);
103
81
  ClipToRange(&result.write_buffer_size, 64 << 10, 1 << 30);
104
81
  ClipToRange(&result.max_file_size, 1 << 20, 1 << 30);
105
81
  ClipToRange(&result.block_size, 1 << 10, 4 << 20);
106
81
  if (result.info_log == nullptr) {
  Branch (106:7): [True: 0, False: 81]
107
    // Open a log file in the same directory as the db
108
0
    src.env->CreateDir(dbname);  // In case it does not exist
109
0
    src.env->RenameFile(InfoLogFileName(dbname), OldInfoLogFileName(dbname));
110
0
    Status s = src.env->NewLogger(InfoLogFileName(dbname), &result.info_log);
111
0
    if (!s.ok()) {
  Branch (111:9): [True: 0, False: 0]
112
      // No place suitable for logging
113
0
      result.info_log = nullptr;
114
0
    }
115
0
  }
116
81
  if (result.block_cache == nullptr) {
  Branch (116:7): [True: 0, False: 81]
117
0
    result.block_cache = NewLRUCache(8 << 20);
118
0
  }
119
81
  return result;
120
81
}
121
122
81
static int TableCacheSize(const Options& sanitized_options) {
123
  // Reserve ten files or so for other uses and give the rest to TableCache.
124
81
  return sanitized_options.max_open_files - kNumNonTableCacheFiles;
125
81
}
126
127
DBImpl::DBImpl(const Options& raw_options, const std::string& dbname)
128
81
    : env_(raw_options.env),
129
81
      internal_comparator_(raw_options.comparator),
130
81
      internal_filter_policy_(raw_options.filter_policy),
131
81
      options_(SanitizeOptions(dbname, &internal_comparator_,
132
81
                               &internal_filter_policy_, raw_options)),
133
81
      owns_info_log_(options_.info_log != raw_options.info_log),
134
81
      owns_cache_(options_.block_cache != raw_options.block_cache),
135
81
      dbname_(dbname),
136
81
      table_cache_(new TableCache(dbname_, options_, TableCacheSize(options_))),
137
81
      db_lock_(nullptr),
138
81
      shutting_down_(false),
139
81
      background_work_finished_signal_(&mutex_),
140
81
      mem_(nullptr),
141
81
      imm_(nullptr),
142
81
      has_imm_(false),
143
81
      logfile_(nullptr),
144
81
      logfile_number_(0),
145
81
      log_(nullptr),
146
81
      seed_(0),
147
81
      tmp_batch_(new WriteBatch),
148
81
      background_compaction_scheduled_(false),
149
81
      manual_compaction_(nullptr),
150
81
      versions_(new VersionSet(dbname_, &options_, table_cache_,
151
81
                               &internal_comparator_)) {}
152
153
450k
DBImpl::~DBImpl() {
154
  // Wait for background work to finish.
155
450k
  mutex_.Lock();
156
450k
  shutting_down_.store(true, std::memory_order_release);
157
450k
  while (background_compaction_scheduled_) {
  Branch (157:10): [True: 0, False: 450k]
158
0
    background_work_finished_signal_.Wait();
159
0
  }
160
450k
  mutex_.Unlock();
161
162
450k
  if (db_lock_ != nullptr) {
  Branch (162:7): [True: 450k, False: 0]
163
450k
    env_->UnlockFile(db_lock_);
164
450k
  }
165
166
450k
  delete versions_;
167
450k
  if (mem_ != nullptr) mem_->Unref();
  Branch (167:7): [True: 450k, False: 0]
168
450k
  if (imm_ != nullptr) imm_->Unref();
  Branch (168:7): [True: 0, False: 450k]
169
450k
  delete tmp_batch_;
170
450k
  delete log_;
171
450k
  delete logfile_;
172
450k
  delete table_cache_;
173
174
450k
  if (owns_info_log_) {
  Branch (174:7): [True: 0, False: 450k]
175
0
    delete options_.info_log;
176
0
  }
177
450k
  if (owns_cache_) {
  Branch (177:7): [True: 0, False: 450k]
178
0
    delete options_.block_cache;
179
0
  }
180
450k
}
181
182
81
Status DBImpl::NewDB() {
183
81
  VersionEdit new_db;
184
81
  new_db.SetComparatorName(user_comparator()->Name());
185
81
  new_db.SetLogNumber(0);
186
81
  new_db.SetNextFile(2);
187
81
  new_db.SetLastSequence(0);
188
189
81
  const std::string manifest = DescriptorFileName(dbname_, 1);
190
81
  WritableFile* file;
191
81
  Status s = env_->NewWritableFile(manifest, &file);
192
81
  if (!s.ok()) {
  Branch (192:7): [True: 0, False: 81]
193
0
    return s;
194
0
  }
195
81
  {
196
81
    log::Writer log(file);
197
81
    std::string record;
198
81
    new_db.EncodeTo(&record);
199
81
    s = log.AddRecord(record);
200
81
    if (s.ok()) {
  Branch (200:9): [True: 81, False: 0]
201
81
      s = file->Close();
202
81
    }
203
81
  }
204
81
  delete file;
205
81
  if (s.ok()) {
  Branch (205:7): [True: 81, False: 0]
206
    // Make "CURRENT" file that points to the new manifest file.
207
81
    s = SetCurrentFile(env_, dbname_, 1);
208
81
  } else {
209
0
    env_->DeleteFile(manifest);
210
0
  }
211
81
  return s;
212
81
}
213
214
0
void DBImpl::MaybeIgnoreError(Status* s) const {
215
0
  if (s->ok() || options_.paranoid_checks) {
  Branch (215:7): [True: 0, False: 0]
  Branch (215:18): [True: 0, False: 0]
216
    // No change needed
217
0
  } else {
218
0
    Log(options_.info_log, "Ignoring error %s", s->ToString().c_str());
219
0
    *s = Status::OK();
220
0
  }
221
0
}
222
223
97
void DBImpl::DeleteObsoleteFiles() {
224
97
  mutex_.AssertHeld();
225
226
97
  if (!bg_error_.ok()) {
  Branch (226:7): [True: 0, False: 97]
227
    // After a background error, we don't know whether a new version may
228
    // or may not have been committed, so we cannot safely garbage collect.
229
0
    return;
230
0
  }
231
232
  // Make a set of all of the live files
233
97
  std::set<uint64_t> live = pending_outputs_;
234
97
  versions_->AddLiveFiles(&live);
235
236
97
  std::vector<std::string> filenames;
237
97
  env_->GetChildren(dbname_, &filenames);  // Ignoring errors on purpose
238
97
  uint64_t number;
239
97
  FileType type;
240
97
  std::vector<std::string> files_to_delete;
241
695
  for (std::string& filename : filenames) {
  Branch (241:30): [True: 695, False: 97]
242
695
    if (ParseFileName(filename, &number, &type)) {
  Branch (242:9): [True: 501, False: 194]
243
501
      bool keep = true;
244
501
      switch (type) {
  Branch (244:15): [True: 0, False: 501]
245
113
        case kLogFile:
  Branch (245:9): [True: 113, False: 388]
246
113
          keep = ((number >= versions_->LogNumber()) ||
  Branch (246:19): [True: 97, False: 16]
247
113
                  (number == versions_->PrevLogNumber()));
  Branch (247:19): [True: 0, False: 16]
248
113
          break;
249
178
        case kDescriptorFile:
  Branch (249:9): [True: 178, False: 323]
250
          // Keep my manifest file, and any newer incarnations'
251
          // (in case there is a race that allows other incarnations)
252
178
          keep = (number >= versions_->ManifestFileNumber());
253
178
          break;
254
16
        case kTableFile:
  Branch (254:9): [True: 16, False: 485]
255
16
          keep = (live.find(number) != live.end());
256
16
          break;
257
0
        case kTempFile:
  Branch (257:9): [True: 0, False: 501]
258
          // Any temp files that are currently being written to must
259
          // be recorded in pending_outputs_, which is inserted into "live"
260
0
          keep = (live.find(number) != live.end());
261
0
          break;
262
97
        case kCurrentFile:
  Branch (262:9): [True: 97, False: 404]
263
194
        case kDBLockFile:
  Branch (263:9): [True: 97, False: 404]
264
194
        case kInfoLogFile:
  Branch (264:9): [True: 0, False: 501]
265
194
          keep = true;
266
194
          break;
267
501
      }
268
269
501
      if (!keep) {
  Branch (269:11): [True: 97, False: 404]
270
97
        files_to_delete.push_back(std::move(filename));
271
97
        if (type == kTableFile) {
  Branch (271:13): [True: 0, False: 97]
272
0
          table_cache_->Evict(number);
273
0
        }
274
97
        Log(options_.info_log, "Delete type=%d #%lld\n", static_cast<int>(type),
275
97
            static_cast<unsigned long long>(number));
276
97
      }
277
501
    }
278
695
  }
279
280
  // While deleting all files unblock other threads. All files being deleted
281
  // have unique names which will not collide with newly created files and
282
  // are therefore safe to delete while allowing other threads to proceed.
283
97
  mutex_.Unlock();
284
97
  for (const std::string& filename : files_to_delete) {
  Branch (284:36): [True: 97, False: 97]
285
97
    env_->DeleteFile(dbname_ + "/" + filename);
286
97
  }
287
97
  mutex_.Lock();
288
97
}
289
290
81
Status DBImpl::Recover(VersionEdit* edit, bool* save_manifest) {
291
81
  mutex_.AssertHeld();
292
293
  // Ignore error from CreateDir since the creation of the DB is
294
  // committed only when the descriptor is created, and this directory
295
  // may already exist from a previous failed creation attempt.
296
81
  env_->CreateDir(dbname_);
297
81
  assert(db_lock_ == nullptr);
  Branch (297:3): [True: 81, False: 0]
298
81
  Status s = env_->LockFile(LockFileName(dbname_), &db_lock_);
299
81
  if (!s.ok()) {
  Branch (299:7): [True: 0, False: 81]
300
0
    return s;
301
0
  }
302
303
81
  if (!env_->FileExists(CurrentFileName(dbname_))) {
  Branch (303:7): [True: 81, False: 0]
304
81
    if (options_.create_if_missing) {
  Branch (304:9): [True: 81, False: 0]
305
81
      s = NewDB();
306
81
      if (!s.ok()) {
  Branch (306:11): [True: 0, False: 81]
307
0
        return s;
308
0
      }
309
81
    } else {
310
0
      return Status::InvalidArgument(
311
0
          dbname_, "does not exist (create_if_missing is false)");
312
0
    }
313
81
  } else {
314
0
    if (options_.error_if_exists) {
  Branch (314:9): [True: 0, False: 0]
315
0
      return Status::InvalidArgument(dbname_,
316
0
                                     "exists (error_if_exists is true)");
317
0
    }
318
0
  }
319
320
81
  s = versions_->Recover(save_manifest);
321
81
  if (!s.ok()) {
  Branch (321:7): [True: 0, False: 81]
322
0
    return s;
323
0
  }
324
81
  SequenceNumber max_sequence(0);
325
326
  // Recover from all newer log files than the ones named in the
327
  // descriptor (new log files may have been added by the previous
328
  // incarnation without registering them in the descriptor).
329
  //
330
  // Note that PrevLogNumber() is no longer used, but we pay
331
  // attention to it in case we are recovering a database
332
  // produced by an older version of leveldb.
333
81
  const uint64_t min_log = versions_->LogNumber();
334
81
  const uint64_t prev_log = versions_->PrevLogNumber();
335
81
  std::vector<std::string> filenames;
336
81
  s = env_->GetChildren(dbname_, &filenames);
337
81
  if (!s.ok()) {
  Branch (337:7): [True: 0, False: 81]
338
0
    return s;
339
0
  }
340
81
  std::set<uint64_t> expected;
341
81
  versions_->AddLiveFiles(&expected);
342
81
  uint64_t number;
343
81
  FileType type;
344
81
  std::vector<uint64_t> logs;
345
486
  for (size_t i = 0; i < filenames.size(); i++) {
  Branch (345:22): [True: 405, False: 81]
346
405
    if (ParseFileName(filenames[i], &number, &type)) {
  Branch (346:9): [True: 243, False: 162]
347
243
      expected.erase(number);
348
243
      if (type == kLogFile && ((number >= min_log) || (number == prev_log)))
  Branch (348:11): [True: 0, False: 243]
  Branch (348:32): [True: 0, False: 0]
  Branch (348:55): [True: 0, False: 0]
349
0
        logs.push_back(number);
350
243
    }
351
405
  }
352
81
  if (!expected.empty()) {
  Branch (352:7): [True: 0, False: 81]
353
0
    char buf[50];
354
0
    snprintf(buf, sizeof(buf), "%d missing files; e.g.",
355
0
             static_cast<int>(expected.size()));
356
0
    return Status::Corruption(buf, TableFileName(dbname_, *(expected.begin())));
357
0
  }
358
359
  // Recover in the order in which the logs were generated
360
81
  std::sort(logs.begin(), logs.end());
361
81
  for (size_t i = 0; i < logs.size(); i++) {
  Branch (361:22): [True: 0, False: 81]
362
0
    s = RecoverLogFile(logs[i], (i == logs.size() - 1), save_manifest, edit,
363
0
                       &max_sequence);
364
0
    if (!s.ok()) {
  Branch (364:9): [True: 0, False: 0]
365
0
      return s;
366
0
    }
367
368
    // The previous incarnation may not have written any MANIFEST
369
    // records after allocating this log number.  So we manually
370
    // update the file number allocation counter in VersionSet.
371
0
    versions_->MarkFileNumberUsed(logs[i]);
372
0
  }
373
374
81
  if (versions_->LastSequence() < max_sequence) {
  Branch (374:7): [True: 0, False: 81]
375
0
    versions_->SetLastSequence(max_sequence);
376
0
  }
377
378
81
  return Status::OK();
379
81
}
380
381
Status DBImpl::RecoverLogFile(uint64_t log_number, bool last_log,
382
                              bool* save_manifest, VersionEdit* edit,
383
0
                              SequenceNumber* max_sequence) {
384
0
  struct LogReporter : public log::Reader::Reporter {
385
0
    Env* env;
386
0
    Logger* info_log;
387
0
    const char* fname;
388
0
    Status* status;  // null if options_.paranoid_checks==false
389
0
    void Corruption(size_t bytes, const Status& s) override {
390
0
      Log(info_log, "%s%s: dropping %d bytes; %s",
391
0
          (this->status == nullptr ? "(ignoring error) " : ""), fname,
  Branch (391:12): [True: 0, False: 0]
392
0
          static_cast<int>(bytes), s.ToString().c_str());
393
0
      if (this->status != nullptr && this->status->ok()) *this->status = s;
  Branch (393:11): [True: 0, False: 0]
  Branch (393:38): [True: 0, False: 0]
394
0
    }
395
0
  };
396
397
0
  mutex_.AssertHeld();
398
399
  // Open the log file
400
0
  std::string fname = LogFileName(dbname_, log_number);
401
0
  SequentialFile* file;
402
0
  Status status = env_->NewSequentialFile(fname, &file);
403
0
  if (!status.ok()) {
  Branch (403:7): [True: 0, False: 0]
404
0
    MaybeIgnoreError(&status);
405
0
    return status;
406
0
  }
407
408
  // Create the log reader.
409
0
  LogReporter reporter;
410
0
  reporter.env = env_;
411
0
  reporter.info_log = options_.info_log;
412
0
  reporter.fname = fname.c_str();
413
0
  reporter.status = (options_.paranoid_checks ? &status : nullptr);
  Branch (413:22): [True: 0, False: 0]
414
  // We intentionally make log::Reader do checksumming even if
415
  // paranoid_checks==false so that corruptions cause entire commits
416
  // to be skipped instead of propagating bad information (like overly
417
  // large sequence numbers).
418
0
  log::Reader reader(file, &reporter, true /*checksum*/, 0 /*initial_offset*/);
419
0
  Log(options_.info_log, "Recovering log #%llu",
420
0
      (unsigned long long)log_number);
421
422
  // Read all the records and add to a memtable
423
0
  std::string scratch;
424
0
  Slice record;
425
0
  WriteBatch batch;
426
0
  int compactions = 0;
427
0
  MemTable* mem = nullptr;
428
0
  while (reader.ReadRecord(&record, &scratch) && status.ok()) {
  Branch (428:10): [True: 0, False: 0]
  Branch (428:50): [True: 0, False: 0]
429
0
    if (record.size() < 12) {
  Branch (429:9): [True: 0, False: 0]
430
0
      reporter.Corruption(record.size(),
431
0
                          Status::Corruption("log record too small", fname));
432
0
      continue;
433
0
    }
434
0
    WriteBatchInternal::SetContents(&batch, record);
435
436
0
    if (mem == nullptr) {
  Branch (436:9): [True: 0, False: 0]
437
0
      mem = new MemTable(internal_comparator_);
438
0
      mem->Ref();
439
0
    }
440
0
    status = WriteBatchInternal::InsertInto(&batch, mem);
441
0
    MaybeIgnoreError(&status);
442
0
    if (!status.ok()) {
  Branch (442:9): [True: 0, False: 0]
443
0
      break;
444
0
    }
445
0
    const SequenceNumber last_seq = WriteBatchInternal::Sequence(&batch) +
446
0
                                    WriteBatchInternal::Count(&batch) - 1;
447
0
    if (last_seq > *max_sequence) {
  Branch (447:9): [True: 0, False: 0]
448
0
      *max_sequence = last_seq;
449
0
    }
450
451
0
    if (mem->ApproximateMemoryUsage() > options_.write_buffer_size) {
  Branch (451:9): [True: 0, False: 0]
452
0
      compactions++;
453
0
      *save_manifest = true;
454
0
      status = WriteLevel0Table(mem, edit, nullptr);
455
0
      mem->Unref();
456
0
      mem = nullptr;
457
0
      if (!status.ok()) {
  Branch (457:11): [True: 0, False: 0]
458
        // Reflect errors immediately so that conditions like full
459
        // file-systems cause the DB::Open() to fail.
460
0
        break;
461
0
      }
462
0
    }
463
0
  }
464
465
0
  delete file;
466
467
  // See if we should keep reusing the last log file.
468
0
  if (status.ok() && options_.reuse_logs && last_log && compactions == 0) {
  Branch (468:7): [True: 0, False: 0]
  Branch (468:22): [True: 0, False: 0]
  Branch (468:45): [True: 0, False: 0]
  Branch (468:57): [True: 0, False: 0]
469
0
    assert(logfile_ == nullptr);
  Branch (469:5): [True: 0, False: 0]
470
0
    assert(log_ == nullptr);
  Branch (470:5): [True: 0, False: 0]
471
0
    assert(mem_ == nullptr);
  Branch (471:5): [True: 0, False: 0]
472
0
    uint64_t lfile_size;
473
0
    if (env_->GetFileSize(fname, &lfile_size).ok() &&
  Branch (473:9): [True: 0, False: 0]
  Branch (473:9): [True: 0, False: 0]
474
0
        env_->NewAppendableFile(fname, &logfile_).ok()) {
  Branch (474:9): [True: 0, False: 0]
475
0
      Log(options_.info_log, "Reusing old log %s \n", fname.c_str());
476
0
      log_ = new log::Writer(logfile_, lfile_size);
477
0
      logfile_number_ = log_number;
478
0
      if (mem != nullptr) {
  Branch (478:11): [True: 0, False: 0]
479
0
        mem_ = mem;
480
0
        mem = nullptr;
481
0
      } else {
482
        // mem can be nullptr if lognum exists but was empty.
483
0
        mem_ = new MemTable(internal_comparator_);
484
0
        mem_->Ref();
485
0
      }
486
0
    }
487
0
  }
488
489
0
  if (mem != nullptr) {
  Branch (489:7): [True: 0, False: 0]
490
    // mem did not get reused; compact it.
491
0
    if (status.ok()) {
  Branch (491:9): [True: 0, False: 0]
492
0
      *save_manifest = true;
493
0
      status = WriteLevel0Table(mem, edit, nullptr);
494
0
    }
495
0
    mem->Unref();
496
0
  }
497
498
0
  return status;
499
0
}
500
501
Status DBImpl::WriteLevel0Table(MemTable* mem, VersionEdit* edit,
502
16
                                Version* base) {
503
16
  mutex_.AssertHeld();
504
16
  const uint64_t start_micros = env_->NowMicros();
505
16
  FileMetaData meta;
506
16
  meta.number = versions_->NewFileNumber();
507
16
  pending_outputs_.insert(meta.number);
508
16
  Iterator* iter = mem->NewIterator();
509
16
  Log(options_.info_log, "Level-0 table #%llu: started",
510
16
      (unsigned long long)meta.number);
511
512
16
  Status s;
513
16
  {
514
16
    mutex_.Unlock();
515
16
    s = BuildTable(dbname_, env_, options_, table_cache_, iter, &meta);
516
16
    mutex_.Lock();
517
16
  }
518
519
16
  Log(options_.info_log, "Level-0 table #%llu: %lld bytes %s",
520
16
      (unsigned long long)meta.number, (unsigned long long)meta.file_size,
521
16
      s.ToString().c_str());
522
16
  delete iter;
523
16
  pending_outputs_.erase(meta.number);
524
525
  // Note that if file_size is zero, the file has been deleted and
526
  // should not be added to the manifest.
527
16
  int level = 0;
528
16
  if (s.ok() && meta.file_size > 0) {
  Branch (528:7): [True: 16, False: 0]
  Branch (528:17): [True: 16, False: 0]
529
16
    const Slice min_user_key = meta.smallest.user_key();
530
16
    const Slice max_user_key = meta.largest.user_key();
531
16
    if (base != nullptr) {
  Branch (531:9): [True: 16, False: 0]
532
16
      level = base->PickLevelForMemTableOutput(min_user_key, max_user_key);
533
16
    }
534
16
    edit->AddFile(level, meta.number, meta.file_size, meta.smallest,
535
16
                  meta.largest);
536
16
  }
537
538
16
  CompactionStats stats;
539
16
  stats.micros = env_->NowMicros() - start_micros;
540
16
  stats.bytes_written = meta.file_size;
541
16
  stats_[level].Add(stats);
542
16
  return s;
543
16
}
544
545
16
void DBImpl::CompactMemTable() {
546
16
  mutex_.AssertHeld();
547
16
  assert(imm_ != nullptr);
  Branch (547:3): [True: 16, False: 0]
548
549
  // Save the contents of the memtable as a new Table
550
16
  VersionEdit edit;
551
16
  Version* base = versions_->current();
552
16
  base->Ref();
553
16
  Status s = WriteLevel0Table(imm_, &edit, base);
554
16
  base->Unref();
555
556
16
  if (s.ok() && shutting_down_.load(std::memory_order_acquire)) {
  Branch (556:7): [True: 16, False: 0]
  Branch (556:17): [True: 0, False: 16]
557
0
    s = Status::IOError("Deleting DB during memtable compaction");
558
0
  }
559
560
  // Replace immutable memtable with the generated Table
561
16
  if (s.ok()) {
  Branch (561:7): [True: 16, False: 0]
562
16
    edit.SetPrevLogNumber(0);
563
16
    edit.SetLogNumber(logfile_number_);  // Earlier logs no longer needed
564
16
    s = versions_->LogAndApply(&edit, &mutex_);
565
16
  }
566
567
16
  if (s.ok()) {
  Branch (567:7): [True: 16, False: 0]
568
    // Commit to the new state
569
16
    imm_->Unref();
570
16
    imm_ = nullptr;
571
16
    has_imm_.store(false, std::memory_order_release);
572
16
    DeleteObsoleteFiles();
573
16
  } else {
574
0
    RecordBackgroundError(s);
575
0
  }
576
16
}
577
578
16
void DBImpl::CompactRange(const Slice* begin, const Slice* end) {
579
16
  int max_level_with_files = 1;
580
16
  {
581
16
    MutexLock l(&mutex_);
582
16
    Version* base = versions_->current();
583
112
    for (int level = 1; level < config::kNumLevels; level++) {
  Branch (583:25): [True: 96, False: 16]
584
96
      if (base->OverlapInLevel(level, begin, end)) {
  Branch (584:11): [True: 0, False: 96]
585
0
        max_level_with_files = level;
586
0
      }
587
96
    }
588
16
  }
589
16
  TEST_CompactMemTable();  // TODO(sanjay): Skip if memtable does not overlap
590
32
  for (int level = 0; level < max_level_with_files; level++) {
  Branch (590:23): [True: 16, False: 16]
591
16
    TEST_CompactRange(level, begin, end);
592
16
  }
593
16
}
594
595
void DBImpl::TEST_CompactRange(int level, const Slice* begin,
596
16
                               const Slice* end) {
597
16
  assert(level >= 0);
  Branch (597:3): [True: 16, False: 0]
598
16
  assert(level + 1 < config::kNumLevels);
  Branch (598:3): [True: 16, False: 0]
599
600
16
  InternalKey begin_storage, end_storage;
601
602
16
  ManualCompaction manual;
603
16
  manual.level = level;
604
16
  manual.done = false;
605
16
  if (begin == nullptr) {
  Branch (605:7): [True: 16, False: 0]
606
16
    manual.begin = nullptr;
607
16
  } else {
608
0
    begin_storage = InternalKey(*begin, kMaxSequenceNumber, kValueTypeForSeek);
609
0
    manual.begin = &begin_storage;
610
0
  }
611
16
  if (end == nullptr) {
  Branch (611:7): [True: 16, False: 0]
612
16
    manual.end = nullptr;
613
16
  } else {
614
0
    end_storage = InternalKey(*end, 0, static_cast<ValueType>(0));
615
0
    manual.end = &end_storage;
616
0
  }
617
618
16
  MutexLock l(&mutex_);
619
48
  while (!manual.done && !shutting_down_.load(std::memory_order_acquire) &&
  Branch (619:10): [True: 32, False: 16]
  Branch (619:26): [True: 32, False: 0]
620
48
         bg_error_.ok()) {
  Branch (620:10): [True: 32, False: 0]
621
32
    if (manual_compaction_ == nullptr) {  // Idle
  Branch (621:9): [True: 16, False: 16]
622
16
      manual_compaction_ = &manual;
623
16
      MaybeScheduleCompaction();
624
16
    } else {  // Running either my compaction or another compaction.
625
16
      background_work_finished_signal_.Wait();
626
16
    }
627
32
  }
628
16
  if (manual_compaction_ == &manual) {
  Branch (628:7): [True: 0, False: 16]
629
    // Cancel my manual compaction since we aborted early for some reason.
630
0
    manual_compaction_ = nullptr;
631
0
  }
632
16
}
633
634
16
Status DBImpl::TEST_CompactMemTable() {
635
  // nullptr batch means just wait for earlier writes to be done
636
16
  Status s = Write(WriteOptions(), nullptr);
637
16
  if (s.ok()) {
  Branch (637:7): [True: 16, False: 0]
638
    // Wait until the compaction completes
639
16
    MutexLock l(&mutex_);
640
32
    while (imm_ != nullptr && bg_error_.ok()) {
  Branch (640:12): [True: 16, False: 16]
  Branch (640:31): [True: 16, False: 0]
641
16
      background_work_finished_signal_.Wait();
642
16
    }
643
16
    if (imm_ != nullptr) {
  Branch (643:9): [True: 0, False: 16]
644
0
      s = bg_error_;
645
0
    }
646
16
  }
647
16
  return s;
648
16
}
649
650
0
void DBImpl::RecordBackgroundError(const Status& s) {
651
0
  mutex_.AssertHeld();
652
0
  if (bg_error_.ok()) {
  Branch (652:7): [True: 0, False: 0]
653
0
    bg_error_ = s;
654
0
    background_work_finished_signal_.SignalAll();
655
0
  }
656
0
}
657
658
145
void DBImpl::MaybeScheduleCompaction() {
659
145
  mutex_.AssertHeld();
660
145
  if (background_compaction_scheduled_) {
  Branch (660:7): [True: 0, False: 145]
661
    // Already scheduled
662
145
  } else if (shutting_down_.load(std::memory_order_acquire)) {
  Branch (662:14): [True: 0, False: 145]
663
    // DB is being deleted; no more background compactions
664
145
  } else if (!bg_error_.ok()) {
  Branch (664:14): [True: 0, False: 145]
665
    // Already got an error; no more changes
666
145
  } else if (imm_ == nullptr && manual_compaction_ == nullptr &&
  Branch (666:14): [True: 129, False: 16]
  Branch (666:33): [True: 113, False: 16]
667
145
             !versions_->NeedsCompaction()) {
  Branch (667:14): [True: 113, False: 0]
668
    // No work to be done
669
113
  } else {
670
32
    background_compaction_scheduled_ = true;
671
32
    env_->Schedule(&DBImpl::BGWork, this);
672
32
  }
673
145
}
674
675
32
void DBImpl::BGWork(void* db) {
676
32
  reinterpret_cast<DBImpl*>(db)->BackgroundCall();
677
32
}
678
679
32
void DBImpl::BackgroundCall() {
680
32
  MutexLock l(&mutex_);
681
32
  assert(background_compaction_scheduled_);
  Branch (681:3): [True: 32, False: 0]
682
32
  if (shutting_down_.load(std::memory_order_acquire)) {
  Branch (682:7): [True: 0, False: 32]
683
    // No more background work when shutting down.
684
32
  } else if (!bg_error_.ok()) {
  Branch (684:14): [True: 0, False: 32]
685
    // No more background work after a background error.
686
32
  } else {
687
32
    BackgroundCompaction();
688
32
  }
689
690
32
  background_compaction_scheduled_ = false;
691
692
  // Previous compaction may have produced too many files in a level,
693
  // so reschedule another compaction if needed.
694
32
  MaybeScheduleCompaction();
695
32
  background_work_finished_signal_.SignalAll();
696
32
}
697
698
32
void DBImpl::BackgroundCompaction() {
699
32
  mutex_.AssertHeld();
700
701
32
  if (imm_ != nullptr) {
  Branch (701:7): [True: 16, False: 16]
702
16
    CompactMemTable();
703
16
    return;
704
16
  }
705
706
16
  Compaction* c;
707
16
  bool is_manual = (manual_compaction_ != nullptr);
708
16
  InternalKey manual_end;
709
16
  if (is_manual) {
  Branch (709:7): [True: 16, False: 0]
710
16
    ManualCompaction* m = manual_compaction_;
711
16
    c = versions_->CompactRange(m->level, m->begin, m->end);
712
16
    m->done = (c == nullptr);
713
16
    if (c != nullptr) {
  Branch (713:9): [True: 0, False: 16]
714
0
      manual_end = c->input(0, c->num_input_files(0) - 1)->largest;
715
0
    }
716
16
    Log(options_.info_log,
717
16
        "Manual compaction at level-%d from %s .. %s; will stop at %s\n",
718
16
        m->level, (m->begin ? m->begin->DebugString().c_str() : "(begin)"),
  Branch (718:20): [True: 0, False: 16]
719
16
        (m->end ? m->end->DebugString().c_str() : "(end)"),
  Branch (719:10): [True: 0, False: 16]
720
16
        (m->done ? "(end)" : manual_end.DebugString().c_str()));
  Branch (720:10): [True: 16, False: 0]
721
16
  } else {
722
0
    c = versions_->PickCompaction();
723
0
  }
724
725
16
  Status status;
726
16
  if (c == nullptr) {
  Branch (726:7): [True: 16, False: 0]
727
    // Nothing to do
728
16
  } else if (!is_manual && c->IsTrivialMove()) {
  Branch (728:14): [True: 0, False: 0]
  Branch (728:28): [True: 0, False: 0]
729
    // Move file to next level
730
0
    assert(c->num_input_files(0) == 1);
  Branch (730:5): [True: 0, False: 0]
731
0
    FileMetaData* f = c->input(0, 0);
732
0
    c->edit()->DeleteFile(c->level(), f->number);
733
0
    c->edit()->AddFile(c->level() + 1, f->number, f->file_size, f->smallest,
734
0
                       f->largest);
735
0
    status = versions_->LogAndApply(c->edit(), &mutex_);
736
0
    if (!status.ok()) {
  Branch (736:9): [True: 0, False: 0]
737
0
      RecordBackgroundError(status);
738
0
    }
739
0
    VersionSet::LevelSummaryStorage tmp;
740
0
    Log(options_.info_log, "Moved #%lld to level-%d %lld bytes %s: %s\n",
741
0
        static_cast<unsigned long long>(f->number), c->level() + 1,
742
0
        static_cast<unsigned long long>(f->file_size),
743
0
        status.ToString().c_str(), versions_->LevelSummary(&tmp));
744
0
  } else {
745
0
    CompactionState* compact = new CompactionState(c);
746
0
    status = DoCompactionWork(compact);
747
0
    if (!status.ok()) {
  Branch (747:9): [True: 0, False: 0]
748
0
      RecordBackgroundError(status);
749
0
    }
750
0
    CleanupCompaction(compact);
751
0
    c->ReleaseInputs();
752
0
    DeleteObsoleteFiles();
753
0
  }
754
16
  delete c;
755
756
16
  if (status.ok()) {
  Branch (756:7): [True: 16, False: 0]
757
    // Done
758
16
  } else if (shutting_down_.load(std::memory_order_acquire)) {
  Branch (758:14): [True: 0, False: 0]
759
    // Ignore compaction errors found during shutting down
760
0
  } else {
761
0
    Log(options_.info_log, "Compaction error: %s", status.ToString().c_str());
762
0
  }
763
764
16
  if (is_manual) {
  Branch (764:7): [True: 16, False: 0]
765
16
    ManualCompaction* m = manual_compaction_;
766
16
    if (!status.ok()) {
  Branch (766:9): [True: 0, False: 16]
767
0
      m->done = true;
768
0
    }
769
16
    if (!m->done) {
  Branch (769:9): [True: 0, False: 16]
770
      // We only compacted part of the requested range.  Update *m
771
      // to the range that is left to be compacted.
772
0
      m->tmp_storage = manual_end;
773
0
      m->begin = &m->tmp_storage;
774
0
    }
775
16
    manual_compaction_ = nullptr;
776
16
  }
777
16
}
778
779
0
void DBImpl::CleanupCompaction(CompactionState* compact) {
780
0
  mutex_.AssertHeld();
781
0
  if (compact->builder != nullptr) {
  Branch (781:7): [True: 0, False: 0]
782
    // May happen if we get a shutdown call in the middle of compaction
783
0
    compact->builder->Abandon();
784
0
    delete compact->builder;
785
0
  } else {
786
0
    assert(compact->outfile == nullptr);
  Branch (786:5): [True: 0, False: 0]
787
0
  }
788
0
  delete compact->outfile;
789
0
  for (size_t i = 0; i < compact->outputs.size(); i++) {
  Branch (789:22): [True: 0, False: 0]
790
0
    const CompactionState::Output& out = compact->outputs[i];
791
0
    pending_outputs_.erase(out.number);
792
0
  }
793
0
  delete compact;
794
0
}
795
796
0
Status DBImpl::OpenCompactionOutputFile(CompactionState* compact) {
797
0
  assert(compact != nullptr);
  Branch (797:3): [True: 0, False: 0]
798
0
  assert(compact->builder == nullptr);
  Branch (798:3): [True: 0, False: 0]
799
0
  uint64_t file_number;
800
0
  {
801
0
    mutex_.Lock();
802
0
    file_number = versions_->NewFileNumber();
803
0
    pending_outputs_.insert(file_number);
804
0
    CompactionState::Output out;
805
0
    out.number = file_number;
806
0
    out.file_size = 0;
807
0
    out.smallest.Clear();
808
0
    out.largest.Clear();
809
0
    compact->outputs.push_back(out);
810
0
    mutex_.Unlock();
811
0
  }
812
813
  // Make the output file
814
0
  std::string fname = TableFileName(dbname_, file_number);
815
0
  Status s = env_->NewWritableFile(fname, &compact->outfile);
816
0
  if (s.ok()) {
  Branch (816:7): [True: 0, False: 0]
817
0
    compact->builder = new TableBuilder(options_, compact->outfile);
818
0
  }
819
0
  return s;
820
0
}
821
822
Status DBImpl::FinishCompactionOutputFile(CompactionState* compact,
823
0
                                          Iterator* input) {
824
0
  assert(compact != nullptr);
  Branch (824:3): [True: 0, False: 0]
825
0
  assert(compact->outfile != nullptr);
  Branch (825:3): [True: 0, False: 0]
826
0
  assert(compact->builder != nullptr);
  Branch (826:3): [True: 0, False: 0]
827
828
0
  const uint64_t output_number = compact->current_output()->number;
829
0
  assert(output_number != 0);
  Branch (829:3): [True: 0, False: 0]
830
831
  // Check for iterator errors
832
0
  Status s = input->status();
833
0
  const uint64_t current_entries = compact->builder->NumEntries();
834
0
  if (s.ok()) {
  Branch (834:7): [True: 0, False: 0]
835
0
    s = compact->builder->Finish();
836
0
  } else {
837
0
    compact->builder->Abandon();
838
0
  }
839
0
  const uint64_t current_bytes = compact->builder->FileSize();
840
0
  compact->current_output()->file_size = current_bytes;
841
0
  compact->total_bytes += current_bytes;
842
0
  delete compact->builder;
843
0
  compact->builder = nullptr;
844
845
  // Finish and check for file errors
846
0
  if (s.ok()) {
  Branch (846:7): [True: 0, False: 0]
847
0
    s = compact->outfile->Sync();
848
0
  }
849
0
  if (s.ok()) {
  Branch (849:7): [True: 0, False: 0]
850
0
    s = compact->outfile->Close();
851
0
  }
852
0
  delete compact->outfile;
853
0
  compact->outfile = nullptr;
854
855
0
  if (s.ok() && current_entries > 0) {
  Branch (855:7): [True: 0, False: 0]
  Branch (855:17): [True: 0, False: 0]
856
    // Verify that the table is usable
857
0
    Iterator* iter =
858
0
        table_cache_->NewIterator(ReadOptions(), output_number, current_bytes);
859
0
    s = iter->status();
860
0
    delete iter;
861
0
    if (s.ok()) {
  Branch (861:9): [True: 0, False: 0]
862
0
      Log(options_.info_log, "Generated table #%llu@%d: %lld keys, %lld bytes",
863
0
          (unsigned long long)output_number, compact->compaction->level(),
864
0
          (unsigned long long)current_entries,
865
0
          (unsigned long long)current_bytes);
866
0
    }
867
0
  }
868
0
  return s;
869
0
}
870
871
0
Status DBImpl::InstallCompactionResults(CompactionState* compact) {
872
0
  mutex_.AssertHeld();
873
0
  Log(options_.info_log, "Compacted %d@%d + %d@%d files => %lld bytes",
874
0
      compact->compaction->num_input_files(0), compact->compaction->level(),
875
0
      compact->compaction->num_input_files(1), compact->compaction->level() + 1,
876
0
      static_cast<long long>(compact->total_bytes));
877
878
  // Add compaction outputs
879
0
  compact->compaction->AddInputDeletions(compact->compaction->edit());
880
0
  const int level = compact->compaction->level();
881
0
  for (size_t i = 0; i < compact->outputs.size(); i++) {
  Branch (881:22): [True: 0, False: 0]
882
0
    const CompactionState::Output& out = compact->outputs[i];
883
0
    compact->compaction->edit()->AddFile(level + 1, out.number, out.file_size,
884
0
                                         out.smallest, out.largest);
885
0
  }
886
0
  return versions_->LogAndApply(compact->compaction->edit(), &mutex_);
887
0
}
888
889
0
Status DBImpl::DoCompactionWork(CompactionState* compact) {
890
0
  const uint64_t start_micros = env_->NowMicros();
891
0
  int64_t imm_micros = 0;  // Micros spent doing imm_ compactions
892
893
0
  Log(options_.info_log, "Compacting %d@%d + %d@%d files",
894
0
      compact->compaction->num_input_files(0), compact->compaction->level(),
895
0
      compact->compaction->num_input_files(1),
896
0
      compact->compaction->level() + 1);
897
898
0
  assert(versions_->NumLevelFiles(compact->compaction->level()) > 0);
  Branch (898:3): [True: 0, False: 0]
899
0
  assert(compact->builder == nullptr);
  Branch (899:3): [True: 0, False: 0]
900
0
  assert(compact->outfile == nullptr);
  Branch (900:3): [True: 0, False: 0]
901
0
  if (snapshots_.empty()) {
  Branch (901:7): [True: 0, False: 0]
902
0
    compact->smallest_snapshot = versions_->LastSequence();
903
0
  } else {
904
0
    compact->smallest_snapshot = snapshots_.oldest()->sequence_number();
905
0
  }
906
907
0
  Iterator* input = versions_->MakeInputIterator(compact->compaction);
908
909
  // Release mutex while we're actually doing the compaction work
910
0
  mutex_.Unlock();
911
912
0
  input->SeekToFirst();
913
0
  Status status;
914
0
  ParsedInternalKey ikey;
915
0
  std::string current_user_key;
916
0
  bool has_current_user_key = false;
917
0
  SequenceNumber last_sequence_for_key = kMaxSequenceNumber;
918
0
  while (input->Valid() && !shutting_down_.load(std::memory_order_acquire)) {
  Branch (918:10): [True: 0, False: 0]
  Branch (918:28): [True: 0, False: 0]
919
    // Prioritize immutable compaction work
920
0
    if (has_imm_.load(std::memory_order_relaxed)) {
  Branch (920:9): [True: 0, False: 0]
921
0
      const uint64_t imm_start = env_->NowMicros();
922
0
      mutex_.Lock();
923
0
      if (imm_ != nullptr) {
  Branch (923:11): [True: 0, False: 0]
924
0
        CompactMemTable();
925
        // Wake up MakeRoomForWrite() if necessary.
926
0
        background_work_finished_signal_.SignalAll();
927
0
      }
928
0
      mutex_.Unlock();
929
0
      imm_micros += (env_->NowMicros() - imm_start);
930
0
    }
931
932
0
    Slice key = input->key();
933
0
    if (compact->compaction->ShouldStopBefore(key) &&
  Branch (933:9): [True: 0, False: 0]
934
0
        compact->builder != nullptr) {
  Branch (934:9): [True: 0, False: 0]
935
0
      status = FinishCompactionOutputFile(compact, input);
936
0
      if (!status.ok()) {
  Branch (936:11): [True: 0, False: 0]
937
0
        break;
938
0
      }
939
0
    }
940
941
    // Handle key/value, add to state, etc.
942
0
    bool drop = false;
943
0
    if (!ParseInternalKey(key, &ikey)) {
  Branch (943:9): [True: 0, False: 0]
944
      // Do not hide error keys
945
0
      current_user_key.clear();
946
0
      has_current_user_key = false;
947
0
      last_sequence_for_key = kMaxSequenceNumber;
948
0
    } else {
949
0
      if (!has_current_user_key ||
  Branch (949:11): [True: 0, False: 0]
  Branch (949:11): [True: 0, False: 0]
950
0
          user_comparator()->Compare(ikey.user_key, Slice(current_user_key)) !=
  Branch (950:11): [True: 0, False: 0]
951
0
              0) {
952
        // First occurrence of this user key
953
0
        current_user_key.assign(ikey.user_key.data(), ikey.user_key.size());
954
0
        has_current_user_key = true;
955
0
        last_sequence_for_key = kMaxSequenceNumber;
956
0
      }
957
958
0
      if (last_sequence_for_key <= compact->smallest_snapshot) {
  Branch (958:11): [True: 0, False: 0]
959
        // Hidden by an newer entry for same user key
960
0
        drop = true;  // (A)
961
0
      } else if (ikey.type == kTypeDeletion &&
  Branch (961:18): [True: 0, False: 0]
962
0
                 ikey.sequence <= compact->smallest_snapshot &&
  Branch (962:18): [True: 0, False: 0]
963
0
                 compact->compaction->IsBaseLevelForKey(ikey.user_key)) {
  Branch (963:18): [True: 0, False: 0]
964
        // For this user key:
965
        // (1) there is no data in higher levels
966
        // (2) data in lower levels will have larger sequence numbers
967
        // (3) data in layers that are being compacted here and have
968
        //     smaller sequence numbers will be dropped in the next
969
        //     few iterations of this loop (by rule (A) above).
970
        // Therefore this deletion marker is obsolete and can be dropped.
971
0
        drop = true;
972
0
      }
973
974
0
      last_sequence_for_key = ikey.sequence;
975
0
    }
976
#if 0
977
    Log(options_.info_log,
978
        "  Compact: %s, seq %d, type: %d %d, drop: %d, is_base: %d, "
979
        "%d smallest_snapshot: %d",
980
        ikey.user_key.ToString().c_str(),
981
        (int)ikey.sequence, ikey.type, kTypeValue, drop,
982
        compact->compaction->IsBaseLevelForKey(ikey.user_key),
983
        (int)last_sequence_for_key, (int)compact->smallest_snapshot);
984
#endif
985
986
0
    if (!drop) {
  Branch (986:9): [True: 0, False: 0]
987
      // Open output file if necessary
988
0
      if (compact->builder == nullptr) {
  Branch (988:11): [True: 0, False: 0]
989
0
        status = OpenCompactionOutputFile(compact);
990
0
        if (!status.ok()) {
  Branch (990:13): [True: 0, False: 0]
991
0
          break;
992
0
        }
993
0
      }
994
0
      if (compact->builder->NumEntries() == 0) {
  Branch (994:11): [True: 0, False: 0]
995
0
        compact->current_output()->smallest.DecodeFrom(key);
996
0
      }
997
0
      compact->current_output()->largest.DecodeFrom(key);
998
0
      compact->builder->Add(key, input->value());
999
1000
      // Close output file if it is big enough
1001
0
      if (compact->builder->FileSize() >=
  Branch (1001:11): [True: 0, False: 0]
1002
0
          compact->compaction->MaxOutputFileSize()) {
1003
0
        status = FinishCompactionOutputFile(compact, input);
1004
0
        if (!status.ok()) {
  Branch (1004:13): [True: 0, False: 0]
1005
0
          break;
1006
0
        }
1007
0
      }
1008
0
    }
1009
1010
0
    input->Next();
1011
0
  }
1012
1013
0
  if (status.ok() && shutting_down_.load(std::memory_order_acquire)) {
  Branch (1013:7): [True: 0, False: 0]
  Branch (1013:22): [True: 0, False: 0]
1014
0
    status = Status::IOError("Deleting DB during compaction");
1015
0
  }
1016
0
  if (status.ok() && compact->builder != nullptr) {
  Branch (1016:7): [True: 0, False: 0]
  Branch (1016:22): [True: 0, False: 0]
1017
0
    status = FinishCompactionOutputFile(compact, input);
1018
0
  }
1019
0
  if (status.ok()) {
  Branch (1019:7): [True: 0, False: 0]
1020
0
    status = input->status();
1021
0
  }
1022
0
  delete input;
1023
0
  input = nullptr;
1024
1025
0
  CompactionStats stats;
1026
0
  stats.micros = env_->NowMicros() - start_micros - imm_micros;
1027
0
  for (int which = 0; which < 2; which++) {
  Branch (1027:23): [True: 0, False: 0]
1028
0
    for (int i = 0; i < compact->compaction->num_input_files(which); i++) {
  Branch (1028:21): [True: 0, False: 0]
1029
0
      stats.bytes_read += compact->compaction->input(which, i)->file_size;
1030
0
    }
1031
0
  }
1032
0
  for (size_t i = 0; i < compact->outputs.size(); i++) {
  Branch (1032:22): [True: 0, False: 0]
1033
0
    stats.bytes_written += compact->outputs[i].file_size;
1034
0
  }
1035
1036
0
  mutex_.Lock();
1037
0
  stats_[compact->compaction->level() + 1].Add(stats);
1038
1039
0
  if (status.ok()) {
  Branch (1039:7): [True: 0, False: 0]
1040
0
    status = InstallCompactionResults(compact);
1041
0
  }
1042
0
  if (!status.ok()) {
  Branch (1042:7): [True: 0, False: 0]
1043
0
    RecordBackgroundError(status);
1044
0
  }
1045
0
  VersionSet::LevelSummaryStorage tmp;
1046
0
  Log(options_.info_log, "compacted to: %s", versions_->LevelSummary(&tmp));
1047
0
  return status;
1048
0
}
1049
1050
namespace {
1051
1052
struct IterState {
1053
  port::Mutex* const mu;
1054
  Version* const version GUARDED_BY(mu);
1055
  MemTable* const mem GUARDED_BY(mu);
1056
  MemTable* const imm GUARDED_BY(mu);
1057
1058
  IterState(port::Mutex* mutex, MemTable* mem, MemTable* imm, Version* version)
1059
24.8k
      : mu(mutex), version(version), mem(mem), imm(imm) {}
1060
};
1061
1062
24.8k
static void CleanupIteratorState(void* arg1, void* arg2) {
1063
24.8k
  IterState* state = reinterpret_cast<IterState*>(arg1);
1064
24.8k
  state->mu->Lock();
1065
24.8k
  state->mem->Unref();
1066
24.8k
  if (state->imm != nullptr) state->imm->Unref();
  Branch (1066:7): [True: 0, False: 24.8k]
1067
24.8k
  state->version->Unref();
1068
24.8k
  state->mu->Unlock();
1069
24.8k
  delete state;
1070
24.8k
}
1071
1072
}  // anonymous namespace
1073
1074
Iterator* DBImpl::NewInternalIterator(const ReadOptions& options,
1075
                                      SequenceNumber* latest_snapshot,
1076
24.8k
                                      uint32_t* seed) {
1077
24.8k
  mutex_.Lock();
1078
24.8k
  *latest_snapshot = versions_->LastSequence();
1079
1080
  // Collect together all needed child iterators
1081
24.8k
  std::vector<Iterator*> list;
1082
24.8k
  list.push_back(mem_->NewIterator());
1083
24.8k
  mem_->Ref();
1084
24.8k
  if (imm_ != nullptr) {
  Branch (1084:7): [True: 0, False: 24.8k]
1085
0
    list.push_back(imm_->NewIterator());
1086
0
    imm_->Ref();
1087
0
  }
1088
24.8k
  versions_->current()->AddIterators(options, &list);
1089
24.8k
  Iterator* internal_iter =
1090
24.8k
      NewMergingIterator(&internal_comparator_, &list[0], list.size());
1091
24.8k
  versions_->current()->Ref();
1092
1093
24.8k
  IterState* cleanup = new IterState(&mutex_, mem_, imm_, versions_->current());
1094
24.8k
  internal_iter->RegisterCleanup(CleanupIteratorState, cleanup, nullptr);
1095
1096
24.8k
  *seed = ++seed_;
1097
24.8k
  mutex_.Unlock();
1098
24.8k
  return internal_iter;
1099
24.8k
}
1100
1101
0
Iterator* DBImpl::TEST_NewInternalIterator() {
1102
0
  SequenceNumber ignored;
1103
0
  uint32_t ignored_seed;
1104
0
  return NewInternalIterator(ReadOptions(), &ignored, &ignored_seed);
1105
0
}
1106
1107
0
int64_t DBImpl::TEST_MaxNextLevelOverlappingBytes() {
1108
0
  MutexLock l(&mutex_);
1109
0
  return versions_->MaxNextLevelOverlappingBytes();
1110
0
}
1111
1112
Status DBImpl::Get(const ReadOptions& options, const Slice& key,
1113
2.52M
                   std::string* value) {
1114
2.52M
  Status s;
1115
2.52M
  MutexLock l(&mutex_);
1116
2.52M
  SequenceNumber snapshot;
1117
2.52M
  if (options.snapshot != nullptr) {
  Branch (1117:7): [True: 0, False: 2.52M]
1118
0
    snapshot =
1119
0
        static_cast<const SnapshotImpl*>(options.snapshot)->sequence_number();
1120
2.52M
  } else {
1121
2.52M
    snapshot = versions_->LastSequence();
1122
2.52M
  }
1123
1124
2.52M
  MemTable* mem = mem_;
1125
2.52M
  MemTable* imm = imm_;
1126
2.52M
  Version* current = versions_->current();
1127
2.52M
  mem->Ref();
1128
2.52M
  if (imm != nullptr) imm->Ref();
  Branch (1128:7): [True: 385, False: 2.52M]
1129
2.52M
  current->Ref();
1130
1131
2.52M
  bool have_stat_update = false;
1132
2.52M
  Version::GetStats stats;
1133
1134
  // Unlock while reading from files and memtables
1135
2.52M
  {
1136
2.52M
    mutex_.Unlock();
1137
    // First look in the memtable, then in the immutable memtable (if any).
1138
2.52M
    LookupKey lkey(key, snapshot);
1139
2.52M
    if (mem->Get(lkey, value, &s)) {
  Branch (1139:9): [True: 180k, False: 2.34M]
1140
      // Done
1141
2.34M
    } else if (imm != nullptr && imm->Get(lkey, value, &s)) {
  Branch (1141:16): [True: 385, False: 2.34M]
  Branch (1141:34): [True: 0, False: 385]
1142
      // Done
1143
2.34M
    } else {
1144
2.34M
      s = current->Get(options, lkey, value, &stats);
1145
2.34M
      have_stat_update = true;
1146
2.34M
    }
1147
2.52M
    mutex_.Lock();
1148
2.52M
  }
1149
1150
2.52M
  if (have_stat_update && current->UpdateStats(stats)) {
  Branch (1150:7): [True: 2.34M, False: 180k]
  Branch (1150:27): [True: 0, False: 2.34M]
1151
0
    MaybeScheduleCompaction();
1152
0
  }
1153
2.52M
  mem->Unref();
1154
2.52M
  if (imm != nullptr) imm->Unref();
  Branch (1154:7): [True: 385, False: 2.52M]
1155
2.52M
  current->Unref();
1156
2.52M
  return s;
1157
2.52M
}
1158
1159
24.8k
Iterator* DBImpl::NewIterator(const ReadOptions& options) {
1160
24.8k
  SequenceNumber latest_snapshot;
1161
24.8k
  uint32_t seed;
1162
24.8k
  Iterator* iter = NewInternalIterator(options, &latest_snapshot, &seed);
1163
24.8k
  return NewDBIterator(this, user_comparator(), iter,
1164
24.8k
                       (options.snapshot != nullptr
  Branch (1164:25): [True: 0, False: 24.8k]
1165
24.8k
                            ? static_cast<const SnapshotImpl*>(options.snapshot)
1166
0
                                  ->sequence_number()
1167
24.8k
                            : latest_snapshot),
1168
24.8k
                       seed);
1169
24.8k
}
1170
1171
553
void DBImpl::RecordReadSample(Slice key) {
1172
553
  MutexLock l(&mutex_);
1173
553
  if (versions_->current()->RecordReadSample(key)) {
  Branch (1173:7): [True: 0, False: 553]
1174
0
    MaybeScheduleCompaction();
1175
0
  }
1176
553
}
1177
1178
0
const Snapshot* DBImpl::GetSnapshot() {
1179
0
  MutexLock l(&mutex_);
1180
0
  return snapshots_.New(versions_->LastSequence());
1181
0
}
1182
1183
0
void DBImpl::ReleaseSnapshot(const Snapshot* snapshot) {
1184
0
  MutexLock l(&mutex_);
1185
0
  snapshots_.Delete(static_cast<const SnapshotImpl*>(snapshot));
1186
0
}
1187
1188
// Convenience methods
1189
0
Status DBImpl::Put(const WriteOptions& o, const Slice& key, const Slice& val) {
1190
0
  return DB::Put(o, key, val);
1191
0
}
1192
1193
0
Status DBImpl::Delete(const WriteOptions& options, const Slice& key) {
1194
0
  return DB::Delete(options, key);
1195
0
}
1196
1197
831k
Status DBImpl::Write(const WriteOptions& options, WriteBatch* updates) {
1198
831k
  Writer w(&mutex_);
1199
831k
  w.batch = updates;
1200
831k
  w.sync = options.sync;
1201
831k
  w.done = false;
1202
1203
831k
  MutexLock l(&mutex_);
1204
831k
  writers_.push_back(&w);
1205
831k
  while (!w.done && &w != writers_.front()) {
  Branch (1205:10): [True: 831k, False: 0]
  Branch (1205:21): [True: 0, False: 831k]
1206
0
    w.cv.Wait();
1207
0
  }
1208
831k
  if (w.done) {
  Branch (1208:7): [True: 0, False: 831k]
1209
0
    return w.status;
1210
0
  }
1211
1212
  // May temporarily unlock and wait.
1213
831k
  Status status = MakeRoomForWrite(updates == nullptr);
1214
831k
  uint64_t last_sequence = versions_->LastSequence();
1215
831k
  Writer* last_writer = &w;
1216
831k
  if (status.ok() && updates != nullptr) {  // nullptr batch is for compactions
  Branch (1216:7): [True: 831k, False: 0]
  Branch (1216:22): [True: 831k, False: 16]
1217
831k
    WriteBatch* write_batch = BuildBatchGroup(&last_writer);
1218
831k
    WriteBatchInternal::SetSequence(write_batch, last_sequence + 1);
1219
831k
    last_sequence += WriteBatchInternal::Count(write_batch);
1220
1221
    // Add to log and apply to memtable.  We can release the lock
1222
    // during this phase since &w is currently responsible for logging
1223
    // and protects against concurrent loggers and concurrent writes
1224
    // into mem_.
1225
831k
    {
1226
831k
      mutex_.Unlock();
1227
831k
      status = log_->AddRecord(WriteBatchInternal::Contents(write_batch));
1228
831k
      bool sync_error = false;
1229
831k
      if (status.ok() && options.sync) {
  Branch (1229:11): [True: 831k, False: 18.4E]
  Branch (1229:26): [True: 305k, False: 525k]
1230
305k
        status = logfile_->Sync();
1231
305k
        if (!status.ok()) {
  Branch (1231:13): [True: 0, False: 305k]
1232
0
          sync_error = true;
1233
0
        }
1234
305k
      }
1235
831k
      if (status.ok()) {
  Branch (1235:11): [True: 831k, False: 18.4E]
1236
831k
        status = WriteBatchInternal::InsertInto(write_batch, mem_);
1237
831k
      }
1238
831k
      mutex_.Lock();
1239
831k
      if (sync_error) {
  Branch (1239:11): [True: 0, False: 831k]
1240
        // The state of the log file is indeterminate: the log record we
1241
        // just added may or may not show up when the DB is re-opened.
1242
        // So we force the DB into a mode where all future writes fail.
1243
0
        RecordBackgroundError(status);
1244
0
      }
1245
831k
    }
1246
831k
    if (write_batch == tmp_batch_) tmp_batch_->Clear();
  Branch (1246:9): [True: 0, False: 831k]
1247
1248
831k
    versions_->SetLastSequence(last_sequence);
1249
831k
  }
1250
1251
831k
  while (true) {
  Branch (1251:10): [Folded - Ignored]
1252
831k
    Writer* ready = writers_.front();
1253
831k
    writers_.pop_front();
1254
831k
    if (ready != &w) {
  Branch (1254:9): [True: 0, False: 831k]
1255
0
      ready->status = status;
1256
0
      ready->done = true;
1257
0
      ready->cv.Signal();
1258
0
    }
1259
831k
    if (ready == last_writer) break;
  Branch (1259:9): [True: 831k, False: 0]
1260
831k
  }
1261
1262
  // Notify new head of write queue
1263
831k
  if (!writers_.empty()) {
  Branch (1263:7): [True: 0, False: 831k]
1264
0
    writers_.front()->cv.Signal();
1265
0
  }
1266
1267
831k
  return status;
1268
831k
}
1269
1270
// REQUIRES: Writer list must be non-empty
1271
// REQUIRES: First writer must have a non-null batch
1272
831k
WriteBatch* DBImpl::BuildBatchGroup(Writer** last_writer) {
1273
831k
  mutex_.AssertHeld();
1274
831k
  assert(!writers_.empty());
  Branch (1274:3): [True: 831k, False: 0]
1275
831k
  Writer* first = writers_.front();
1276
831k
  WriteBatch* result = first->batch;
1277
831k
  assert(result != nullptr);
  Branch (1277:3): [True: 831k, False: 0]
1278
1279
831k
  size_t size = WriteBatchInternal::ByteSize(first->batch);
1280
1281
  // Allow the group to grow up to a maximum size, but if the
1282
  // original write is small, limit the growth so we do not slow
1283
  // down the small write too much.
1284
831k
  size_t max_size = 1 << 20;
1285
831k
  if (size <= (128 << 10)) {
  Branch (1285:7): [True: 831k, False: 0]
1286
831k
    max_size = size + (128 << 10);
1287
831k
  }
1288
1289
831k
  *last_writer = first;
1290
831k
  std::deque<Writer*>::iterator iter = writers_.begin();
1291
831k
  ++iter;  // Advance past "first"
1292
831k
  for (; iter != writers_.end(); ++iter) {
  Branch (1292:10): [True: 0, False: 831k]
1293
0
    Writer* w = *iter;
1294
0
    if (w->sync && !first->sync) {
  Branch (1294:9): [True: 0, False: 0]
  Branch (1294:20): [True: 0, False: 0]
1295
      // Do not include a sync write into a batch handled by a non-sync write.
1296
0
      break;
1297
0
    }
1298
1299
0
    if (w->batch != nullptr) {
  Branch (1299:9): [True: 0, False: 0]
1300
0
      size += WriteBatchInternal::ByteSize(w->batch);
1301
0
      if (size > max_size) {
  Branch (1301:11): [True: 0, False: 0]
1302
        // Do not make batch too big
1303
0
        break;
1304
0
      }
1305
1306
      // Append to *result
1307
0
      if (result == first->batch) {
  Branch (1307:11): [True: 0, False: 0]
1308
        // Switch to temporary batch instead of disturbing caller's batch
1309
0
        result = tmp_batch_;
1310
0
        assert(WriteBatchInternal::Count(result) == 0);
  Branch (1310:9): [True: 0, False: 0]
1311
0
        WriteBatchInternal::Append(result, first->batch);
1312
0
      }
1313
0
      WriteBatchInternal::Append(result, w->batch);
1314
0
    }
1315
0
    *last_writer = w;
1316
0
  }
1317
831k
  return result;
1318
831k
}
1319
1320
// REQUIRES: mutex_ is held
1321
// REQUIRES: this thread is currently at the front of the writer queue
1322
831k
Status DBImpl::MakeRoomForWrite(bool force) {
1323
831k
  mutex_.AssertHeld();
1324
831k
  assert(!writers_.empty());
  Branch (1324:3): [True: 831k, False: 0]
1325
831k
  bool allow_delay = !force;
1326
831k
  Status s;
1327
831k
  while (true) {
  Branch (1327:10): [Folded - Ignored]
1328
831k
    if (!bg_error_.ok()) {
  Branch (1328:9): [True: 0, False: 831k]
1329
      // Yield previous error
1330
0
      s = bg_error_;
1331
0
      break;
1332
831k
    } else if (allow_delay && versions_->NumLevelFiles(0) >=
  Branch (1332:16): [True: 831k, False: 32]
  Branch (1332:31): [True: 0, False: 831k]
1333
831k
                                  config::kL0_SlowdownWritesTrigger) {
1334
      // We are getting close to hitting a hard limit on the number of
1335
      // L0 files.  Rather than delaying a single write by several
1336
      // seconds when we hit the hard limit, start delaying each
1337
      // individual write by 1ms to reduce latency variance.  Also,
1338
      // this delay hands over some CPU to the compaction thread in
1339
      // case it is sharing the same core as the writer.
1340
0
      mutex_.Unlock();
1341
0
      env_->SleepForMicroseconds(1000);
1342
0
      allow_delay = false;  // Do not delay a single write more than once
1343
0
      mutex_.Lock();
1344
831k
    } else if (!force &&
  Branch (1344:16): [True: 831k, False: 16]
1345
831k
               (mem_->ApproximateMemoryUsage() <= options_.write_buffer_size)) {
  Branch (1345:16): [True: 831k, False: 0]
1346
      // There is room in current memtable
1347
831k
      break;
1348
831k
    } else if (imm_ != nullptr) {
  Branch (1348:16): [True: 0, False: 16]
1349
      // We have filled up the current memtable, but the previous
1350
      // one is still being compacted, so we wait.
1351
0
      Log(options_.info_log, "Current memtable full; waiting...\n");
1352
0
      background_work_finished_signal_.Wait();
1353
16
    } else if (versions_->NumLevelFiles(0) >= config::kL0_StopWritesTrigger) {
  Branch (1353:16): [True: 0, False: 16]
1354
      // There are too many level-0 files.
1355
0
      Log(options_.info_log, "Too many L0 files; waiting...\n");
1356
0
      background_work_finished_signal_.Wait();
1357
16
    } else {
1358
      // Attempt to switch to a new memtable and trigger compaction of old
1359
16
      assert(versions_->PrevLogNumber() == 0);
  Branch (1359:7): [True: 16, False: 0]
1360
16
      uint64_t new_log_number = versions_->NewFileNumber();
1361
16
      WritableFile* lfile = nullptr;
1362
16
      s = env_->NewWritableFile(LogFileName(dbname_, new_log_number), &lfile);
1363
16
      if (!s.ok()) {
  Branch (1363:11): [True: 0, False: 16]
1364
        // Avoid chewing through file number space in a tight loop.
1365
0
        versions_->ReuseFileNumber(new_log_number);
1366
0
        break;
1367
0
      }
1368
16
      delete log_;
1369
16
      delete logfile_;
1370
16
      logfile_ = lfile;
1371
16
      logfile_number_ = new_log_number;
1372
16
      log_ = new log::Writer(lfile);
1373
16
      imm_ = mem_;
1374
16
      has_imm_.store(true, std::memory_order_release);
1375
16
      mem_ = new MemTable(internal_comparator_);
1376
16
      mem_->Ref();
1377
16
      force = false;  // Do not force another compaction if have room
1378
16
      MaybeScheduleCompaction();
1379
16
    }
1380
831k
  }
1381
831k
  return s;
1382
831k
}
1383
1384
0
bool DBImpl::GetProperty(const Slice& property, std::string* value) {
1385
0
  value->clear();
1386
1387
0
  MutexLock l(&mutex_);
1388
0
  Slice in = property;
1389
0
  Slice prefix("leveldb.");
1390
0
  if (!in.starts_with(prefix)) return false;
  Branch (1390:7): [True: 0, False: 0]
1391
0
  in.remove_prefix(prefix.size());
1392
1393
0
  if (in.starts_with("num-files-at-level")) {
  Branch (1393:7): [True: 0, False: 0]
1394
0
    in.remove_prefix(strlen("num-files-at-level"));
1395
0
    uint64_t level;
1396
0
    bool ok = ConsumeDecimalNumber(&in, &level) && in.empty();
  Branch (1396:15): [True: 0, False: 0]
  Branch (1396:52): [True: 0, False: 0]
1397
0
    if (!ok || level >= config::kNumLevels) {
  Branch (1397:9): [True: 0, False: 0]
  Branch (1397:16): [True: 0, False: 0]
1398
0
      return false;
1399
0
    } else {
1400
0
      char buf[100];
1401
0
      snprintf(buf, sizeof(buf), "%d",
1402
0
               versions_->NumLevelFiles(static_cast<int>(level)));
1403
0
      *value = buf;
1404
0
      return true;
1405
0
    }
1406
0
  } else if (in == "stats") {
  Branch (1406:14): [True: 0, False: 0]
1407
0
    char buf[200];
1408
0
    snprintf(buf, sizeof(buf),
1409
0
             "                               Compactions\n"
1410
0
             "Level  Files Size(MB) Time(sec) Read(MB) Write(MB)\n"
1411
0
             "--------------------------------------------------\n");
1412
0
    value->append(buf);
1413
0
    for (int level = 0; level < config::kNumLevels; level++) {
  Branch (1413:25): [True: 0, False: 0]
1414
0
      int files = versions_->NumLevelFiles(level);
1415
0
      if (stats_[level].micros > 0 || files > 0) {
  Branch (1415:11): [True: 0, False: 0]
  Branch (1415:39): [True: 0, False: 0]
1416
0
        snprintf(buf, sizeof(buf), "%3d %8d %8.0f %9.0f %8.0f %9.0f\n", level,
1417
0
                 files, versions_->NumLevelBytes(level) / 1048576.0,
1418
0
                 stats_[level].micros / 1e6,
1419
0
                 stats_[level].bytes_read / 1048576.0,
1420
0
                 stats_[level].bytes_written / 1048576.0);
1421
0
        value->append(buf);
1422
0
      }
1423
0
    }
1424
0
    return true;
1425
0
  } else if (in == "sstables") {
  Branch (1425:14): [True: 0, False: 0]
1426
0
    *value = versions_->current()->DebugString();
1427
0
    return true;
1428
0
  } else if (in == "approximate-memory-usage") {
  Branch (1428:14): [True: 0, False: 0]
1429
0
    size_t total_usage = options_.block_cache->TotalCharge();
1430
0
    if (mem_) {
  Branch (1430:9): [True: 0, False: 0]
1431
0
      total_usage += mem_->ApproximateMemoryUsage();
1432
0
    }
1433
0
    if (imm_) {
  Branch (1433:9): [True: 0, False: 0]
1434
0
      total_usage += imm_->ApproximateMemoryUsage();
1435
0
    }
1436
0
    char buf[50];
1437
0
    snprintf(buf, sizeof(buf), "%llu",
1438
0
             static_cast<unsigned long long>(total_usage));
1439
0
    value->append(buf);
1440
0
    return true;
1441
0
  }
1442
1443
0
  return false;
1444
0
}
1445
1446
0
void DBImpl::GetApproximateSizes(const Range* range, int n, uint64_t* sizes) {
1447
  // TODO(opt): better implementation
1448
0
  MutexLock l(&mutex_);
1449
0
  Version* v = versions_->current();
1450
0
  v->Ref();
1451
1452
0
  for (int i = 0; i < n; i++) {
  Branch (1452:19): [True: 0, False: 0]
1453
    // Convert user_key into a corresponding internal key.
1454
0
    InternalKey k1(range[i].start, kMaxSequenceNumber, kValueTypeForSeek);
1455
0
    InternalKey k2(range[i].limit, kMaxSequenceNumber, kValueTypeForSeek);
1456
0
    uint64_t start = versions_->ApproximateOffsetOf(v, k1);
1457
0
    uint64_t limit = versions_->ApproximateOffsetOf(v, k2);
1458
0
    sizes[i] = (limit >= start ? limit - start : 0);
  Branch (1458:17): [True: 0, False: 0]
1459
0
  }
1460
1461
0
  v->Unref();
1462
0
}
1463
1464
// Default implementations of convenience methods that subclasses of DB
1465
// can call if they wish
1466
0
Status DB::Put(const WriteOptions& opt, const Slice& key, const Slice& value) {
1467
0
  WriteBatch batch;
1468
0
  batch.Put(key, value);
1469
0
  return Write(opt, &batch);
1470
0
}
1471
1472
0
Status DB::Delete(const WriteOptions& opt, const Slice& key) {
1473
0
  WriteBatch batch;
1474
0
  batch.Delete(key);
1475
0
  return Write(opt, &batch);
1476
0
}
1477
1478
450k
DB::~DB() = default;
1479
1480
81
Status DB::Open(const Options& options, const std::string& dbname, DB** dbptr) {
1481
81
  *dbptr = nullptr;
1482
1483
81
  DBImpl* impl = new DBImpl(options, dbname);
1484
81
  impl->mutex_.Lock();
1485
81
  VersionEdit edit;
1486
  // Recover handles create_if_missing, error_if_exists
1487
81
  bool save_manifest = false;
1488
81
  Status s = impl->Recover(&edit, &save_manifest);
1489
81
  if (s.ok() && impl->mem_ == nullptr) {
  Branch (1489:7): [True: 81, False: 0]
  Branch (1489:17): [True: 81, False: 0]
1490
    // Create new log and a corresponding memtable.
1491
81
    uint64_t new_log_number = impl->versions_->NewFileNumber();
1492
81
    WritableFile* lfile;
1493
81
    s = options.env->NewWritableFile(LogFileName(dbname, new_log_number),
1494
81
                                     &lfile);
1495
81
    if (s.ok()) {
  Branch (1495:9): [True: 81, False: 0]
1496
81
      edit.SetLogNumber(new_log_number);
1497
81
      impl->logfile_ = lfile;
1498
81
      impl->logfile_number_ = new_log_number;
1499
81
      impl->log_ = new log::Writer(lfile);
1500
81
      impl->mem_ = new MemTable(impl->internal_comparator_);
1501
81
      impl->mem_->Ref();
1502
81
    }
1503
81
  }
1504
81
  if (s.ok() && save_manifest) {
  Branch (1504:7): [True: 81, False: 0]
  Branch (1504:17): [True: 81, False: 0]
1505
81
    edit.SetPrevLogNumber(0);  // No older logs needed after recovery.
1506
81
    edit.SetLogNumber(impl->logfile_number_);
1507
81
    s = impl->versions_->LogAndApply(&edit, &impl->mutex_);
1508
81
  }
1509
81
  if (s.ok()) {
  Branch (1509:7): [True: 81, False: 0]
1510
81
    impl->DeleteObsoleteFiles();
1511
81
    impl->MaybeScheduleCompaction();
1512
81
  }
1513
81
  impl->mutex_.Unlock();
1514
81
  if (s.ok()) {
  Branch (1514:7): [True: 81, False: 0]
1515
81
    assert(impl->mem_ != nullptr);
  Branch (1515:5): [True: 81, False: 0]
1516
81
    *dbptr = impl;
1517
81
  } else {
1518
0
    delete impl;
1519
0
  }
1520
81
  return s;
1521
81
}
1522
1523
450k
Snapshot::~Snapshot() = default;
1524
1525
0
Status DestroyDB(const std::string& dbname, const Options& options) {
1526
0
  Env* env = options.env;
1527
0
  std::vector<std::string> filenames;
1528
0
  Status result = env->GetChildren(dbname, &filenames);
1529
0
  if (!result.ok()) {
  Branch (1529:7): [True: 0, False: 0]
1530
    // Ignore error in case directory does not exist
1531
0
    return Status::OK();
1532
0
  }
1533
1534
0
  FileLock* lock;
1535
0
  const std::string lockname = LockFileName(dbname);
1536
0
  result = env->LockFile(lockname, &lock);
1537
0
  if (result.ok()) {
  Branch (1537:7): [True: 0, False: 0]
1538
0
    uint64_t number;
1539
0
    FileType type;
1540
0
    for (size_t i = 0; i < filenames.size(); i++) {
  Branch (1540:24): [True: 0, False: 0]
1541
0
      if (ParseFileName(filenames[i], &number, &type) &&
  Branch (1541:11): [True: 0, False: 0]
1542
0
          type != kDBLockFile) {  // Lock file will be deleted at end
  Branch (1542:11): [True: 0, False: 0]
1543
0
        Status del = env->DeleteFile(dbname + "/" + filenames[i]);
1544
0
        if (result.ok() && !del.ok()) {
  Branch (1544:13): [True: 0, False: 0]
  Branch (1544:28): [True: 0, False: 0]
1545
0
          result = del;
1546
0
        }
1547
0
      }
1548
0
    }
1549
0
    env->UnlockFile(lock);  // Ignore error since state is already gone
1550
0
    env->DeleteFile(lockname);
1551
0
    env->DeleteDir(dbname);  // Ignore error in case dir contains other files
1552
0
  }
1553
0
  return result;
1554
0
}
1555
1556
}  // namespace leveldb