Track pseudo tags popping for individual IDs
For each log router ID, we track the popped version of each pseudo tag so that the popping only applied to the minimum of these versions. Also add more tracing for popping and epochs.
This commit is contained in:
parent
3509209d3f
commit
73824faf65
|
@ -197,9 +197,13 @@ ACTOR Future<Void> backupWorker(BackupInterface interf, InitializeBackupRequest
|
|||
Reference<ILogSystem> ls = ILogSystem::fromServerDBInfo(self.myId, db->get(), true);
|
||||
if (ls && ls->hasPseudoLocality(tagLocalityBackup)) {
|
||||
self.logSystem.set(ls);
|
||||
TraceEvent("BackupWorkerLogSystem", interf.id()).detail("HasBackupLocality", true);
|
||||
TraceEvent("BackupWorkerLogSystem", interf.id())
|
||||
.detail("HasBackupLocality", true)
|
||||
.detail("Tag", self.tag.toString());
|
||||
} else {
|
||||
TraceEvent("BackupWorkerLogSystem", interf.id()).detail("HasBackupLocality", false);
|
||||
TraceEvent("BackupWorkerLogSystem", interf.id())
|
||||
.detail("HasBackupLocality", false)
|
||||
.detail("Tag", self.tag.toString());
|
||||
}
|
||||
}
|
||||
when(HaltBackupRequest req = waitNext(interf.haltBackup.getFuture())) {
|
||||
|
|
|
@ -757,7 +757,10 @@ struct ILogSystem {
|
|||
|
||||
virtual bool hasPseudoLocality(int8_t locality) = 0;
|
||||
|
||||
virtual Version popPseudoLocalityTag(int8_t locality, Version upTo) = 0;
|
||||
// Returns the actual version to be popped from the log router tag for the given pseudo tag.
|
||||
// For instance, a pseudo tag (-8, 2) means the actual popping tag is (-2, 2). Assuming there
|
||||
// are multiple pseudo tags, the returned version is the min(all pseudo tags' "upTo" versions).
|
||||
virtual Version popPseudoLocalityTag(Tag tag, Version upTo) = 0;
|
||||
|
||||
virtual void setBackupWorkers(
|
||||
std::vector<Reference<AsyncVar<OptionalInterface<BackupInterface>>>> backupWorkers) = 0;
|
||||
|
|
|
@ -697,7 +697,7 @@ ACTOR Future<Void> tLogPopCore( TLogData* self, Tag inputTag, Version to, Refere
|
|||
int8_t tagLocality = inputTag.locality;
|
||||
if (isPseudoLocality(tagLocality)) {
|
||||
if (logData->logSystem->get().isValid()) {
|
||||
upTo = logData->logSystem->get()->popPseudoLocalityTag(tagLocality, to);
|
||||
upTo = logData->logSystem->get()->popPseudoLocalityTag(inputTag, to);
|
||||
tagLocality = tagLocalityLogRouter;
|
||||
} else {
|
||||
TraceEvent("TLogPopNoLogSystem", self->dbgid).detail("Locality", tagLocality).detail("Version", upTo);
|
||||
|
|
|
@ -985,7 +985,7 @@ ACTOR Future<Void> tLogPopCore( TLogData* self, Tag inputTag, Version to, Refere
|
|||
int8_t tagLocality = inputTag.locality;
|
||||
if (isPseudoLocality(tagLocality)) {
|
||||
if (logData->logSystem->get().isValid()) {
|
||||
upTo = logData->logSystem->get()->popPseudoLocalityTag(tagLocality, to);
|
||||
upTo = logData->logSystem->get()->popPseudoLocalityTag(inputTag, to);
|
||||
tagLocality = tagLocalityLogRouter;
|
||||
} else {
|
||||
// TODO: if this happens, need to save the popped version? or discard the pop?
|
||||
|
@ -995,6 +995,7 @@ ACTOR Future<Void> tLogPopCore( TLogData* self, Tag inputTag, Version to, Refere
|
|||
}
|
||||
state Tag tag(tagLocality, inputTag.id);
|
||||
auto tagData = logData->getTagData(tag);
|
||||
TraceEvent("TLogPop0", logData->logId).detail("Tag", tag.toString()).detail("To", to).detail("UpTo", upTo).detail("Popped", tagData ? tagData->popped : -1);
|
||||
if (!tagData) {
|
||||
tagData = logData->createTagData(tag, upTo, true, true, false);
|
||||
} else if (upTo > tagData->popped) {
|
||||
|
@ -1012,7 +1013,7 @@ ACTOR Future<Void> tLogPopCore( TLogData* self, Tag inputTag, Version to, Refere
|
|||
|
||||
if (upTo > logData->persistentDataDurableVersion)
|
||||
wait(tagData->eraseMessagesBefore(upTo, self, logData, TaskPriority::TLogPop));
|
||||
//TraceEvent("TLogPop", self->dbgid).detail("Tag", tag.toString()).detail("To", upTo);
|
||||
TraceEvent("TLogPop", logData->logId).detail("Tag", tag.toString()).detail("To", upTo);
|
||||
}
|
||||
return Void();
|
||||
}
|
||||
|
|
|
@ -181,7 +181,7 @@ struct TagPartitionedLogSystem : ILogSystem, ReferenceCounted<TagPartitionedLogS
|
|||
const LogEpoch epoch;
|
||||
|
||||
// new members
|
||||
std::map<int8_t, Version> pseudoLocalityPopVersion;
|
||||
std::map<Tag, Version> pseudoLocalityPopVersion;
|
||||
Future<Void> rejoins;
|
||||
Future<Void> recoveryComplete;
|
||||
Future<Void> remoteRecovery;
|
||||
|
@ -238,7 +238,9 @@ struct TagPartitionedLogSystem : ILogSystem, ReferenceCounted<TagPartitionedLogS
|
|||
void addPseudoLocality(int8_t locality) {
|
||||
ASSERT(locality < 0);
|
||||
pseudoLocalities.insert(locality);
|
||||
pseudoLocalityPopVersion[locality] = 0;
|
||||
for (uint16_t i = 0; i < logRouterTags; i++) {
|
||||
pseudoLocalityPopVersion[Tag(locality, i)] = 0;
|
||||
}
|
||||
}
|
||||
|
||||
Tag getPseudoPopTag(Tag tag, ProcessClass::ClassType type) override {
|
||||
|
@ -264,15 +266,16 @@ struct TagPartitionedLogSystem : ILogSystem, ReferenceCounted<TagPartitionedLogS
|
|||
|
||||
bool hasPseudoLocality(int8_t locality) override { return pseudoLocalities.count(locality) > 0; }
|
||||
|
||||
Version popPseudoLocalityTag(int8_t locality, Version upTo) override {
|
||||
ASSERT(hasPseudoLocality(locality));
|
||||
auto& localityVersion = pseudoLocalityPopVersion[locality];
|
||||
Version popPseudoLocalityTag(Tag tag, Version upTo) override {
|
||||
ASSERT(isPseudoLocality(tag.locality) && hasPseudoLocality(tag.locality));
|
||||
|
||||
Version& localityVersion = pseudoLocalityPopVersion[tag];
|
||||
localityVersion = std::max(localityVersion, upTo);
|
||||
Version minVersion = localityVersion;
|
||||
for (const auto& it : pseudoLocalityPopVersion) {
|
||||
minVersion = std::min(minVersion, it.second);
|
||||
for (const int8_t locality : pseudoLocalities) {
|
||||
minVersion = std::min(minVersion, pseudoLocalityPopVersion[Tag(locality, tag.id)]);
|
||||
}
|
||||
TraceEvent("Pop").detail("L", locality).detail("Version", upTo).detail("PopVersion", minVersion);
|
||||
TraceEvent("Pop", dbgid).detail("Tag", tag.toString()).detail("Version", upTo).detail("PopVersion", minVersion);
|
||||
return minVersion;
|
||||
}
|
||||
|
||||
|
@ -304,6 +307,9 @@ struct TagPartitionedLogSystem : ILogSystem, ReferenceCounted<TagPartitionedLogS
|
|||
|
||||
for (const auto& oldTlogConf : lsConf.oldTLogs) {
|
||||
logSystem->oldLogData.emplace_back(oldTlogConf);
|
||||
//TraceEvent("BWFromLSConf")
|
||||
// .detail("Epoch", logSystem->oldLogData.back().epoch)
|
||||
// .detail("Version", logSystem->oldLogData.back().epochEnd);
|
||||
}
|
||||
|
||||
logSystem->logSystemType = lsConf.logSystemType;
|
||||
|
@ -326,6 +332,9 @@ struct TagPartitionedLogSystem : ILogSystem, ReferenceCounted<TagPartitionedLogS
|
|||
|
||||
for (int i = 1; i < lsConf.oldTLogs.size(); i++ ) {
|
||||
logSystem->oldLogData.emplace_back(lsConf.oldTLogs[i]);
|
||||
//TraceEvent("BWFromOldLSConf")
|
||||
// .detail("Epoch", logSystem->oldLogData.back().epoch)
|
||||
// .detail("Version", logSystem->oldLogData.back().epochEnd);
|
||||
}
|
||||
}
|
||||
logSystem->logSystemType = lsConf.logSystemType;
|
||||
|
@ -360,6 +369,9 @@ struct TagPartitionedLogSystem : ILogSystem, ReferenceCounted<TagPartitionedLogS
|
|||
if(!recoveryComplete.isValid() || !recoveryComplete.isReady() || (repopulateRegionAntiQuorum == 0 && (!remoteRecoveryComplete.isValid() || !remoteRecoveryComplete.isReady()))) {
|
||||
for (const auto& oldData : oldLogData) {
|
||||
newState.oldTLogData.emplace_back(oldData);
|
||||
TraceEvent("BWToCore")
|
||||
.detail("Epoch", newState.oldTLogData.back().epoch)
|
||||
.detail("Version", newState.oldTLogData.back().epochEnd);
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -1235,6 +1247,9 @@ struct TagPartitionedLogSystem : ILogSystem, ReferenceCounted<TagPartitionedLogS
|
|||
if(!recoveryCompleteWrittenToCoreState.get()) {
|
||||
for (const auto& oldData : oldLogData) {
|
||||
logSystemConfig.oldTLogs.emplace_back(oldData);
|
||||
//TraceEvent("BWGetLSConf")
|
||||
// .detail("Epoch", logSystemConfig.oldTLogs.back().epoch)
|
||||
// .detail("Version", logSystemConfig.oldTLogs.back().epochEnd);
|
||||
}
|
||||
}
|
||||
return logSystemConfig;
|
||||
|
@ -2115,6 +2130,9 @@ struct TagPartitionedLogSystem : ILogSystem, ReferenceCounted<TagPartitionedLogS
|
|||
logSystem->oldLogData[0].epoch = oldLogSystem->epoch;
|
||||
}
|
||||
logSystem->oldLogData.insert(logSystem->oldLogData.end(), oldLogSystem->oldLogData.begin(), oldLogSystem->oldLogData.end());
|
||||
//for (const auto& old : logSystem->oldLogData) {
|
||||
// TraceEvent("BWEndVersion").detail("Epoch", old.epoch).detail("Version", old.epochEnd);
|
||||
//}
|
||||
|
||||
logSystem->tLogs[0]->startVersion = oldLogSystem->knownCommittedVersion + 1;
|
||||
state int lockNum = 0;
|
||||
|
|
|
@ -1399,6 +1399,9 @@ ACTOR Future<Void> masterCore( Reference<MasterData> self ) {
|
|||
.detail("MyRecoveryCount", self->cstate.prevDBState.recoveryCount+2)
|
||||
.detail("ForceRecovery", self->forceRecovery)
|
||||
.trackLatest("MasterRecoveryState");
|
||||
//for (const auto& old : self->cstate.prevDBState.oldTLogData) {
|
||||
// TraceEvent("BWReadCoreState", self->dbgid).detail("Epoch", old.epoch).detail("Version", old.epochEnd);
|
||||
//}
|
||||
|
||||
state Reference<AsyncVar<Reference<ILogSystem>>> oldLogSystems( new AsyncVar<Reference<ILogSystem>> );
|
||||
state Future<Void> recoverAndEndEpoch = ILogSystem::recoverAndEndEpoch(oldLogSystems, self->dbgid, self->cstate.prevDBState, self->myInterface.tlogRejoin.getFuture(), self->myInterface.locality, &self->forceRecovery);
|
||||
|
|
Loading…
Reference in New Issue