wbtiger
/
foundationdb
mirror of https://github.com/apple/foundationdb.git
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Limit the number of clearRanges per commit to a knob value only for rocksdb storage engine and added few metrics. (#12054) (#12062)

This commit is contained in:
neethuhaneesha 2025-04-01 21:38:09 -07:00 committed by GitHub
parent 3f3c898bab
commit 3b98e53c70
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 76 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
fdbclient/ServerKnobs.cpp
View File

@ -492,9 +492,10 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi
int64_t memtableBytes = isSimulated ? 1024 * 1024 : 512 * 1024 * 1024;
init( ROCKSDB_MEMTABLE_BYTES, memtableBytes );
init( ROCKSDB_UNSAFE_AUTO_FSYNC, false );
init( ROCKSDB_PERIODIC_COMPACTION_SECONDS, 0 );
init( ROCKSDB_TTL_COMPACTION_SECONDS, 0 );
init( ROCKSDB_MAX_COMPACTION_BYTES, 0 );
init( ROCKSDB_PERIODIC_COMPACTION_SECONDS, 0 ); if( isSimulated ) ROCKSDB_PERIODIC_COMPACTION_SECONDS = deterministicRandom()->randomInt(5*60, 24*60*60);
init( ROCKSDB_TTL_COMPACTION_SECONDS, 0 ); if( isSimulated ) ROCKSDB_TTL_COMPACTION_SECONDS = deterministicRandom()->randomInt(5*60, 24*60*60);
int64_t maxCompactionBytes = 160LL * 64 * 1024 * 1024;
init( ROCKSDB_MAX_COMPACTION_BYTES, 0 ); /* default = 25*64MB */ if( randomize && BUGGIFY ) ROCKSDB_MAX_COMPACTION_BYTES = deterministicRandom()->randomInt64(5*64*1024*1024, maxCompactionBytes);
init( ROCKSDB_PREFIX_LEN, 11 ); if( randomize && BUGGIFY ) ROCKSDB_PREFIX_LEN = deterministicRandom()->randomInt(1, 20);
init( ROCKSDB_MEMTABLE_PREFIX_BLOOM_SIZE_RATIO, 0.1 );
init( ROCKSDB_BLOOM_BITS_PER_KEY, 10 );
@ -555,6 +556,7 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi
init( ROCKSDB_SINGLEKEY_DELETES_MAX, 200 ); // Max rocksdb::delete calls in a transaction
init( ROCKSDB_ENABLE_CLEAR_RANGE_EAGER_READS, false );
init( ROCKSDB_FORCE_DELETERANGE_FOR_CLEARRANGE, false );
init( ROCKSDB_CLEARRANGES_LIMIT_PER_COMMIT, 0 ); if( isSimulated ) ROCKSDB_CLEARRANGES_LIMIT_PER_COMMIT = deterministicRandom()->randomInt(1000, 10000); // Default: 0 (disabled).
// ROCKSDB_STATS_LEVEL=1 indicates rocksdb::StatsLevel::kExceptHistogramOrTimers
// Refer StatsLevel: https://github.com/facebook/rocksdb/blob/main/include/rocksdb/statistics.h#L594
init( ROCKSDB_STATS_LEVEL, 1 );
@ -590,7 +592,7 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi
init( ROCKSDB_VERIFY_CHECKSUM_BEFORE_RESTORE, true );
init( ROCKSDB_ENABLE_CHECKPOINT_VALIDATION, false ); if ( randomize && BUGGIFY ) ROCKSDB_ENABLE_CHECKPOINT_VALIDATION = deterministicRandom()->coinflip();
init( ROCKSDB_RETURN_OVERLOADED_ON_TIMEOUT, true );
init( ROCKSDB_COMPACTION_PRI, 3 ); // kMinOverlappingRatio, RocksDB default.
init( ROCKSDB_COMPACTION_PRI, 3 ); /* kMinOverlappingRatio, RocksDB default. */ if ( randomize && BUGGIFY ) ROCKSDB_COMPACTION_PRI = deterministicRandom()->randomInt(0, 4);
init( ROCKSDB_WAL_RECOVERY_MODE, 2 ); // kPointInTimeRecovery, RocksDB default.
init( ROCKSDB_TARGET_FILE_SIZE_BASE, 0 ); // If 0, pick RocksDB default.
init( ROCKSDB_TARGET_FILE_SIZE_MULTIPLIER, 1 ); // RocksDB default.

1
fdbclient/include/fdbclient/ServerKnobs.h
View File

@ -541,6 +541,7 @@ public:
int ROCKSDB_SINGLEKEY_DELETES_MAX;
bool ROCKSDB_ENABLE_CLEAR_RANGE_EAGER_READS;
bool ROCKSDB_FORCE_DELETERANGE_FOR_CLEARRANGE;
int ROCKSDB_CLEARRANGES_LIMIT_PER_COMMIT; // Max number of clearranges per commit with rocksdb kvstore.
bool ROCKSDB_ENABLE_COMPACT_ON_DELETION;
int64_t ROCKSDB_CDCF_SLIDING_WINDOW_SIZE; // CDCF: CompactOnDeletionCollectorFactory
int64_t ROCKSDB_CDCF_DELETION_TRIGGER; // CDCF: CompactOnDeletionCollectorFactory

44
fdbserver/KeyValueStoreRocksDB.actor.cpp
View File

@ -165,10 +165,14 @@ rocksdb::ColumnFamilyOptions SharedRocksDBState::initialCfOptions() {
options.disable_auto_compactions = SERVER_KNOBS->ROCKSDB_DISABLE_AUTO_COMPACTIONS;
}
if (SERVER_KNOBS->ROCKSDB_PERIODIC_COMPACTION_SECONDS > 0) {
options.periodic_compaction_seconds = SERVER_KNOBS->ROCKSDB_PERIODIC_COMPACTION_SECONDS;
// Adding two days range of jitter.
int64_t jitter = 2 * 24 * 60 * 60 * deterministicRandom()->random01();
options.periodic_compaction_seconds = SERVER_KNOBS->ROCKSDB_PERIODIC_COMPACTION_SECONDS + jitter;
}
if (SERVER_KNOBS->ROCKSDB_TTL_COMPACTION_SECONDS > 0) {
options.ttl = SERVER_KNOBS->ROCKSDB_TTL_COMPACTION_SECONDS;
// Adding two days range of jitter.
int64_t jitter = 2 * 24 * 60 * 60 * deterministicRandom()->random01();
options.ttl = SERVER_KNOBS->ROCKSDB_TTL_COMPACTION_SECONDS + jitter;
}
if (SERVER_KNOBS->ROCKSDB_MAX_COMPACTION_BYTES > 0) {
options.max_compaction_bytes = SERVER_KNOBS->ROCKSDB_MAX_COMPACTION_BYTES;
@ -434,6 +438,8 @@ const StringRef ROCKSDB_READVALUE_GET_HISTOGRAM = "RocksDBReadValueGet"_sr;
const StringRef ROCKSDB_READPREFIX_GET_HISTOGRAM = "RocksDBReadPrefixGet"_sr;
const StringRef ROCKSDB_READ_RANGE_BYTES_RETURNED_HISTOGRAM = "RocksDBReadRangeBytesReturned"_sr;
const StringRef ROCKSDB_READ_RANGE_KV_PAIRS_RETURNED_HISTOGRAM = "RocksDBReadRangeKVPairsReturned"_sr;
const StringRef ROCKSDB_DELETES_PER_COMMIT_HISTOGRAM = "RocksDBDeletesPerCommit"_sr;
const StringRef ROCKSDB_DELETE_RANGES_PER_COMMIT_HISTOGRAM = "RocksDBDeleteRangesPerCommit"_sr;
rocksdb::ExportImportFilesMetaData getMetaData(const CheckpointMetaData& checkpoint) {
rocksdb::ExportImportFilesMetaData metaData;
@ -1936,7 +1942,17 @@ struct RocksDBKeyValueStore : IKeyValueStore {
fetchSemaphore(SERVER_KNOBS->ROCKSDB_FETCH_QUEUE_SOFT_MAX),
numReadWaiters(SERVER_KNOBS->ROCKSDB_READ_QUEUE_HARD_MAX - SERVER_KNOBS->ROCKSDB_READ_QUEUE_SOFT_MAX),
numFetchWaiters(SERVER_KNOBS->ROCKSDB_FETCH_QUEUE_HARD_MAX - SERVER_KNOBS->ROCKSDB_FETCH_QUEUE_SOFT_MAX),
errorListener(std::make_shared<RocksDBErrorListener>(id)), errorFuture(errorListener->getFuture()) {
errorListener(std::make_shared<RocksDBErrorListener>(id)), errorFuture(errorListener->getFuture()),
deletesPerCommitHistogram(Histogram::getHistogram(ROCKSDBSTORAGE_HISTOGRAM_GROUP,
ROCKSDB_DELETES_PER_COMMIT_HISTOGRAM,
Histogram::Unit::countLinear,
0,
10000)),
deleteRangesPerCommitHistogram(Histogram::getHistogram(ROCKSDBSTORAGE_HISTOGRAM_GROUP,
ROCKSDB_DELETE_RANGES_PER_COMMIT_HISTOGRAM,
Histogram::Unit::countLinear,
0,
10000)) {
eventListener = std::make_shared<RocksDBEventListener>(sharedState);
// In simluation, run the reader/writer threads as Coro threads (i.e. in the network thread. The storage engine
// is still multi-threaded as background compaction threads are still present. Reads/writes to disk will also
@ -2146,6 +2162,8 @@ struct RocksDBKeyValueStore : IKeyValueStore {
0 /* default_cf_ts_sz default:0 */));
keysSet.clear();
maxDeletes = SERVER_KNOBS->ROCKSDB_SINGLEKEY_DELETES_MAX;
deletesPerCommit = 0;
deleteRangesPerCommit = 0;
}
ASSERT(defaultFdbCF != nullptr);
writeBatch->Put(defaultFdbCF, toSlice(kv.key), toSlice(kv.value));
@ -2163,6 +2181,8 @@ struct RocksDBKeyValueStore : IKeyValueStore {
0 /* default_cf_ts_sz default:0 */));
keysSet.clear();
maxDeletes = SERVER_KNOBS->ROCKSDB_SINGLEKEY_DELETES_MAX;
deletesPerCommit = 0;
deleteRangesPerCommit = 0;
}
ASSERT(defaultFdbCF != nullptr);
@ -2172,6 +2192,7 @@ struct RocksDBKeyValueStore : IKeyValueStore {
writeBatch->Delete(defaultFdbCF, toSlice(keyRange.begin));
++counters.deleteKeyReqs;
--maxDeletes;
++deletesPerCommit;
} else {
++counters.deleteRangeReqs;
if (SERVER_KNOBS->ROCKSDB_SINGLEKEY_DELETES_ON_CLEARRANGE &&
@ -2188,11 +2209,13 @@ struct RocksDBKeyValueStore : IKeyValueStore {
writeBatch->Delete(defaultFdbCF, cursor->key());
++counters.convertedDeleteKeyReqs;
--maxDeletes;
++deletesPerCommit;
cursor->Next();
}
if (!cursor->status().ok() || maxDeletes <= 0) {
// if readrange iteration fails, then do a deleteRange.
writeBatch->DeleteRange(defaultFdbCF, toSlice(keyRange.begin), toSlice(keyRange.end));
++deleteRangesPerCommit;
} else {
auto it = keysSet.lower_bound(keyRange.begin);
while (it != keysSet.end() && *it < keyRange.end) {
@ -2200,6 +2223,7 @@ struct RocksDBKeyValueStore : IKeyValueStore {
++counters.convertedDeleteKeyReqs;
--maxDeletes;
it++;
++deletesPerCommit;
}
it = previousCommitKeysSet.lower_bound(keyRange.begin);
while (it != previousCommitKeysSet.end() && *it < keyRange.end) {
@ -2207,10 +2231,12 @@ struct RocksDBKeyValueStore : IKeyValueStore {
++counters.convertedDeleteKeyReqs;
--maxDeletes;
it++;
++deletesPerCommit;
}
}
} else {
writeBatch->DeleteRange(defaultFdbCF, toSlice(keyRange.begin), toSlice(keyRange.end));
++deleteRangesPerCommit;
}
}
}
@ -2259,6 +2285,14 @@ struct RocksDBKeyValueStore : IKeyValueStore {
a->batchToCommit = std::move(self->writeBatch);
self->previousCommitKeysSet = std::move(self->keysSet);
self->maxDeletes = SERVER_KNOBS->ROCKSDB_SINGLEKEY_DELETES_MAX;
self->deletesPerCommitHistogram->sampleRecordCounter(self->deletesPerCommit);
self->deleteRangesPerCommitHistogram->sampleRecordCounter(self->deleteRangesPerCommit);
if (self->deletesPerCommit > 1000 || self->deleteRangesPerCommit > 1000)
TraceEvent("RocksDBDeletesCount", self->id)
.detail("DeletesPerCommit", self->deletesPerCommit)
.detail("DeleteRangesPerCommit", self->deleteRangesPerCommit);
self->deletesPerCommit = 0;
self->deleteRangesPerCommit = 0;
state Future<Void> fut = a->done.getFuture();
self->writeThread->post(a);
wait(fut);
@ -2481,6 +2515,10 @@ struct RocksDBKeyValueStore : IKeyValueStore {
std::set<Key> previousCommitKeysSet;
// maximum number of single key deletes in a commit, if ROCKSDB_SINGLEKEY_DELETES_ON_CLEARRANGE is enabled.
int maxDeletes;
int deletesPerCommit;
int deleteRangesPerCommit;
Reference<Histogram> deletesPerCommitHistogram;
Reference<Histogram> deleteRangesPerCommitHistogram;
Optional<Future<Void>> metrics;
FlowLock readSemaphore;
int numReadWaiters;

37
fdbserver/storageserver.actor.cpp
View File

@ -598,7 +598,8 @@ struct StorageServerDisk {
bool makeVersionMutationsDurable(Version& prevStorageVersion,
Version newStorageVersion,
int64_t& bytesLeft,
UnlimitedCommitBytes unlimitedCommitBytes);
UnlimitedCommitBytes unlimitedCommitBytes,
int64_t& clearRangesLeft);
void makeVersionDurable(Version version);
void makeAccumulativeChecksumDurable(const AccumulativeChecksumState& acsState);
void clearAccumulativeChecksumState(const AccumulativeChecksumState& acsState);
@ -1487,6 +1488,8 @@ public:
// bytesInput, instead of the actual bytes taken in the storages, so that (bytesInput - bytesDurable) can
// reflect the current memory footprint of MVCC.
Counter bytesDurable;
// Count of all fetchKey clearRange operations to the storage engine.
Counter kvClearRangesInFetchKeys;
// Bytes fetched by fetchChangeFeed for data movements.
Counter feedBytesFetched;
@ -1579,6 +1582,7 @@ public:
pTreeSets("PTreeSets", cc), pTreeClears("PTreeClears", cc), pTreeClearSplits("PTreeClearSplits", cc),
changeServerKeysAssigned("ChangeServerKeysAssigned", cc),
changeServerKeysUnassigned("ChangeServerKeysUnassigned", cc),
kvClearRangesInFetchKeys("KvClearRangesInFetchKeys", cc),
readLatencySample("ReadLatencyMetrics",
self->thisServerID,
SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL,
@ -9223,6 +9227,7 @@ ACTOR Future<Void> fetchKeys(StorageServer* data, AddingShard* shard) {
}
this_block = RangeResult();
++data->counters.kvClearRangesInFetchKeys;
data->fetchKeysTotalCommitBytes += expectedBlockSize;
data->fetchKeysBytesBudget -= expectedBlockSize;
data->fetchKeysBudgetUsed.set(data->fetchKeysBytesBudget <= 0);
@ -12677,10 +12682,11 @@ struct UpdateStorageCommitStats {
uint64_t mutationBytes;
uint64_t fetchKeyBytes;
int64_t seqId;
int64_t clearRangesLeft;
UpdateStorageCommitStats()
: seqId(0), whenCommit(0), beforeStorageUpdates(0), beforeStorageCommit(0), duration(0), commitDuration(0),
incompleteCommitDuration(0), mutationBytes(0), fetchKeyBytes(0) {}
incompleteCommitDuration(0), mutationBytes(0), fetchKeyBytes(0), clearRangesLeft(0) {}
void log(UID ssid, std::string reason) const {
TraceEvent(SevInfo, "UpdateStorageCommitStats", ssid)
@ -12693,7 +12699,8 @@ struct UpdateStorageCommitStats {
.detail("BeforeStorageCommit", beforeStorageCommit)
.detail("WhenCommit", whenCommit)
.detail("MutationBytes", mutationBytes)
.detail("FetchKeyBytes", fetchKeyBytes);
.detail("FetchKeyBytes", fetchKeyBytes)
.detail("ClearRangesLeft", clearRangesLeft);
}
};
@ -12738,6 +12745,11 @@ ACTOR Future<Void> updateStorage(StorageServer* data) {
state Version newOldestVersion = data->storageVersion();
state Version desiredVersion = data->desiredOldestVersion.get();
state int64_t bytesLeft = SERVER_KNOBS->STORAGE_COMMIT_BYTES;
state int64_t clearRangesLeft = data->storage.getKeyValueStoreType() == KeyValueStoreType::SSD_ROCKSDB_V1
? (SERVER_KNOBS->ROCKSDB_CLEARRANGES_LIMIT_PER_COMMIT > 0
? SERVER_KNOBS->ROCKSDB_CLEARRANGES_LIMIT_PER_COMMIT
: INT_MAX)
: INT_MAX;
// Clean up stale checkpoint requests, this is not supposed to happen, since checkpoints are cleaned up on
// failures. This is kept as a safeguard.
@ -12813,10 +12825,11 @@ ACTOR Future<Void> updateStorage(StorageServer* data) {
}
// Write mutations to storage until we reach the desiredVersion or have written too much (bytesleft)
// or until we reach clearRanges limit, in case of rocksdb.
state double beforeStorageUpdates = now();
loop {
state bool done = data->storage.makeVersionMutationsDurable(
newOldestVersion, desiredVersion, bytesLeft, unlimitedCommitBytes);
newOldestVersion, desiredVersion, bytesLeft, unlimitedCommitBytes, clearRangesLeft);
if (data->tenantMap.getLatestVersion() < newOldestVersion) {
data->tenantMap.createNewVersion(newOldestVersion);
}
@ -12834,10 +12847,11 @@ ACTOR Future<Void> updateStorage(StorageServer* data) {
}
recentCommitStats.back().mutationBytes = SERVER_KNOBS->STORAGE_COMMIT_BYTES - bytesLeft;
recentCommitStats.back().clearRangesLeft = clearRangesLeft;
recentCommitStats.back().beforeStorageUpdates = beforeStorageUpdates;
// Allow data fetch to use an additional bytesLeft but don't penalize fetch budget if bytesLeft is negative
if (SERVER_KNOBS->STORAGE_FETCH_KEYS_USE_COMMIT_BUDGET && bytesLeft > 0) {
if (SERVER_KNOBS->STORAGE_FETCH_KEYS_USE_COMMIT_BUDGET && bytesLeft > 0 && clearRangesLeft > 0) {
data->fetchKeysBytesBudget += bytesLeft;
data->fetchKeysBudgetUsed.set(data->fetchKeysBytesBudget <= 0);
@ -12991,7 +13005,8 @@ ACTOR Future<Void> updateStorage(StorageServer* data) {
when(wait(delay(60.0))) {
TraceEvent(SevWarn, "CommitTooLong", data->thisServerID)
.detail("FetchBytes", data->fetchKeysTotalCommitBytes)
.detail("CommitBytes", SERVER_KNOBS->STORAGE_COMMIT_BYTES - bytesLeft);
.detail("CommitBytes", SERVER_KNOBS->STORAGE_COMMIT_BYTES - bytesLeft)
.detail("ClearRangesLeft", clearRangesLeft);
if (data->storage.getKeyValueStoreType() == KeyValueStoreType::SSD_SHARDED_ROCKSDB &&
SERVER_KNOBS->LOGGING_ROCKSDB_BG_WORK_WHEN_IO_TIMEOUT) {
@ -13401,8 +13416,9 @@ void StorageServerDisk::writeMutations(const VectorRef<MutationRef>& mutations,
bool StorageServerDisk::makeVersionMutationsDurable(Version& prevStorageVersion,
Version newStorageVersion,
int64_t& bytesLeft,
UnlimitedCommitBytes unlimitedCommitBytes) {
if (!unlimitedCommitBytes && bytesLeft <= 0)
UnlimitedCommitBytes unlimitedCommitBytes,
int64_t& clearRangesLeft) {
if ((!unlimitedCommitBytes && bytesLeft <= 0) || clearRangesLeft <= 0)
return true;
// Apply mutations from the mutationLog
@ -13417,8 +13433,11 @@ bool StorageServerDisk::makeVersionMutationsDurable(Version& prevStorageVersion,
} else {
writeMutationsBuggy(v.mutations, v.version, "makeVersionDurable");
}
for (const auto& m : v.mutations)
for (const auto& m : v.mutations) {
bytesLeft -= mvccStorageBytes(m);
if (m.type == MutationRef::ClearRange)
--clearRangesLeft;
}
prevStorageVersion = v.version;
return false;
} else {