Inside Bitcoin

751    c->ReleaseInputs();
752    DeleteObsoleteFiles();
753  }
754  delete c;
755
756  if (status.ok()) {
757    // Done
758  } else if (shutting_down_.load(std::memory_order_acquire)) {
759    // Ignore compaction errors found during shutting down
760  } else {
761    Log(options_.info_log, "Compaction error: %s", status.ToString().c_str());
762  }
763
764  if (is_manual) {
765    ManualCompaction* m = manual_compaction_;
766    if (!status.ok()) {
767      m->done = true;
768    }
769    if (!m->done) {
770      // We only compacted part of the requested range.  Update *m
771      // to the range that is left to be compacted.
772      m->tmp_storage = manual_end;
773      m->begin = &m->tmp_storage;
774    }
775    manual_compaction_ = nullptr;
776  }
777}
778
779void DBImpl::CleanupCompaction(CompactionState* compact) {
780  mutex_.AssertHeld();
781  if (compact->builder != nullptr) {
782    // May happen if we get a shutdown call in the middle of compaction
783    compact->builder->Abandon();
784    delete compact->builder;
785  } else {
786    assert(compact->outfile == nullptr);
787  }
788  delete compact->outfile;
789  for (size_t i = 0; i < compact->outputs.size(); i++) {
790    const CompactionState::Output& out = compact->outputs[i];
791    pending_outputs_.erase(out.number);
792  }
793  delete compact;
794}
795
796Status DBImpl::OpenCompactionOutputFile(CompactionState* compact) {
797  assert(compact != nullptr);
798  assert(compact->builder == nullptr);
799  uint64_t file_number;
800  {
801    mutex_.Lock();
802    file_number = versions_->NewFileNumber();
803    pending_outputs_.insert(file_number);
804    CompactionState::Output out;
805    out.number = file_number;
806    out.file_size = 0;
807    out.smallest.Clear();
808    out.largest.Clear();
809    compact->outputs.push_back(out);
810    mutex_.Unlock();
811  }
812
813  // Make the output file
814  std::string fname = TableFileName(dbname_, file_number);
815  Status s = env_->NewWritableFile(fname, &compact->outfile);
816  if (s.ok()) {
817    compact->builder = new TableBuilder(options_, compact->outfile);
818  }
819  return s;
820}
821
822Status DBImpl::FinishCompactionOutputFile(CompactionState* compact,
823                                          Iterator* input) {
824  assert(compact != nullptr);
825  assert(compact->outfile != nullptr);
826  assert(compact->builder != nullptr);
827
828  const uint64_t output_number = compact->current_output()->number;
829  assert(output_number != 0);
830
831  // Check for iterator errors
832  Status s = input->status();
833  const uint64_t current_entries = compact->builder->NumEntries();
834  if (s.ok()) {
835    s = compact->builder->Finish();
836  } else {
837    compact->builder->Abandon();
838  }
839  const uint64_t current_bytes = compact->builder->FileSize();
840  compact->current_output()->file_size = current_bytes;
841  compact->total_bytes += current_bytes;
842  delete compact->builder;
843  compact->builder = nullptr;
844
845  // Finish and check for file errors
846  if (s.ok()) {
847    s = compact->outfile->Sync();
848  }
849  if (s.ok()) {
850    s = compact->outfile->Close();
851  }
852  delete compact->outfile;
853  compact->outfile = nullptr;
854
855  if (s.ok() && current_entries > 0) {
856    // Verify that the table is usable
857    Iterator* iter =
858        table_cache_->NewIterator(ReadOptions(), output_number, current_bytes);
859    s = iter->status();
860    delete iter;
861    if (s.ok()) {
862      Log(options_.info_log, "Generated table #%llu@%d: %lld keys, %lld bytes",
863          (unsigned long long)output_number, compact->compaction->level(),
864          (unsigned long long)current_entries,
865          (unsigned long long)current_bytes);
866    }
867  }
868  return s;
869}
870
871Status DBImpl::InstallCompactionResults(CompactionState* compact) {
872  mutex_.AssertHeld();
873  Log(options_.info_log, "Compacted %d@%d + %d@%d files => %lld bytes",
874      compact->compaction->num_input_files(0), compact->compaction->level(),
875      compact->compaction->num_input_files(1), compact->compaction->level() + 1,
876      static_cast<long long>(compact->total_bytes));
877
878  // Add compaction outputs
879  compact->compaction->AddInputDeletions(compact->compaction->edit());
880  const int level = compact->compaction->level();
881  for (size_t i = 0; i < compact->outputs.size(); i++) {
882    const CompactionState::Output& out = compact->outputs[i];
883    compact->compaction->edit()->AddFile(level + 1, out.number, out.file_size,
884                                         out.smallest, out.largest);
885  }
886  return versions_->LogAndApply(compact->compaction->edit(), &mutex_);
887}
888
889Status DBImpl::DoCompactionWork(CompactionState* compact) {
890  const uint64_t start_micros = env_->NowMicros();
891  int64_t imm_micros = 0;  // Micros spent doing imm_ compactions
892
893  Log(options_.info_log, "Compacting %d@%d + %d@%d files",
894      compact->compaction->num_input_files(0), compact->compaction->level(),
895      compact->compaction->num_input_files(1),
896      compact->compaction->level() + 1);
897
898  assert(versions_->NumLevelFiles(compact->compaction->level()) > 0);
899  assert(compact->builder == nullptr);
900  assert(compact->outfile == nullptr);
901  if (snapshots_.empty()) {
902    compact->smallest_snapshot = versions_->LastSequence();
903  } else {
904    compact->smallest_snapshot = snapshots_.oldest()->sequence_number();
905  }
906
907  Iterator* input = versions_->MakeInputIterator(compact->compaction);
908
909  // Release mutex while we're actually doing the compaction work
910  mutex_.Unlock();
911
912  input->SeekToFirst();
913  Status status;
914  ParsedInternalKey ikey;
915  std::string current_user_key;
916  bool has_current_user_key = false;
917  SequenceNumber last_sequence_for_key = kMaxSequenceNumber;
918  while (input->Valid() && !shutting_down_.load(std::memory_order_acquire)) {
919    // Prioritize immutable compaction work
920    if (has_imm_.load(std::memory_order_relaxed)) {
921      const uint64_t imm_start = env_->NowMicros();
922      mutex_.Lock();
923      if (imm_ != nullptr) {
924        CompactMemTable();
925        // Wake up MakeRoomForWrite() if necessary.
926        background_work_finished_signal_.SignalAll();
927      }
928      mutex_.Unlock();
929      imm_micros += (env_->NowMicros() - imm_start);
930    }
931
932    Slice key = input->key();
933    if (compact->compaction->ShouldStopBefore(key) &&
934        compact->builder != nullptr) {
935      status = FinishCompactionOutputFile(compact, input);
936      if (!status.ok()) {
937        break;
938      }
939    }
940
941    // Handle key/value, add to state, etc.
942    bool drop = false;
943    if (!ParseInternalKey(key, &ikey)) {
944      // Do not hide error keys
945      current_user_key.clear();
946      has_current_user_key = false;
947      last_sequence_for_key = kMaxSequenceNumber;
948    } else {
949      if (!has_current_user_key ||
950          user_comparator()->Compare(ikey.user_key, Slice(current_user_key)) !=
951              0) {
952        // First occurrence of this user key
953        current_user_key.assign(ikey.user_key.data(), ikey.user_key.size());
954        has_current_user_key = true;
955        last_sequence_for_key = kMaxSequenceNumber;
956      }
957
958      if (last_sequence_for_key <= compact->smallest_snapshot) {
959        // Hidden by an newer entry for same user key
960        drop = true;  // (A)
961      } else if (ikey.type == kTypeDeletion &&
962                 ikey.sequence <= compact->smallest_snapshot &&
963                 compact->compaction->IsBaseLevelForKey(ikey.user_key)) {
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        drop = true;
972      }
973
974      last_sequence_for_key = ikey.sequence;
975    }
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    if (!drop) {
987      // Open output file if necessary
988      if (compact->builder == nullptr) {
989        status = OpenCompactionOutputFile(compact);
990        if (!status.ok()) {
991          break;
992        }
993      }
994      if (compact->builder->NumEntries() == 0) {
995        compact->current_output()->smallest.DecodeFrom(key);
996      }
997      compact->current_output()->largest.DecodeFrom(key);
998      compact->builder->Add(key, input->value());
999
1000      // Close output file if it is big enough