diff --git a/CMakeLists.txt b/CMakeLists.txt index b31cc810b55..606cc9e04db 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -23,7 +23,7 @@ else() endif() project(foundationdb - VERSION 7.1.29 + VERSION 7.1.31 DESCRIPTION "FoundationDB is a scalable, fault-tolerant, ordered key-value store with full ACID transactions." HOMEPAGE_URL "http://www.foundationdb.org/" LANGUAGES C CXX ASM) diff --git a/cmake/CompileRocksDB.cmake b/cmake/CompileRocksDB.cmake index 62405d04df9..b1f5d8e8e35 100644 --- a/cmake/CompileRocksDB.cmake +++ b/cmake/CompileRocksDB.cmake @@ -1,6 +1,6 @@ # FindRocksDB -find_package(RocksDB 7.7.3) +find_package(RocksDB 7.10.2) include(ExternalProject) @@ -54,8 +54,8 @@ if(ROCKSDB_FOUND) ${BINARY_DIR}/librocksdb.a) else() ExternalProject_Add(rocksdb - URL https://github.com/facebook/rocksdb/archive/refs/tags/v7.7.3.tar.gz - URL_HASH SHA256=b8ac9784a342b2e314c821f6d701148912215666ac5e9bdbccd93cf3767cb611 + URL https://github.com/facebook/rocksdb/archive/refs/tags/v7.10.2.tar.gz + URL_HASH SHA256=4619ae7308cd3d11cdd36f0bfad3fb03a1ad399ca333f192b77b6b95b08e2f78 CMAKE_ARGS ${RocksDB_CMAKE_ARGS} BUILD_BYPRODUCTS /librocksdb.a INSTALL_COMMAND "" diff --git a/documentation/sphinx/source/release-notes/release-notes-710.rst b/documentation/sphinx/source/release-notes/release-notes-710.rst index b2922f80818..eb711318fb5 100644 --- a/documentation/sphinx/source/release-notes/release-notes-710.rst +++ b/documentation/sphinx/source/release-notes/release-notes-710.rst @@ -4,6 +4,23 @@ Release Notes ############# +7.1.31 +====== +* Same as 7.1.30 release with AVX enabled. + +7.1.30 +====== +* Released with AVX disabled. +* Fixed storage server finishedQueries metric when using getMappedRange. `(PR #9785) `_ +* Fixed unnecessary transaction system recovery when excluding the servers that are already excluded/failed. `(PR #9809) `_ and `(PR #9878) `_ +* Fixed the exclusion of stateless processes by skipping the free capacity check. `(PR #9789) `_ and `(PR #9769) `_ +* Fixed an issue where the new worker cannot get ServerDBInfo update. `(PR #9778) `_ +* Added RocksDB bloom filter knobs. `(PR #9770) `_ +* Upgraded RocksDB to version 7.10.2. `(PR #9829) `_ +* Fixed an issue where ExclusionSafetyCheckRequest could be blocked forever. `(PR #9871) `_ +* Fixed fdbserver not able to join the cluster if the majority of coordinators in its connection string have failed. `(PR #9883) `_ +* Fixed stuck data movement when moving to removed a storage server. `(PR #9904) `_ + 7.1.29 ====== * Same as 7.1.28 release with AVX enabled. diff --git a/fdbclient/ManagementAPI.actor.cpp b/fdbclient/ManagementAPI.actor.cpp index ff06b3e18ac..dd0ce41d7be 100644 --- a/fdbclient/ManagementAPI.actor.cpp +++ b/fdbclient/ManagementAPI.actor.cpp @@ -1124,7 +1124,7 @@ struct AutoQuorumChange final : IQuorumChange { desiredCount = redundancy * 2 - 1; } - std::vector excl = wait(getExcludedServers(tr)); + std::vector excl = wait(getAllExcludedServers(tr)); state std::set excluded(excl.begin(), excl.end()); std::vector _workers = wait(getWorkers(tr)); @@ -1273,23 +1273,37 @@ Reference autoQuorumChange(int desired) { return Reference(new AutoQuorumChange(desired)); } -void excludeServers(Transaction& tr, std::vector& servers, bool failed) { - tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS); - tr.setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); - tr.setOption(FDBTransactionOptions::LOCK_AWARE); - tr.setOption(FDBTransactionOptions::USE_PROVISIONAL_PROXIES); - std::string excludeVersionKey = deterministicRandom()->randomUniqueID().toString(); - auto serversVersionKey = failed ? failedServersVersionKey : excludedServersVersionKey; - tr.addReadConflictRange(singleKeyRange(serversVersionKey)); // To conflict with parallel includeServers - tr.set(serversVersionKey, excludeVersionKey); +ACTOR Future excludeServers(Transaction* tr, std::vector servers, bool failed) { + tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS); + tr->setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); + tr->setOption(FDBTransactionOptions::LOCK_AWARE); + tr->setOption(FDBTransactionOptions::USE_PROVISIONAL_PROXIES); + std::vector excl = wait(failed ? getExcludedFailedServerList(tr) : getExcludedServerList(tr)); + std::set exclusions(excl.begin(), excl.end()); + bool containNewExclusion = false; for (auto& s : servers) { + if (exclusions.find(s) != exclusions.end()) { + continue; + } + containNewExclusion = true; if (failed) { - tr.set(encodeFailedServersKey(s), StringRef()); + tr->set(encodeFailedServersKey(s), StringRef()); } else { - tr.set(encodeExcludedServersKey(s), StringRef()); + tr->set(encodeExcludedServersKey(s), StringRef()); } } - TraceEvent("ExcludeServersCommit").detail("Servers", describe(servers)).detail("ExcludeFailed", failed); + + if (containNewExclusion) { + std::string excludeVersionKey = deterministicRandom()->randomUniqueID().toString(); + auto serversVersionKey = failed ? failedServersVersionKey : excludedServersVersionKey; + tr->addReadConflictRange(singleKeyRange(serversVersionKey)); // To conflict with parallel includeServers + tr->set(serversVersionKey, excludeVersionKey); + } + TraceEvent("ExcludeServersCommit") + .detail("Servers", describe(servers)) + .detail("ExcludeFailed", failed) + .detail("ExclusionUpdated", containNewExclusion); + return Void(); } ACTOR Future excludeServers(Database cx, std::vector servers, bool failed) { @@ -1321,7 +1335,7 @@ ACTOR Future excludeServers(Database cx, std::vector ser state Transaction tr(cx); loop { try { - excludeServers(tr, servers, failed); + wait(excludeServers(&tr, servers, failed)); wait(tr.commit()); return Void(); } catch (Error& e) { @@ -1333,23 +1347,36 @@ ACTOR Future excludeServers(Database cx, std::vector ser } // excludes localities by setting the keys in api version below 7.0 -void excludeLocalities(Transaction& tr, std::unordered_set localities, bool failed) { - tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS); - tr.setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); - tr.setOption(FDBTransactionOptions::LOCK_AWARE); - tr.setOption(FDBTransactionOptions::USE_PROVISIONAL_PROXIES); - std::string excludeVersionKey = deterministicRandom()->randomUniqueID().toString(); - auto localityVersionKey = failed ? failedLocalityVersionKey : excludedLocalityVersionKey; - tr.addReadConflictRange(singleKeyRange(localityVersionKey)); // To conflict with parallel includeLocalities - tr.set(localityVersionKey, excludeVersionKey); +ACTOR Future excludeLocalities(Transaction* tr, std::unordered_set localities, bool failed) { + tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS); + tr->setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); + tr->setOption(FDBTransactionOptions::LOCK_AWARE); + tr->setOption(FDBTransactionOptions::USE_PROVISIONAL_PROXIES); + std::vector excl = wait(failed ? getExcludedFailedLocalityList(tr) : getExcludedLocalityList(tr)); + std::set exclusion(excl.begin(), excl.end()); + bool containNewExclusion = false; for (const auto& l : localities) { + if (exclusion.find(l) != exclusion.end()) { + continue; + } + containNewExclusion = true; if (failed) { - tr.set(encodeFailedLocalityKey(l), StringRef()); + tr->set(encodeFailedLocalityKey(l), StringRef()); } else { - tr.set(encodeExcludedLocalityKey(l), StringRef()); + tr->set(encodeExcludedLocalityKey(l), StringRef()); } } - TraceEvent("ExcludeLocalitiesCommit").detail("Localities", describe(localities)).detail("ExcludeFailed", failed); + if (containNewExclusion) { + std::string excludeVersionKey = deterministicRandom()->randomUniqueID().toString(); + auto localityVersionKey = failed ? failedLocalityVersionKey : excludedLocalityVersionKey; + tr->addReadConflictRange(singleKeyRange(localityVersionKey)); // To conflict with parallel includeLocalities + tr->set(localityVersionKey, excludeVersionKey); + } + TraceEvent("ExcludeLocalitiesCommit") + .detail("Localities", describe(localities)) + .detail("ExcludeFailed", failed) + .detail("ExclusionUpdated", containNewExclusion); + return Void(); } // Exclude the servers matching the given set of localities from use as state servers. @@ -1384,7 +1411,7 @@ ACTOR Future excludeLocalities(Database cx, std::unordered_set setClass(Database cx, AddressExclusion server, ProcessClass p } } -ACTOR Future> getExcludedServers(Transaction* tr) { +ACTOR Future> getExcludedServerList(Transaction* tr) { state RangeResult r = wait(tr->getRange(excludedServersKeys, CLIENT_KNOBS->TOO_MANY)); ASSERT(!r.more && r.size() < CLIENT_KNOBS->TOO_MANY); - state RangeResult r2 = wait(tr->getRange(failedServersKeys, CLIENT_KNOBS->TOO_MANY)); - ASSERT(!r2.more && r2.size() < CLIENT_KNOBS->TOO_MANY); std::vector exclusions; for (auto i = r.begin(); i != r.end(); ++i) { @@ -1629,7 +1654,16 @@ ACTOR Future> getExcludedServers(Transaction* tr) if (a.isValid()) exclusions.push_back(a); } - for (auto i = r2.begin(); i != r2.end(); ++i) { + uniquify(exclusions); + return exclusions; +} + +ACTOR Future> getExcludedFailedServerList(Transaction* tr) { + state RangeResult r = wait(tr->getRange(failedServersKeys, CLIENT_KNOBS->TOO_MANY)); + ASSERT(!r.more && r.size() < CLIENT_KNOBS->TOO_MANY); + + std::vector exclusions; + for (auto i = r.begin(); i != r.end(); ++i) { auto a = decodeFailedServersKey(i->key); if (a.isValid()) exclusions.push_back(a); @@ -1638,14 +1672,24 @@ ACTOR Future> getExcludedServers(Transaction* tr) return exclusions; } -ACTOR Future> getExcludedServers(Database cx) { +ACTOR Future> getAllExcludedServers(Transaction* tr) { + state std::vector exclusions; + std::vector excludedServers = wait(getExcludedServerList(tr)); + exclusions.insert(exclusions.end(), excludedServers.begin(), excludedServers.end()); + std::vector excludedFailed = wait(getExcludedFailedServerList(tr)); + exclusions.insert(exclusions.end(), excludedFailed.begin(), excludedFailed.end()); + uniquify(exclusions); + return exclusions; +} + +ACTOR Future> getAllExcludedServers(Database cx) { state Transaction tr(cx); loop { try { tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS); tr.setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); // necessary? tr.setOption(FDBTransactionOptions::LOCK_AWARE); - std::vector exclusions = wait(getExcludedServers(&tr)); + std::vector exclusions = wait(getAllExcludedServers(&tr)); return exclusions; } catch (Error& e) { wait(tr.onError(e)); @@ -1653,19 +1697,25 @@ ACTOR Future> getExcludedServers(Database cx) { } } -// Get the current list of excluded localities by reading the keys. -ACTOR Future> getExcludedLocalities(Transaction* tr) { +ACTOR Future> getExcludedLocalityList(Transaction* tr) { state RangeResult r = wait(tr->getRange(excludedLocalityKeys, CLIENT_KNOBS->TOO_MANY)); ASSERT(!r.more && r.size() < CLIENT_KNOBS->TOO_MANY); - state RangeResult r2 = wait(tr->getRange(failedLocalityKeys, CLIENT_KNOBS->TOO_MANY)); - ASSERT(!r2.more && r2.size() < CLIENT_KNOBS->TOO_MANY); std::vector excludedLocalities; for (const auto& i : r) { auto a = decodeExcludedLocalityKey(i.key); excludedLocalities.push_back(a); } - for (const auto& i : r2) { + uniquify(excludedLocalities); + return excludedLocalities; +} + +ACTOR Future> getExcludedFailedLocalityList(Transaction* tr) { + state RangeResult r = wait(tr->getRange(failedLocalityKeys, CLIENT_KNOBS->TOO_MANY)); + ASSERT(!r.more && r.size() < CLIENT_KNOBS->TOO_MANY); + + std::vector excludedLocalities; + for (const auto& i : r) { auto a = decodeFailedLocalityKey(i.key); excludedLocalities.push_back(a); } @@ -1673,15 +1723,25 @@ ACTOR Future> getExcludedLocalities(Transaction* tr) { return excludedLocalities; } +ACTOR Future> getAllExcludedLocalities(Transaction* tr) { + state std::vector exclusions; + std::vector excludedLocalities = wait(getExcludedLocalityList(tr)); + exclusions.insert(exclusions.end(), excludedLocalities.begin(), excludedLocalities.end()); + std::vector failedLocalities = wait(getExcludedFailedLocalityList(tr)); + exclusions.insert(exclusions.end(), failedLocalities.begin(), failedLocalities.end()); + uniquify(exclusions); + return exclusions; +} + // Get the list of excluded localities by reading the keys. -ACTOR Future> getExcludedLocalities(Database cx) { +ACTOR Future> getAllExcludedLocalities(Database cx) { state Transaction tr(cx); loop { try { tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS); tr.setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); tr.setOption(FDBTransactionOptions::LOCK_AWARE); - std::vector exclusions = wait(getExcludedLocalities(&tr)); + std::vector exclusions = wait(getAllExcludedLocalities(&tr)); return exclusions; } catch (Error& e) { wait(tr.onError(e)); diff --git a/fdbclient/ManagementAPI.actor.h b/fdbclient/ManagementAPI.actor.h index 5ddaf2ffbd3..f9d204bdb39 100644 --- a/fdbclient/ManagementAPI.actor.h +++ b/fdbclient/ManagementAPI.actor.h @@ -65,12 +65,12 @@ Reference nameQuorumChange(std::string const& name, Reference excludeServers(Database cx, std::vector servers, bool failed = false); -void excludeServers(Transaction& tr, std::vector& servers, bool failed = false); +ACTOR Future excludeServers(Transaction* tr, std::vector servers, bool failed = false); // Exclude the servers matching the given set of localities from use as state servers. Returns as soon as the change // is durable, without necessarily waiting for the servers to be evacuated. ACTOR Future excludeLocalities(Database cx, std::unordered_set localities, bool failed = false); -void excludeLocalities(Transaction& tr, std::unordered_set localities, bool failed = false); +ACTOR Future excludeLocalities(Transaction* tr, std::unordered_set localities, bool failed = false); // Remove the given servers from the exclusion list. A NetworkAddress with a port of 0 means all servers on the given // IP. A NetworkAddress() means all servers (don't exclude anything) @@ -86,13 +86,25 @@ ACTOR Future includeLocalities(Database cx, // the given IP. ACTOR Future setClass(Database cx, AddressExclusion server, ProcessClass processClass); -// Get the current list of excluded servers -ACTOR Future> getExcludedServers(Database cx); -ACTOR Future> getExcludedServers(Transaction* tr); +// Get the current list of excluded servers including both "exclude" and "failed". +ACTOR Future> getAllExcludedServers(Database cx); +ACTOR Future> getAllExcludedServers(Transaction* tr); + +// Get the current list of excluded servers. +ACTOR Future> getExcludedServerList(Transaction* tr); + +// Get the current list of failed servers. +ACTOR Future> getExcludedFailedServerList(Transaction* tr); // Get the current list of excluded localities -ACTOR Future> getExcludedLocalities(Database cx); -ACTOR Future> getExcludedLocalities(Transaction* tr); +ACTOR Future> getAllExcludedLocalities(Database cx); +ACTOR Future> getAllExcludedLocalities(Transaction* tr); + +// Get the current list of excluded localities. +ACTOR Future> getExcludedLocalityList(Transaction* tr); + +// Get the current list of failed localities. +ACTOR Future> getExcludedFailedLocalityList(Transaction* tr); std::set getAddressesByLocality(const std::vector& workers, const std::string& locality); diff --git a/fdbclient/NativeAPI.actor.cpp b/fdbclient/NativeAPI.actor.cpp index cd6865028bd..9bbd0e70d0d 100644 --- a/fdbclient/NativeAPI.actor.cpp +++ b/fdbclient/NativeAPI.actor.cpp @@ -8264,7 +8264,6 @@ ACTOR Future checkSafeExclusions(Database cx, std::vector checkSafeExclusions(Database cx, std::vectorgetCommitProxies(UseProvisionalProxies::False), &CommitProxyInterface::exclusionSafetyCheckReq, - req, + ExclusionSafetyCheckRequest(exclusions), cx->taskID))) { ddCheck = _ddCheck.safe; break; diff --git a/fdbclient/ServerKnobs.cpp b/fdbclient/ServerKnobs.cpp index c9358ae66f5..20d16833100 100644 --- a/fdbclient/ServerKnobs.cpp +++ b/fdbclient/ServerKnobs.cpp @@ -377,6 +377,10 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi init( ROCKSDB_UNSAFE_AUTO_FSYNC, false ); init( ROCKSDB_PERIODIC_COMPACTION_SECONDS, 0 ); init( ROCKSDB_PREFIX_LEN, 0 ); + init( ROCKSDB_MEMTABLE_PREFIX_BLOOM_SIZE_RATIO, 0.1 ); + init( ROCKSDB_BLOOM_BITS_PER_KEY, 10 ); + init( ROCKSDB_BLOOM_WHOLE_KEY_FILTERING, false ); + init( ROCKSDB_MAX_AUTO_READAHEAD_SIZE, 0 ); // If rocksdb block cache size is 0, the default 8MB is used. int64_t blockCacheSize = isSimulated ? 16 * 1024 * 1024 : 2147483648 /* 2GB */; init( ROCKSDB_BLOCK_CACHE_SIZE, blockCacheSize ); @@ -417,7 +421,9 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi // These knobs have contrary functionality. init( ROCKSDB_SINGLEKEY_DELETES_ON_CLEARRANGE, true ); init( ROCKSDB_SINGLEKEY_DELETES_BYTES_LIMIT, 10000 ); // 10KB + 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 ); // 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 ); if( randomize && BUGGIFY ) ROCKSDB_STATS_LEVEL = deterministicRandom()->randomInt(0, 6); diff --git a/fdbclient/ServerKnobs.h b/fdbclient/ServerKnobs.h index 33fdcaf4f37..7295ac7466e 100644 --- a/fdbclient/ServerKnobs.h +++ b/fdbclient/ServerKnobs.h @@ -306,6 +306,10 @@ class ServerKnobs : public KnobsImpl { bool ROCKSDB_UNSAFE_AUTO_FSYNC; int64_t ROCKSDB_PERIODIC_COMPACTION_SECONDS; int ROCKSDB_PREFIX_LEN; + double ROCKSDB_MEMTABLE_PREFIX_BLOOM_SIZE_RATIO; + int ROCKSDB_BLOOM_BITS_PER_KEY; + bool ROCKSDB_BLOOM_WHOLE_KEY_FILTERING; + int ROCKSDB_MAX_AUTO_READAHEAD_SIZE; int64_t ROCKSDB_BLOCK_CACHE_SIZE; double ROCKSDB_METRICS_DELAY; double ROCKSDB_READ_VALUE_TIMEOUT; @@ -338,7 +342,9 @@ class ServerKnobs : public KnobsImpl { bool ROCKSDB_DISABLE_WAL_EXPERIMENTAL; bool ROCKSDB_SINGLEKEY_DELETES_ON_CLEARRANGE; int64_t ROCKSDB_SINGLEKEY_DELETES_BYTES_LIMIT; + int ROCKSDB_SINGLEKEY_DELETES_MAX; bool ROCKSDB_ENABLE_CLEAR_RANGE_EAGER_READS; + bool ROCKSDB_FORCE_DELETERANGE_FOR_CLEARRANGE; 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/fdbclient/SpecialKeySpace.actor.cpp b/fdbclient/SpecialKeySpace.actor.cpp index 451febfbeeb..1f0970b668d 100644 --- a/fdbclient/SpecialKeySpace.actor.cpp +++ b/fdbclient/SpecialKeySpace.actor.cpp @@ -979,6 +979,9 @@ ACTOR Future checkExclusion(Database db, state int64_t totalKvStoreFreeBytes = 0; state int64_t totalKvStoreUsedBytes = 0; state int64_t totalKvStoreUsedBytesNonExcluded = 0; + // Keep track if we exclude any storage process with the provided adddresses + state bool excludedAddressesContainsStorageRole = false; + try { for (auto proc : processesMap.obj()) { StatusObjectReader process(proc.second); @@ -988,8 +991,8 @@ ACTOR Future checkExclusion(Database db, return false; } NetworkAddress addr = NetworkAddress::parse(addrStr); - bool excluded = - (process.has("excluded") && process.last().get_bool()) || addressExcluded(*exclusions, addr); + bool includedInExclusion = addressExcluded(*exclusions, addr); + bool excluded = (process.has("excluded") && process.last().get_bool()) || includedInExclusion; StatusObjectReader localityObj; std::string disk_id; @@ -1002,6 +1005,12 @@ ACTOR Future checkExclusion(Database db, if (role["role"].get_str() == "storage") { ssTotalCount++; + // Check if we are excluding a process that serves the storage role. We only have to check the free + // capacity if we are excluding at least one process that serves the storage role. + if (!excludedAddressesContainsStorageRole && includedInExclusion) { + excludedAddressesContainsStorageRole = true; + } + int64_t used_bytes; if (!role.get("kvstore_used_bytes", used_bytes)) { *msg = ManagementAPIError::toJsonString( @@ -1041,6 +1050,12 @@ ACTOR Future checkExclusion(Database db, return false; } + // If the exclusion command only contains processes that serve a non storage role we can skip the free capacity + // check in order to not block those exclusions. + if (!excludedAddressesContainsStorageRole) { + return true; + } + double finalFreeRatio = 1 - (totalKvStoreUsedBytes / (totalKvStoreUsedBytesNonExcluded + totalKvStoreFreeBytes)); if (ssExcludedCount == ssTotalCount || finalFreeRatio <= 0.1) { std::string temp = "ERROR: This exclude may cause the total free space in the cluster to drop below 10%.\n" @@ -1106,7 +1121,7 @@ ACTOR Future> excludeCommitActor(ReadYourWritesTransaction if (!safe) return result; } - excludeServers(ryw->getTransaction(), addresses, failed); + wait(excludeServers(&(ryw->getTransaction()), addresses, failed)); includeServers(ryw); return result; @@ -1151,7 +1166,7 @@ ACTOR Future ExclusionInProgressActor(ReadYourWritesTransaction* ry tr.setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); // necessary? tr.setOption(FDBTransactionOptions::LOCK_AWARE); - state std::vector excl = wait((getExcludedServers(&tr))); + state std::vector excl = wait((getAllExcludedServers(&tr))); state std::set exclusions(excl.begin(), excl.end()); state std::set inProgressExclusion; // Just getting a consistent read version proves that a set of tlogs satisfying the exclusions has completed @@ -2684,7 +2699,7 @@ ACTOR Future> excludeLocalityCommitActor(ReadYourWritesTra return result; } - excludeLocalities(ryw->getTransaction(), localities, failed); + wait(excludeLocalities(&ryw->getTransaction(), localities, failed)); includeLocalities(ryw); return result; diff --git a/fdbclient/StorageServerInterface.cpp b/fdbclient/StorageServerInterface.cpp index b7efbfbc836..71097c939e3 100644 --- a/fdbclient/StorageServerInterface.cpp +++ b/fdbclient/StorageServerInterface.cpp @@ -48,7 +48,7 @@ void TSS_traceMismatch(TraceEvent& event, const GetValueRequest& req, const GetValueReply& src, const GetValueReply& tss) { - event.detail("Key", req.key.printable()) + event.detail("Key", req.key) .detail("Tenant", req.tenantInfo.name) .detail("Version", req.version) .detail("SSReply", src.value.present() ? traceChecksumValue(src.value.get()) : "missing") @@ -167,10 +167,10 @@ static void traceKeyValuesDiff(TraceEvent& event, if (i >= ssKV.size() || i >= tssKV.size() || ssKV[i] != tssKV[i]) { event.detail("MismatchIndex", i); if (i >= ssKV.size() || i >= tssKV.size() || ssKV[i].key != tssKV[i].key) { - event.detail("MismatchSSKey", i < ssKV.size() ? ssKV[i].key.printable() : "missing"); - event.detail("MismatchTSSKey", i < tssKV.size() ? tssKV[i].key.printable() : "missing"); + event.detail("MismatchSSKey", i < ssKV.size() ? ssKV[i].key : "missing"_sr); + event.detail("MismatchTSSKey", i < tssKV.size() ? tssKV[i].key : "missing"_sr); } else { - event.detail("MismatchKey", ssKV[i].key.printable()); + event.detail("MismatchKey", ssKV[i].key); event.detail("MismatchSSValue", traceChecksumValue(ssKV[i].value)); event.detail("MismatchTSSValue", traceChecksumValue(tssKV[i].value)); } diff --git a/fdbserver/ClusterController.actor.cpp b/fdbserver/ClusterController.actor.cpp index 0136192da7e..cd6b78e7600 100644 --- a/fdbserver/ClusterController.actor.cpp +++ b/fdbserver/ClusterController.actor.cpp @@ -2389,7 +2389,7 @@ ACTOR Future dbInfoUpdater(ClusterControllerData* self) { when(wait(dbInfoChange)) {} } - UpdateServerDBInfoRequest req; + state UpdateServerDBInfoRequest req; if (dbInfoChange.isReady()) { for (auto& it : self->id_worker) { req.broadcastInfo.push_back(it.second.details.interf.updateServerDBInfo.getEndpoint()); @@ -2411,7 +2411,9 @@ ACTOR Future dbInfoUpdater(ClusterControllerData* self) { req.serializedDbInfo = BinaryWriter::toValue(self->db.serverInfo->get(), AssumeVersion(g_network->protocolVersion())); - TraceEvent("DBInfoStartBroadcast", self->id).log(); + TraceEvent("DBInfoStartBroadcast", self->id) + .detail("MasterLifetime", self->db.serverInfo->get().masterLifetime.toString()); + ; choose { when(std::vector notUpdated = wait(broadcastDBInfoRequest(req, SERVER_KNOBS->DBINFO_SEND_AMOUNT, Optional(), false))) { @@ -2422,6 +2424,11 @@ ACTOR Future dbInfoUpdater(ClusterControllerData* self) { } } when(wait(dbInfoChange)) {} + when(wait(updateDBInfo)) { + // The current round of broadcast hasn't finished yet. So we need to include all the current broadcast + // endpoints in the new round as well. + self->updateDBInfoEndpoints.insert(req.broadcastInfo.begin(), req.broadcastInfo.end()); + } } } } diff --git a/fdbserver/DataDistributionQueue.actor.cpp b/fdbserver/DataDistributionQueue.actor.cpp index 081efd8a307..016f40d1fcf 100644 --- a/fdbserver/DataDistributionQueue.actor.cpp +++ b/fdbserver/DataDistributionQueue.actor.cpp @@ -386,15 +386,19 @@ void launchDest(RelocateData& relocation, } } +void completeDest(const RelocateData& relocation, std::map& destBusymap) { + int destWorkFactor = getDestWorkFactor(); + for (UID id : relocation.completeDests) { + destBusymap[id].removeWork(relocation.priority, destWorkFactor); + } +} + void complete(RelocateData const& relocation, std::map& busymap, std::map& destBusymap) { ASSERT(relocation.workFactor > 0); for (int i = 0; i < relocation.src.size(); i++) busymap[relocation.src[i]].removeWork(relocation.priority, relocation.workFactor); - int destWorkFactor = getDestWorkFactor(); - for (UID id : relocation.completeDests) { - destBusymap[id].removeWork(relocation.priority, destWorkFactor); - } + completeDest(relocation, destBusymap); } ACTOR Future dataDistributionRelocator(struct DDQueueData* self, @@ -1388,6 +1392,12 @@ ACTOR Future dataDistributionRelocator(DDQueueData* self, RelocateData rd, } else { TEST(true); // move to removed server healthyDestinations.addDataInFlightToTeam(-metrics.bytes); + if (!signalledTransferComplete) { + // Update destination busyness map to allow further movement to them. + // signalling transferComplete calls completeDest() in complete(), so + // doing so here would double-complete the work. + completeDest(rd, self->destBusymap); + } rd.completeDests.clear(); wait(delay(SERVER_KNOBS->RETRY_RELOCATESHARD_DELAY, TaskPriority::DataDistributionLaunch)); } diff --git a/fdbserver/IKeyValueStore.h b/fdbserver/IKeyValueStore.h index 9c0f3afaa91..74b93b9620c 100644 --- a/fdbserver/IKeyValueStore.h +++ b/fdbserver/IKeyValueStore.h @@ -62,9 +62,7 @@ class IKeyValueStore : public IClosable { public: virtual KeyValueStoreType getType() const = 0; virtual void set(KeyValueRef keyValue, const Arena* arena = nullptr) = 0; - virtual void clear(KeyRangeRef range, - const StorageServerMetrics* storageMetrics = nullptr, - const Arena* arena = nullptr) = 0; + virtual void clear(KeyRangeRef range, const Arena* arena = nullptr) = 0; virtual Future canCommit() { return Void(); } virtual Future commit( bool sequential = false) = 0; // returns when prior sets and clears are (atomically) durable diff --git a/fdbserver/KeyValueStoreCompressTestData.actor.cpp b/fdbserver/KeyValueStoreCompressTestData.actor.cpp index 38ddd785e65..3121232b581 100644 --- a/fdbserver/KeyValueStoreCompressTestData.actor.cpp +++ b/fdbserver/KeyValueStoreCompressTestData.actor.cpp @@ -53,11 +53,7 @@ struct KeyValueStoreCompressTestData final : IKeyValueStore { void set(KeyValueRef keyValue, const Arena* arena = nullptr) override { store->set(KeyValueRef(keyValue.key, pack(keyValue.value)), arena); } - void clear(KeyRangeRef range, - const StorageServerMetrics* storageMetrics = nullptr, - const Arena* arena = nullptr) override { - store->clear(range, storageMetrics, arena); - } + void clear(KeyRangeRef range, const Arena* arena = nullptr) override { store->clear(range, arena); } Future commit(bool sequential = false) override { return store->commit(sequential); } Future> readValue(KeyRef key, IKeyValueStore::ReadType, Optional debugID) override { diff --git a/fdbserver/KeyValueStoreMemory.actor.cpp b/fdbserver/KeyValueStoreMemory.actor.cpp index 616aaed8585..4ded4f67492 100644 --- a/fdbserver/KeyValueStoreMemory.actor.cpp +++ b/fdbserver/KeyValueStoreMemory.actor.cpp @@ -124,7 +124,7 @@ class KeyValueStoreMemory final : public IKeyValueStore, NonCopyable { } } - void clear(KeyRangeRef range, const StorageServerMetrics* storageMetrics, const Arena* arena) override { + void clear(KeyRangeRef range, const Arena* arena) override { // A commit that occurs with no available space returns Never, so we can throw out all modifications if (getAvailableSize() <= 0) return; diff --git a/fdbserver/KeyValueStoreRocksDB.actor.cpp b/fdbserver/KeyValueStoreRocksDB.actor.cpp index 83f7697bf98..1a320ef68ab 100644 --- a/fdbserver/KeyValueStoreRocksDB.actor.cpp +++ b/fdbserver/KeyValueStoreRocksDB.actor.cpp @@ -68,9 +68,9 @@ #ifdef SSD_ROCKSDB_EXPERIMENTAL -// Enforcing rocksdb version to be 7.7.3. -static_assert((ROCKSDB_MAJOR == 7 && ROCKSDB_MINOR == 7 && ROCKSDB_PATCH == 3), - "Unsupported rocksdb version. Update the rocksdb to 7.7.3 version"); +// Enforcing rocksdb version to be 7.10.2 +static_assert((ROCKSDB_MAJOR == 7 && ROCKSDB_MINOR == 10 && ROCKSDB_PATCH == 2), + "Unsupported rocksdb version. Update the rocksdb to 7.10.2 version"); namespace { using rocksdb::BackgroundErrorReason; @@ -226,10 +226,14 @@ rocksdb::ExportImportFilesMetaData getMetaData(const CheckpointMetaData& checkpo for (const LiveFileMetaData& fileMetaData : rocksCF.sstFiles) { rocksdb::LiveFileMetaData liveFileMetaData; - liveFileMetaData.size = fileMetaData.size; - liveFileMetaData.name = fileMetaData.name; + liveFileMetaData.relative_filename = fileMetaData.relative_filename; + liveFileMetaData.directory = fileMetaData.directory; liveFileMetaData.file_number = fileMetaData.file_number; - liveFileMetaData.db_path = fileMetaData.db_path; + liveFileMetaData.file_type = static_cast(fileMetaData.file_type); + liveFileMetaData.size = fileMetaData.size; + liveFileMetaData.temperature = static_cast(fileMetaData.temperature); + liveFileMetaData.file_checksum = fileMetaData.file_checksum; + liveFileMetaData.file_checksum_func_name = fileMetaData.file_checksum_func_name; liveFileMetaData.smallest_seqno = fileMetaData.smallest_seqno; liveFileMetaData.largest_seqno = fileMetaData.largest_seqno; liveFileMetaData.smallestkey = fileMetaData.smallestkey; @@ -238,12 +242,12 @@ rocksdb::ExportImportFilesMetaData getMetaData(const CheckpointMetaData& checkpo liveFileMetaData.being_compacted = fileMetaData.being_compacted; liveFileMetaData.num_entries = fileMetaData.num_entries; liveFileMetaData.num_deletions = fileMetaData.num_deletions; - liveFileMetaData.temperature = static_cast(fileMetaData.temperature); liveFileMetaData.oldest_blob_file_number = fileMetaData.oldest_blob_file_number; liveFileMetaData.oldest_ancester_time = fileMetaData.oldest_ancester_time; liveFileMetaData.file_creation_time = fileMetaData.file_creation_time; - liveFileMetaData.file_checksum = fileMetaData.file_checksum; - liveFileMetaData.file_checksum_func_name = fileMetaData.file_checksum_func_name; + liveFileMetaData.epoch_number = fileMetaData.epoch_number; + liveFileMetaData.name = fileMetaData.name; + liveFileMetaData.db_path = fileMetaData.db_path; liveFileMetaData.column_family_name = fileMetaData.column_family_name; liveFileMetaData.level = fileMetaData.level; metaData.files.push_back(liveFileMetaData); @@ -257,10 +261,14 @@ void populateMetaData(CheckpointMetaData* checkpoint, const rocksdb::ExportImpor rocksCF.dbComparatorName = metaData.db_comparator_name; for (const rocksdb::LiveFileMetaData& fileMetaData : metaData.files) { LiveFileMetaData liveFileMetaData; - liveFileMetaData.size = fileMetaData.size; - liveFileMetaData.name = fileMetaData.name; + liveFileMetaData.relative_filename = fileMetaData.relative_filename; + liveFileMetaData.directory = fileMetaData.directory; liveFileMetaData.file_number = fileMetaData.file_number; - liveFileMetaData.db_path = fileMetaData.db_path; + liveFileMetaData.file_type = static_cast(fileMetaData.file_type); + liveFileMetaData.size = fileMetaData.size; + liveFileMetaData.temperature = static_cast(fileMetaData.temperature); + liveFileMetaData.file_checksum = fileMetaData.file_checksum; + liveFileMetaData.file_checksum_func_name = fileMetaData.file_checksum_func_name; liveFileMetaData.smallest_seqno = fileMetaData.smallest_seqno; liveFileMetaData.largest_seqno = fileMetaData.largest_seqno; liveFileMetaData.smallestkey = fileMetaData.smallestkey; @@ -269,12 +277,12 @@ void populateMetaData(CheckpointMetaData* checkpoint, const rocksdb::ExportImpor liveFileMetaData.being_compacted = fileMetaData.being_compacted; liveFileMetaData.num_entries = fileMetaData.num_entries; liveFileMetaData.num_deletions = fileMetaData.num_deletions; - liveFileMetaData.temperature = fileMetaData.temperature; liveFileMetaData.oldest_blob_file_number = fileMetaData.oldest_blob_file_number; liveFileMetaData.oldest_ancester_time = fileMetaData.oldest_ancester_time; liveFileMetaData.file_creation_time = fileMetaData.file_creation_time; - liveFileMetaData.file_checksum = fileMetaData.file_checksum; - liveFileMetaData.file_checksum_func_name = fileMetaData.file_checksum_func_name; + liveFileMetaData.epoch_number = fileMetaData.epoch_number; + liveFileMetaData.name = fileMetaData.name; + liveFileMetaData.db_path = fileMetaData.db_path; liveFileMetaData.column_family_name = fileMetaData.column_family_name; liveFileMetaData.level = fileMetaData.level; rocksCF.sstFiles.push_back(liveFileMetaData); @@ -340,8 +348,7 @@ rocksdb::ColumnFamilyOptions getCFOptions() { options.prefix_extractor.reset(rocksdb::NewFixedPrefixTransform(SERVER_KNOBS->ROCKSDB_PREFIX_LEN)); // Also turn on bloom filters in the memtable. - // TODO: Make a knob for this as well. - options.memtable_prefix_bloom_size_ratio = 0.1; + options.memtable_prefix_bloom_size_ratio = SERVER_KNOBS->ROCKSDB_MEMTABLE_PREFIX_BLOOM_SIZE_RATIO; // 5 -- Can be read by RocksDB's versions since 6.6.0. Full and partitioned // filters use a generally faster and more accurate Bloom filter @@ -352,11 +359,11 @@ rocksdb::ColumnFamilyOptions getCFOptions() { // Create and apply a bloom filter using the 10 bits // which should yield a ~1% false positive rate: // https://github.com/facebook/rocksdb/wiki/RocksDB-Bloom-Filter#full-filters-new-format - bbOpts.filter_policy.reset(rocksdb::NewBloomFilterPolicy(10)); + bbOpts.filter_policy.reset(rocksdb::NewBloomFilterPolicy(SERVER_KNOBS->ROCKSDB_BLOOM_BITS_PER_KEY)); // The whole key blooms are only used for point lookups. // https://github.com/facebook/rocksdb/wiki/RocksDB-Bloom-Filter#prefix-vs-whole-key - bbOpts.whole_key_filtering = false; + bbOpts.whole_key_filtering = SERVER_KNOBS->ROCKSDB_BLOOM_WHOLE_KEY_FILTERING; } if (SERVER_KNOBS->ROCKSDB_BLOCK_CACHE_SIZE > 0) { @@ -369,6 +376,12 @@ rocksdb::ColumnFamilyOptions getCFOptions() { bbOpts.block_size = SERVER_KNOBS->ROCKSDB_BLOCK_SIZE; } + // The readahead size starts with 8KB and is exponentially increased on each additional sequential IO, + // up to a max of BlockBasedTableOptions.max_auto_readahead_size (default 256 KB) + if (SERVER_KNOBS->ROCKSDB_MAX_AUTO_READAHEAD_SIZE > 0) { + bbOpts.max_auto_readahead_size = SERVER_KNOBS->ROCKSDB_MAX_AUTO_READAHEAD_SIZE; + } + options.table_factory.reset(rocksdb::NewBlockBasedTableFactory(bbOpts)); return options; @@ -411,13 +424,14 @@ struct Counters { Counter convertedDeleteKeyReqs; Counter convertedDeleteRangeReqs; Counter rocksdbReadRangeQueries; + Counter commitDelayed; Counters() : cc("RocksDBThrottle"), immediateThrottle("ImmediateThrottle", cc), failedToAcquire("FailedToAcquire", cc), deleteKeyReqs("DeleteKeyRequests", cc), deleteRangeReqs("DeleteRangeRequests", cc), convertedDeleteKeyReqs("ConvertedDeleteKeyRequests", cc), convertedDeleteRangeReqs("ConvertedDeleteRangeRequests", cc), - rocksdbReadRangeQueries("RocksdbReadRangeQueries", cc) {} + rocksdbReadRangeQueries("RocksdbReadRangeQueries", cc), commitDelayed("CommitDelayed", cc) {} }; struct ReadIterator { @@ -935,6 +949,8 @@ ACTOR Future rocksDBMetricLogger(UID id, { "RowCacheHit", rocksdb::ROW_CACHE_HIT, 0 }, { "RowCacheMiss", rocksdb::ROW_CACHE_MISS, 0 }, { "CountIterSkippedKeys", rocksdb::NUMBER_ITER_SKIP, 0 }, + { "NoIteratorCreated", rocksdb::NO_ITERATOR_CREATED, 0 }, + { "NoIteratorDeleted", rocksdb::NO_ITERATOR_DELETED, 0 }, }; // To control the rocksdb::StatsLevel, use ROCKSDB_STATS_LEVEL knob. @@ -1868,6 +1884,8 @@ struct RocksDBKeyValueStore : IKeyValueStore { Future openFuture; std::unique_ptr writeBatch; std::set keysSet; + // maximum number of single key deletes in a commit, if ROCKSDB_SINGLEKEY_DELETES_ON_CLEARRANGE is enabled. + int maxDeletes; Optional> metrics; FlowLock readSemaphore; int numReadWaiters; @@ -2056,6 +2074,7 @@ struct RocksDBKeyValueStore : IKeyValueStore { if (writeBatch == nullptr) { writeBatch.reset(new rocksdb::WriteBatch()); keysSet.clear(); + maxDeletes = SERVER_KNOBS->ROCKSDB_SINGLEKEY_DELETES_MAX; } ASSERT(defaultFdbCF != nullptr); writeBatch->Put(defaultFdbCF, toSlice(kv.key), toSlice(kv.value)); @@ -2064,23 +2083,24 @@ struct RocksDBKeyValueStore : IKeyValueStore { } } - void clear(KeyRangeRef keyRange, const StorageServerMetrics* storageMetrics, const Arena*) override { + void clear(KeyRangeRef keyRange, const Arena*) override { if (writeBatch == nullptr) { writeBatch.reset(new rocksdb::WriteBatch()); keysSet.clear(); + maxDeletes = SERVER_KNOBS->ROCKSDB_SINGLEKEY_DELETES_MAX; } ASSERT(defaultFdbCF != nullptr); // Number of deletes to rocksdb = counters.deleteKeyReqs + convertedDeleteKeyReqs; // Number of deleteRanges to rocksdb = counters.deleteRangeReqs - counters.convertedDeleteRangeReqs; - if (keyRange.singleKeyRange()) { + if (keyRange.singleKeyRange() && !SERVER_KNOBS->ROCKSDB_FORCE_DELETERANGE_FOR_CLEARRANGE) { writeBatch->Delete(defaultFdbCF, toSlice(keyRange.begin)); ++counters.deleteKeyReqs; + --maxDeletes; } else { ++counters.deleteRangeReqs; - if (SERVER_KNOBS->ROCKSDB_SINGLEKEY_DELETES_ON_CLEARRANGE && storageMetrics != nullptr && - storageMetrics->byteSample.getEstimate(keyRange) < - SERVER_KNOBS->ROCKSDB_SINGLEKEY_DELETES_BYTES_LIMIT) { + if (SERVER_KNOBS->ROCKSDB_SINGLEKEY_DELETES_ON_CLEARRANGE && + !SERVER_KNOBS->ROCKSDB_FORCE_DELETERANGE_FOR_CLEARRANGE && maxDeletes > 0) { ++counters.convertedDeleteRangeReqs; rocksdb::ReadOptions options = getReadOptions(); auto beginSlice = toSlice(keyRange.begin); @@ -2089,12 +2109,13 @@ struct RocksDBKeyValueStore : IKeyValueStore { options.iterate_upper_bound = &endSlice; auto cursor = std::unique_ptr(db->NewIterator(options, defaultFdbCF)); cursor->Seek(toSlice(keyRange.begin)); - while (cursor->Valid() && toStringRef(cursor->key()) < keyRange.end) { + while (cursor->Valid() && toStringRef(cursor->key()) < keyRange.end && maxDeletes > 0) { writeBatch->Delete(defaultFdbCF, cursor->key()); ++counters.convertedDeleteKeyReqs; + --maxDeletes; cursor->Next(); } - if (!cursor->status().ok()) { + if (!cursor->status().ok() || maxDeletes <= 0) { // if readrange iteration fails, then do a deleteRange. writeBatch->DeleteRange(defaultFdbCF, toSlice(keyRange.begin), toSlice(keyRange.end)); } else { @@ -2102,6 +2123,7 @@ struct RocksDBKeyValueStore : IKeyValueStore { while (it != keysSet.end() && *it < keyRange.end) { writeBatch->Delete(defaultFdbCF, toSlice(*it)); ++counters.convertedDeleteKeyReqs; + --maxDeletes; it++; } } @@ -2119,9 +2141,12 @@ struct RocksDBKeyValueStore : IKeyValueStore { &estPendCompactBytes); while (count && estPendCompactBytes > SERVER_KNOBS->ROCKSDB_CAN_COMMIT_COMPACT_BYTES_LIMIT) { wait(delay(SERVER_KNOBS->ROCKSDB_CAN_COMMIT_DELAY_ON_OVERLOAD)); + ++self->counters.commitDelayed; count--; self->db->GetAggregatedIntProperty(rocksdb::DB::Properties::kEstimatePendingCompactionBytes, &estPendCompactBytes); + if (deterministicRandom()->random01() < 0.001) + TraceEvent(SevWarn, "RocksDBCommitsDelayed1000x", self->id); } return Void(); @@ -2137,6 +2162,7 @@ struct RocksDBKeyValueStore : IKeyValueStore { auto a = new Writer::CommitAction(); a->batchToCommit = std::move(writeBatch); keysSet.clear(); + maxDeletes = SERVER_KNOBS->ROCKSDB_SINGLEKEY_DELETES_MAX; auto res = a->done.getFuture(); writeThread->post(a); return res; @@ -2461,7 +2487,7 @@ TEST_CASE("noSim/fdbserver/KeyValueStoreRocksDB/CheckpointRestore") { TEST_CASE("noSim/fdbserver/KeyValueStoreRocksDB/RocksDBTypes") { // If the following assertion fails, update SstFileMetaData and LiveFileMetaData in RocksDBCheckpointUtils.actor.h // to be the same as rocksdb::SstFileMetaData and rocksdb::LiveFileMetaData. - ASSERT_EQ(sizeof(rocksdb::LiveFileMetaData), 184); + ASSERT_EQ(sizeof(rocksdb::LiveFileMetaData), 192); ASSERT_EQ(sizeof(rocksdb::ExportImportFilesMetaData), 32); return Void(); } diff --git a/fdbserver/KeyValueStoreSQLite.actor.cpp b/fdbserver/KeyValueStoreSQLite.actor.cpp index dd56e412433..435a9b2eb5d 100644 --- a/fdbserver/KeyValueStoreSQLite.actor.cpp +++ b/fdbserver/KeyValueStoreSQLite.actor.cpp @@ -1586,9 +1586,7 @@ class KeyValueStoreSQLite final : public IKeyValueStore { StorageBytes getStorageBytes() const override; void set(KeyValueRef keyValue, const Arena* arena = nullptr) override; - void clear(KeyRangeRef range, - const StorageServerMetrics* storageMetrics = nullptr, - const Arena* arena = nullptr) override; + void clear(KeyRangeRef range, const Arena* arena = nullptr) override; Future commit(bool sequential = false) override; Future> readValue(KeyRef key, IKeyValueStore::ReadType, Optional debugID) override; @@ -2204,7 +2202,7 @@ void KeyValueStoreSQLite::set(KeyValueRef keyValue, const Arena* arena) { ++writesRequested; writeThread->post(new Writer::SetAction(keyValue)); } -void KeyValueStoreSQLite::clear(KeyRangeRef range, const StorageServerMetrics* storageMetrics, const Arena* arena) { +void KeyValueStoreSQLite::clear(KeyRangeRef range, const Arena* arena) { ++writesRequested; writeThread->post(new Writer::ClearAction(range)); } diff --git a/fdbserver/LeaderElection.actor.cpp b/fdbserver/LeaderElection.actor.cpp index e641f3d0f31..d010f7d3d19 100644 --- a/fdbserver/LeaderElection.actor.cpp +++ b/fdbserver/LeaderElection.actor.cpp @@ -158,7 +158,13 @@ ACTOR Future tryBecomeLeaderInternal(ServerCoordinators coordinators, // These coordinators are forwarded to another set. But before we change our own cluster file, we need // to make sure that a majority of coordinators know that. SOMEDAY: Wait briefly to see if other // coordinators will tell us they already know, to save communication? - wait(changeLeaderCoordinators(coordinators, leader.get().first.serializedInfo)); + // NOTE: If a majority of coordinators (in the current connection string) have failed then we can + // end up waiting here indefinitely. Try to make progress in that scenario by proceeding with the + // connection string that we have received. Not a great solution, but can help in certain scenarios. + choose { + when(wait(changeLeaderCoordinators(coordinators, leader.get().first.serializedInfo))) {} + when(wait(delay(20))) {} + } if (!hasConnected) { TraceEvent(SevWarnAlways, "IncorrectClusterFileContentsAtConnection") diff --git a/fdbserver/RemoteIKeyValueStore.actor.h b/fdbserver/RemoteIKeyValueStore.actor.h index de7cd362649..0465e4afbb2 100644 --- a/fdbserver/RemoteIKeyValueStore.actor.h +++ b/fdbserver/RemoteIKeyValueStore.actor.h @@ -392,9 +392,7 @@ struct RemoteIKeyValueStore : public IKeyValueStore { void set(KeyValueRef keyValue, const Arena* arena = nullptr) override { interf.set.send(IKVSSetRequest{ keyValue, ReplyPromise() }); } - void clear(KeyRangeRef range, - const StorageServerMetrics* storageMetrics = nullptr, - const Arena* arena = nullptr) override { + void clear(KeyRangeRef range, const Arena* arena = nullptr) override { interf.clear.send(IKVSClearRequest{ range, ReplyPromise() }); } diff --git a/fdbserver/RocksDBCheckpointUtils.actor.h b/fdbserver/RocksDBCheckpointUtils.actor.h index db812998ff3..23979f22e08 100644 --- a/fdbserver/RocksDBCheckpointUtils.actor.h +++ b/fdbserver/RocksDBCheckpointUtils.actor.h @@ -28,108 +28,81 @@ #include "fdbclient/NativeAPI.actor.h" #include "fdbserver/ServerCheckpoint.actor.h" #include "flow/flow.h" -#include "rocksdb/types.h" -#include "rocksdb/options.h" #include "flow/actorcompiler.h" // has to be last include // Copied from rocksdb/metadata.h, so that we can add serializer. -// -// Basic identifiers and metadata for a file in a DB. This only includes -// information considered relevant for taking backups, checkpoints, or other -// services relating to DB file storage. -// This is only appropriate for immutable files, such as SST files or all -// files in a backup. See also LiveFileStorageInfo. struct SstFileMetaData { constexpr static FileIdentifier file_identifier = 3804347; SstFileMetaData() - : file_number(0), size(0), temperature(rocksdb::Temperature::kUnknown), - smallest_seqno(0), largest_seqno(0), num_reads_sampled(0), - being_compacted(false), num_entries(0), num_deletions(0), - oldest_blob_file_number(0), oldest_ancester_time(0), - file_creation_time(0) {} + : file_number(0), file_type(2), size(0), temperature(0), smallest_seqno(0), largest_seqno(0), + num_reads_sampled(0), being_compacted(false), num_entries(0), num_deletions(0), oldest_blob_file_number(0), + oldest_ancester_time(0), file_creation_time(0), epoch_number(0) {} - SstFileMetaData(const std::string& _file_name, - uint64_t _file_number, + SstFileMetaData(const std::string& _relative_filename, const std::string& _directory, - size_t _size, - rocksdb::SequenceNumber _smallest_seqno, - rocksdb::SequenceNumber _largest_seqno, + uint64_t _file_number, + int _file_type, + uint64_t _size, + int _temperature, + std::string& _file_checksum, + std::string& _file_checksum_func_name, + uint64_t _smallest_seqno, + uint64_t _largest_seqno, const std::string& _smallestkey, const std::string& _largestkey, uint64_t _num_reads_sampled, bool _being_compacted, - rocksdb::Temperature _temperature, + uint64_t _num_entries, + uint64_t _num_deletions, uint64_t _oldest_blob_file_number, uint64_t _oldest_ancester_time, uint64_t _file_creation_time, - std::string& _file_checksum, - std::string& _file_checksum_func_name) - : smallest_seqno(_smallest_seqno), largest_seqno(_largest_seqno), smallestkey(_smallestkey), - largestkey(_largestkey), num_reads_sampled(_num_reads_sampled), being_compacted(_being_compacted), - num_entries(0), num_deletions(0), oldest_blob_file_number(_oldest_blob_file_number), - oldest_ancester_time(_oldest_ancester_time), file_creation_time(_file_creation_time) { - if (!_file_name.empty()) { - if (_file_name[0] == '/') { - relative_filename = _file_name.substr(1); - name = _file_name; // Deprecated field - } else { - relative_filename = _file_name; - name = std::string("/") + _file_name; // Deprecated field - } - assert(relative_filename.size() + 1 == name.size()); - assert(relative_filename[0] != '/'); - assert(name[0] == '/'); - } - directory = _directory; - db_path = _directory; // Deprecated field - file_number = _file_number; - file_type = rocksdb::kTableFile; - size = _size; - temperature = _temperature; - file_checksum = _file_checksum; - file_checksum_func_name = _file_checksum_func_name; - } + uint64_t _epoch_number, + const std::string& _name, + const std::string& _db_path) + : relative_filename(_relative_filename), directory(_directory), file_number(_file_number), file_type(_file_type), + size(_size), temperature(_temperature), file_checksum(_file_checksum), + file_checksum_func_name(_file_checksum_func_name), smallest_seqno(_smallest_seqno), + largest_seqno(_largest_seqno), smallestkey(_smallestkey), largestkey(_largestkey), + num_reads_sampled(_num_reads_sampled), being_compacted(_being_compacted), num_entries(_num_entries), + num_deletions(_num_deletions), oldest_blob_file_number(_oldest_blob_file_number), + oldest_ancester_time(_oldest_ancester_time), file_creation_time(_file_creation_time), + epoch_number(_epoch_number), name(_name), db_path(_db_path) {} // The name of the file within its directory (e.g. "123456.sst") std::string relative_filename; // The directory containing the file, without a trailing '/'. This could be // a DB path, wal_dir, etc. std::string directory; - // The id of the file within a single DB. Set to 0 if the file does not have // a number (e.g. CURRENT) - uint64_t file_number = 0; + uint64_t file_number; // The type of the file as part of a DB. - rocksdb::FileType file_type = rocksdb::kTempFile; - + int file_type; // File size in bytes. See also `trim_to_size`. - uint64_t size = 0; - + uint64_t size; // This feature is experimental and subject to change. - rocksdb::Temperature temperature = rocksdb::Temperature::kUnknown; - + int temperature; // The checksum of a SST file, the value is decided by the file content and // the checksum algorithm used for this SST file. The checksum function is // identified by the file_checksum_func_name. If the checksum function is // not specified, file_checksum is "0" by default. std::string file_checksum; - // The name of the checksum function used to generate the file checksum // value. If file checksum is not enabled (e.g., sst_file_checksum_func is // null), file_checksum_func_name is UnknownFileChecksumFuncName, which is // "Unknown". std::string file_checksum_func_name; - rocksdb::SequenceNumber smallest_seqno; // Smallest sequence number in file. - rocksdb::SequenceNumber largest_seqno; // Largest sequence number in file. + + uint64_t smallest_seqno; // Smallest sequence number in file. + uint64_t largest_seqno; // Largest sequence number in file. std::string smallestkey; // Smallest user defined key in the file. std::string largestkey; // Largest user defined key in the file. uint64_t num_reads_sampled; // How many times the file is read. bool being_compacted; // true if the file is currently being compacted. - uint64_t num_entries; uint64_t num_deletions; - uint64_t oldest_blob_file_number; // The id of the oldest blob file // referenced by the file. // An SST file may be generated by compactions whose input files may @@ -143,22 +116,30 @@ struct SstFileMetaData { // Timestamp when the SST file is created, provided by // SystemClock::GetCurrentTime(). 0 if the information is not available. uint64_t file_creation_time; - + // The order of a file being flushed or ingested/imported. + // Compaction output file will be assigned with the minimum `epoch_number` + // among input files'. + // For L0, larger `epoch_number` indicates newer L0 file. + // 0 if the information is not available. + uint64_t epoch_number; // DEPRECATED: The name of the file within its directory with a // leading slash (e.g. "/123456.sst"). Use relative_filename from base struct // instead. std::string name; - // DEPRECATED: replaced by `directory` in base struct std::string db_path; template void serialize(Ar& ar) { serializer(ar, - size, - name, + relative_filename, + directory, file_number, - db_path, + file_type, + size, + temperature, + file_checksum, + file_checksum_func_name, smallest_seqno, largest_seqno, smallestkey, @@ -167,12 +148,12 @@ struct SstFileMetaData { being_compacted, num_entries, num_deletions, - temperature, oldest_blob_file_number, oldest_ancester_time, file_creation_time, - file_checksum, - file_checksum_func_name); + epoch_number, + name, + db_path); } }; @@ -187,10 +168,14 @@ struct LiveFileMetaData : public SstFileMetaData { template void serialize(Ar& ar) { serializer(ar, - SstFileMetaData::size, - SstFileMetaData::name, + SstFileMetaData::relative_filename, + SstFileMetaData::directory, SstFileMetaData::file_number, - SstFileMetaData::db_path, + SstFileMetaData::file_type, + SstFileMetaData::size, + SstFileMetaData::temperature, + SstFileMetaData::file_checksum, + SstFileMetaData::file_checksum_func_name, SstFileMetaData::smallest_seqno, SstFileMetaData::largest_seqno, SstFileMetaData::smallestkey, @@ -199,12 +184,12 @@ struct LiveFileMetaData : public SstFileMetaData { SstFileMetaData::being_compacted, SstFileMetaData::num_entries, SstFileMetaData::num_deletions, - SstFileMetaData::temperature, SstFileMetaData::oldest_blob_file_number, SstFileMetaData::oldest_ancester_time, SstFileMetaData::file_creation_time, - SstFileMetaData::file_checksum, - SstFileMetaData::file_checksum_func_name, + SstFileMetaData::epoch_number, + SstFileMetaData::name, + SstFileMetaData::db_path, column_family_name, level, fetched); @@ -248,4 +233,4 @@ ACTOR Future deleteRocksCFCheckpoint(CheckpointMetaData checkpoint); ICheckpointReader* newRocksDBCheckpointReader(const CheckpointMetaData& checkpoint, UID logID); RocksDBColumnFamilyCheckpoint getRocksCF(const CheckpointMetaData& checkpoint); -#endif +#endif \ No newline at end of file diff --git a/fdbserver/VersionedBTree.actor.cpp b/fdbserver/VersionedBTree.actor.cpp index c20f3f656d9..cf8211ed015 100644 --- a/fdbserver/VersionedBTree.actor.cpp +++ b/fdbserver/VersionedBTree.actor.cpp @@ -7822,9 +7822,7 @@ class KeyValueStoreRedwood : public IKeyValueStore { Future getError() const override { return delayed(m_error.getFuture()); }; - void clear(KeyRangeRef range, - const StorageServerMetrics* storageMetrics = nullptr, - const Arena* arena = 0) override { + void clear(KeyRangeRef range, const Arena* arena = 0) override { debug_printf("CLEAR %s\n", printable(range).c_str()); m_tree->clear(range); } diff --git a/fdbserver/storageserver.actor.cpp b/fdbserver/storageserver.actor.cpp index 7e141468302..fed740f79bd 100644 --- a/fdbserver/storageserver.actor.cpp +++ b/fdbserver/storageserver.actor.cpp @@ -4173,6 +4173,7 @@ ACTOR Future getMappedKeyValuesQ(StorageServer* data, GetMappedKeyValuesRe } data->transactionTagCounter.addRequest(req.tags, resultSize); + ++data->counters.finishedQueries; ++data->counters.finishedGetMappedRangeQueries; --data->readQueueSizeMetric; @@ -7869,7 +7870,7 @@ void setAssignedStatus(StorageServer* self, KeyRangeRef keys, bool nowAssigned) } void StorageServerDisk::clearRange(KeyRangeRef keys) { - storage->clear(keys, &data->metrics); + storage->clear(keys); ++(*kvClearRanges); if (keys.singleKeyRange()) { ++(*kvClearSingleKey); @@ -7886,7 +7887,7 @@ void StorageServerDisk::writeMutation(MutationRef mutation) { storage->set(KeyValueRef(mutation.param1, mutation.param2)); *kvCommitLogicalBytes += mutation.expectedSize(); } else if (mutation.type == MutationRef::ClearRange) { - storage->clear(KeyRangeRef(mutation.param1, mutation.param2), &data->metrics); + storage->clear(KeyRangeRef(mutation.param1, mutation.param2)); ++(*kvClearRanges); if (KeyRangeRef(mutation.param1, mutation.param2).singleKeyRange()) { ++(*kvClearSingleKey); @@ -7904,7 +7905,7 @@ void StorageServerDisk::writeMutations(const VectorRef& mutations, storage->set(KeyValueRef(m.param1, m.param2)); *kvCommitLogicalBytes += m.expectedSize(); } else if (m.type == MutationRef::ClearRange) { - storage->clear(KeyRangeRef(m.param1, m.param2), &data->metrics); + storage->clear(KeyRangeRef(m.param1, m.param2)); ++(*kvClearRanges); if (KeyRangeRef(m.param1, m.param2).singleKeyRange()) { ++(*kvClearSingleKey); @@ -8317,7 +8318,7 @@ ACTOR Future restoreDurableState(StorageServer* data, IKeyValueStore* stor ++data->counters.kvSystemClearRanges; // TODO(alexmiller): Figure out how to selectively enable spammy data distribution events. // DEBUG_KEY_RANGE("clearInvalidVersion", invalidVersion, clearRange); - storage->clear(clearRange, &data->metrics); + storage->clear(clearRange); ++data->counters.kvSystemClearRanges; data->byteSampleApplyClear(clearRange, invalidVersion); } diff --git a/fdbserver/workloads/DataLossRecovery.actor.cpp b/fdbserver/workloads/DataLossRecovery.actor.cpp index dce83880804..dcb34b4435b 100644 --- a/fdbserver/workloads/DataLossRecovery.actor.cpp +++ b/fdbserver/workloads/DataLossRecovery.actor.cpp @@ -152,7 +152,7 @@ struct DataLossRecoveryWorkload : TestWorkload { servers.push_back(AddressExclusion(addr.ip, addr.port)); loop { try { - excludeServers(tr, servers, true); + wait(excludeServers(&tr, servers, true)); wait(tr.commit()); break; } catch (Error& e) { diff --git a/fdbserver/workloads/GetMappedRange.actor.cpp b/fdbserver/workloads/GetMappedRange.actor.cpp index abbf2c248f5..b735b3748d7 100644 --- a/fdbserver/workloads/GetMappedRange.actor.cpp +++ b/fdbserver/workloads/GetMappedRange.actor.cpp @@ -23,6 +23,7 @@ #include #include "fdbrpc/simulator.h" #include "fdbclient/MutationLogReader.actor.h" +#include "fdbclient/StatusClient.h" #include "fdbclient/Tuple.h" #include "fdbserver/workloads/ApiWorkload.h" #include "fdbserver/workloads/workloads.actor.h" @@ -49,9 +50,13 @@ struct GetMappedRangeWorkload : ApiWorkload { // const bool SPLIT_RECORDS = deterministicRandom()->random01() < 0.5; const bool SPLIT_RECORDS = true; const static int SPLIT_SIZE = 3; + double checkStorageQueueSeconds; + double queueMaxLength; GetMappedRangeWorkload(WorkloadContext const& wcx) : ApiWorkload(wcx) { enabled = !clientId; // only do this on the "first" client + checkStorageQueueSeconds = getOption(options, "checkStorageQueueSeconds"_sr, 60.0); + queueMaxLength = getOption(options, "queueMaxLength"_sr, 200); } std::string description() const override { return "GetMappedRange"; } @@ -248,8 +253,13 @@ struct GetMappedRangeWorkload : ApiWorkload { e.code() == error_code_quick_get_key_values_miss)) { TraceEvent("GetMappedRangeWorkloadExpectedErrorDetected").error(e); return MappedRangeResult(); + } else if (e.code() == error_code_proxy_memory_limit_exceeded || + e.code() == error_code_operation_cancelled) { + // requests have overwhelmed commit proxy, rest a bit + wait(delay(FLOW_KNOBS->PREVENT_FAST_SPIN_DELAY)); + continue; } else { - std::cout << "error " << e.what() << std::endl; + std::cout << "error " << e.what() << " code is " << e.code() << std::endl; wait(tr->onError(e)); } std::cout << "failed scanMappedRangeWithLimits" << std::endl; @@ -257,6 +267,30 @@ struct GetMappedRangeWorkload : ApiWorkload { } } + // if sendFirstRequestIndefinitely is true, then this method would send the first request indefinitly + // it is in order to test the metric + ACTOR Future submitSmallRequestIndefinitely(Database cx, + int beginId, + int endId, + Key mapper, + GetMappedRangeWorkload* self) { + Key beginTuple = Tuple().append(prefix).append(INDEX).append(indexKey(beginId)).getDataAsStandalone(); + state KeySelector beginSelector = KeySelector(firstGreaterOrEqual(beginTuple)); + Key endTuple = Tuple().append(prefix).append(INDEX).append(indexKey(endId)).getDataAsStandalone(); + state KeySelector endSelector = KeySelector(firstGreaterOrEqual(endTuple)); + state int limit = 1; + state int byteLimit = 10000; + while (true) { + MappedRangeResult result = wait(self->scanMappedRangeWithLimits( + cx, beginSelector, endSelector, mapper, limit, byteLimit, beginId, self)); + if (result.empty()) { + TraceEvent("EmptyResult"); + } + // to avoid requests make proxy memory overwhelmed + wait(delay(FLOW_KNOBS->PREVENT_FAST_SPIN_DELAY)); + } + } + ACTOR Future scanMappedRange(Database cx, int beginId, int endId, Key mapper, GetMappedRangeWorkload* self) { Key beginTuple = Tuple().append(prefix).append(INDEX).append(indexKey(beginId)).getDataAsStandalone(); state KeySelector beginSelector = KeySelector(firstGreaterOrEqual(beginTuple)); @@ -282,7 +316,6 @@ struct GetMappedRangeWorkload : ApiWorkload { int indexByteLimit = byteLimit * SERVER_KNOBS->FRACTION_INDEX_BYTELIMIT_PREFETCH; int indexCountByteLimit = indexByteLimit / indexSize + (indexByteLimit % indexSize != 0); int indexCount = std::min(limit, indexCountByteLimit); - std::cout << "indexCount: " << indexCount << std::endl; // result set cannot be larger than the number of index fetched ASSERT(result.size() <= indexCount); @@ -298,7 +331,6 @@ struct GetMappedRangeWorkload : ApiWorkload { boundByRecord = round * SERVER_KNOBS->MAX_PARALLEL_QUICK_GET_VALUE; } expectedCnt = std::min(expectedCnt, boundByRecord); - std::cout << "boundByRecord: " << boundByRecord << std::endl; ASSERT(result.size() == expectedCnt); beginSelector = KeySelector(firstGreaterThan(result.back().key)); } @@ -389,6 +421,46 @@ struct GetMappedRangeWorkload : ApiWorkload { } } + // checking the max storage queue length is bounded + ACTOR static Future reportMetric(GetMappedRangeWorkload* self, Database cx) { + loop { + StatusObject result = wait(StatusClient::statusFetcher(cx)); + StatusObjectReader statusObj(result); + state StatusObjectReader statusObjCluster; + state StatusObjectReader processesMap; + state int64_t queryQueueMax = 0; + state int waitInterval = 2; + if (!statusObj.get("cluster", statusObjCluster)) { + TraceEvent("NoCluster"); + wait(delay(waitInterval)); + continue; + } + + if (!statusObjCluster.get("processes", processesMap)) { + TraceEvent("NoProcesses"); + wait(delay(waitInterval)); + continue; + } + for (auto proc : processesMap.obj()) { + StatusObjectReader process(proc.second); + if (process.has("roles")) { + StatusArray rolesArray = proc.second.get_obj()["roles"].get_array(); + for (StatusObjectReader role : rolesArray) { + if (role["role"].get_str() == "storage") { + role.get("query_queue_max", queryQueueMax); + TEST(queryQueueMax > 0); // SS query queue is non-empty + TraceEvent(SevDebug, "QueryQueueMax").detail("Value", queryQueueMax); + ASSERT(queryQueueMax < self->queueMaxLength); + } + } + } else { + TraceEvent("NoRoles"); + } + } + wait(delay(waitInterval)); + } + } + // If the same transaction writes to the read set (the scanned ranges) before reading, it should throw read your // write exception. ACTOR Future testRYW(GetMappedRangeWorkload* self) { @@ -413,6 +485,27 @@ struct GetMappedRangeWorkload : ApiWorkload { } } + ACTOR static Future testMetric(Database cx, + GetMappedRangeWorkload* self, + int beginId, + int endId, + Key mapper, + int seconds) { + loop choose { + when(wait(reportMetric(self, cx))) { + TraceEvent(SevError, "Error: ReportMetric has ended"); + return Void(); + } + when(wait(self->submitSmallRequestIndefinitely(cx, 10, 490, mapper, self))) { + TraceEvent(SevError, "Error: submitSmallRequestIndefinitely has ended"); + return Void(); + } + when(wait(delay(seconds))) { + return Void(); + } + } + } + ACTOR Future _start(Database cx, GetMappedRangeWorkload* self) { TraceEvent("GetMappedRangeWorkloadConfig").detail("BadMapper", self->BAD_MAPPER); @@ -440,12 +533,13 @@ struct GetMappedRangeWorkload : ApiWorkload { std::cout << "Test configuration: transactionType:" << self->transactionType << " snapshot:" << self->snapshot << "bad_mapper:" << self->BAD_MAPPER << std::endl; - Key mapper = getMapper(self); + state Key mapper = getMapper(self); // The scanned range cannot be too large to hit get_mapped_key_values_has_more. We have a unit validating the // error is thrown when the range is large. state bool originalStrictlyEnforeByteLimit = SERVER_KNOBS->STRICTLY_ENFORCE_BYTE_LIMIT; (const_cast SERVER_KNOBS)->STRICTLY_ENFORCE_BYTE_LIMIT = deterministicRandom()->coinflip(); wait(self->scanMappedRange(cx, 10, 490, mapper, self)); + wait(testMetric(cx, self, 10, 490, mapper, self->checkStorageQueueSeconds)); (const_cast SERVER_KNOBS)->STRICTLY_ENFORCE_BYTE_LIMIT = originalStrictlyEnforeByteLimit; return Void(); } diff --git a/fdbserver/workloads/SpecialKeySpaceCorrectness.actor.cpp b/fdbserver/workloads/SpecialKeySpaceCorrectness.actor.cpp index 05b509bc252..84f36de5f7f 100644 --- a/fdbserver/workloads/SpecialKeySpaceCorrectness.actor.cpp +++ b/fdbserver/workloads/SpecialKeySpaceCorrectness.actor.cpp @@ -1203,142 +1203,6 @@ struct SpecialKeySpaceCorrectnessWorkload : TestWorkload { } catch (Error& e) { wait(tx->onError(e)); } - // profile client get - loop { - try { - tx->setOption(FDBTransactionOptions::READ_SYSTEM_KEYS); - // client_txn_sample_rate - state Optional txnSampleRate = - wait(tx->get(LiteralStringRef("client_txn_sample_rate") - .withPrefix(SpecialKeySpace::getManagementApiCommandPrefix("profile")))); - ASSERT(txnSampleRate.present()); - Optional txnSampleRateKey = wait(tx->get(GlobalConfig::prefixedKey(fdbClientInfoTxnSampleRate))); - if (txnSampleRateKey.present()) { - const double sampleRateDbl = - BinaryReader::fromStringRef(txnSampleRateKey.get(), Unversioned()); - if (!std::isinf(sampleRateDbl)) { - ASSERT(txnSampleRate.get().toString() == boost::lexical_cast(sampleRateDbl)); - } else { - ASSERT(txnSampleRate.get().toString() == "default"); - } - } else { - ASSERT(txnSampleRate.get().toString() == "default"); - } - // client_txn_size_limit - state Optional txnSizeLimit = - wait(tx->get(LiteralStringRef("client_txn_size_limit") - .withPrefix(SpecialKeySpace::getManagementApiCommandPrefix("profile")))); - ASSERT(txnSizeLimit.present()); - Optional txnSizeLimitKey = wait(tx->get(GlobalConfig::prefixedKey(fdbClientInfoTxnSizeLimit))); - if (txnSizeLimitKey.present()) { - const int64_t sizeLimit = - BinaryReader::fromStringRef(txnSizeLimitKey.get(), Unversioned()); - if (sizeLimit != -1) { - ASSERT(txnSizeLimit.get().toString() == boost::lexical_cast(sizeLimit)); - } else { - ASSERT(txnSizeLimit.get().toString() == "default"); - } - } else { - ASSERT(txnSizeLimit.get().toString() == "default"); - } - tx->reset(); - break; - } catch (Error& e) { - TraceEvent(SevDebug, "ProfileClientGet").error(e); - wait(tx->onError(e)); - } - } - { - state double r_sample_rate = deterministicRandom()->random01(); - state int64_t r_size_limit = deterministicRandom()->randomInt64(1e3, 1e6); - // update the sample rate and size limit - loop { - try { - tx->setOption(FDBTransactionOptions::RAW_ACCESS); - tx->setOption(FDBTransactionOptions::SPECIAL_KEY_SPACE_ENABLE_WRITES); - tx->set(LiteralStringRef("client_txn_sample_rate") - .withPrefix(SpecialKeySpace::getManagementApiCommandPrefix("profile")), - Value(boost::lexical_cast(r_sample_rate))); - tx->set(LiteralStringRef("client_txn_size_limit") - .withPrefix(SpecialKeySpace::getManagementApiCommandPrefix("profile")), - Value(boost::lexical_cast(r_size_limit))); - wait(tx->commit()); - tx->reset(); - break; - } catch (Error& e) { - wait(tx->onError(e)); - } - } - // commit successfully, verify the system key changed - loop { - try { - state Optional sampleRate = - wait(tx->get(GlobalConfig::prefixedKey(fdbClientInfoTxnSampleRate))); - if (!sampleRate.present()) { - // Wait for GlobalConfig synchronization - wait(delayJittered(1.0)); - continue; - } - ASSERT(sampleRate.present()); - ASSERT_LT(r_sample_rate - boost::lexical_cast(sampleRate.get().toString()), 0.001); - Optional sizeLimit = wait(tx->get(GlobalConfig::prefixedKey(fdbClientInfoTxnSizeLimit))); - ASSERT(sizeLimit.present()); - ASSERT(r_size_limit == boost::lexical_cast(sizeLimit.get().toString())); - tx->reset(); - break; - } catch (Error& e) { - wait(tx->onError(e)); - } - } - // Change back to default - loop { - try { - tx->setOption(FDBTransactionOptions::RAW_ACCESS); - tx->setOption(FDBTransactionOptions::SPECIAL_KEY_SPACE_ENABLE_WRITES); - tx->set(LiteralStringRef("client_txn_sample_rate") - .withPrefix(SpecialKeySpace::getManagementApiCommandPrefix("profile")), - LiteralStringRef("default")); - tx->set(LiteralStringRef("client_txn_size_limit") - .withPrefix(SpecialKeySpace::getManagementApiCommandPrefix("profile")), - LiteralStringRef("default")); - wait(tx->commit()); - tx->reset(); - break; - } catch (Error& e) { - wait(tx->onError(e)); - } - } - // Test invalid values - loop { - try { - tx->setOption(FDBTransactionOptions::RAW_ACCESS); - tx->setOption(FDBTransactionOptions::SPECIAL_KEY_SPACE_ENABLE_WRITES); - tx->set((deterministicRandom()->coinflip() ? LiteralStringRef("client_txn_sample_rate") - : LiteralStringRef("client_txn_size_limit")) - .withPrefix(SpecialKeySpace::getManagementApiCommandPrefix("profile")), - LiteralStringRef("invalid_value")); - wait(tx->commit()); - ASSERT(false); - } catch (Error& e) { - if (e.code() == error_code_special_keys_api_failure) { - Optional errorMsg = - wait(tx->get(SpecialKeySpace::getModuleRange(SpecialKeySpace::MODULE::ERRORMSG).begin)); - ASSERT(errorMsg.present()); - std::string errorStr; - auto valueObj = readJSONStrictly(errorMsg.get().toString()).get_obj(); - auto schema = readJSONStrictly(JSONSchemas::managementApiErrorSchema.toString()).get_obj(); - // special_key_space_management_api_error_msg schema validation - ASSERT(schemaMatch(schema, valueObj, errorStr, SevError, true)); - ASSERT(valueObj["command"].get_str() == "profile" && !valueObj["retriable"].get_bool()); - tx->reset(); - break; - } else { - wait(tx->onError(e)); - } - wait(delay(FLOW_KNOBS->PREVENT_FAST_SPIN_DELAY)); - } - } - } // data_distribution & maintenance get loop { try { diff --git a/tests/fast/GetMappedRange.toml b/tests/fast/GetMappedRange.toml index 2dd4672592a..36067edd063 100644 --- a/tests/fast/GetMappedRange.toml +++ b/tests/fast/GetMappedRange.toml @@ -4,3 +4,5 @@ useDB = true [[test.workload]] testName = 'GetMappedRange' + checkStorageQueueSeconds=60 + queueMaxLength=200 \ No newline at end of file