From 3b98e53c705cdaaa851b53d44ccb70769482ae2f Mon Sep 17 00:00:00 2001 From: neethuhaneesha Date: Tue, 1 Apr 2025 21:38:09 -0700 Subject: [PATCH] Limit the number of clearRanges per commit to a knob value only for rocksdb storage engine and added few metrics. (#12054) (#12062) --- fdbclient/ServerKnobs.cpp | 10 +++--- fdbclient/include/fdbclient/ServerKnobs.h | 1 + fdbserver/KeyValueStoreRocksDB.actor.cpp | 44 +++++++++++++++++++++-- fdbserver/storageserver.actor.cpp | 37 ++++++++++++++----- 4 files changed, 76 insertions(+), 16 deletions(-) diff --git a/fdbclient/ServerKnobs.cpp b/fdbclient/ServerKnobs.cpp index 105ba4b694..ef203eda31 100644 --- a/fdbclient/ServerKnobs.cpp +++ b/fdbclient/ServerKnobs.cpp @@ -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. diff --git a/fdbclient/include/fdbclient/ServerKnobs.h b/fdbclient/include/fdbclient/ServerKnobs.h index e02686f380..094d0aca73 100644 --- a/fdbclient/include/fdbclient/ServerKnobs.h +++ b/fdbclient/include/fdbclient/ServerKnobs.h @@ -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 diff --git a/fdbserver/KeyValueStoreRocksDB.actor.cpp b/fdbserver/KeyValueStoreRocksDB.actor.cpp index 0c154b3ae4..7b457839dc 100644 --- a/fdbserver/KeyValueStoreRocksDB.actor.cpp +++ b/fdbserver/KeyValueStoreRocksDB.actor.cpp @@ -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(id)), errorFuture(errorListener->getFuture()) { + errorListener(std::make_shared(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(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 fut = a->done.getFuture(); self->writeThread->post(a); wait(fut); @@ -2481,6 +2515,10 @@ struct RocksDBKeyValueStore : IKeyValueStore { std::set 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 deletesPerCommitHistogram; + Reference deleteRangesPerCommitHistogram; Optional> metrics; FlowLock readSemaphore; int numReadWaiters; diff --git a/fdbserver/storageserver.actor.cpp b/fdbserver/storageserver.actor.cpp index 92e807b158..a709b012f7 100644 --- a/fdbserver/storageserver.actor.cpp +++ b/fdbserver/storageserver.actor.cpp @@ -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 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 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 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 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 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& 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 {