diff --git a/bindings/python/tests/fdbcli_tests.py b/bindings/python/tests/fdbcli_tests.py index 49fd45c632..8004f77f30 100755 --- a/bindings/python/tests/fdbcli_tests.py +++ b/bindings/python/tests/fdbcli_tests.py @@ -381,12 +381,19 @@ def exclude(logger): while True: logger.debug("Excluding process: {}".format(excluded_address)) error_message = run_fdbcli_command_and_get_error('exclude', excluded_address) - if not error_message: + if error_message == 'WARNING: {} is a coordinator!'.format(excluded_address): + # exclude coordinator will print the warning, verify the randomly selected process is the coordinator + coordinator_list = get_value_from_status_json(True, 'client', 'coordinators', 'coordinators') + assert len(coordinator_list) == 1 + assert coordinator_list[0]['address'] == excluded_address break + elif not error_message: + break + else: + logger.debug("Error message: {}\n".format(error_message)) logger.debug("Retry exclude after 1 second") time.sleep(1) output2 = run_fdbcli_command('exclude') - # logger.debug(output3) assert 'There are currently 1 servers or localities being excluded from the database' in output2 assert excluded_address in output2 run_fdbcli_command('include', excluded_address) @@ -416,6 +423,6 @@ if __name__ == '__main__': else: assert process_number > 1, "Process number should be positive" coordinators() - # exclude() + exclude() diff --git a/cmake/AddFdbTest.cmake b/cmake/AddFdbTest.cmake index d292679e91..752fcb8ebe 100644 --- a/cmake/AddFdbTest.cmake +++ b/cmake/AddFdbTest.cmake @@ -416,14 +416,14 @@ function(add_fdbclient_test) message(STATUS "Adding Client test ${T_NAME}") if (T_PROCESS_NUMBER) add_test(NAME "${T_NAME}" - COMMAND ${CMAKE_SOURCE_DIR}/tests/TestRunner/tmp_cluster.py + COMMAND ${Python_EXECUTABLE} ${CMAKE_SOURCE_DIR}/tests/TestRunner/tmp_cluster.py --build-dir ${CMAKE_BINARY_DIR} --process-number ${T_PROCESS_NUMBER} -- ${T_COMMAND}) else() add_test(NAME "${T_NAME}" - COMMAND ${CMAKE_SOURCE_DIR}/tests/TestRunner/tmp_cluster.py + COMMAND ${Python_EXECUTABLE} ${CMAKE_SOURCE_DIR}/tests/TestRunner/tmp_cluster.py --build-dir ${CMAKE_BINARY_DIR} -- ${T_COMMAND}) @@ -459,7 +459,7 @@ function(add_multi_fdbclient_test) endif() message(STATUS "Adding Client test ${T_NAME}") add_test(NAME "${T_NAME}" - COMMAND ${CMAKE_SOURCE_DIR}/tests/TestRunner/tmp_multi_cluster.py + COMMAND ${Python_EXECUTABLE} ${CMAKE_SOURCE_DIR}/tests/TestRunner/tmp_multi_cluster.py --build-dir ${CMAKE_BINARY_DIR} --clusters 3 -- diff --git a/fdbclient/BackupContainerAzureBlobStore.actor.cpp b/fdbclient/BackupContainerAzureBlobStore.actor.cpp index e2733bf7ce..6018c4d5cb 100644 --- a/fdbclient/BackupContainerAzureBlobStore.actor.cpp +++ b/fdbclient/BackupContainerAzureBlobStore.actor.cpp @@ -19,9 +19,7 @@ */ #include "fdbclient/BackupContainerAzureBlobStore.h" -#if (!defined(TLS_DISABLED) && !defined(_WIN32)) #include "fdbrpc/AsyncFileEncrypted.h" -#emdif #include "flow/actorcompiler.h" // This must be the last #include. @@ -250,9 +248,7 @@ BackupContainerAzureBlobStore::BackupContainerAzureBlobStore(const NetworkAddres const std::string& containerName, const Optional& encryptionKeyFileName) : containerName(containerName) { -#if (!defined(TLS_DISABLED) && !defined(_WIN32)) setEncryptionKey(encryptionKeyFileName); -#endif std::string accountKey = std::getenv("AZURE_KEY"); auto credential = std::make_shared(accountName, accountKey); auto storageAccount = std::make_shared( diff --git a/fdbclient/BackupContainerFileSystem.actor.cpp b/fdbclient/BackupContainerFileSystem.actor.cpp index 5417056fed..040c759956 100644 --- a/fdbclient/BackupContainerFileSystem.actor.cpp +++ b/fdbclient/BackupContainerFileSystem.actor.cpp @@ -23,9 +23,7 @@ #include "fdbclient/BackupContainerFileSystem.h" #include "fdbclient/BackupContainerLocalDirectory.h" #include "fdbclient/JsonBuilder.h" -#if (!defined(TLS_DISABLED) && !defined(_WIN32)) #include "flow/StreamCipher.h" -#endif #include "flow/UnitTest.h" #include @@ -1481,7 +1479,6 @@ Future BackupContainerFileSystem::encryptionSetupComplete() const { return encryptionSetupFuture; } -#if (!defined(TLS_DISABLED) && !defined(_WIN32)) void BackupContainerFileSystem::setEncryptionKey(Optional const& encryptionKeyFileName) { if (encryptionKeyFileName.present()) { #if ENCRYPTION_ENABLED @@ -1498,11 +1495,6 @@ Future BackupContainerFileSystem::createTestEncryptionKeyFile(std::string return Void(); #endif } -#else -Future BackupContainerFileSystem::createTestEncryptionKeyFile(std::string const& filename) { - return Void(); -} -#endif namespace backup_test { diff --git a/fdbclient/BackupContainerFileSystem.h b/fdbclient/BackupContainerFileSystem.h index 4b92b2f409..292fc67abb 100644 --- a/fdbclient/BackupContainerFileSystem.h +++ b/fdbclient/BackupContainerFileSystem.h @@ -156,9 +156,7 @@ public: protected: bool usesEncryption() const; -#if (!defined(TLS_DISABLED) && !defined(_WIN32)) void setEncryptionKey(Optional const& encryptionKeyFileName); -#endif Future encryptionSetupComplete() const; private: diff --git a/fdbclient/BackupContainerLocalDirectory.actor.cpp b/fdbclient/BackupContainerLocalDirectory.actor.cpp index fa0f351d5a..b89d085a64 100644 --- a/fdbclient/BackupContainerLocalDirectory.actor.cpp +++ b/fdbclient/BackupContainerLocalDirectory.actor.cpp @@ -133,9 +133,7 @@ std::string BackupContainerLocalDirectory::getURLFormat() { BackupContainerLocalDirectory::BackupContainerLocalDirectory(const std::string& url, const Optional& encryptionKeyFileName) { -#if (!defined(TLS_DISABLED) && !defined(_WIN32)) setEncryptionKey(encryptionKeyFileName); -#endif std::string path; if (url.find("file://") != 0) { diff --git a/fdbclient/BackupContainerS3BlobStore.actor.cpp b/fdbclient/BackupContainerS3BlobStore.actor.cpp index c48e66e597..b915701a3f 100644 --- a/fdbclient/BackupContainerS3BlobStore.actor.cpp +++ b/fdbclient/BackupContainerS3BlobStore.actor.cpp @@ -147,9 +147,7 @@ BackupContainerS3BlobStore::BackupContainerS3BlobStore(Reference& encryptionKeyFileName) : m_bstore(bstore), m_name(name), m_bucket("FDB_BACKUPS_V2") { -#if (!defined(TLS_DISABLED) && !defined(_WIN32)) setEncryptionKey(encryptionKeyFileName); -#endif // Currently only one parameter is supported, "bucket" for (const auto& [name, value] : params) { if (name == "bucket") { diff --git a/fdbclient/DatabaseContext.h b/fdbclient/DatabaseContext.h index 2b1383cb98..751e523af5 100644 --- a/fdbclient/DatabaseContext.h +++ b/fdbclient/DatabaseContext.h @@ -196,7 +196,7 @@ public: Reference getCommitProxies(bool useProvisionalProxies); Future> getCommitProxiesFuture(bool useProvisionalProxies); Reference getGrvProxies(bool useProvisionalProxies); - Future onProxiesChanged(); + Future onProxiesChanged() const; Future getHealthMetrics(bool detailed); // Returns the protocol version reported by the coordinator this client is connected to @@ -255,7 +255,7 @@ public: // private: explicit DatabaseContext(Reference>> connectionFile, Reference> clientDBInfo, - Reference>> coordinator, + Reference> const> coordinator, Future clientInfoMonitor, TaskPriority taskID, LocalityData const& clientLocality, @@ -307,7 +307,7 @@ public: // trust that the read version (possibly set manually by the application) is actually from the correct cluster. // Updated everytime we get a GRV response Version minAcceptableReadVersion = std::numeric_limits::max(); - void validateVersion(Version); + void validateVersion(Version) const; // Client status updater struct ClientStatusUpdater { @@ -399,7 +399,7 @@ public: Future connected; // An AsyncVar that reports the coordinator this DatabaseContext is interacting with - Reference>> coordinator; + Reference> const> coordinator; Reference>> statusClusterInterface; Future statusLeaderMon; @@ -428,7 +428,6 @@ public: static bool debugUseTags; static const std::vector debugTransactionTagChoices; - std::unordered_map> watchMap; // Adds or updates the specified (SS, TSS) pair in the TSS mapping (if not already present). // Requests to the storage server will be duplicated to the TSS. @@ -437,6 +436,9 @@ public: // Removes the storage server and its TSS pair from the TSS mapping (if present). // Requests to the storage server will no longer be duplicated to its pair TSS. void removeTssMapping(StorageServerInterface const& ssi); + +private: + std::unordered_map> watchMap; }; #endif diff --git a/fdbclient/GlobalConfig.actor.h b/fdbclient/GlobalConfig.actor.h index 2d63d8de60..444f1ab697 100644 --- a/fdbclient/GlobalConfig.actor.h +++ b/fdbclient/GlobalConfig.actor.h @@ -72,7 +72,7 @@ public: // to allow global configuration to run transactions on the latest // database. template - static void create(Database& cx, Reference> db, const ClientDBInfo* dbInfo) { + static void create(Database& cx, Reference const> db, const ClientDBInfo* dbInfo) { if (g_network->global(INetwork::enGlobalConfig) == nullptr) { auto config = new GlobalConfig{ cx }; g_network->setGlobal(INetwork::enGlobalConfig, config); diff --git a/fdbclient/MonitorLeader.h b/fdbclient/MonitorLeader.h index b9b195a9da..22ef1a5300 100644 --- a/fdbclient/MonitorLeader.h +++ b/fdbclient/MonitorLeader.h @@ -49,7 +49,7 @@ struct ClientData { OpenDatabaseRequest getRequest(); - ClientData() : clientInfo(new AsyncVar>(CachedSerialization())) {} + ClientData() : clientInfo(makeReference>>()) {} }; struct MonitorLeaderInfo { diff --git a/fdbclient/NativeAPI.actor.cpp b/fdbclient/NativeAPI.actor.cpp index 305ec8fc63..20d2b9343d 100644 --- a/fdbclient/NativeAPI.actor.cpp +++ b/fdbclient/NativeAPI.actor.cpp @@ -285,7 +285,7 @@ std::string unprintable(std::string const& val) { return s; } -void DatabaseContext::validateVersion(Version version) { +void DatabaseContext::validateVersion(Version version) const { // Version could be 0 if the INITIALIZE_NEW_DATABASE option is set. In that case, it is illegal to perform any // reads. We throw client_invalid_operation because the caller didn't directly set the version, so the // version_invalid error might be confusing. @@ -650,7 +650,7 @@ ACTOR static Future clientStatusUpdateActor(DatabaseContext* cx) { } } -ACTOR static Future monitorProxiesChange(Reference> clientDBInfo, +ACTOR static Future monitorProxiesChange(Reference const> clientDBInfo, AsyncTrigger* triggerVar) { state vector curCommitProxies; state vector curGrvProxies; @@ -1085,7 +1085,7 @@ Future HealthMetricsRangeImpl::getRange(ReadYourWritesTransaction* DatabaseContext::DatabaseContext(Reference>> connectionFile, Reference> clientInfo, - Reference>> coordinator, + Reference> const> coordinator, Future clientInfoMonitor, TaskPriority taskID, LocalityData const& clientLocality, @@ -1482,7 +1482,7 @@ void DatabaseContext::invalidateCache(const KeyRangeRef& keys) { locationCache.insert(KeyRangeRef(begin, end), Reference()); } -Future DatabaseContext::onProxiesChanged() { +Future DatabaseContext::onProxiesChanged() const { return this->proxiesChangeTrigger.onTrigger(); } @@ -1759,7 +1759,8 @@ Database Database::createDatabase(Reference connFile, } auto database = Database(db); - GlobalConfig::create(database, clientInfo, std::addressof(clientInfo->get())); + GlobalConfig::create( + database, Reference const>(clientInfo), std::addressof(clientInfo->get())); return database; } @@ -5760,7 +5761,7 @@ ACTOR Future> getCoordinatorProtocolFromConnectPacket( NetworkAddress coordinatorAddress, Optional expectedVersion) { - state Reference>> protocolVersion = + state Reference> const> protocolVersion = FlowTransport::transport().getPeerProtocolAsyncVar(coordinatorAddress); loop { @@ -5785,7 +5786,7 @@ ACTOR Future> getCoordinatorProtocolFromConnectPacket( // Returns the protocol version reported by the given coordinator // If an expected version is given, the future won't return until the protocol version is different than expected ACTOR Future getClusterProtocolImpl( - Reference>> coordinator, + Reference> const> coordinator, Optional expectedVersion) { state bool needToConnect = true; diff --git a/fdbrpc/FlowTransport.actor.cpp b/fdbrpc/FlowTransport.actor.cpp index 8a6b32df56..d44483da12 100644 --- a/fdbrpc/FlowTransport.actor.cpp +++ b/fdbrpc/FlowTransport.actor.cpp @@ -1698,7 +1698,7 @@ Reference> FlowTransport::getDegraded() { // // Note that this function does not establish a connection to the peer. In order to obtain a peer's protocol // version, some other mechanism should be used to connect to that peer. -Reference>> FlowTransport::getPeerProtocolAsyncVar(NetworkAddress addr) { +Reference> const> FlowTransport::getPeerProtocolAsyncVar(NetworkAddress addr) { return self->peers.at(addr)->protocolVersion; } @@ -1723,4 +1723,4 @@ void FlowTransport::createInstance(bool isClient, uint64_t transportId) { HealthMonitor* FlowTransport::healthMonitor() { return &self->healthMonitor; -} \ No newline at end of file +} diff --git a/fdbrpc/FlowTransport.h b/fdbrpc/FlowTransport.h index 7ae82b8ef7..0ba5a605aa 100644 --- a/fdbrpc/FlowTransport.h +++ b/fdbrpc/FlowTransport.h @@ -252,7 +252,7 @@ public: // // Note that this function does not establish a connection to the peer. In order to obtain a peer's protocol // version, some other mechanism should be used to connect to that peer. - Reference>> getPeerProtocolAsyncVar(NetworkAddress addr); + Reference> const> getPeerProtocolAsyncVar(NetworkAddress addr); static FlowTransport& transport() { return *static_cast((void*)g_network->global(INetwork::enFlowTransport)); diff --git a/fdbserver/BackupWorker.actor.cpp b/fdbserver/BackupWorker.actor.cpp index ff783dafca..d6bd6a0ebb 100644 --- a/fdbserver/BackupWorker.actor.cpp +++ b/fdbserver/BackupWorker.actor.cpp @@ -237,7 +237,7 @@ struct BackupData { CounterCollection cc; Future logger; - explicit BackupData(UID id, Reference> db, const InitializeBackupRequest& req) + explicit BackupData(UID id, Reference const> db, const InitializeBackupRequest& req) : myId(id), tag(req.routerTag), totalTags(req.totalTags), startVersion(req.startVersion), endVersion(req.endVersion), recruitedEpoch(req.recruitedEpoch), backupEpoch(req.backupEpoch), minKnownCommittedVersion(invalidVersion), savedVersion(req.startVersion - 1), popVersion(req.startVersion - 1), @@ -987,7 +987,7 @@ ACTOR Future monitorBackupKeyOrPullData(BackupData* self, bool keyPresent) } } -ACTOR Future checkRemoved(Reference> db, LogEpoch recoveryCount, BackupData* self) { +ACTOR Future checkRemoved(Reference const> db, LogEpoch recoveryCount, BackupData* self) { loop { bool isDisplaced = db->get().recoveryCount > recoveryCount && db->get().recoveryState != RecoveryState::UNINITIALIZED; @@ -1033,7 +1033,7 @@ ACTOR static Future monitorWorkerPause(BackupData* self) { ACTOR Future backupWorker(BackupInterface interf, InitializeBackupRequest req, - Reference> db) { + Reference const> db) { state BackupData self(interf.id(), db, req); state PromiseStream> addActor; state Future error = actorCollection(addActor.getFuture()); diff --git a/fdbserver/CommitProxyServer.actor.cpp b/fdbserver/CommitProxyServer.actor.cpp index 337a24e956..71316dcdb7 100644 --- a/fdbserver/CommitProxyServer.actor.cpp +++ b/fdbserver/CommitProxyServer.actor.cpp @@ -1596,7 +1596,7 @@ ACTOR static Future rejoinServer(CommitProxyInterface proxy, ProxyCommitDa } } -ACTOR Future ddMetricsRequestServer(CommitProxyInterface proxy, Reference> db) { +ACTOR Future ddMetricsRequestServer(CommitProxyInterface proxy, Reference const> db) { loop { choose { when(state GetDDMetricsRequest req = waitNext(proxy.getDDMetrics.getFuture())) { @@ -1754,7 +1754,8 @@ ACTOR Future proxySnapCreate(ProxySnapRequest snapReq, ProxyCommitData* co return Void(); } -ACTOR Future proxyCheckSafeExclusion(Reference> db, ExclusionSafetyCheckRequest req) { +ACTOR Future proxyCheckSafeExclusion(Reference const> db, + ExclusionSafetyCheckRequest req) { TraceEvent("SafetyCheckCommitProxyBegin"); state ExclusionSafetyCheckReply reply(false); if (!db->get().distributor.present()) { @@ -1783,7 +1784,7 @@ ACTOR Future proxyCheckSafeExclusion(Reference> db, } ACTOR Future reportTxnTagCommitCost(UID myID, - Reference> db, + Reference const> db, UIDTransactionTagMap* ssTrTagCommitCost) { state Future nextRequestTimer = Never(); state Future nextReply = Never(); @@ -1818,7 +1819,7 @@ ACTOR Future reportTxnTagCommitCost(UID myID, ACTOR Future commitProxyServerCore(CommitProxyInterface proxy, MasterInterface master, - Reference> db, + Reference const> db, LogEpoch epoch, Version recoveryTransactionVersion, bool firstProxy, @@ -2037,7 +2038,7 @@ ACTOR Future commitProxyServerCore(CommitProxyInterface proxy, } } -ACTOR Future checkRemoved(Reference> db, +ACTOR Future checkRemoved(Reference const> db, uint64_t recoveryCount, CommitProxyInterface myInterface) { loop { @@ -2051,7 +2052,7 @@ ACTOR Future checkRemoved(Reference> db, ACTOR Future commitProxyServer(CommitProxyInterface proxy, InitializeCommitProxyRequest req, - Reference> db, + Reference const> db, std::string whitelistBinPaths) { try { state Future core = commitProxyServerCore(proxy, diff --git a/fdbserver/ConfigDatabaseUnitTests.actor.cpp b/fdbserver/ConfigDatabaseUnitTests.actor.cpp index a99f4560cd..39d7be0ac1 100644 --- a/fdbserver/ConfigDatabaseUnitTests.actor.cpp +++ b/fdbserver/ConfigDatabaseUnitTests.actor.cpp @@ -126,7 +126,7 @@ class ReadFromLocalConfigEnvironment { UID id; std::string dataDir; LocalConfiguration localConfiguration; - Reference const> cbfi; + Reference const> cbfi; Future consumer; ACTOR static Future checkEventually(LocalConfiguration const* localConfiguration, @@ -168,7 +168,7 @@ public: return setup(); } - void connectToBroadcaster(Reference const> const& cbfi) { + void connectToBroadcaster(Reference const> const& cbfi) { ASSERT(!this->cbfi); this->cbfi = cbfi; consumer = localConfiguration.consume(cbfi); @@ -228,7 +228,7 @@ class BroadcasterToLocalConfigEnvironment { ACTOR static Future setup(BroadcasterToLocalConfigEnvironment* self) { wait(self->readFrom.setup()); - self->readFrom.connectToBroadcaster(IDependentAsyncVar::create(self->cbfi)); + self->readFrom.connectToBroadcaster(IAsyncListener::create(self->cbfi)); self->broadcastServer = self->broadcaster.serve(self->cbfi->get()); return Void(); } @@ -364,7 +364,7 @@ class TransactionToLocalConfigEnvironment { ACTOR static Future setup(TransactionToLocalConfigEnvironment* self) { wait(self->readFrom.setup()); - self->readFrom.connectToBroadcaster(IDependentAsyncVar::create(self->cbfi)); + self->readFrom.connectToBroadcaster(IAsyncListener::create(self->cbfi)); self->broadcastServer = self->broadcaster.serve(self->cbfi->get()); return Void(); } diff --git a/fdbserver/DataDistribution.actor.cpp b/fdbserver/DataDistribution.actor.cpp index 175b1518a2..ac6538596e 100644 --- a/fdbserver/DataDistribution.actor.cpp +++ b/fdbserver/DataDistribution.actor.cpp @@ -5218,7 +5218,7 @@ ACTOR Future initializeStorage(DDTeamCollection* self, } ACTOR Future storageRecruiter(DDTeamCollection* self, - Reference> db, + Reference const> db, const DDEnabledState* ddEnabledState) { state Future fCandidateWorker; state RecruitStorageRequest lastRequest; @@ -5490,7 +5490,7 @@ ACTOR Future serverGetTeamRequests(TeamCollectionInterface tci, DDTeamColl } } -ACTOR Future remoteRecovered(Reference> db) { +ACTOR Future remoteRecovered(Reference const> db) { TraceEvent("DDTrackerStarting"); while (db->get().recoveryState < RecoveryState::ALL_LOGS_RECRUITED) { TraceEvent("DDTrackerStarting").detail("RecoveryState", (int)db->get().recoveryState); @@ -5516,8 +5516,8 @@ ACTOR Future monitorHealthyTeams(DDTeamCollection* self) { ACTOR Future dataDistributionTeamCollection(Reference teamCollection, Reference initData, TeamCollectionInterface tci, - Reference> db, - const DDEnabledState* ddEnabledState) { + Reference const> db, + DDEnabledState const* ddEnabledState) { state DDTeamCollection* self = teamCollection.getPtr(); state Future loggingTrigger = Void(); state PromiseStream serverRemoved; @@ -5744,16 +5744,16 @@ ACTOR Future pollMoveKeysLock(Database cx, MoveKeysLock lock, const DDEnab } struct DataDistributorData : NonCopyable, ReferenceCounted { - Reference> dbInfo; + Reference const> dbInfo; UID ddId; PromiseStream> addActor; DDTeamCollection* teamCollection; - DataDistributorData(Reference> const& db, UID id) + DataDistributorData(Reference const> const& db, UID id) : dbInfo(db), ddId(id), teamCollection(nullptr) {} }; -ACTOR Future monitorBatchLimitedTime(Reference> db, double* lastLimited) { +ACTOR Future monitorBatchLimitedTime(Reference const> db, double* lastLimited) { loop { wait(delay(SERVER_KNOBS->METRIC_UPDATE_RATE)); @@ -6121,7 +6121,7 @@ static std::set const& normalDataDistributorErrors() { return s; } -ACTOR Future ddSnapCreateCore(DistributorSnapRequest snapReq, Reference> db) { +ACTOR Future ddSnapCreateCore(DistributorSnapRequest snapReq, Reference const> db) { state Database cx = openDBOnServer(db, TaskPriority::DefaultDelay, LockAware::True); state ReadYourWritesTransaction tr(cx); loop { @@ -6265,7 +6265,7 @@ ACTOR Future ddSnapCreateCore(DistributorSnapRequest snapReq, Reference ddSnapCreate(DistributorSnapRequest snapReq, - Reference> db, + Reference const> db, DDEnabledState* ddEnabledState) { state Future dbInfoChange = db->onChange(); if (!ddEnabledState->setDDEnabled(false, snapReq.snapUID)) { @@ -6459,7 +6459,7 @@ ACTOR Future ddGetMetrics(GetDataDistributorMetricsRequest req, return Void(); } -ACTOR Future dataDistributor(DataDistributorInterface di, Reference> db) { +ACTOR Future dataDistributor(DataDistributorInterface di, Reference const> db) { state Reference self(new DataDistributorData(db, di.id())); state Future collection = actorCollection(self->addActor.getFuture()); state PromiseStream getShardMetricsList; diff --git a/fdbserver/GrvProxyServer.actor.cpp b/fdbserver/GrvProxyServer.actor.cpp index 415ae9a310..3ebf2931e6 100644 --- a/fdbserver/GrvProxyServer.actor.cpp +++ b/fdbserver/GrvProxyServer.actor.cpp @@ -222,7 +222,7 @@ struct GrvProxyData { Reference logSystem; Database cx; - Reference> db; + Reference const> db; Optional latencyBandConfig; double lastStartCommit; @@ -251,7 +251,7 @@ struct GrvProxyData { GrvProxyData(UID dbgid, MasterInterface master, RequestStream getConsistentReadVersion, - Reference> db) + Reference const> db) : dbgid(dbgid), stats(dbgid), master(master), getConsistentReadVersion(getConsistentReadVersion), cx(openDBOnServer(db, TaskPriority::DefaultEndpoint, LockAware::True)), db(db), lastStartCommit(0), lastCommitLatency(SERVER_KNOBS->REQUIRED_MIN_RECOVERY_DURATION), updateCommitRequests(0), lastCommitTime(0), @@ -275,7 +275,7 @@ ACTOR Future healthMetricsRequestServer(GrvProxyInterface grvProxy, // Get transaction rate info from RateKeeper. ACTOR Future getRate(UID myID, - Reference> db, + Reference const> db, int64_t* inTransactionCount, int64_t* inBatchTransactionCount, GrvTransactionRateInfo* transactionRateInfo, @@ -375,7 +375,7 @@ void dropRequestFromQueue(Deque* queue, GrvProxyStats* st } // Put a GetReadVersion request into the queue corresponding to its priority. -ACTOR Future queueGetReadVersionRequests(Reference> db, +ACTOR Future queueGetReadVersionRequests(Reference const> db, SpannedDeque* systemQueue, SpannedDeque* defaultQueue, SpannedDeque* batchQueue, @@ -634,7 +634,7 @@ ACTOR Future sendGrvReplies(Future replyFuture, return Void(); } -ACTOR Future monitorDDMetricsChanges(int64_t* midShardSize, Reference> db) { +ACTOR Future monitorDDMetricsChanges(int64_t* midShardSize, Reference const> db) { state Future nextRequestTimer = Never(); state Future nextReply = Never(); @@ -680,7 +680,7 @@ ACTOR Future monitorDDMetricsChanges(int64_t* midShardSize, Reference transactionStarter(GrvProxyInterface proxy, - Reference> db, + Reference const> db, PromiseStream> addActor, GrvProxyData* grvProxyData, GetHealthMetricsReply* healthMetricsReply, @@ -898,7 +898,7 @@ ACTOR static Future transactionStarter(GrvProxyInterface proxy, ACTOR Future grvProxyServerCore(GrvProxyInterface proxy, MasterInterface master, - Reference> db) { + Reference const> db) { state GrvProxyData grvProxyData(proxy.id(), master, proxy.getConsistentReadVersion, db); state PromiseStream> addActor; @@ -945,7 +945,7 @@ ACTOR Future grvProxyServerCore(GrvProxyInterface proxy, } } -ACTOR Future checkRemoved(Reference> db, +ACTOR Future checkRemoved(Reference const> db, uint64_t recoveryCount, GrvProxyInterface myInterface) { loop { @@ -959,7 +959,7 @@ ACTOR Future checkRemoved(Reference> db, ACTOR Future grvProxyServer(GrvProxyInterface proxy, InitializeGrvProxyRequest req, - Reference> db) { + Reference const> db) { try { state Future core = grvProxyServerCore(proxy, req.master, db); wait(core || checkRemoved(db, req.recoveryCount, proxy)); diff --git a/fdbserver/IPager.h b/fdbserver/IPager.h index 22ee4b94ca..c5558613bc 100644 --- a/fdbserver/IPager.h +++ b/fdbserver/IPager.h @@ -42,31 +42,31 @@ typedef uint32_t QueueID; // Pager Events enum class PagerEvents { CacheLookup = 0, CacheHit, CacheMiss, PageWrite, MAXEVENTS }; -static const std::string PagerEventsStrings[] = { "Lookup", "Hit", "Miss", "Write", "Unknown" }; +static const char* const PagerEventsStrings[] = { "Lookup", "Hit", "Miss", "Write", "Unknown" }; // Reasons for page level events. enum class PagerEventReasons { PointRead = 0, RangeRead, RangePrefetch, Commit, LazyClear, MetaData, MAXEVENTREASONS }; -static const std::string PagerEventReasonsStrings[] = { "Get", "GetR", "GetRPF", "Commit", "LazyClr", "Meta", "Unknown" }; +static const char* const PagerEventReasonsStrings[] = { + "Get", "GetR", "GetRPF", "Commit", "LazyClr", "Meta", "Unknown" +}; static const int nonBtreeLevel = 0; -static const std::pair possibleEventReasonPairs[] = { +static const std::vector> possibleEventReasonPairs = { + { PagerEvents::CacheLookup, PagerEventReasons::Commit }, + { PagerEvents::CacheLookup, PagerEventReasons::LazyClear }, { PagerEvents::CacheLookup, PagerEventReasons::PointRead }, { PagerEvents::CacheLookup, PagerEventReasons::RangeRead }, - { PagerEvents::CacheLookup, PagerEventReasons::LazyClear }, - { PagerEvents::CacheLookup, PagerEventReasons::MetaData }, + { PagerEvents::CacheHit, PagerEventReasons::Commit }, + { PagerEvents::CacheHit, PagerEventReasons::LazyClear }, { PagerEvents::CacheHit, PagerEventReasons::PointRead }, { PagerEvents::CacheHit, PagerEventReasons::RangeRead }, - { PagerEvents::CacheHit, PagerEventReasons::LazyClear }, - { PagerEvents::CacheHit, PagerEventReasons::MetaData }, - { PagerEvents::CacheHit, PagerEventReasons::Commit }, + { PagerEvents::CacheMiss, PagerEventReasons::Commit }, + { PagerEvents::CacheMiss, PagerEventReasons::LazyClear }, { PagerEvents::CacheMiss, PagerEventReasons::PointRead }, { PagerEvents::CacheMiss, PagerEventReasons::RangeRead }, - { PagerEvents::CacheMiss, PagerEventReasons::LazyClear }, - { PagerEvents::CacheMiss, PagerEventReasons::MetaData }, - { PagerEvents::CacheMiss, PagerEventReasons::Commit }, - { PagerEvents::PageWrite, PagerEventReasons::MetaData }, + { PagerEvents::PageWrite, PagerEventReasons::Commit }, { PagerEvents::PageWrite, PagerEventReasons::LazyClear }, }; -static const std::pair L0PossibleEventReasonPairs[] = { +static const std::vector> L0PossibleEventReasonPairs = { { PagerEvents::CacheLookup, PagerEventReasons::RangePrefetch }, { PagerEvents::CacheLookup, PagerEventReasons::MetaData }, { PagerEvents::CacheHit, PagerEventReasons::RangePrefetch }, diff --git a/fdbserver/LocalConfiguration.actor.cpp b/fdbserver/LocalConfiguration.actor.cpp index b522ca1fca..238db7041e 100644 --- a/fdbserver/LocalConfiguration.actor.cpp +++ b/fdbserver/LocalConfiguration.actor.cpp @@ -309,9 +309,8 @@ class LocalConfigurationImpl { } } - ACTOR static Future consume( - LocalConfigurationImpl* self, - Reference const> broadcaster) { + ACTOR static Future consume(LocalConfigurationImpl* self, + Reference const> broadcaster) { ASSERT(self->initFuture.isValid() && self->initFuture.isReady()); loop { choose { @@ -371,7 +370,7 @@ public: return getKnobs().getTestKnobs(); } - Future consume(Reference const> const& broadcaster) { + Future consume(Reference const> const& broadcaster) { return consume(this, broadcaster); } @@ -453,7 +452,7 @@ TestKnobs const& LocalConfiguration::getTestKnobs() const { } Future LocalConfiguration::consume( - Reference const> const& broadcaster) { + Reference const> const& broadcaster) { return impl().consume(broadcaster); } diff --git a/fdbserver/LocalConfiguration.h b/fdbserver/LocalConfiguration.h index b2f73641c3..95e43eb20a 100644 --- a/fdbserver/LocalConfiguration.h +++ b/fdbserver/LocalConfiguration.h @@ -60,7 +60,7 @@ public: ClientKnobs const& getClientKnobs() const; ServerKnobs const& getServerKnobs() const; TestKnobs const& getTestKnobs() const; - Future consume(Reference const> const& broadcaster); + Future consume(Reference const> const& broadcaster); UID getID() const; public: // Testing diff --git a/fdbserver/LogRouter.actor.cpp b/fdbserver/LogRouter.actor.cpp index f91a138b9e..8527a7a01e 100644 --- a/fdbserver/LogRouter.actor.cpp +++ b/fdbserver/LogRouter.actor.cpp @@ -669,7 +669,7 @@ ACTOR Future logRouterPop(LogRouterData* self, TLogPopRequest req) { ACTOR Future logRouterCore(TLogInterface interf, InitializeLogRouterRequest req, - Reference> db) { + Reference const> db) { state LogRouterData logRouterData(interf.id(), req); state PromiseStream> addActor; state Future error = actorCollection(addActor.getFuture()); @@ -700,7 +700,7 @@ ACTOR Future logRouterCore(TLogInterface interf, } } -ACTOR Future checkRemoved(Reference> db, +ACTOR Future checkRemoved(Reference const> db, uint64_t recoveryCount, TLogInterface myInterface) { loop { @@ -717,7 +717,7 @@ ACTOR Future checkRemoved(Reference> db, ACTOR Future logRouter(TLogInterface interf, InitializeLogRouterRequest req, - Reference> db) { + Reference const> db) { try { TraceEvent("LogRouterStart", interf.id()) .detail("Start", req.startVersion) diff --git a/fdbserver/OldTLogServer_4_6.actor.cpp b/fdbserver/OldTLogServer_4_6.actor.cpp index 7c3c067b92..2b8c8b2cc5 100644 --- a/fdbserver/OldTLogServer_4_6.actor.cpp +++ b/fdbserver/OldTLogServer_4_6.actor.cpp @@ -291,7 +291,7 @@ struct TLogData : NonCopyable { AsyncVar largeDiskQueueCommitBytes; // becomes true when diskQueueCommitBytes is greater than MAX_QUEUE_COMMIT_BYTES - Reference> dbInfo; + Reference const> dbInfo; NotifiedVersion queueCommitEnd; Version queueCommitBegin; @@ -322,7 +322,7 @@ struct TLogData : NonCopyable { UID workerID, IKeyValueStore* persistentData, IDiskQueue* persistentQueue, - Reference> const& dbInfo) + Reference const> const& dbInfo) : dbgid(dbgid), workerID(workerID), instanceID(deterministicRandom()->randomUniqueID().first()), persistentData(persistentData), rawPersistentQueue(persistentQueue), persistentQueue(new TLogQueue(persistentQueue, dbgid)), dbInfo(dbInfo), queueCommitBegin(0), queueCommitEnd(0), @@ -1618,7 +1618,7 @@ ACTOR Future restorePersistentState(TLogData* self, LocalityData locality) ACTOR Future tLog(IKeyValueStore* persistentData, IDiskQueue* persistentQueue, - Reference> db, + Reference const> db, LocalityData locality, UID tlogId, UID workerID) { diff --git a/fdbserver/OldTLogServer_6_0.actor.cpp b/fdbserver/OldTLogServer_6_0.actor.cpp index abc7c37517..3de98dda8e 100644 --- a/fdbserver/OldTLogServer_6_0.actor.cpp +++ b/fdbserver/OldTLogServer_6_0.actor.cpp @@ -264,7 +264,7 @@ struct TLogData : NonCopyable { AsyncVar largeDiskQueueCommitBytes; // becomes true when diskQueueCommitBytes is greater than MAX_QUEUE_COMMIT_BYTES - Reference> dbInfo; + Reference const> dbInfo; Database cx; NotifiedVersion queueCommitEnd; @@ -302,7 +302,7 @@ struct TLogData : NonCopyable { UID workerID, IKeyValueStore* persistentData, IDiskQueue* persistentQueue, - Reference> dbInfo, + Reference const> dbInfo, Reference> degraded, std::string folder) : dbgid(dbgid), workerID(workerID), instanceID(deterministicRandom()->randomUniqueID().first()), @@ -2778,7 +2778,7 @@ ACTOR Future startSpillingInTenSeconds(TLogData* self, UID tlogId, Referen // New tLog (if !recoverFrom.size()) or restore from network ACTOR Future tLog(IKeyValueStore* persistentData, IDiskQueue* persistentQueue, - Reference> db, + Reference const> db, LocalityData locality, PromiseStream tlogRequests, UID tlogId, diff --git a/fdbserver/OldTLogServer_6_2.actor.cpp b/fdbserver/OldTLogServer_6_2.actor.cpp index 3d9749e0d5..64b3ce4008 100644 --- a/fdbserver/OldTLogServer_6_2.actor.cpp +++ b/fdbserver/OldTLogServer_6_2.actor.cpp @@ -327,7 +327,7 @@ struct TLogData : NonCopyable { AsyncVar largeDiskQueueCommitBytes; // becomes true when diskQueueCommitBytes is greater than MAX_QUEUE_COMMIT_BYTES - Reference> dbInfo; + Reference const> dbInfo; Database cx; NotifiedVersion queueCommitEnd; @@ -365,7 +365,7 @@ struct TLogData : NonCopyable { UID workerID, IKeyValueStore* persistentData, IDiskQueue* persistentQueue, - Reference> dbInfo, + Reference const> dbInfo, Reference> degraded, std::string folder) : dbgid(dbgid), workerID(workerID), instanceID(deterministicRandom()->randomUniqueID().first()), @@ -3271,7 +3271,7 @@ ACTOR Future startSpillingInTenSeconds(TLogData* self, UID tlogId, Referen // New tLog (if !recoverFrom.size()) or restore from network ACTOR Future tLog(IKeyValueStore* persistentData, IDiskQueue* persistentQueue, - Reference> db, + Reference const> db, LocalityData locality, PromiseStream tlogRequests, UID tlogId, diff --git a/fdbserver/ProxyCommitData.actor.h b/fdbserver/ProxyCommitData.actor.h index 43cff6bd47..29f7802bd7 100644 --- a/fdbserver/ProxyCommitData.actor.h +++ b/fdbserver/ProxyCommitData.actor.h @@ -161,7 +161,7 @@ struct ProxyCommitData { RequestStream getConsistentReadVersion; RequestStream commit; Database cx; - Reference> db; + Reference const> db; EventMetricHandle singleKeyMutationEvent; std::map> storageCache; @@ -239,7 +239,7 @@ struct ProxyCommitData { RequestStream getConsistentReadVersion, Version recoveryTransactionVersion, RequestStream commit, - Reference> db, + Reference const> db, bool firstProxy) : dbgid(dbgid), stats(dbgid, &version, &committedVersion, &commitBatchesMemBytesCount), master(master), logAdapter(nullptr), txnStateStore(nullptr), popRemoteTxs(false), committedVersion(recoveryTransactionVersion), diff --git a/fdbserver/QuietDatabase.actor.cpp b/fdbserver/QuietDatabase.actor.cpp index d633352088..f5e5443ca0 100644 --- a/fdbserver/QuietDatabase.actor.cpp +++ b/fdbserver/QuietDatabase.actor.cpp @@ -35,7 +35,7 @@ #include #include "flow/actorcompiler.h" // This must be the last #include. -ACTOR Future> getWorkers(Reference> dbInfo, int flags = 0) { +ACTOR Future> getWorkers(Reference const> dbInfo, int flags = 0) { loop { choose { when(vector w = wait(brokenPromiseToNever( @@ -48,7 +48,7 @@ ACTOR Future> getWorkers(Reference> } // Gets the WorkerInterface representing the Master server. -ACTOR Future getMasterWorker(Database cx, Reference> dbInfo) { +ACTOR Future getMasterWorker(Database cx, Reference const> dbInfo) { TraceEvent("GetMasterWorker").detail("Stage", "GettingWorkers"); loop { @@ -75,7 +75,7 @@ ACTOR Future getMasterWorker(Database cx, Reference getDataDistributorWorker(Database cx, Reference> dbInfo) { +ACTOR Future getDataDistributorWorker(Database cx, Reference const> dbInfo) { TraceEvent("GetDataDistributorWorker").detail("Stage", "GettingWorkers"); loop { @@ -118,7 +118,7 @@ ACTOR Future getDataInFlight(Database cx, WorkerInterface distributorWo } // Gets the number of bytes in flight from the data distributor. -ACTOR Future getDataInFlight(Database cx, Reference> dbInfo) { +ACTOR Future getDataInFlight(Database cx, Reference const> dbInfo) { WorkerInterface distributorInterf = wait(getDataDistributorWorker(cx, dbInfo)); int64_t dataInFlight = wait(getDataInFlight(cx, distributorInterf)); return dataInFlight; @@ -144,7 +144,7 @@ int64_t getPoppedVersionLag(const TraceEventFields& md) { return persistentDataDurableVersion - queuePoppedVersion; } -ACTOR Future> getCoordWorkers(Database cx, Reference> dbInfo) { +ACTOR Future> getCoordWorkers(Database cx, Reference const> dbInfo) { state std::vector workers = wait(getWorkers(dbInfo)); Optional coordinators = @@ -177,7 +177,8 @@ ACTOR Future> getCoordWorkers(Database cx, Reference> getTLogQueueInfo(Database cx, Reference> dbInfo) { +ACTOR Future> getTLogQueueInfo(Database cx, + Reference const> dbInfo) { TraceEvent("MaxTLogQueueSize").detail("Stage", "ContactingLogs"); state std::vector workers = wait(getWorkers(dbInfo)); @@ -245,7 +246,7 @@ ACTOR Future> getStorageServers(Database cx, bool } ACTOR Future> getStorageWorkers(Database cx, - Reference> dbInfo, + Reference const> dbInfo, bool localOnly) { state std::vector servers = wait(getStorageServers(cx)); state std::map workersMap; @@ -335,7 +336,7 @@ ACTOR Future getStorageMetricsTimeout(UID storage, WorkerInter }; // Gets the maximum size of all the storage server queues -ACTOR Future getMaxStorageServerQueueSize(Database cx, Reference> dbInfo) { +ACTOR Future getMaxStorageServerQueueSize(Database cx, Reference const> dbInfo) { TraceEvent("MaxStorageServerQueueSize").detail("Stage", "ContactingStorageServers"); Future> serversFuture = getStorageServers(cx); @@ -399,7 +400,7 @@ ACTOR Future getDataDistributionQueueSize(Database cx, // Gets the size of the data distribution queue. If reportInFlight is true, then data in flight is considered part of // the queue Convenience method that first finds the master worker from a zookeeper interface ACTOR Future getDataDistributionQueueSize(Database cx, - Reference> dbInfo, + Reference const> dbInfo, bool reportInFlight) { WorkerInterface distributorInterf = wait(getDataDistributorWorker(cx, dbInfo)); int64_t inQueue = wait(getDataDistributionQueueSize(cx, distributorInterf, reportInFlight)); @@ -516,7 +517,7 @@ ACTOR Future getTeamCollectionValid(Database cx, WorkerInterface dataDistr // Gets if the number of process and machine teams does not exceed the maximum allowed number of teams // Convenience method that first finds the master worker from a zookeeper interface -ACTOR Future getTeamCollectionValid(Database cx, Reference> dbInfo) { +ACTOR Future getTeamCollectionValid(Database cx, Reference const> dbInfo) { WorkerInterface dataDistributorWorker = wait(getDataDistributorWorker(cx, dbInfo)); bool valid = wait(getTeamCollectionValid(cx, dataDistributorWorker)); return valid; @@ -565,7 +566,9 @@ ACTOR Future getStorageServersRecruiting(Database cx, WorkerInterface dist } } -ACTOR Future repairDeadDatacenter(Database cx, Reference> dbInfo, std::string context) { +ACTOR Future repairDeadDatacenter(Database cx, + Reference const> dbInfo, + std::string context) { if (g_network->isSimulated() && g_simulator.usableRegions > 1) { bool primaryDead = g_simulator.datacenterDead(g_simulator.primaryDcId); bool remoteDead = g_simulator.datacenterDead(g_simulator.remoteDcId); @@ -601,7 +604,7 @@ ACTOR Future repairDeadDatacenter(Database cx, Reference reconfigureAfter(Database cx, double time, - Reference> dbInfo, + Reference const> dbInfo, std::string context) { wait(delay(time)); wait(repairDeadDatacenter(cx, dbInfo, context)); @@ -611,7 +614,7 @@ ACTOR Future reconfigureAfter(Database cx, // Waits until a database quiets down (no data in flight, small tlog queue, low SQ, no active data distribution). This // requires the database to be available and healthy in order to succeed. ACTOR Future waitForQuietDatabase(Database cx, - Reference> dbInfo, + Reference const> dbInfo, std::string phase, int64_t dataInFlightGate = 2e6, int64_t maxTLogQueueGate = 5e6, @@ -748,7 +751,7 @@ ACTOR Future waitForQuietDatabase(Database cx, } Future quietDatabase(Database const& cx, - Reference> const& dbInfo, + Reference const> const& dbInfo, std::string phase, int64_t dataInFlightGate, int64_t maxTLogQueueGate, diff --git a/fdbserver/QuietDatabase.h b/fdbserver/QuietDatabase.h index 37897e63fe..6a7ddc6d5e 100644 --- a/fdbserver/QuietDatabase.h +++ b/fdbserver/QuietDatabase.h @@ -28,25 +28,26 @@ #include "fdbserver/WorkerInterface.actor.h" #include "flow/actorcompiler.h" -Future getDataInFlight(Database const& cx, Reference> const&); +Future getDataInFlight(Database const& cx, Reference const> const&); Future> getTLogQueueInfo(Database const& cx, - Reference> const&); -Future getMaxStorageServerQueueSize(Database const& cx, Reference> const&); + Reference const> const&); +Future getMaxStorageServerQueueSize(Database const& cx, Reference const> const&); Future getDataDistributionQueueSize(Database const& cx, - Reference> const&, + Reference const> const&, bool const& reportInFlight); Future getTeamCollectionValid(Database const& cx, WorkerInterface const&); -Future getTeamCollectionValid(Database const& cx, Reference> const&); +Future getTeamCollectionValid(Database const& cx, Reference const> const&); Future> getStorageServers(Database const& cx, bool const& use_system_priority = false); -Future> getWorkers(Reference> const& dbInfo, int const& flags = 0); -Future getMasterWorker(Database const& cx, Reference> const& dbInfo); +Future> getWorkers(Reference const> const& dbInfo, int const& flags = 0); +Future getMasterWorker(Database const& cx, Reference const> const& dbInfo); Future repairDeadDatacenter(Database const& cx, - Reference> const& dbInfo, + Reference const> const& dbInfo, std::string const& context); Future> getStorageWorkers(Database const& cx, - Reference> const& dbInfo, + Reference const> const& dbInfo, bool const& localOnly); -Future> getCoordWorkers(Database const& cx, Reference> const& dbInfo); +Future> getCoordWorkers(Database const& cx, + Reference const> const& dbInfo); #include "flow/unactorcompiler.h" #endif diff --git a/fdbserver/Ratekeeper.actor.cpp b/fdbserver/Ratekeeper.actor.cpp index a49c4f476d..a13f9583be 100644 --- a/fdbserver/Ratekeeper.actor.cpp +++ b/fdbserver/Ratekeeper.actor.cpp @@ -1408,7 +1408,7 @@ ACTOR Future configurationMonitor(RatekeeperData* self) { } } -ACTOR Future ratekeeper(RatekeeperInterface rkInterf, Reference> dbInfo) { +ACTOR Future ratekeeper(RatekeeperInterface rkInterf, Reference const> dbInfo) { state RatekeeperData self(rkInterf.id(), openDBOnServer(dbInfo, TaskPriority::DefaultEndpoint, LockAware::True)); state Future timeout = Void(); state std::vector> tlogTrackers; diff --git a/fdbserver/Resolver.actor.cpp b/fdbserver/Resolver.actor.cpp index 351439a947..ff09179bf1 100644 --- a/fdbserver/Resolver.actor.cpp +++ b/fdbserver/Resolver.actor.cpp @@ -354,7 +354,7 @@ ACTOR Future resolverCore(ResolverInterface resolver, InitializeResolverRe } } -ACTOR Future checkRemoved(Reference> db, +ACTOR Future checkRemoved(Reference const> db, uint64_t recoveryCount, ResolverInterface myInterface) { loop { @@ -367,7 +367,7 @@ ACTOR Future checkRemoved(Reference> db, ACTOR Future resolver(ResolverInterface resolver, InitializeResolverRequest initReq, - Reference> db) { + Reference const> db) { try { state Future core = resolverCore(resolver, initReq); loop choose { diff --git a/fdbserver/StorageCache.actor.cpp b/fdbserver/StorageCache.actor.cpp index 3344636683..8f44f054d6 100644 --- a/fdbserver/StorageCache.actor.cpp +++ b/fdbserver/StorageCache.actor.cpp @@ -162,7 +162,7 @@ public: ProtocolVersion logProtocol; Reference logSystem; Key ck; // cacheKey - Reference> const& db; + Reference const> db; Database cx; StorageCacheUpdater* updater; @@ -238,7 +238,7 @@ public: } } counters; - explicit StorageCacheData(UID thisServerID, uint16_t index, Reference> const& db) + explicit StorageCacheData(UID thisServerID, uint16_t index, Reference const> const& db) : /*versionedData(FastAllocPTree{std::make_shared(0)}), */ thisServerID(thisServerID), index(index), logProtocol(0), db(db), cacheRangeChangeCounter(0), lastTLogVersion(0), lastVersionWithData(0), peekVersion(0), compactionInProgress(Void()), @@ -2165,7 +2165,9 @@ ACTOR Future watchInterface(StorageCacheData* self, StorageServerInterface } } -ACTOR Future storageCacheServer(StorageServerInterface ssi, uint16_t id, Reference> db) { +ACTOR Future storageCacheServer(StorageServerInterface ssi, + uint16_t id, + Reference const> db) { state StorageCacheData self(ssi.id(), id, db); state ActorCollection actors(false); state Future dbInfoChange = Void(); diff --git a/fdbserver/TLogServer.actor.cpp b/fdbserver/TLogServer.actor.cpp index 3e50dc9132..799c8a52da 100644 --- a/fdbserver/TLogServer.actor.cpp +++ b/fdbserver/TLogServer.actor.cpp @@ -329,7 +329,7 @@ struct TLogData : NonCopyable { AsyncVar largeDiskQueueCommitBytes; // becomes true when diskQueueCommitBytes is greater than MAX_QUEUE_COMMIT_BYTES - Reference> dbInfo; + Reference const> dbInfo; Database cx; NotifiedVersion queueCommitEnd; @@ -373,7 +373,7 @@ struct TLogData : NonCopyable { UID workerID, IKeyValueStore* persistentData, IDiskQueue* persistentQueue, - Reference> dbInfo, + Reference const> dbInfo, Reference> degraded, std::string folder) : dbgid(dbgid), workerID(workerID), instanceID(deterministicRandom()->randomUniqueID().first()), @@ -3337,7 +3337,7 @@ ACTOR Future startSpillingInTenSeconds(TLogData* self, UID tlogId, Referen // New tLog (if !recoverFrom.size()) or restore from network ACTOR Future tLog(IKeyValueStore* persistentData, IDiskQueue* persistentQueue, - Reference> db, + Reference const> db, LocalityData locality, PromiseStream tlogRequests, UID tlogId, diff --git a/fdbserver/VersionedBTree.actor.cpp b/fdbserver/VersionedBTree.actor.cpp index d42a65d3c9..ecd6540ffb 100644 --- a/fdbserver/VersionedBTree.actor.cpp +++ b/fdbserver/VersionedBTree.actor.cpp @@ -1447,85 +1447,59 @@ int nextPowerOf2(uint32_t x) { } struct RedwoodMetrics { - static constexpr int btreeLevels = 5; + static constexpr unsigned int btreeLevels = 5; static int maxRecordCount; struct EventReasonsArray { unsigned int eventReasons[(size_t)PagerEvents::MAXEVENTS][(size_t)PagerEventReasons::MAXEVENTREASONS]; EventReasonsArray() { clear(); } - void clear() { - for (size_t i = 0; i < (size_t)PagerEvents::MAXEVENTS; i++) { - for (size_t j = 0; j < (size_t)PagerEventReasons::MAXEVENTREASONS; j++) { - eventReasons[i][j] = 0; - } - } - } + void clear() { memset(eventReasons, 0, sizeof(eventReasons)); } + void addEventReason(PagerEvents event, PagerEventReasons reason) { eventReasons[(size_t)event][(size_t)reason] += 1; } - const unsigned int& getEventReason(PagerEvents event, PagerEventReasons reason) { + + unsigned int getEventReason(PagerEvents event, PagerEventReasons reason) const { return eventReasons[(size_t)event][(size_t)reason]; } - std::string ouputSummary(int currLevel) { - std::string result = ""; - PagerEvents prevEvent = PagerEvents::MAXEVENTS; - if (currLevel == 0) { - for (const auto& ER : L0PossibleEventReasonPairs) { - if (prevEvent != ER.first) { - result += "\n"; - result += PagerEventsStrings[(size_t)ER.first]; - result += "\n\t"; - prevEvent = ER.first; - } - std::string num = std::to_string(eventReasons[(size_t)ER.first][(size_t)ER.second]); - result += PagerEventReasonsStrings[(size_t)ER.second]; - result.append(16 - PagerEventReasonsStrings[(size_t)ER.second].length(), ' '); - result.append(8 - num.length(), ' '); - result += num; - result.append(13, ' '); - } - } else { - for (const auto& ER : possibleEventReasonPairs) { - if (prevEvent != ER.first) { - result += "\n"; - result += PagerEventsStrings[(size_t)ER.first]; - result += "\n\t"; - prevEvent = ER.first; - } - std::string num = std::to_string(eventReasons[(size_t)ER.first][(size_t)ER.second]); - result += PagerEventReasonsStrings[(size_t)ER.second]; - result.append(16 - PagerEventReasonsStrings[(size_t)ER.second].length(), ' '); - result.append(8 - num.length(), ' '); - result += num; - result.append(13, ' '); + std::string toString(int level, double elapsed) const { + std::string result; + + const auto& pairs = (level == 0 ? L0PossibleEventReasonPairs : possibleEventReasonPairs); + PagerEvents prevEvent = pairs.front().first; + std::string lineStart = (level == 0) ? "" : "\t"; + + for (const auto& p : pairs) { + if (p.first != prevEvent) { + result += "\n"; + result += lineStart; } + + std::string name = + format("%s%s", PagerEventsStrings[(int)p.first], PagerEventReasonsStrings[(int)p.second]); + int count = getEventReason(p.first, p.second); + result += format("%-15s %8u %8u/s ", name.c_str(), count, int(count / elapsed)); + + prevEvent = p.first; } + return result; } - void reportTrace(TraceEvent* t, int h) { - if (h == 0) { - for (const auto& ER : L0PossibleEventReasonPairs) { - t->detail( - format("L%d%s", - h, - (PagerEventsStrings[(size_t)ER.first] + PagerEventReasonsStrings[(size_t)ER.second]) - .c_str()), - eventReasons[(size_t)ER.first][(size_t)ER.second]); - } - } else { - for (const auto& ER : possibleEventReasonPairs) { - t->detail( - format("L%d%s", - h, - (PagerEventsStrings[(size_t)ER.first] + PagerEventReasonsStrings[(size_t)ER.second]) - .c_str()), - eventReasons[(size_t)ER.first][(size_t)ER.second]); - } + + void toTraceEvent(TraceEvent* t, int level) const { + const auto& pairs = (level == 0 ? L0PossibleEventReasonPairs : possibleEventReasonPairs); + for (const auto& p : pairs) { + std::string name = + format(level == 0 ? "" : "L%d", level) + + format("%s%s", PagerEventsStrings[(int)p.first], PagerEventReasonsStrings[(int)p.second]); + int count = getEventReason(p.first, p.second); + t->detail(std::move(name), count); } } }; + // Metrics by level struct Level { struct Counters { @@ -1542,7 +1516,7 @@ struct RedwoodMetrics { unsigned int lazyClearFreeExt; unsigned int forceUpdate; unsigned int detachChild; - EventReasonsArray eventReasons; + EventReasonsArray events; }; Counters metrics; Reference buildFillPctSketch; @@ -1554,37 +1528,33 @@ struct RedwoodMetrics { Level() { clear(); } - void clear(int levelCounter = -1) { + void clear(int level = 0) { metrics = {}; - if (!buildFillPctSketch.isValid() || - buildFillPctSketch->name() != ("buildFillPct:" + std::to_string(levelCounter))) { - std::string levelCounterStr = std::to_string(levelCounter); - buildFillPctSketch = Histogram::getHistogram( - LiteralStringRef("buildFillPct"), StringRef(levelCounterStr), Histogram::Unit::percentage); - modifyFillPctSketch = Histogram::getHistogram( - LiteralStringRef("modifyFillPct"), StringRef(levelCounterStr), Histogram::Unit::percentage); - buildStoredPctSketch = Histogram::getHistogram( - LiteralStringRef("buildStoredPct"), StringRef(levelCounterStr), Histogram::Unit::percentage); - modifyStoredPctSketch = Histogram::getHistogram( - LiteralStringRef("modifyStoredPct"), StringRef(levelCounterStr), Histogram::Unit::percentage); - buildItemCountSketch = Histogram::getHistogram(LiteralStringRef("buildItemCount"), - StringRef(levelCounterStr), - Histogram::Unit::count, - 0, - maxRecordCount); - modifyItemCountSketch = Histogram::getHistogram(LiteralStringRef("modifyItemCount"), - StringRef(levelCounterStr), - Histogram::Unit::count, - 0, - maxRecordCount); + + if (level > 0) { + if (!buildFillPctSketch) { + std::string levelString = format("L%d", level); + buildFillPctSketch = Histogram::getHistogram( + LiteralStringRef("buildFillPct"), levelString, Histogram::Unit::percentage); + modifyFillPctSketch = Histogram::getHistogram( + LiteralStringRef("modifyFillPct"), levelString, Histogram::Unit::percentage); + buildStoredPctSketch = Histogram::getHistogram( + LiteralStringRef("buildStoredPct"), levelString, Histogram::Unit::percentage); + modifyStoredPctSketch = Histogram::getHistogram( + LiteralStringRef("modifyStoredPct"), levelString, Histogram::Unit::percentage); + buildItemCountSketch = Histogram::getHistogram( + LiteralStringRef("buildItemCount"), levelString, Histogram::Unit::count, 0, maxRecordCount); + modifyItemCountSketch = Histogram::getHistogram( + LiteralStringRef("modifyItemCount"), levelString, Histogram::Unit::count, 0, maxRecordCount); + } + + buildFillPctSketch->clear(); + modifyFillPctSketch->clear(); + buildStoredPctSketch->clear(); + modifyStoredPctSketch->clear(); + buildItemCountSketch->clear(); + modifyItemCountSketch->clear(); } - metrics.eventReasons.clear(); - buildFillPctSketch->clear(); - modifyFillPctSketch->clear(); - buildStoredPctSketch->clear(); - modifyStoredPctSketch->clear(); - buildItemCountSketch->clear(); - modifyItemCountSketch->clear(); } }; @@ -1652,18 +1622,17 @@ struct RedwoodMetrics { } Level& level(unsigned int level) { - static Level outOfBound; // Valid levels are from 0 - btreeLevels - if (level < 0 || level > btreeLevels) { - return outOfBound; - } - return levels[level]; + // Level 0 is for operations that are not BTree level specific, as many of the metrics are the same + // Level 0 - btreeLevels correspond to BTree node height, however heights above btreeLevels are combined + // into the level at btreeLevels + return levels[std::min(level, btreeLevels)]; } void updateMaxRecordCount(int maxRecords) { if (maxRecordCount != maxRecords) { maxRecordCount = maxRecords; - for (int i = 0; i < btreeLevels + 1; ++i) { + for (int i = 1; i <= btreeLevels; ++i) { auto& level = levels[i]; level.buildItemCountSketch->updateUpperBound(maxRecordCount); level.modifyItemCountSketch->updateUpperBound(maxRecordCount); @@ -1699,8 +1668,8 @@ struct RedwoodMetrics { { "", 0 }, { "PagerRemapFree", metric.pagerRemapFree }, { "PagerRemapCopy", metric.pagerRemapCopy }, - { "PagerRemapSkip", metric.pagerRemapSkip } }; - GetHistogramRegistry().logReport(); + { "PagerRemapSkip", metric.pagerRemapSkip }, + { "", 0 } }; double elapsed = now() - startTime; @@ -1711,6 +1680,7 @@ struct RedwoodMetrics { e->detail(m.first, m.second); } } + levels[0].metrics.events.toTraceEvent(e, 0); } if (s != nullptr) { @@ -1721,10 +1691,10 @@ struct RedwoodMetrics { *s += format("%-15s %-8u %8" PRId64 "/s ", m.first, m.second, int64_t(m.second / elapsed)); } } - *s += "\n"; + *s += levels[0].metrics.events.toString(0, elapsed); } - for (int i = 0; i < btreeLevels + 1; ++i) { + for (int i = 1; i < btreeLevels + 1; ++i) { auto& metric = levels[i].metrics; std::pair metrics[] = { @@ -1754,7 +1724,7 @@ struct RedwoodMetrics { e->detail(format("L%d%s", i + 1, m.first + (c == '-' ? 1 : 0)), m.second); } } - metric.eventReasons.reportTrace(e, i); + metric.events.toTraceEvent(e, i); } if (s != nullptr) { @@ -1774,8 +1744,7 @@ struct RedwoodMetrics { *s += format("%-15s %8u %8u/s ", name, m.second, rate ? int(m.second / elapsed) : 0); } } - *s += '\n'; - *s += metric.eventReasons.ouputSummary(i); + *s += metric.events.toString(i, elapsed); } } } @@ -2516,7 +2485,7 @@ public: state PriorityMultiLock::Lock lock = wait(self->ioLock.lock(header ? ioMaxPriority : ioMinPriority)); ++g_redwoodMetrics.metric.pagerDiskWrite; - g_redwoodMetrics.level(level).metrics.eventReasons.addEventReason(PagerEvents::PageWrite, reason); + g_redwoodMetrics.level(level).metrics.events.addEventReason(PagerEvents::PageWrite, reason); if (self->memoryOnly) { return Void(); @@ -2792,7 +2761,7 @@ public: bool noHit) override { // Use cached page if present, without triggering a cache hit. // Otherwise, read the page and return it but don't add it to the cache - auto& eventReasons = g_redwoodMetrics.level(level).metrics.eventReasons; + auto& eventReasons = g_redwoodMetrics.level(level).metrics.events; eventReasons.addEventReason(PagerEvents::CacheLookup, reason); if (!cacheable) { debug_printf("DWALPager(%s) op=readUncached %s\n", filename.c_str(), toString(pageID).c_str()); @@ -2950,7 +2919,7 @@ public: Future> readExtent(LogicalPageID pageID) override { debug_printf("DWALPager(%s) op=readExtent %s\n", filename.c_str(), toString(pageID).c_str()); PageCacheEntry* pCacheEntry = extentCache.getIfExists(pageID); - auto& eventReasons = g_redwoodMetrics.level(0).metrics.eventReasons; + auto& eventReasons = g_redwoodMetrics.level(0).metrics.events; if (pCacheEntry != nullptr) { eventReasons.addEventReason(PagerEvents::CacheLookup, PagerEventReasons::MetaData); debug_printf("DWALPager(%s) Cache Entry exists for %s\n", filename.c_str(), toString(pageID).c_str()); @@ -5412,8 +5381,8 @@ private: } debug_printf("readPage() op=readComplete %s @%" PRId64 " \n", toString(id).c_str(), snapshot->getVersion()); - const BTreePage* pTreePage = (const BTreePage*)page->begin(); - auto& metrics = g_redwoodMetrics.level(pTreePage->height).metrics; + const BTreePage* btPage = (const BTreePage*)page->begin(); + auto& metrics = g_redwoodMetrics.level(btPage->height).metrics; metrics.pageRead += 1; metrics.pageReadExt += (id.size() - 1); @@ -6674,7 +6643,7 @@ public: std::string toString() const { std::string r = format("{ptr=%p reason=%s %s ", this, - PagerEventsStrings[(int)reason].c_str(), + PagerEventsStrings[(int)reason], ::toString(pager->getVersion()).c_str()); for (int i = 0; i < path.size(); ++i) { std::string id = ""; @@ -6738,11 +6707,11 @@ public: // Initialize or reinitialize cursor Future init(VersionedBTree* btree_in, - PagerEventReasons reason, + PagerEventReasons reason_in, Reference pager_in, BTreePageIDRef root) { btree = btree_in; - reason = reason; + reason = reason_in; pager = pager_in; path.clear(); path.reserve(6); @@ -9056,7 +9025,7 @@ TEST_CASE("/redwood/correctness/btree") { mutationBytesThisCommit >= mutationBytesTargetThisCommit) { // Wait for previous commit to finish wait(commit); - printf("Last commit complete. Next commit %d bytes, %" PRId64 " bytes committed so far.", + printf("Commit complete. Next commit %d bytes, %" PRId64 " bytes committed so far.", mutationBytesThisCommit, mutationBytes.get() - mutationBytesThisCommit); printf(" Stats: Insert %.2f MB/s ClearedKeys %.2f MB/s Total %.2f\n", diff --git a/fdbserver/WorkerInterface.actor.h b/fdbserver/WorkerInterface.actor.h index 682801a90c..ddd62d6aff 100644 --- a/fdbserver/WorkerInterface.actor.h +++ b/fdbserver/WorkerInterface.actor.h @@ -831,7 +831,7 @@ ACTOR Future traceRole(Role role, UID roleId); struct ServerDBInfo; -class Database openDBOnServer(Reference> const& db, +class Database openDBOnServer(Reference const> const& db, TaskPriority taskID = TaskPriority::DefaultEndpoint, LockAware = LockAware::False, EnableLocalityLoadBalance = EnableLocalityLoadBalance::True); @@ -868,32 +868,32 @@ ACTOR Future storageServer(IKeyValueStore* persistentData, Tag seedTag, Version tssSeedVersion, ReplyPromise recruitReply, - Reference> db, + Reference const> db, std::string folder); ACTOR Future storageServer( IKeyValueStore* persistentData, StorageServerInterface ssi, - Reference> db, + Reference const> db, std::string folder, Promise recovered, Reference connFile); // changes pssi->id() to be the recovered ID); // changes pssi->id() to be the recovered ID ACTOR Future masterServer(MasterInterface mi, - Reference> db, + Reference const> db, Reference>> ccInterface, ServerCoordinators serverCoordinators, LifetimeToken lifetime, bool forceRecovery); ACTOR Future commitProxyServer(CommitProxyInterface proxy, InitializeCommitProxyRequest req, - Reference> db, + Reference const> db, std::string whitelistBinPaths); ACTOR Future grvProxyServer(GrvProxyInterface proxy, InitializeGrvProxyRequest req, - Reference> db); + Reference const> db); ACTOR Future tLog(IKeyValueStore* persistentData, IDiskQueue* persistentQueue, - Reference> db, + Reference const> db, LocalityData locality, PromiseStream tlogRequests, UID tlogId, @@ -906,14 +906,18 @@ ACTOR Future tLog(IKeyValueStore* persistentData, Reference> activeSharedTLog); ACTOR Future resolver(ResolverInterface resolver, InitializeResolverRequest initReq, - Reference> db); + Reference const> db); ACTOR Future logRouter(TLogInterface interf, InitializeLogRouterRequest req, - Reference> db); -ACTOR Future dataDistributor(DataDistributorInterface ddi, Reference> db); -ACTOR Future ratekeeper(RatekeeperInterface rki, Reference> db); -ACTOR Future storageCacheServer(StorageServerInterface interf, uint16_t id, Reference> db); -ACTOR Future backupWorker(BackupInterface bi, InitializeBackupRequest req, Reference> db); + Reference const> db); +ACTOR Future dataDistributor(DataDistributorInterface ddi, Reference const> db); +ACTOR Future ratekeeper(RatekeeperInterface rki, Reference const> db); +ACTOR Future storageCacheServer(StorageServerInterface interf, + uint16_t id, + Reference const> db); +ACTOR Future backupWorker(BackupInterface bi, + InitializeBackupRequest req, + Reference const> db); void registerThreadForProfiling(); void updateCpuProfiler(ProfilerRequest req); @@ -921,7 +925,7 @@ void updateCpuProfiler(ProfilerRequest req); namespace oldTLog_4_6 { ACTOR Future tLog(IKeyValueStore* persistentData, IDiskQueue* persistentQueue, - Reference> db, + Reference const> db, LocalityData locality, UID tlogId, UID workerID); @@ -929,7 +933,7 @@ ACTOR Future tLog(IKeyValueStore* persistentData, namespace oldTLog_6_0 { ACTOR Future tLog(IKeyValueStore* persistentData, IDiskQueue* persistentQueue, - Reference> db, + Reference const> db, LocalityData locality, PromiseStream tlogRequests, UID tlogId, @@ -944,7 +948,7 @@ ACTOR Future tLog(IKeyValueStore* persistentData, namespace oldTLog_6_2 { ACTOR Future tLog(IKeyValueStore* persistentData, IDiskQueue* persistentQueue, - Reference> db, + Reference const> db, LocalityData locality, PromiseStream tlogRequests, UID tlogId, diff --git a/fdbserver/masterserver.actor.cpp b/fdbserver/masterserver.actor.cpp index 167491efb5..f3fb50bfff 100644 --- a/fdbserver/masterserver.actor.cpp +++ b/fdbserver/masterserver.actor.cpp @@ -228,7 +228,7 @@ struct MasterData : NonCopyable, ReferenceCounted { ReusableCoordinatedState cstate; Promise cstateUpdated; - Reference> dbInfo; + Reference const> dbInfo; int64_t registrationCount; // Number of different MasterRegistrationRequests sent to clusterController RecoveryState recoveryState; @@ -255,7 +255,7 @@ struct MasterData : NonCopyable, ReferenceCounted { Future logger; - MasterData(Reference> const& dbInfo, + MasterData(Reference const> const& dbInfo, MasterInterface const& myInterface, ServerCoordinators const& coordinators, ClusterControllerFullInterface const& clusterController, @@ -1978,7 +1978,7 @@ ACTOR Future masterCore(Reference self) { } ACTOR Future masterServer(MasterInterface mi, - Reference> db, + Reference const> db, Reference>> ccInterface, ServerCoordinators coordinators, LifetimeToken lifetime, diff --git a/fdbserver/storageserver.actor.cpp b/fdbserver/storageserver.actor.cpp index 3f454e79f4..ffd0ee4f40 100644 --- a/fdbserver/storageserver.actor.cpp +++ b/fdbserver/storageserver.actor.cpp @@ -614,7 +614,7 @@ public: bool tssInQuarantine; Key sk; - Reference> db; + Reference const> db; Database cx; ActorCollection actors; @@ -806,7 +806,7 @@ public: } counters; StorageServer(IKeyValueStore* storage, - Reference> const& db, + Reference const> const& db, StorageServerInterface const& ssi) : fetchKeysHistograms(), instanceID(deterministicRandom()->randomUniqueID().first()), storage(this, storage), db(db), actors(false), lastTLogVersion(0), lastVersionWithData(0), restoredVersion(0), @@ -5134,7 +5134,7 @@ ACTOR Future storageServer(IKeyValueStore* persistentData, Tag seedTag, Version tssSeedVersion, ReplyPromise recruitReply, - Reference> db, + Reference const> db, std::string folder) { state StorageServer self(persistentData, db, ssi); if (ssi.isTss()) { @@ -5328,7 +5328,7 @@ ACTOR Future replaceTSSInterface(StorageServer* self, StorageServerInterfa // for recovering an existing storage server ACTOR Future storageServer(IKeyValueStore* persistentData, StorageServerInterface ssi, - Reference> db, + Reference const> db, std::string folder, Promise recovered, Reference connFile) { diff --git a/fdbserver/worker.actor.cpp b/fdbserver/worker.actor.cpp index 561e212bd9..426dd51b59 100644 --- a/fdbserver/worker.actor.cpp +++ b/fdbserver/worker.actor.cpp @@ -122,7 +122,7 @@ ACTOR Future> broadcastDBInfoRequest(UpdateServerDBInfoReq return notUpdated; } -ACTOR static Future extractClientInfo(Reference> db, +ACTOR static Future extractClientInfo(Reference const> db, Reference> info) { state std::vector lastCommitProxyUIDs; state std::vector lastCommitProxies; @@ -136,7 +136,7 @@ ACTOR static Future extractClientInfo(Reference> db } } -Database openDBOnServer(Reference> const& db, +Database openDBOnServer(Reference const> const& db, TaskPriority taskID, LockAware lockAware, EnableLocalityLoadBalance enableLocalityLoadBalance) { @@ -502,15 +502,15 @@ std::vector getDiskStores(std::string folder) { // Register the worker interf to cluster controller (cc) and // re-register the worker when key roles interface, e.g., cc, dd, ratekeeper, change. -ACTOR Future registrationClient(Reference>> ccInterface, +ACTOR Future registrationClient(Reference> const> ccInterface, WorkerInterface interf, Reference> asyncPriorityInfo, ProcessClass initialClass, - Reference>> ddInterf, - Reference>> rkInterf, - Reference> degraded, + Reference> const> ddInterf, + Reference> const> rkInterf, + Reference const> degraded, Reference connFile, - Reference>> issues) { + Reference> const> issues) { // Keeps the cluster controller (as it may be re-elected) informed that this worker exists // The cluster controller uses waitFailureClient to find out if we die, and returns from registrationReply // (requiring us to re-register) The registration request piggybacks optional distributor interface if it exists. @@ -2304,10 +2304,9 @@ ACTOR Future fdbd(Reference connFile, auto dbInfo = makeReference>(); if (useConfigDB != UseConfigDB::DISABLED) { - actors.push_back( - reportErrors(localConfig.consume(IDependentAsyncVar::create( - dbInfo, [](auto const& info) { return info.configBroadcaster; })), - "LocalConfiguration")); + actors.push_back(reportErrors(localConfig.consume(IAsyncListener::create( + dbInfo, [](auto const& info) { return info.configBroadcaster; })), + "LocalConfiguration")); } actors.push_back(reportErrors(monitorAndWriteCCPriorityInfo(fitnessFilePath, asyncPriorityInfo), "MonitorAndWriteCCPriorityInfo")); diff --git a/fdbserver/workloads/UnitTests.actor.cpp b/fdbserver/workloads/UnitTests.actor.cpp index 482a89ff13..6bf9eb39dc 100644 --- a/fdbserver/workloads/UnitTests.actor.cpp +++ b/fdbserver/workloads/UnitTests.actor.cpp @@ -55,7 +55,7 @@ struct UnitTestWorkload : TestWorkload { if (g_network->isSimulated()) { testParams.setDataDir(getOption(options, "dataDir"_sr, "simfdb/unittests/"_sr).toString()); } else { - testParams.setDataDir(getOption(options, "dataDir"_sr, "/private/tmp/"_sr).toString()); + testParams.setDataDir(getOption(options, "dataDir"_sr, "unittests/"_sr).toString()); } cleanupAfterTests = getOption(options, "cleanupAfterTests"_sr, true); diff --git a/fdbserver/workloads/workloads.actor.h b/fdbserver/workloads/workloads.actor.h index d47b981409..5829b19fa6 100644 --- a/fdbserver/workloads/workloads.actor.h +++ b/fdbserver/workloads/workloads.actor.h @@ -222,7 +222,7 @@ double testKeyToDouble(const KeyRef& p, const KeyRef& prefix); ACTOR Future databaseWarmer(Database cx); Future quietDatabase(Database const& cx, - Reference> const&, + Reference const> const&, std::string phase, int64_t dataInFlightGate = 2e6, int64_t maxTLogQueueGate = 5e6, diff --git a/flow/genericactors.actor.cpp b/flow/genericactors.actor.cpp index b199175af7..9b7f906713 100644 --- a/flow/genericactors.actor.cpp +++ b/flow/genericactors.actor.cpp @@ -158,7 +158,7 @@ ACTOR Future testPublisher(Reference> input) { return Void(); } -ACTOR Future testSubscriber(Reference> output, Optional expected) { +ACTOR Future testSubscriber(Reference> output, Optional expected) { loop { wait(output->onChange()); ASSERT(expected.present()); @@ -170,12 +170,12 @@ ACTOR Future testSubscriber(Reference> output, Opt } // namespace -TEST_CASE("/flow/genericactors/DependentAsyncVar") { +TEST_CASE("/flow/genericactors/AsyncListener") { auto input = makeReference>(); state Future subscriber1 = - testSubscriber(IDependentAsyncVar::create(input, [](auto const& var) { return var.changed; }), 100); + testSubscriber(IAsyncListener::create(input, [](auto const& var) { return var.changed; }), 100); state Future subscriber2 = - testSubscriber(IDependentAsyncVar::create(input, [](auto const& var) { return var.unchanged; }), {}); + testSubscriber(IAsyncListener::create(input, [](auto const& var) { return var.unchanged; }), {}); wait(subscriber1 && testPublisher(input)); ASSERT(!subscriber2.isReady()); return Void(); diff --git a/flow/genericactors.actor.h b/flow/genericactors.actor.h index d04a0478f8..30794d9791 100644 --- a/flow/genericactors.actor.h +++ b/flow/genericactors.actor.h @@ -690,7 +690,7 @@ public: AsyncTrigger() {} AsyncTrigger(AsyncTrigger&& at) : v(std::move(at.v)) {} void operator=(AsyncTrigger&& at) { v = std::move(at.v); } - Future onTrigger() { return v.onChange(); } + Future onTrigger() const { return v.onChange(); } void trigger() { v.trigger(); } private: @@ -700,7 +700,7 @@ private: // Binds an AsyncTrigger object to an AsyncVar, so when the AsyncVar changes // the AsyncTrigger is triggered. ACTOR template -void forward(Reference> from, AsyncTrigger* to) { +void forward(Reference const> from, AsyncTrigger* to) { loop { wait(from->onChange()); to->trigger(); @@ -1957,25 +1957,28 @@ Future operator>>(Future const& lhs, Future const& rhs) { } /* - * IDependentAsyncVar is similar to AsyncVar, but it decouples the input and output, so the translation unit + * IAsyncListener is similar to AsyncVar, but it decouples the input and output, so the translation unit * responsible for handling the output does not need to have knowledge of how the output is generated */ template -class IDependentAsyncVar : public ReferenceCounted> { +class IAsyncListener : public ReferenceCounted> { public: - virtual ~IDependentAsyncVar() = default; + virtual ~IAsyncListener() = default; virtual Output const& get() const = 0; virtual Future onChange() const = 0; template - static Reference create(Reference> const& input, F const& f); - static Reference create(Reference> const& output); + static Reference create(Reference> const& input, F const& f); + static Reference create(Reference> const& output); }; +namespace IAsyncListenerImpl { + template -class DependentAsyncVar final : public IDependentAsyncVar { - Reference> output; +class AsyncListener final : public IAsyncListener { + // Order matters here, output must outlive monitorActor + AsyncVar output; Future monitorActor; - ACTOR static Future monitor(Reference> input, Reference> output, F f) { + ACTOR static Future monitor(Reference const> input, AsyncVar* output, F f) { loop { wait(input->onChange()); output->set(f(input->get())); @@ -1983,23 +1986,24 @@ class DependentAsyncVar final : public IDependentAsyncVar { } public: - DependentAsyncVar(Reference> const& input, F const& f) - : output(makeReference>(f(input->get()))), monitorActor(monitor(input, output, f)) {} - Output const& get() const override { return output->get(); } - Future onChange() const override { return output->onChange(); } + AsyncListener(Reference const> const& input, F const& f) + : output(f(input->get())), monitorActor(monitor(input, &output, f)) {} + Output const& get() const override { return output.get(); } + Future onChange() const override { return output.onChange(); } }; +} // namespace IAsyncListenerImpl + template template -Reference> IDependentAsyncVar::create(Reference> const& input, - F const& f) { - return makeReference>(input, f); +Reference> IAsyncListener::create(Reference> const& input, F const& f) { + return makeReference>(input, f); } template -Reference> IDependentAsyncVar::create(Reference> const& input) { +Reference> IAsyncListener::create(Reference> const& input) { auto identity = [](const auto& x) { return x; }; - return makeReference>(input, identity); + return makeReference>(input, identity); } // A weak reference type to wrap a future Reference object.