Inside Bitcoin

1251  while (true) {
1252    Writer* ready = writers_.front();
1253    writers_.pop_front();
1254    if (ready != &w) {
1255      ready->status = status;
1256      ready->done = true;
1257      ready->cv.Signal();
1258    }
1259    if (ready == last_writer) break;
1260  }
1261
1262  // Notify new head of write queue
1263  if (!writers_.empty()) {
1264    writers_.front()->cv.Signal();
1265  }
1266
1267  return status;
1268}
1269
1270// REQUIRES: Writer list must be non-empty
1271// REQUIRES: First writer must have a non-null batch
1272WriteBatch* DBImpl::BuildBatchGroup(Writer** last_writer) {
1273  mutex_.AssertHeld();
1274  assert(!writers_.empty());
1275  Writer* first = writers_.front();
1276  WriteBatch* result = first->batch;
1277  assert(result != nullptr);
1278
1279  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  size_t max_size = 1 << 20;
1285  if (size <= (128 << 10)) {
1286    max_size = size + (128 << 10);
1287  }
1288
1289  *last_writer = first;
1290  std::deque<Writer*>::iterator iter = writers_.begin();
1291  ++iter;  // Advance past "first"
1292  for (; iter != writers_.end(); ++iter) {
1293    Writer* w = *iter;
1294    if (w->sync && !first->sync) {
1295      // Do not include a sync write into a batch handled by a non-sync write.
1296      break;
1297    }
1298
1299    if (w->batch != nullptr) {
1300      size += WriteBatchInternal::ByteSize(w->batch);
1301      if (size > max_size) {
1302        // Do not make batch too big
1303        break;
1304      }
1305
1306      // Append to *result
1307      if (result == first->batch) {
1308        // Switch to temporary batch instead of disturbing caller's batch
1309        result = tmp_batch_;
1310        assert(WriteBatchInternal::Count(result) == 0);
1311        WriteBatchInternal::Append(result, first->batch);
1312      }
1313      WriteBatchInternal::Append(result, w->batch);
1314    }
1315    *last_writer = w;
1316  }
1317  return result;
1318}
1319
1320// REQUIRES: mutex_ is held
1321// REQUIRES: this thread is currently at the front of the writer queue
1322Status DBImpl::MakeRoomForWrite(bool force) {
1323  mutex_.AssertHeld();
1324  assert(!writers_.empty());
1325  bool allow_delay = !force;
1326  Status s;
1327  while (true) {
1328    if (!bg_error_.ok()) {
1329      // Yield previous error
1330      s = bg_error_;
1331      break;
1332    } else if (allow_delay && versions_->NumLevelFiles(0) >=
1333                                  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      mutex_.Unlock();
1341      env_->SleepForMicroseconds(1000);
1342      allow_delay = false;  // Do not delay a single write more than once
1343      mutex_.Lock();
1344    } else if (!force &&
1345               (mem_->ApproximateMemoryUsage() <= options_.write_buffer_size)) {
1346      // There is room in current memtable
1347      break;
1348    } else if (imm_ != nullptr) {
1349      // We have filled up the current memtable, but the previous
1350      // one is still being compacted, so we wait.
1351      Log(options_.info_log, "Current memtable full; waiting...\n");
1352      background_work_finished_signal_.Wait();
1353    } else if (versions_->NumLevelFiles(0) >= config::kL0_StopWritesTrigger) {
1354      // There are too many level-0 files.
1355      Log(options_.info_log, "Too many L0 files; waiting...\n");
1356      background_work_finished_signal_.Wait();
1357    } else {
1358      // Attempt to switch to a new memtable and trigger compaction of old
1359      assert(versions_->PrevLogNumber() == 0);
1360      uint64_t new_log_number = versions_->NewFileNumber();
1361      WritableFile* lfile = nullptr;
1362      s = env_->NewWritableFile(LogFileName(dbname_, new_log_number), &lfile);
1363      if (!s.ok()) {
1364        // Avoid chewing through file number space in a tight loop.
1365        versions_->ReuseFileNumber(new_log_number);
1366        break;
1367      }
1368      delete log_;
1369      delete logfile_;
1370      logfile_ = lfile;
1371      logfile_number_ = new_log_number;
1372      log_ = new log::Writer(lfile);
1373      imm_ = mem_;
1374      has_imm_.store(true, std::memory_order_release);
1375      mem_ = new MemTable(internal_comparator_);
1376      mem_->Ref();
1377      force = false;  // Do not force another compaction if have room
1378      MaybeScheduleCompaction();
1379    }
1380  }
1381  return s;
1382}
1383
1384bool DBImpl::GetProperty(const Slice& property, std::string* value) {
1385  value->clear();
1386
1387  MutexLock l(&mutex_);
1388  Slice in = property;
1389  Slice prefix("leveldb.");
1390  if (!in.starts_with(prefix)) return false;
1391  in.remove_prefix(prefix.size());
1392
1393  if (in.starts_with("num-files-at-level")) {
1394    in.remove_prefix(strlen("num-files-at-level"));
1395    uint64_t level;
1396    bool ok = ConsumeDecimalNumber(&in, &level) && in.empty();
1397    if (!ok || level >= config::kNumLevels) {
1398      return false;
1399    } else {
1400      char buf[100];
1401      snprintf(buf, sizeof(buf), "%d",
1402               versions_->NumLevelFiles(static_cast<int>(level)));
1403      *value = buf;
1404      return true;
1405    }
1406  } else if (in == "stats") {
1407    char buf[200];
1408    snprintf(buf, sizeof(buf),
1409             "                               Compactions\n"
1410             "Level  Files Size(MB) Time(sec) Read(MB) Write(MB)\n"
1411             "--------------------------------------------------\n");
1412    value->append(buf);
1413    for (int level = 0; level < config::kNumLevels; level++) {
1414      int files = versions_->NumLevelFiles(level);
1415      if (stats_[level].micros > 0 || files > 0) {
1416        snprintf(buf, sizeof(buf), "%3d %8d %8.0f %9.0f %8.0f %9.0f\n", level,
1417                 files, versions_->NumLevelBytes(level) / 1048576.0,
1418                 stats_[level].micros / 1e6,
1419                 stats_[level].bytes_read / 1048576.0,
1420                 stats_[level].bytes_written / 1048576.0);
1421        value->append(buf);
1422      }
1423    }
1424    return true;
1425  } else if (in == "sstables") {
1426    *value = versions_->current()->DebugString();
1427    return true;
1428  } else if (in == "approximate-memory-usage") {
1429    size_t total_usage = options_.block_cache->TotalCharge();
1430    if (mem_) {
1431      total_usage += mem_->ApproximateMemoryUsage();
1432    }
1433    if (imm_) {
1434      total_usage += imm_->ApproximateMemoryUsage();
1435    }
1436    char buf[50];
1437    snprintf(buf, sizeof(buf), "%llu",
1438             static_cast<unsigned long long>(total_usage));
1439    value->append(buf);
1440    return true;
1441  }
1442
1443  return false;
1444}
1445
1446void DBImpl::GetApproximateSizes(const Range* range, int n, uint64_t* sizes) {
1447  // TODO(opt): better implementation
1448  MutexLock l(&mutex_);
1449  Version* v = versions_->current();
1450  v->Ref();
1451
1452  for (int i = 0; i < n; i++) {
1453    // Convert user_key into a corresponding internal key.
1454    InternalKey k1(range[i].start, kMaxSequenceNumber, kValueTypeForSeek);
1455    InternalKey k2(range[i].limit, kMaxSequenceNumber, kValueTypeForSeek);
1456    uint64_t start = versions_->ApproximateOffsetOf(v, k1);
1457    uint64_t limit = versions_->ApproximateOffsetOf(v, k2);
1458    sizes[i] = (limit >= start ? limit - start : 0);
1459  }
1460
1461  v->Unref();
1462}
1463
1464// Default implementations of convenience methods that subclasses of DB
1465// can call if they wish
1466Status DB::Put(const WriteOptions& opt, const Slice& key, const Slice& value) {
1467  WriteBatch batch;
1468  batch.Put(key, value);
1469  return Write(opt, &batch);
1470}
1471
1472Status DB::Delete(const WriteOptions& opt, const Slice& key) {
1473  WriteBatch batch;
1474  batch.Delete(key);
1475  return Write(opt, &batch);
1476}
1477
1478DB::~DB() = default;
1479
1480Status DB::Open(const Options& options, const std::string& dbname, DB** dbptr) {
1481  *dbptr = nullptr;
1482
1483  DBImpl* impl = new DBImpl(options, dbname);
1484  impl->mutex_.Lock();
1485  VersionEdit edit;
1486  // Recover handles create_if_missing, error_if_exists
1487  bool save_manifest = false;
1488  Status s = impl->Recover(&edit, &save_manifest);
1489  if (s.ok() && impl->mem_ == nullptr) {
1490    // Create new log and a corresponding memtable.
1491    uint64_t new_log_number = impl->versions_->NewFileNumber();
1492    WritableFile* lfile;
1493    s = options.env->NewWritableFile(LogFileName(dbname, new_log_number),
1494                                     &lfile);
1495    if (s.ok()) {
1496      edit.SetLogNumber(new_log_number);
1497      impl->logfile_ = lfile;
1498      impl->logfile_number_ = new_log_number;
1499      impl->log_ = new log::Writer(lfile);
1500      impl->mem_ = new MemTable(impl->internal_comparator_);