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

tpcv messaging

This commit is contained in:
Dan Lambright 2021-06-29 16:41:08 -04:00
parent 0b0093b8be
commit a107dd655e
8 changed files with 114 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
fdbclient/ServerKnobs.cpp
View File

@ -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<int>(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

1
fdbclient/ServerKnobs.h
View File

@ -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

8
fdbclient/VersionVector.h
View File

@ -43,6 +43,14 @@ struct VersionVector {
maxVersion = version;
}
void setVersions(const std::set<Tag>& 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();

29
fdbserver/CommitProxyServer.actor.cpp
View File

@ -480,6 +480,9 @@ struct CommitBatchContext {
double commitStartTime;
std::set<uint16_t> locSet; // the set of tlog locations written to in the mutation.
std::set<Tag> tagSet; // the set of tags written to in the mutation.
CommitBatchContext(ProxyCommitData*, const std::vector<CommitTransactionRequest>*, const int);
void setupTraceBatch();
@ -873,6 +876,15 @@ ACTOR Future<Void> applyMetadataToCommittedTransactions(CommitBatchContext* self
return Void();
}
// Message the sequencer to obtain the previous commit version for each storage server's tag
ACTOR Future<Void> 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<Void> assignMutationsToStorageServers(CommitBatchContext* self) {
@ -949,7 +961,9 @@ ACTOR Future<Void> 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<Void> 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<Void> 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<Void> 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<std::set<Tag>> 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()) {

16
fdbserver/LogSystem.h
View File

@ -995,6 +995,22 @@ struct LogPushData : NonCopyable {
writtenLocations.clear();
}
// copy next_message_tags into given set
void saveTags(std::set<Tag>& 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<uint16_t>& locSet, std::set<Tag>& 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();

36
fdbserver/MasterInterface.h
View File

@ -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<struct GetRawCommittedVersionRequest> getLiveCommittedVersion;
// Report a proxy's committed version.
RequestStream<struct ReportRawCommittedVersionRequest> reportLiveCommittedVersion;
RequestStream<struct GetTlogPrevCommitVersionRequest> getTlogPrevCommitVersion;
NetworkAddress address() const { return changeCoordinators.getEndpoint().getPrimaryAddress(); }
NetworkAddressList addresses() const { return changeCoordinators.getEndpoint().addresses; }
@ -66,6 +68,8 @@ struct MasterInterface {
RequestStream<struct GetRawCommittedVersionRequest>(waitFailure.getEndpoint().getAdjustedEndpoint(5));
reportLiveCommittedVersion = RequestStream<struct ReportRawCommittedVersionRequest>(
waitFailure.getEndpoint().getAdjustedEndpoint(6));
getTlogPrevCommitVersion =
RequestStream<struct GetTlogPrevCommitVersionRequest>(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<uint16_t, Version> tpcvMap;
GetTlogPrevCommitVersionReply() {}
template <class Ar>
void serialize(Ar& ar) {
serializer(ar, tpcvMap);
}
};
struct GetTlogPrevCommitVersionRequest {
constexpr static FileIdentifier file_identifier = 16683184;
std::set<uint16_t> locSet;
ReplyPromise<GetTlogPrevCommitVersionReply> reply;
GetTlogPrevCommitVersionRequest() {}
GetTlogPrevCommitVersionRequest(std::set<uint16_t>& locSet) : locSet(locSet) {}
template <class Ar>
void serialize(Ar& ar) {
serializer(ar, locSet, reply);
}
};
struct ReportRawCommittedVersionRequest {
constexpr static FileIdentifier file_identifier = 1853148;
Version version;
bool locked;
Optional<Value> metadataVersion;
Version minKnownCommittedVersion;
Optional<std::set<Tag>> tagSet;
ReplyPromise<Void> reply;
ReportRawCommittedVersionRequest() : version(invalidVersion), locked(false), minKnownCommittedVersion(0) {}
ReportRawCommittedVersionRequest(Version version,
bool locked,
Optional<Value> metadataVersion,
Version minKnownCommittedVersion)
Version minKnownCommittedVersion,
Optional<std::set<Tag>> tagSet = Optional<std::set<Tag>>())
: version(version), locked(locked), metadataVersion(metadataVersion),
minKnownCommittedVersion(minKnownCommittedVersion) {}
minKnownCommittedVersion(minKnownCommittedVersion), tagSet(tagSet) {}
template <class Ar>
void serialize(Ar& ar) {
serializer(ar, version, locked, metadataVersion, minKnownCommittedVersion, reply);
serializer(ar, version, locked, metadataVersion, minKnownCommittedVersion, tagSet, reply);
}
};

36
fdbserver/masterserver.actor.cpp
View File

@ -250,6 +250,9 @@ struct MasterData : NonCopyable, ReferenceCounted<MasterData> {
// Captures the latest commit version targeted for each storage server in the cluster.
VersionVector ssVersionVector;
// The previous commit versions per tlog
std::map<uint16_t, Version> tpcvMap;
CounterCollection cc;
Counter changeCoordinatorsRequests;
Counter getCommitVersionRequests;
@ -299,8 +302,8 @@ ACTOR Future<Void> newCommitProxies(Reference<MasterData> 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<Void> sendInitialCommitToResolvers(Reference<MasterData> 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<Future<ResolveTransactionBatchReply>> replies;
for (auto& r : self->resolvers) {
@ -976,9 +979,9 @@ ACTOR Future<Void> sendInitialCommitToResolvers(Reference<MasterData> 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<Void> serveLiveCommittedVersion(Reference<MasterData> 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<Void> serveLiveCommittedVersion(Reference<MasterData> 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);
}
}
}
}

1
flow/network.h
View File

@ -81,6 +81,7 @@ enum class TaskPriority {
GetConsistentReadVersion = 8500,
GetLiveCommittedVersionReply = 8490,
GetLiveCommittedVersion = 8480,
GetTlogPrevCommitVersion = 8400,
DefaultPromiseEndpoint = 8000,
DefaultOnMainThread = 7500,
DefaultDelay = 7010,