Skip to content

Make log servers return an empty version range only when it is correct to do so #12188

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
+181 −47
Open

Make log servers return an empty version range only when it is correct to do so #12188

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions fdbserver/LogRouter.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,8 @@ struct LogRouterData {
int64_t generation = -1;
Reference<Histogram> peekLatencyDist;
Optional<Version> recoverAt = Optional<Version>();
Optional<std::map<uint8_t, std::vector<uint16_t>>> knownLockedTLogIds =
Optional<std::map<uint8_t, std::vector<uint16_t>>>();

struct PeekTrackerData {
std::map<int, Promise<std::pair<Version, bool>>> sequence_version;
Expand Down Expand Up @@ -140,6 +142,7 @@ struct LogRouterData {
logSet.updateLocalitySet(req.tLogLocalities);

recoverAt = req.recoverAt;
knownLockedTLogIds = req.knownLockedTLogIds;

for (int i = 0; i < req.tLogLocalities.size(); i++) {
Tag tag(tagLocalityRemoteLog, i);
Expand Down Expand Up @@ -390,8 +393,8 @@ Future<Reference<ILogSystem::IPeekCursor>> LogRouterData::getPeekCursorData(Refe
.When(logSystemChanged,
[&](const Void&) {
if (logSystem->get()) {
result =
logSystem->get()->peekLogRouter(dbgid, beginVersion, routerTag, useSatellite, recoverAt);
result = logSystem->get()->peekLogRouter(
dbgid, beginVersion, routerTag, useSatellite, recoverAt, knownLockedTLogIds);
primaryPeekLocation = result->getPrimaryPeekLocation();
TraceEvent("LogRouterPeekLocation", dbgid)
.detail("LogID", result->getPrimaryPeekLocation())
Expand All @@ -407,7 +410,8 @@ Future<Reference<ILogSystem::IPeekCursor>> LogRouterData::getPeekCursorData(Refe
CODE_PROBE(true, "Detect log router slow peeks");
TraceEvent(SevWarnAlways, "LogRouterSlowPeek", dbgid).detail("NextTrySatellite", !useSatellite);
useSatellite = !useSatellite;
result = logSystem->get()->peekLogRouter(dbgid, beginVersion, routerTag, useSatellite, recoverAt);
result = logSystem->get()->peekLogRouter(
dbgid, beginVersion, routerTag, useSatellite, recoverAt, knownLockedTLogIds);
primaryPeekLocation = result->getPrimaryPeekLocation();
TraceEvent("LogRouterPeekLocation", dbgid)
.detail("LogID", result->getPrimaryPeekLocation())
Expand Down
60 changes: 48 additions & 12 deletions fdbserver/LogSystemPeekCursor.actor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,8 @@ ACTOR Future<Void> tryEstablishPeekStream(ILogSystem::ServerPeekCursor* self) {
self->tag,
self->returnIfBlocked,
std::numeric_limits<int>::max(),
self->end.version);
self->end.version,
self->returnEmptyIfStopped);
self->peekReplyStream = self->interf->get().interf().peekStreamMessages.getReplyStream(req);
DebugLogTraceEvent(SevDebug, "SPC_StreamCreated", self->randomID)
.detail("Tag", self->tag)
Expand All @@ -56,12 +57,13 @@ ILogSystem::ServerPeekCursor::ServerPeekCursor(Reference<AsyncVar<OptionalInterf
Version begin,
Version end,
bool returnIfBlocked,
bool parallelGetMore)
bool parallelGetMore,
bool returnEmptyIfStopped)
: interf(interf), tag(tag), rd(results.arena, results.messages, Unversioned()), messageVersion(begin), end(end),
poppedVersion(0), hasMsg(false), randomID(deterministicRandom()->randomUniqueID()),
returnIfBlocked(returnIfBlocked), onlySpilled(false), parallelGetMore(parallelGetMore),
usePeekStream(SERVER_KNOBS->PEEK_USING_STREAMING), sequence(0), lastReset(0), resetCheck(Void()), slowReplies(0),
fastReplies(0), unknownReplies(0) {
fastReplies(0), unknownReplies(0), returnEmptyIfStopped(returnEmptyIfStopped) {
this->results.maxKnownVersion = 0;
this->results.minKnownCommittedVersion = 0;
DebugLogTraceEvent(SevDebug, "SPC_Starting", randomID)
Expand All @@ -84,7 +86,7 @@ ILogSystem::ServerPeekCursor::ServerPeekCursor(TLogPeekReply const& results,
end(end), poppedVersion(poppedVersion), messageAndTags(message), hasMsg(hasMsg),
randomID(deterministicRandom()->randomUniqueID()), returnIfBlocked(false), onlySpilled(false),
parallelGetMore(false), usePeekStream(false), sequence(0), lastReset(0), resetCheck(Void()), slowReplies(0),
fastReplies(0), unknownReplies(0) {
fastReplies(0), unknownReplies(0), returnEmptyIfStopped(false) {
//TraceEvent("SPC_Clone", randomID);
this->results.maxKnownVersion = 0;
this->results.minKnownCommittedVersion = 0;
Expand Down Expand Up @@ -279,6 +281,7 @@ ACTOR Future<Void> serverPeekParallelGetMore(ILogSystem::ServerPeekCursor* self,
.detail("Seq", self->sequence)
.detail("Sizes", self->futureResults.size())
.detail("Interf", self->interf->get().present() ? self->interf->get().id() : UID());

state Version expectedBegin = self->messageVersion.version;
try {
if (self->parallelGetMore || self->onlySpilled) {
Expand All @@ -293,7 +296,8 @@ ACTOR Future<Void> serverPeekParallelGetMore(ILogSystem::ServerPeekCursor* self,
self->returnIfBlocked,
self->onlySpilled,
std::make_pair(self->randomID, self->sequence++),
self->end.version),
self->end.version,
self->returnEmptyIfStopped),
taskID)));
}
if (self->sequence == std::numeric_limits<decltype(self->sequence)>::max()) {
Expand Down Expand Up @@ -451,7 +455,8 @@ ACTOR Future<Void> serverPeekGetMore(ILogSystem::ServerPeekCursor* self, TaskPri
self->returnIfBlocked,
self->onlySpilled,
Optional<std::pair<UID, int>>(),
self->end.version),
self->end.version,
self->returnEmptyIfStopped),
taskID))
: Never())) {
updateCursorWithReply(self, res);
Expand Down Expand Up @@ -523,11 +528,9 @@ Future<Void> ILogSystem::ServerPeekCursor::onFailed() const {
return serverPeekOnFailed(this);
}

bool ILogSystem::ServerPeekCursor::isActive() const {
bool isAvailable(Reference<AsyncVar<OptionalInterface<TLogInterface>>> const& interf) {
if (!interf->get().present())
return false;
if (isExhausted())
return false;
return IFailureMonitor::failureMonitor()
.getState(interf->get().interf().peekMessages.getEndpoint())
.isAvailable() &&
Expand All @@ -536,6 +539,12 @@ bool ILogSystem::ServerPeekCursor::isActive() const {
.isAvailable();
}

bool ILogSystem::ServerPeekCursor::isActive() const {
if (isExhausted())
return false;
return isAvailable(interf);
}

bool ILogSystem::ServerPeekCursor::isExhausted() const {
return messageVersion >= end;
}
Expand Down Expand Up @@ -564,6 +573,17 @@ Version ILogSystem::ServerPeekCursor::popped() const {
return poppedVersion;
}

void resetBestServerIfNotAvailable(std::vector<Reference<AsyncVar<OptionalInterface<TLogInterface>>>> const& logServers,
int& bestServer,
Version end) {
ASSERT(SERVER_KNOBS->ENABLE_VERSION_VECTOR_TLOG_UNICAST);
if (bestServer >= 0 && end != std::numeric_limits<Version>::max()) {
if (!isAvailable(logServers[bestServer])) {
bestServer = -1;
}
}
}

ILogSystem::MergedPeekCursor::MergedPeekCursor(std::vector<Reference<ILogSystem::IPeekCursor>> const& serverCursors,
Version begin)
: serverCursors(serverCursors), tag(invalidTag), bestServer(-1), currentCursor(0), readQuorum(serverCursors.size()),
Expand Down Expand Up @@ -594,9 +614,17 @@ ILogSystem::MergedPeekCursor::MergedPeekCursor(
logSet->updateLocalitySet(logSet->tLogLocalities);
}

if (SERVER_KNOBS->ENABLE_VERSION_VECTOR_TLOG_UNICAST) {
resetBestServerIfNotAvailable(logServers, bestServer, end);
}

for (int i = 0; i < logServers.size(); i++) {
bool returnEmptyIfStopped =
((SERVER_KNOBS->ENABLE_VERSION_VECTOR_TLOG_UNICAST && end != std::numeric_limits<Version>::max())
? (bestServer < 0 || (i == bestServer))
: false);
auto cursor = makeReference<ILogSystem::ServerPeekCursor>(
logServers[i], tag, begin, end, bestServer >= 0, parallelGetMore);
logServers[i], tag, begin, end, bestServer >= 0, parallelGetMore, returnEmptyIfStopped);
//TraceEvent("MPC_Starting", randomID).detail("Cursor", cursor->randomID).detail("End", end);
serverCursors.push_back(cursor);
}
Expand Down Expand Up @@ -856,13 +884,21 @@ ILogSystem::SetPeekCursor::SetPeekCursor(std::vector<Reference<LogSet>> const& l
Version end,
bool parallelGetMore)
: logSets(logSets), tag(tag), bestSet(bestSet), bestServer(bestServer), currentSet(bestSet), currentCursor(0),
messageVersion(begin), hasNextMessage(false), useBestSet(true), randomID(deterministicRandom()->randomUniqueID()) {
messageVersion(begin), hasNextMessage(false), useBestSet(true), randomID(deterministicRandom()->randomUniqueID()),
end(end) {
if (SERVER_KNOBS->ENABLE_VERSION_VECTOR_TLOG_UNICAST && bestSet >= 0) {
resetBestServerIfNotAvailable(logSets[bestSet]->logServers, bestServer, end);
}
serverCursors.resize(logSets.size());
int maxServers = 0;
for (int i = 0; i < logSets.size(); i++) {
for (int j = 0; j < logSets[i]->logServers.size(); j++) {
bool returnEmptyIfStopped =
((SERVER_KNOBS->ENABLE_VERSION_VECTOR_TLOG_UNICAST && end != std::numeric_limits<Version>::max())
? (bestSet < 0 || bestServer < 0 || (i == bestSet && j == bestServer))
: false);
auto cursor = makeReference<ILogSystem::ServerPeekCursor>(
logSets[i]->logServers[j], tag, begin, end, true, parallelGetMore);
logSets[i]->logServers[j], tag, begin, end, true, parallelGetMore, returnEmptyIfStopped);
serverCursors[i].push_back(cursor);
}
maxServers = std::max<int>(maxServers, serverCursors[i].size());
Expand Down
13 changes: 9 additions & 4 deletions fdbserver/TLogServer.actor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1750,7 +1750,8 @@ Future<Void> tLogPeekMessages(PromiseType replyPromise,
bool reqReturnIfBlocked = false,
bool reqOnlySpilled = false,
Optional<std::pair<UID, int>> reqSequence = Optional<std::pair<UID, int>>(),
Optional<Version> reqEnd = Optional<Version>()) {
Optional<Version> reqEnd = Optional<Version>(),
Optional<bool> reqReturnEmptyIfStopped = Optional<bool>()) {
state BinaryWriter messages(Unversioned());
state BinaryWriter messages2(Unversioned());
state int sequence = -1;
Expand Down Expand Up @@ -1828,7 +1829,9 @@ Future<Void> tLogPeekMessages(PromiseType replyPromise,
ASSERT(!SERVER_KNOBS->ENABLE_VERSION_VECTOR_REPLY_RECOVERY || SERVER_KNOBS->ENABLE_VERSION_VECTOR);
if (logData->version.get() < reqBegin) {
if (SERVER_KNOBS->ENABLE_VERSION_VECTOR_REPLY_RECOVERY && SERVER_KNOBS->ENABLE_VERSION_VECTOR &&
logData->stopped() && reqEnd.present() && reqEnd.get() != std::numeric_limits<Version>::max()) {
logData->stopped() && reqEnd.present() && reqEnd.get() != std::numeric_limits<Version>::max() &&
reqReturnEmptyIfStopped.present() && reqReturnEmptyIfStopped.get()) {
ASSERT(SERVER_KNOBS->ENABLE_VERSION_VECTOR_TLOG_UNICAST);
replyWithRecoveryVersion = reqEnd;
} else if (reqReturnIfBlocked) {
replyPromise.sendError(end_of_stream());
Expand Down Expand Up @@ -2214,7 +2217,8 @@ ACTOR Future<Void> tLogPeekStream(TLogData* self, TLogPeekStreamRequest req, Ref
req.returnIfBlocked,
onlySpilled,
Optional<std::pair<UID, int>>(),
req.end));
req.end,
req.returnEmptyIfStopped));

reply.rep.begin = begin;
req.reply.send(reply);
Expand Down Expand Up @@ -2849,7 +2853,8 @@ ACTOR Future<Void> serveTLogInterface(TLogData* self,
req.returnIfBlocked,
req.onlySpilled,
req.sequence,
req.end));
req.end,
req.returnEmptyIfStopped));
}
when(TLogPopRequest req = waitNext(tli.popMessages.getFuture())) {
logData->addActor.send(tLogPop(self, req, logData));
Expand Down
Loading