Inside Bitcoin

251          // (in case there is a race that allows other incarnations)
252          keep = (number >= versions_->ManifestFileNumber());
253          break;
254        case kTableFile:
255          keep = (live.find(number) != live.end());
256          break;
257        case kTempFile:
258          // Any temp files that are currently being written to must
259          // be recorded in pending_outputs_, which is inserted into "live"
260          keep = (live.find(number) != live.end());
261          break;
262        case kCurrentFile:
263        case kDBLockFile:
264        case kInfoLogFile:
265          keep = true;
266          break;
267      }
268
269      if (!keep) {
270        files_to_delete.push_back(std::move(filename));
271        if (type == kTableFile) {
272          table_cache_->Evict(number);
273        }
274        Log(options_.info_log, "Delete type=%d #%lld\n", static_cast<int>(type),
275            static_cast<unsigned long long>(number));
276      }
277    }
278  }
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  mutex_.Unlock();
284  for (const std::string& filename : files_to_delete) {
285    env_->DeleteFile(dbname_ + "/" + filename);
286  }
287  mutex_.Lock();
288}
289
290Status DBImpl::Recover(VersionEdit* edit, bool* save_manifest) {
291  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  env_->CreateDir(dbname_);
297  assert(db_lock_ == nullptr);
298  Status s = env_->LockFile(LockFileName(dbname_), &db_lock_);
299  if (!s.ok()) {
300    return s;
301  }
302
303  if (!env_->FileExists(CurrentFileName(dbname_))) {
304    if (options_.create_if_missing) {
305      s = NewDB();
306      if (!s.ok()) {
307        return s;
308      }
309    } else {
310      return Status::InvalidArgument(
311          dbname_, "does not exist (create_if_missing is false)");
312    }
313  } else {
314    if (options_.error_if_exists) {
315      return Status::InvalidArgument(dbname_,
316                                     "exists (error_if_exists is true)");
317    }
318  }
319
320  s = versions_->Recover(save_manifest);
321  if (!s.ok()) {
322    return s;
323  }
324  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  const uint64_t min_log = versions_->LogNumber();
334  const uint64_t prev_log = versions_->PrevLogNumber();
335  std::vector<std::string> filenames;
336  s = env_->GetChildren(dbname_, &filenames);
337  if (!s.ok()) {
338    return s;
339  }
340  std::set<uint64_t> expected;
341  versions_->AddLiveFiles(&expected);
342  uint64_t number;
343  FileType type;
344  std::vector<uint64_t> logs;
345  for (size_t i = 0; i < filenames.size(); i++) {
346    if (ParseFileName(filenames[i], &number, &type)) {
347      expected.erase(number);
348      if (type == kLogFile && ((number >= min_log) || (number == prev_log)))
349        logs.push_back(number);
350    }
351  }
352  if (!expected.empty()) {
353    char buf[50];
354    snprintf(buf, sizeof(buf), "%d missing files; e.g.",
355             static_cast<int>(expected.size()));
356    return Status::Corruption(buf, TableFileName(dbname_, *(expected.begin())));
357  }
358
359  // Recover in the order in which the logs were generated
360  std::sort(logs.begin(), logs.end());
361  for (size_t i = 0; i < logs.size(); i++) {
362    s = RecoverLogFile(logs[i], (i == logs.size() - 1), save_manifest, edit,
363                       &max_sequence);
364    if (!s.ok()) {
365      return s;
366    }
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    versions_->MarkFileNumberUsed(logs[i]);
372  }
373
374  if (versions_->LastSequence() < max_sequence) {
375    versions_->SetLastSequence(max_sequence);
376  }
377
378  return Status::OK();
379}
380
381Status DBImpl::RecoverLogFile(uint64_t log_number, bool last_log,
382                              bool* save_manifest, VersionEdit* edit,
383                              SequenceNumber* max_sequence) {
384  struct LogReporter : public log::Reader::Reporter {
385    Env* env;
386    Logger* info_log;
387    const char* fname;
388    Status* status;  // null if options_.paranoid_checks==false
389    void Corruption(size_t bytes, const Status& s) override {
390      Log(info_log, "%s%s: dropping %d bytes; %s",
391          (this->status == nullptr ? "(ignoring error) " : ""), fname,
392          static_cast<int>(bytes), s.ToString().c_str());
393      if (this->status != nullptr && this->status->ok()) *this->status = s;
394    }
395  };
396
397  mutex_.AssertHeld();
398
399  // Open the log file
400  std::string fname = LogFileName(dbname_, log_number);
401  SequentialFile* file;
402  Status status = env_->NewSequentialFile(fname, &file);
403  if (!status.ok()) {
404    MaybeIgnoreError(&status);
405    return status;
406  }
407
408  // Create the log reader.
409  LogReporter reporter;
410  reporter.env = env_;
411  reporter.info_log = options_.info_log;
412  reporter.fname = fname.c_str();
413  reporter.status = (options_.paranoid_checks ? &status : nullptr);
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  log::Reader reader(file, &reporter, true /*checksum*/, 0 /*initial_offset*/);
419  Log(options_.info_log, "Recovering log #%llu",
420      (unsigned long long)log_number);
421
422  // Read all the records and add to a memtable
423  std::string scratch;
424  Slice record;
425  WriteBatch batch;
426  int compactions = 0;
427  MemTable* mem = nullptr;
428  while (reader.ReadRecord(&record, &scratch) && status.ok()) {
429    if (record.size() < 12) {
430      reporter.Corruption(record.size(),
431                          Status::Corruption("log record too small", fname));
432      continue;
433    }
434    WriteBatchInternal::SetContents(&batch, record);
435
436    if (mem == nullptr) {
437      mem = new MemTable(internal_comparator_);
438      mem->Ref();
439    }
440    status = WriteBatchInternal::InsertInto(&batch, mem);
441    MaybeIgnoreError(&status);
442    if (!status.ok()) {
443      break;
444    }
445    const SequenceNumber last_seq = WriteBatchInternal::Sequence(&batch) +
446                                    WriteBatchInternal::Count(&batch) - 1;
447    if (last_seq > *max_sequence) {
448      *max_sequence = last_seq;
449    }
450
451    if (mem->ApproximateMemoryUsage() > options_.write_buffer_size) {
452      compactions++;
453      *save_manifest = true;
454      status = WriteLevel0Table(mem, edit, nullptr);
455      mem->Unref();
456      mem = nullptr;
457      if (!status.ok()) {
458        // Reflect errors immediately so that conditions like full
459        // file-systems cause the DB::Open() to fail.
460        break;
461      }
462    }
463  }
464
465  delete file;
466
467  // See if we should keep reusing the last log file.
468  if (status.ok() && options_.reuse_logs && last_log && compactions == 0) {
469    assert(logfile_ == nullptr);
470    assert(log_ == nullptr);
471    assert(mem_ == nullptr);
472    uint64_t lfile_size;
473    if (env_->GetFileSize(fname, &lfile_size).ok() &&
474        env_->NewAppendableFile(fname, &logfile_).ok()) {
475      Log(options_.info_log, "Reusing old log %s \n", fname.c_str());
476      log_ = new log::Writer(logfile_, lfile_size);
477      logfile_number_ = log_number;
478      if (mem != nullptr) {
479        mem_ = mem;
480        mem = nullptr;
481      } else {
482        // mem can be nullptr if lognum exists but was empty.
483        mem_ = new MemTable(internal_comparator_);
484        mem_->Ref();
485      }
486    }
487  }
488
489  if (mem != nullptr) {
490    // mem did not get reused; compact it.
491    if (status.ok()) {
492      *save_manifest = true;
493      status = WriteLevel0Table(mem, edit, nullptr);
494    }
495    mem->Unref();
496  }
497
498  return status;
499}
500