Inside Bitcoin

1001      if (compact->builder->FileSize() >=
1002          compact->compaction->MaxOutputFileSize()) {
1003        status = FinishCompactionOutputFile(compact, input);
1004        if (!status.ok()) {
1005          break;
1006        }
1007      }
1008    }
1009
1010    input->Next();
1011  }
1012
1013  if (status.ok() && shutting_down_.load(std::memory_order_acquire)) {
1014    status = Status::IOError("Deleting DB during compaction");
1015  }
1016  if (status.ok() && compact->builder != nullptr) {
1017    status = FinishCompactionOutputFile(compact, input);
1018  }
1019  if (status.ok()) {
1020    status = input->status();
1021  }
1022  delete input;
1023  input = nullptr;
1024
1025  CompactionStats stats;
1026  stats.micros = env_->NowMicros() - start_micros - imm_micros;
1027  for (int which = 0; which < 2; which++) {
1028    for (int i = 0; i < compact->compaction->num_input_files(which); i++) {
1029      stats.bytes_read += compact->compaction->input(which, i)->file_size;
1030    }
1031  }
1032  for (size_t i = 0; i < compact->outputs.size(); i++) {
1033    stats.bytes_written += compact->outputs[i].file_size;
1034  }
1035
1036  mutex_.Lock();
1037  stats_[compact->compaction->level() + 1].Add(stats);
1038
1039  if (status.ok()) {
1040    status = InstallCompactionResults(compact);
1041  }
1042  if (!status.ok()) {
1043    RecordBackgroundError(status);
1044  }
1045  VersionSet::LevelSummaryStorage tmp;
1046  Log(options_.info_log, "compacted to: %s", versions_->LevelSummary(&tmp));
1047  return status;
1048}
1049
1050namespace {
1051
1052struct 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      : mu(mutex), version(version), mem(mem), imm(imm) {}
1060};
1061
1062static void CleanupIteratorState(void* arg1, void* arg2) {
1063  IterState* state = reinterpret_cast<IterState*>(arg1);
1064  state->mu->Lock();
1065  state->mem->Unref();
1066  if (state->imm != nullptr) state->imm->Unref();
1067  state->version->Unref();
1068  state->mu->Unlock();
1069  delete state;
1070}
1071
1072}  // anonymous namespace
1073
1074Iterator* DBImpl::NewInternalIterator(const ReadOptions& options,
1075                                      SequenceNumber* latest_snapshot,
1076                                      uint32_t* seed) {
1077  mutex_.Lock();
1078  *latest_snapshot = versions_->LastSequence();
1079
1080  // Collect together all needed child iterators
1081  std::vector<Iterator*> list;
1082  list.push_back(mem_->NewIterator());
1083  mem_->Ref();
1084  if (imm_ != nullptr) {
1085    list.push_back(imm_->NewIterator());
1086    imm_->Ref();
1087  }
1088  versions_->current()->AddIterators(options, &list);
1089  Iterator* internal_iter =
1090      NewMergingIterator(&internal_comparator_, &list[0], list.size());
1091  versions_->current()->Ref();
1092
1093  IterState* cleanup = new IterState(&mutex_, mem_, imm_, versions_->current());
1094  internal_iter->RegisterCleanup(CleanupIteratorState, cleanup, nullptr);
1095
1096  *seed = ++seed_;
1097  mutex_.Unlock();
1098  return internal_iter;
1099}
1100
1101Iterator* DBImpl::TEST_NewInternalIterator() {
1102  SequenceNumber ignored;
1103  uint32_t ignored_seed;
1104  return NewInternalIterator(ReadOptions(), &ignored, &ignored_seed);
1105}
1106
1107int64_t DBImpl::TEST_MaxNextLevelOverlappingBytes() {
1108  MutexLock l(&mutex_);
1109  return versions_->MaxNextLevelOverlappingBytes();
1110}
1111
1112Status DBImpl::Get(const ReadOptions& options, const Slice& key,
1113                   std::string* value) {
1114  Status s;
1115  MutexLock l(&mutex_);
1116  SequenceNumber snapshot;
1117  if (options.snapshot != nullptr) {
1118    snapshot =
1119        static_cast<const SnapshotImpl*>(options.snapshot)->sequence_number();
1120  } else {
1121    snapshot = versions_->LastSequence();
1122  }
1123
1124  MemTable* mem = mem_;
1125  MemTable* imm = imm_;
1126  Version* current = versions_->current();
1127  mem->Ref();
1128  if (imm != nullptr) imm->Ref();
1129  current->Ref();
1130
1131  bool have_stat_update = false;
1132  Version::GetStats stats;
1133
1134  // Unlock while reading from files and memtables
1135  {
1136    mutex_.Unlock();
1137    // First look in the memtable, then in the immutable memtable (if any).
1138    LookupKey lkey(key, snapshot);
1139    if (mem->Get(lkey, value, &s)) {
1140      // Done
1141    } else if (imm != nullptr && imm->Get(lkey, value, &s)) {
1142      // Done
1143    } else {
1144      s = current->Get(options, lkey, value, &stats);
1145      have_stat_update = true;
1146    }
1147    mutex_.Lock();
1148  }
1149
1150  if (have_stat_update && current->UpdateStats(stats)) {
1151    MaybeScheduleCompaction();
1152  }
1153  mem->Unref();
1154  if (imm != nullptr) imm->Unref();
1155  current->Unref();
1156  return s;
1157}
1158
1159Iterator* DBImpl::NewIterator(const ReadOptions& options) {
1160  SequenceNumber latest_snapshot;
1161  uint32_t seed;
1162  Iterator* iter = NewInternalIterator(options, &latest_snapshot, &seed);
1163  return NewDBIterator(this, user_comparator(), iter,
1164                       (options.snapshot != nullptr
1165                            ? static_cast<const SnapshotImpl*>(options.snapshot)
1166                                  ->sequence_number()
1167                            : latest_snapshot),
1168                       seed);
1169}
1170
1171void DBImpl::RecordReadSample(Slice key) {
1172  MutexLock l(&mutex_);
1173  if (versions_->current()->RecordReadSample(key)) {
1174    MaybeScheduleCompaction();
1175  }
1176}
1177
1178const Snapshot* DBImpl::GetSnapshot() {
1179  MutexLock l(&mutex_);
1180  return snapshots_.New(versions_->LastSequence());
1181}
1182
1183void DBImpl::ReleaseSnapshot(const Snapshot* snapshot) {
1184  MutexLock l(&mutex_);
1185  snapshots_.Delete(static_cast<const SnapshotImpl*>(snapshot));
1186}
1187
1188// Convenience methods
1189Status DBImpl::Put(const WriteOptions& o, const Slice& key, const Slice& val) {
1190  return DB::Put(o, key, val);
1191}
1192
1193Status DBImpl::Delete(const WriteOptions& options, const Slice& key) {
1194  return DB::Delete(options, key);
1195}
1196
1197Status DBImpl::Write(const WriteOptions& options, WriteBatch* updates) {
1198  Writer w(&mutex_);
1199  w.batch = updates;
1200  w.sync = options.sync;
1201  w.done = false;
1202
1203  MutexLock l(&mutex_);
1204  writers_.push_back(&w);
1205  while (!w.done && &w != writers_.front()) {
1206    w.cv.Wait();
1207  }
1208  if (w.done) {
1209    return w.status;
1210  }
1211
1212  // May temporarily unlock and wait.
1213  Status status = MakeRoomForWrite(updates == nullptr);
1214  uint64_t last_sequence = versions_->LastSequence();
1215  Writer* last_writer = &w;
1216  if (status.ok() && updates != nullptr) {  // nullptr batch is for compactions
1217    WriteBatch* write_batch = BuildBatchGroup(&last_writer);
1218    WriteBatchInternal::SetSequence(write_batch, last_sequence + 1);
1219    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    {
1226      mutex_.Unlock();
1227      status = log_->AddRecord(WriteBatchInternal::Contents(write_batch));
1228      bool sync_error = false;
1229      if (status.ok() && options.sync) {
1230        status = logfile_->Sync();
1231        if (!status.ok()) {
1232          sync_error = true;
1233        }
1234      }
1235      if (status.ok()) {
1236        status = WriteBatchInternal::InsertInto(write_batch, mem_);
1237      }
1238      mutex_.Lock();
1239      if (sync_error) {
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        RecordBackgroundError(status);
1244      }
1245    }
1246    if (write_batch == tmp_batch_) tmp_batch_->Clear();
1247
1248    versions_->SetLastSequence(last_sequence);
1249  }
1250