@@ -181,7 +181,7 @@ DBImpl::DBImpl(const DBOptions& options, const std::string& dbname,
181
181
log_dir_synced_(false ),
182
182
log_empty_(true ),
183
183
persist_stats_cf_handle_(nullptr ),
184
- log_sync_cv_(&mutex_ ),
184
+ log_sync_cv_(&log_write_mutex_ ),
185
185
total_log_size_(0 ),
186
186
is_snapshot_supported_(true ),
187
187
write_buffer_manager_(immutable_db_options_.write_buffer_manager.get()),
@@ -271,6 +271,8 @@ DBImpl::DBImpl(const DBOptions& options, const std::string& dbname,
271
271
mutable_db_options_.Dump (immutable_db_options_.info_log .get ());
272
272
DumpSupportInfo (immutable_db_options_.info_log .get ());
273
273
274
+ max_total_wal_size_.store (mutable_db_options_.max_total_wal_size ,
275
+ std::memory_order_relaxed);
274
276
if (write_buffer_manager_) {
275
277
wbm_stall_.reset (new WBMStallInterface ());
276
278
}
@@ -641,26 +643,28 @@ Status DBImpl::CloseHelper() {
641
643
job_context.Clean ();
642
644
mutex_.Lock ();
643
645
}
644
-
645
- for (auto l : logs_to_free_) {
646
- delete l;
647
- }
648
- for (auto & log : logs_) {
649
- uint64_t log_number = log.writer ->get_log_number ();
650
- Status s = log.ClearWriter ();
651
- if (!s.ok ()) {
652
- ROCKS_LOG_WARN (
653
- immutable_db_options_.info_log ,
654
- " Unable to Sync WAL file %s with error -- %s" ,
655
- LogFileName (immutable_db_options_.GetWalDir (), log_number).c_str (),
656
- s.ToString ().c_str ());
657
- // Retain the first error
658
- if (ret.ok ()) {
659
- ret = s;
646
+ {
647
+ InstrumentedMutexLock lock (&log_write_mutex_);
648
+ for (auto l : logs_to_free_) {
649
+ delete l;
650
+ }
651
+ for (auto & log : logs_) {
652
+ uint64_t log_number = log.writer ->get_log_number ();
653
+ Status s = log.ClearWriter ();
654
+ if (!s.ok ()) {
655
+ ROCKS_LOG_WARN (
656
+ immutable_db_options_.info_log ,
657
+ " Unable to Sync WAL file %s with error -- %s" ,
658
+ LogFileName (immutable_db_options_.GetWalDir (), log_number).c_str (),
659
+ s.ToString ().c_str ());
660
+ // Retain the first error
661
+ if (ret.ok ()) {
662
+ ret = s;
663
+ }
660
664
}
661
665
}
666
+ logs_.clear ();
662
667
}
663
- logs_.clear ();
664
668
665
669
// Table cache may have table handles holding blocks from the block cache.
666
670
// We need to release them before the block cache is destroyed. The block
@@ -1046,6 +1050,7 @@ Status DBImpl::TablesRangeTombstoneSummary(ColumnFamilyHandle* column_family,
1046
1050
}
1047
1051
1048
1052
void DBImpl::ScheduleBgLogWriterClose (JobContext* job_context) {
1053
+ mutex_.AssertHeld ();
1049
1054
if (!job_context->logs_to_free .empty ()) {
1050
1055
for (auto l : job_context->logs_to_free ) {
1051
1056
AddToLogsToFreeQueue (l);
@@ -1225,6 +1230,11 @@ Status DBImpl::SetDBOptions(
1225
1230
new_options.stats_persist_period_sec );
1226
1231
mutex_.Lock ();
1227
1232
}
1233
+ if (new_options.max_total_wal_size !=
1234
+ mutable_db_options_.max_total_wal_size ) {
1235
+ max_total_wal_size_.store (new_options.max_total_wal_size ,
1236
+ std::memory_order_release);
1237
+ }
1228
1238
write_controller_.set_max_delayed_write_rate (
1229
1239
new_options.delayed_write_rate );
1230
1240
table_cache_.get ()->SetCapacity (new_options.max_open_files == -1
@@ -1345,7 +1355,7 @@ Status DBImpl::SyncWAL() {
1345
1355
uint64_t current_log_number;
1346
1356
1347
1357
{
1348
- InstrumentedMutexLock l (&mutex_ );
1358
+ InstrumentedMutexLock l (&log_write_mutex_ );
1349
1359
assert (!logs_.empty ());
1350
1360
1351
1361
// This SyncWAL() call only cares about logs up to this number.
@@ -1402,19 +1412,37 @@ Status DBImpl::SyncWAL() {
1402
1412
TEST_SYNC_POINT (" DBWALTest::SyncWALNotWaitWrite:2" );
1403
1413
1404
1414
TEST_SYNC_POINT (" DBImpl::SyncWAL:BeforeMarkLogsSynced:1" );
1415
+ VersionEdit synced_wals;
1405
1416
{
1406
- InstrumentedMutexLock l (&mutex_ );
1417
+ InstrumentedMutexLock l (&log_write_mutex_ );
1407
1418
if (status.ok ()) {
1408
- status = MarkLogsSynced (current_log_number, need_log_dir_sync);
1419
+ MarkLogsSynced (current_log_number, need_log_dir_sync, &synced_wals );
1409
1420
} else {
1410
1421
MarkLogsNotSynced (current_log_number);
1411
1422
}
1412
1423
}
1424
+ if (status.ok () && synced_wals.IsWalAddition ()) {
1425
+ InstrumentedMutexLock l (&mutex_);
1426
+ status = ApplyWALToManifest (&synced_wals);
1427
+ }
1428
+
1413
1429
TEST_SYNC_POINT (" DBImpl::SyncWAL:BeforeMarkLogsSynced:2" );
1414
1430
1415
1431
return status;
1416
1432
}
1417
1433
1434
+ Status DBImpl::ApplyWALToManifest (VersionEdit* synced_wals) {
1435
+ // not empty, write to MANIFEST.
1436
+ mutex_.AssertHeld ();
1437
+ Status status =
1438
+ versions_->LogAndApplyToDefaultColumnFamily (synced_wals, &mutex_);
1439
+ if (!status.ok () && versions_->io_status ().IsIOError ()) {
1440
+ status = error_handler_.SetBGError (versions_->io_status (),
1441
+ BackgroundErrorReason::kManifestWrite );
1442
+ }
1443
+ return status;
1444
+ }
1445
+
1418
1446
Status DBImpl::LockWAL () {
1419
1447
log_write_mutex_.Lock ();
1420
1448
auto cur_log_writer = logs_.back ().writer ;
@@ -1434,24 +1462,22 @@ Status DBImpl::UnlockWAL() {
1434
1462
return Status::OK ();
1435
1463
}
1436
1464
1437
- Status DBImpl::MarkLogsSynced (uint64_t up_to, bool synced_dir) {
1438
- mutex_.AssertHeld ();
1465
+ void DBImpl::MarkLogsSynced (uint64_t up_to, bool synced_dir,
1466
+ VersionEdit* synced_wals) {
1467
+ log_write_mutex_.AssertHeld ();
1439
1468
if (synced_dir && logfile_number_ == up_to) {
1440
1469
log_dir_synced_ = true ;
1441
1470
}
1442
- VersionEdit synced_wals;
1443
1471
for (auto it = logs_.begin (); it != logs_.end () && it->number <= up_to;) {
1444
1472
auto & wal = *it;
1445
1473
assert (wal.IsSyncing ());
1446
1474
1447
1475
if (logs_.size () > 1 ) {
1448
1476
if (immutable_db_options_.track_and_verify_wals_in_manifest &&
1449
1477
wal.GetPreSyncSize () > 0 ) {
1450
- synced_wals. AddWal (wal.number , WalMetadata (wal.GetPreSyncSize ()));
1478
+ synced_wals-> AddWal (wal.number , WalMetadata (wal.GetPreSyncSize ()));
1451
1479
}
1452
1480
logs_to_free_.push_back (wal.ReleaseWriter ());
1453
- // To modify logs_ both mutex_ and log_write_mutex_ must be held
1454
- InstrumentedMutexLock l (&log_write_mutex_);
1455
1481
it = logs_.erase (it);
1456
1482
} else {
1457
1483
wal.FinishSync ();
@@ -1460,22 +1486,11 @@ Status DBImpl::MarkLogsSynced(uint64_t up_to, bool synced_dir) {
1460
1486
}
1461
1487
assert (logs_.empty () || logs_[0 ].number > up_to ||
1462
1488
(logs_.size () == 1 && !logs_[0 ].IsSyncing ()));
1463
-
1464
- Status s;
1465
- if (synced_wals.IsWalAddition ()) {
1466
- // not empty, write to MANIFEST.
1467
- s = versions_->LogAndApplyToDefaultColumnFamily (&synced_wals, &mutex_);
1468
- if (!s.ok () && versions_->io_status ().IsIOError ()) {
1469
- s = error_handler_.SetBGError (versions_->io_status (),
1470
- BackgroundErrorReason::kManifestWrite );
1471
- }
1472
- }
1473
1489
log_sync_cv_.SignalAll ();
1474
- return s;
1475
1490
}
1476
1491
1477
1492
void DBImpl::MarkLogsNotSynced (uint64_t up_to) {
1478
- mutex_ .AssertHeld ();
1493
+ log_write_mutex_ .AssertHeld ();
1479
1494
for (auto it = logs_.begin (); it != logs_.end () && it->number <= up_to;
1480
1495
++it) {
1481
1496
auto & wal = *it;
0 commit comments