diff --git a/fdbclient/ServerKnobs.cpp b/fdbclient/ServerKnobs.cpp index af6a294aa4..2b5dcaa1d5 100644 --- a/fdbclient/ServerKnobs.cpp +++ b/fdbclient/ServerKnobs.cpp @@ -38,6 +38,7 @@ void ServerKnobs::initialize(Randomize _randomize, ClientKnobs* clientKnobs, IsS init( MAX_WRITE_TRANSACTION_LIFE_VERSIONS, 5 * VERSIONS_PER_SECOND ); if (randomize && BUGGIFY) MAX_WRITE_TRANSACTION_LIFE_VERSIONS=std::max(1, 1 * VERSIONS_PER_SECOND); init( MAX_COMMIT_BATCH_INTERVAL, 2.0 ); if( randomize && BUGGIFY ) MAX_COMMIT_BATCH_INTERVAL = 0.5; // Each commit proxy generates a CommitTransactionBatchRequest at least this often, so that versions always advance smoothly MAX_COMMIT_BATCH_INTERVAL = std::min(MAX_COMMIT_BATCH_INTERVAL, MAX_READ_TRANSACTION_LIFE_VERSIONS/double(2*VERSIONS_PER_SECOND)); // Ensure that the proxy commits 2 times every MAX_READ_TRANSACTION_LIFE_VERSIONS, otherwise the master will not give out versions fast enough + init( ENABLE_VERSION_VECTOR, true ); // TLogs init( TLOG_TIMEOUT, 0.4 ); //cannot buggify because of availability diff --git a/fdbclient/ServerKnobs.h b/fdbclient/ServerKnobs.h index 83ac569589..5e67f465ab 100644 --- a/fdbclient/ServerKnobs.h +++ b/fdbclient/ServerKnobs.h @@ -36,6 +36,7 @@ public: int64_t MAX_VERSIONS_IN_FLIGHT_FORCED; int64_t MAX_READ_TRANSACTION_LIFE_VERSIONS; int64_t MAX_WRITE_TRANSACTION_LIFE_VERSIONS; + bool ENABLE_VERSION_VECTOR; double MAX_COMMIT_BATCH_INTERVAL; // Each commit proxy generates a CommitTransactionBatchRequest at least this // often, so that versions always advance smoothly diff --git a/fdbclient/VersionVector.h b/fdbclient/VersionVector.h index 86c835b6e5..4293ec2e53 100644 --- a/fdbclient/VersionVector.h +++ b/fdbclient/VersionVector.h @@ -43,6 +43,14 @@ struct VersionVector { maxVersion = version; } + void setVersions(const std::set& tags, Version version) { + ASSERT(version > maxVersion); + for (auto& tag : tags) { + ASSERT(tag != invalidTag); + versions[tag] = version; + } + } + bool hasVersion(const Tag& tag) const { ASSERT(tag != invalidTag); return versions.find(tag) != versions.end(); diff --git a/fdbserver/CommitProxyServer.actor.cpp b/fdbserver/CommitProxyServer.actor.cpp index c88d4c054b..6927eefc20 100644 --- a/fdbserver/CommitProxyServer.actor.cpp +++ b/fdbserver/CommitProxyServer.actor.cpp @@ -480,6 +480,9 @@ struct CommitBatchContext { double commitStartTime; + std::set locSet; // the set of tlog locations written to in the mutation. + std::set tagSet; // the set of tags written to in the mutation. + CommitBatchContext(ProxyCommitData*, const std::vector*, const int); void setupTraceBatch(); @@ -873,6 +876,15 @@ ACTOR Future applyMetadataToCommittedTransactions(CommitBatchContext* self return Void(); } +// Message the sequencer to obtain the previous commit version for each storage server's tag +ACTOR Future getTPCV(CommitBatchContext* self) { + state ProxyCommitData* const pProxyCommitData = self->pProxyCommitData; + GetTlogPrevCommitVersionReply rep = wait(brokenPromiseToNever( + pProxyCommitData->master.getTlogPrevCommitVersion.getReply(GetTlogPrevCommitVersionRequest(self->locSet)))); + // TraceEvent("GetTlogPrevCommitVersionRequest"); + return Void(); +} + /// This second pass through committed transactions assigns the actual mutations to the appropriate storage servers' /// tags ACTOR Future assignMutationsToStorageServers(CommitBatchContext* self) { @@ -949,7 +961,9 @@ ACTOR Future assignMutationsToStorageServers(CommitBatchContext* self) { if (pProxyCommitData->cacheInfo[m.param1]) { self->toCommit.addTag(cacheTag); } + self->toCommit.saveTags(self->tagSet); self->toCommit.writeTypedMessage(m); + self->toCommit.saveLocations(self->locSet, self->tagSet); } else if (m.type == MutationRef::ClearRange) { KeyRangeRef clearRange(KeyRangeRef(m.param1, m.param2)); auto ranges = pProxyCommitData->keyInfo.intersectingRanges(clearRange); @@ -999,14 +1013,15 @@ ACTOR Future assignMutationsToStorageServers(CommitBatchContext* self) { .detail("Dbgid", pProxyCommitData->dbgid) .detail("To", allSources) .detail("Mutation", m); - self->toCommit.addTags(allSources); } if (pProxyCommitData->needsCacheTag(clearRange)) { self->toCommit.addTag(cacheTag); } + self->toCommit.saveTags(self->tagSet); self->toCommit.writeTypedMessage(m); + self->toCommit.saveLocations(self->locSet, self->tagSet); } else { UNREACHABLE(); } @@ -1096,6 +1111,11 @@ ACTOR Future postResolution(CommitBatchContext* self) { // Second pass wait(assignMutationsToStorageServers(self)); + // Obtain previous committed versions for each affected tlog from sequencer + if (SERVER_KNOBS->ENABLE_VERSION_VECTOR) { + wait(getTPCV(self)); + } + // Serialize and backup the mutations as a single mutation if ((pProxyCommitData->vecBackupKeys.size() > 1) && self->logRangeMutations.size()) { wait(addBackupMutations(pProxyCommitData, @@ -1278,11 +1298,16 @@ ACTOR Future reply(CommitBatchContext* self) { // self->committedVersion. TEST(pProxyCommitData->committedVersion.get() > self->commitVersion); // A later version was reported committed first if (self->commitVersion >= pProxyCommitData->committedVersion.get()) { + state Optional> tagSet; + if (SERVER_KNOBS->ENABLE_VERSION_VECTOR) { + tagSet = self->tagSet; + } wait(pProxyCommitData->master.reportLiveCommittedVersion.getReply( ReportRawCommittedVersionRequest(self->commitVersion, self->lockedAfter, self->metadataVersionAfter, - pProxyCommitData->minKnownCommittedVersion), + pProxyCommitData->minKnownCommittedVersion, + tagSet), TaskPriority::ProxyMasterVersionReply)); } if (self->commitVersion > pProxyCommitData->committedVersion.get()) { diff --git a/fdbserver/LogSystem.h b/fdbserver/LogSystem.h index da2fbcf5f2..3413e094d7 100644 --- a/fdbserver/LogSystem.h +++ b/fdbserver/LogSystem.h @@ -995,6 +995,22 @@ struct LogPushData : NonCopyable { writtenLocations.clear(); } + // copy next_message_tags into given set + void saveTags(std::set& tagSet) { + tagSet.insert(next_message_tags.begin(), next_message_tags.end()); + return; + } + + // store tlogs as represented by index + // also store in tag set all replicas + void saveLocations(std::set& locSet, std::set& tagSet) { + locSet.insert(msg_locations.begin(), msg_locations.end()); + for (auto loc: msg_locations) { + tagSet.insert(Tag(0, loc)); // TODO POST DEMO support DC other than primary + } + return; + } + void writeMessage(StringRef rawMessageWithoutLength, bool usePreviousLocations) { if (!usePreviousLocations) { prev_tags.clear(); diff --git a/fdbserver/MasterInterface.h b/fdbserver/MasterInterface.h index eea733b14c..950006c53b 100644 --- a/fdbserver/MasterInterface.h +++ b/fdbserver/MasterInterface.h @@ -26,6 +26,7 @@ #include "fdbclient/StorageServerInterface.h" #include "fdbclient/CommitTransaction.h" #include "fdbclient/DatabaseConfiguration.h" +#include "fdbclient/VersionVector.h" #include "fdbserver/TLogInterface.h" typedef uint64_t DBRecoveryCount; @@ -43,6 +44,7 @@ struct MasterInterface { RequestStream getLiveCommittedVersion; // Report a proxy's committed version. RequestStream reportLiveCommittedVersion; + RequestStream getTlogPrevCommitVersion; NetworkAddress address() const { return changeCoordinators.getEndpoint().getPrimaryAddress(); } NetworkAddressList addresses() const { return changeCoordinators.getEndpoint().addresses; } @@ -66,6 +68,8 @@ struct MasterInterface { RequestStream(waitFailure.getEndpoint().getAdjustedEndpoint(5)); reportLiveCommittedVersion = RequestStream( waitFailure.getEndpoint().getAdjustedEndpoint(6)); + getTlogPrevCommitVersion = + RequestStream(waitFailure.getEndpoint().getAdjustedEndpoint(7)); } } @@ -78,6 +82,7 @@ struct MasterInterface { streams.push_back(notifyBackupWorkerDone.getReceiver()); streams.push_back(getLiveCommittedVersion.getReceiver(TaskPriority::GetLiveCommittedVersion)); streams.push_back(reportLiveCommittedVersion.getReceiver(TaskPriority::ReportLiveCommittedVersion)); + streams.push_back(getTlogPrevCommitVersion.getReceiver(TaskPriority::GetTlogPrevCommitVersion)); FlowTransport::transport().addEndpoints(streams); } }; @@ -184,26 +189,49 @@ struct GetCommitVersionRequest { } }; +struct GetTlogPrevCommitVersionReply { + constexpr static FileIdentifier file_identifier = 16683183; + std::unordered_map tpcvMap; + GetTlogPrevCommitVersionReply() {} + template + void serialize(Ar& ar) { + serializer(ar, tpcvMap); + } +}; + +struct GetTlogPrevCommitVersionRequest { + constexpr static FileIdentifier file_identifier = 16683184; + std::set locSet; + ReplyPromise reply; + GetTlogPrevCommitVersionRequest() {} + GetTlogPrevCommitVersionRequest(std::set& locSet) : locSet(locSet) {} + template + void serialize(Ar& ar) { + serializer(ar, locSet, reply); + } +}; + struct ReportRawCommittedVersionRequest { constexpr static FileIdentifier file_identifier = 1853148; Version version; bool locked; Optional metadataVersion; Version minKnownCommittedVersion; - + Optional> tagSet; ReplyPromise reply; ReportRawCommittedVersionRequest() : version(invalidVersion), locked(false), minKnownCommittedVersion(0) {} ReportRawCommittedVersionRequest(Version version, bool locked, Optional metadataVersion, - Version minKnownCommittedVersion) + Version minKnownCommittedVersion, + Optional> tagSet = Optional>()) : version(version), locked(locked), metadataVersion(metadataVersion), - minKnownCommittedVersion(minKnownCommittedVersion) {} + minKnownCommittedVersion(minKnownCommittedVersion), tagSet(tagSet) {} template void serialize(Ar& ar) { - serializer(ar, version, locked, metadataVersion, minKnownCommittedVersion, reply); + serializer(ar, version, locked, metadataVersion, minKnownCommittedVersion, tagSet, reply); } }; diff --git a/fdbserver/masterserver.actor.cpp b/fdbserver/masterserver.actor.cpp index d9f8b4cc47..7aa6bf3478 100644 --- a/fdbserver/masterserver.actor.cpp +++ b/fdbserver/masterserver.actor.cpp @@ -250,6 +250,9 @@ struct MasterData : NonCopyable, ReferenceCounted { // Captures the latest commit version targeted for each storage server in the cluster. VersionVector ssVersionVector; + // The previous commit versions per tlog + std::map tpcvMap; + CounterCollection cc; Counter changeCoordinatorsRequests; Counter getCommitVersionRequests; @@ -299,8 +302,8 @@ ACTOR Future newCommitProxies(Reference self, RecruitFromConfi req.recoveryTransactionVersion = self->recoveryTransactionVersion; req.firstProxy = i == 0; TraceEvent("CommitProxyReplies", self->dbgid) - .detail("WorkerID", recr.commitProxies[i].id()) - .detail("FirstProxy", req.firstProxy ? "True" : "False"); + .detail("WorkerID", recr.commitProxies[i].id()) + .detail("FirstProxy", req.firstProxy ? "True" : "False"); initializationReplies.push_back( transformErrors(throwErrorOr(recr.commitProxies[i].commitProxy.getReplyUnlessFailedFor( req, SERVER_KNOBS->TLOG_TIMEOUT, SERVER_KNOBS->MASTER_FAILURE_SLOPE_DURING_RECOVERY)), @@ -960,9 +963,9 @@ ACTOR Future sendInitialCommitToResolvers(Reference self) { } wait(waitForAll(txnReplies)); TraceEvent("RecoveryInternal", self->dbgid) - .detail("StatusCode", RecoveryStatus::recovery_transaction) - .detail("Status", RecoveryStatus::names[RecoveryStatus::recovery_transaction]) - .detail("Step", "SentTxnStateStoreToCommitProxies"); + .detail("StatusCode", RecoveryStatus::recovery_transaction) + .detail("Status", RecoveryStatus::names[RecoveryStatus::recovery_transaction]) + .detail("Step", "SentTxnStateStoreToCommitProxies"); vector> replies; for (auto& r : self->resolvers) { @@ -976,9 +979,9 @@ ACTOR Future sendInitialCommitToResolvers(Reference self) { wait(waitForAll(replies)); TraceEvent("RecoveryInternal", self->dbgid) - .detail("StatusCode", RecoveryStatus::recovery_transaction) - .detail("Status", RecoveryStatus::names[RecoveryStatus::recovery_transaction]) - .detail("Step", "InitializedAllResolvers"); + .detail("StatusCode", RecoveryStatus::recovery_transaction) + .detail("Status", RecoveryStatus::names[RecoveryStatus::recovery_transaction]) + .detail("Step", "InitializedAllResolvers"); return Void(); } @@ -1243,6 +1246,12 @@ ACTOR Future serveLiveCommittedVersion(Reference self) { when(ReportRawCommittedVersionRequest req = waitNext(self->myInterface.reportLiveCommittedVersion.getFuture())) { self->minKnownCommittedVersion = std::max(self->minKnownCommittedVersion, req.minKnownCommittedVersion); + if (SERVER_KNOBS->ENABLE_VERSION_VECTOR && req.tagSet.present()) { + if (req.version > self->ssVersionVector.maxVersion) { + // TraceEvent("Received ReportRawCommittedVersionRequest").detail("Version",req.version); + self->ssVersionVector.setVersions(req.tagSet.get(), req.version); + } + } if (req.version > self->liveCommittedVersion) { self->liveCommittedVersion = req.version; self->databaseLocked = req.locked; @@ -1251,6 +1260,17 @@ ACTOR Future serveLiveCommittedVersion(Reference self) { ++self->reportLiveCommittedVersionRequests; req.reply.send(Void()); } + when(GetTlogPrevCommitVersionRequest req = + waitNext(self->myInterface.getTlogPrevCommitVersion.getFuture())) { + GetTlogPrevCommitVersionReply reply; + for (uint16_t loc : req.locSet) { + // TraceEvent("Received GetTlogPrevCommitVersionRequest").detail("Loc", loc); + if (self->tpcvMap.find(loc) != self->tpcvMap.end()) { + reply.tpcvMap[loc] = self->tpcvMap[loc]; + } + } + req.reply.send(reply); + } } } } diff --git a/flow/network.h b/flow/network.h index 00f430fb86..a4de7bde29 100644 --- a/flow/network.h +++ b/flow/network.h @@ -81,6 +81,7 @@ enum class TaskPriority { GetConsistentReadVersion = 8500, GetLiveCommittedVersionReply = 8490, GetLiveCommittedVersion = 8480, + GetTlogPrevCommitVersion = 8400, DefaultPromiseEndpoint = 8000, DefaultOnMainThread = 7500, DefaultDelay = 7010,