From 79704aa05fc400a7d47befe79045a39b729afb86 Mon Sep 17 00:00:00 2001 From: Josh Slocum Date: Tue, 18 Jan 2022 20:17:28 -0600 Subject: [PATCH] Fixing change feed private mutation handling --- fdbclient/NativeAPI.actor.cpp | 56 +++++++++++++++++++++++++------ fdbserver/storageserver.actor.cpp | 13 ++++++- 2 files changed, 57 insertions(+), 12 deletions(-) diff --git a/fdbclient/NativeAPI.actor.cpp b/fdbclient/NativeAPI.actor.cpp index 8caad964ca..6377a09d90 100644 --- a/fdbclient/NativeAPI.actor.cpp +++ b/fdbclient/NativeAPI.actor.cpp @@ -7160,6 +7160,10 @@ Version ChangeFeedData::getVersion() { #define DEBUG_CF_VERSION(cfId, v) \ DEBUG_CF_START_VERSION <= v&& v <= DEBUG_CF_END_VERSION && (""_sr == DEBUG_CF_ID || cfId.printable() == DEBUG_CF_ID) +#define DEBUG_CF_VERSION_RANGE(cfId, vStart, vEnd) \ + DEBUG_CF_START_VERSION <= vEnd&& vStart <= DEBUG_CF_END_VERSION && \ + (""_sr == DEBUG_CF_ID || cfId.printable() == DEBUG_CF_ID) + #define DEBUG_CF_WAIT(cfId, v) DEBUG_CF_WAIT_VERSION == v && (""_sr == DEBUG_CF_ID || cfId.printable() == DEBUG_CF_ID) // This function is essentially bubbling the information about what has been processed from the server through the @@ -7314,7 +7318,8 @@ ACTOR Future partialChangeFeedStream(StorageServerInterface interf, Version end, Reference feedData, Reference storageData, - int idx /* TODO REMOVE this param after correctness clean */) { + int idx /* TODO REMOVE this param after correctness clean */, + KeyRange range /* TODO REMOVE this param after correctness clean */) { // calling lastReturnedVersion's callbacks could cause us to be cancelled state Promise refresh = feedData->refresh; @@ -7323,12 +7328,20 @@ ACTOR Future partialChangeFeedStream(StorageServerInterface interf, try { loop { if (nextVersion >= end) { + if (DEBUG_CF_VERSION(feedData->id, end)) { + fmt::print(" single {0} {1} [{2} - {3}): sending EOS\n", + idx, + interf.id().toString().substr(0, 4), + range.begin.printable(), + range.end.printable()); + } results.sendError(end_of_stream()); return Void(); } choose { when(state ChangeFeedStreamReply rep = waitNext(replyStream.getFuture())) { - if (DEBUG_CF_VERSION(feedData->id, rep.mutations.back().version)) { + if (DEBUG_CF_VERSION_RANGE( + feedData->id, rep.mutations.front().version, rep.mutations.back().version)) { fmt::print(" single {0} {1}: response {2} - {3} ({4}), atLatest={5}, rep.atLatest={6}, " "notAtLatest={7}, " "minSV={8}\n", @@ -7343,6 +7356,12 @@ ACTOR Future partialChangeFeedStream(StorageServerInterface interf, rep.minStreamVersion); } + // TODO REMOVE, just for debugging + // set next version so debug statements trigger + if (nextVersion == 0) { + nextVersion = rep.mutations.front().version; + } + if (rep.mutations.back().version > feedData->maxSeenVersion) { feedData->maxSeenVersion = rep.mutations.back().version; } @@ -7351,9 +7370,11 @@ ACTOR Future partialChangeFeedStream(StorageServerInterface interf, while (resultLoc < rep.mutations.size()) { wait(results.onEmpty()); if (DEBUG_CF_VERSION(feedData->id, rep.mutations[resultLoc].version)) { - fmt::print(" single {0} {1}: sending {2}/{3} {4} ({5})\n", + fmt::print(" single {0} {1} [{2} - {3}): sending {4}/{5} {6} ({7})\n", idx, interf.id().toString().substr(0, 4), + range.begin.printable(), + range.end.printable(), resultLoc, rep.mutations.size(), rep.mutations[resultLoc].version, @@ -7423,7 +7444,14 @@ ACTOR Future partialChangeFeedStream(StorageServerInterface interf, } } catch (Error& e) { // TODO REMOVE eventually, useful for debugging for now - fmt::print("NAS: CFError {}\n", e.name()); + if (DEBUG_CF_VERSION(feedData->id, nextVersion)) { + fmt::print(" single {0} {1} [{2} - {3}): CFError {4}\n", + idx, + interf.id().toString().substr(0, 4), + range.begin.printable(), + range.end.printable(), + e.name()); + } if (e.code() == error_code_actor_cancelled) { throw; } @@ -7451,10 +7479,6 @@ ACTOR Future doCFMerge(Reference results, wait(delay(0)); ASSERT(results->mutations.isEmpty()); - // push one initial mutation from each stream - - // Without this delay, weird issues with the last stream getting on another stream's callstack can happen - wait(delay(0)); state int interfNum = 0; // TODO minor optimization - could make this just a vector of indexes if each MutationAndVersionStream remembered @@ -7465,6 +7489,7 @@ ACTOR Future doCFMerge(Reference results, streamsUsed.push_back(stream); } + state Version nextVersion; loop { // bring all of the streams up to date to ensure we have the latest element from each stream in mutations interfNum = 0; @@ -7495,7 +7520,7 @@ ACTOR Future doCFMerge(Reference results, // pop first item off queue - this will be mutation with the lowest version Standalone> nextOut; - state Version nextVersion = nextVersion = mutations.top().next.version; + nextVersion = mutations.top().next.version; streamsUsed.push_back(mutations.top()); nextOut.push_back_deep(nextOut.arena(), mutations.top().next); @@ -7593,10 +7618,19 @@ ACTOR Future mergeChangeFeedStream(Reference db, results->notAtLatest.set(interfs.size()); refresh.send(Void()); + if (DEBUG_CF_START_VERSION != invalidVersion) { + fmt::print("Starting merge cursor for {0} @ {1} - {2}\n", interfs.size(), *begin, end); + } for (int i = 0; i < interfs.size(); i++) { onErrors[i] = results->streams[i].onError(); - fetchers[i] = partialChangeFeedStream( - interfs[i].first, streams[i].results, results->streams[i], end, results, results->storageData[i], i); + fetchers[i] = partialChangeFeedStream(interfs[i].first, + streams[i].results, + results->streams[i], + end, + results, + results->storageData[i], + i, + interfs[i].second); } wait(onCFErrors(onErrors) || doCFMerge(results, interfs, streams, begin, end)); diff --git a/fdbserver/storageserver.actor.cpp b/fdbserver/storageserver.actor.cpp index 53c8d2c70d..f4692c89ad 100644 --- a/fdbserver/storageserver.actor.cpp +++ b/fdbserver/storageserver.actor.cpp @@ -4023,6 +4023,7 @@ ACTOR Future fetchChangeFeed(StorageServer* data, TraceEvent(SevDebug, "FetchChangeFeed", data->thisServerID) .detail("RangeID", rangeId.printable()) .detail("Range", range.toString()) + .detail("FetchVersion", fetchVersion) .detail("Existing", existing); if (!existing) { @@ -4068,6 +4069,7 @@ ACTOR Future dispatchChangeFeeds(StorageServer* data, UID fetchKeysID, Key try { state std::vector feeds = wait(data->cx->getOverlappingChangeFeeds(keys, fetchVersion + 1)); + // TODO add trace events for some of these for (auto& feed : feeds) { feedFetches[feed.rangeId] = fetchChangeFeed(data, feed.rangeId, feed.range, feed.stopped, fetchVersion); } @@ -4492,6 +4494,10 @@ void AddingShard::addMutation(Version version, bool fromFetch, MutationRef const } it->mutations.back().mutations.push_back_deep(it->mutations.back().arena(), mutation); server->currentChangeFeeds.insert(it->id); + DEBUG_MUTATION("ChangeFeedWriteSet", version, mutation, server->thisServerID) + .detail("Range", it->range) + .detail("ChangeFeedID", it->id) + .detail("Source", "Adding"); } } } else if (mutation.type == MutationRef::ClearRange) { @@ -4504,6 +4510,10 @@ void AddingShard::addMutation(Version version, bool fromFetch, MutationRef const } it->mutations.back().mutations.push_back_deep(it->mutations.back().arena(), mutation); server->currentChangeFeeds.insert(it->id); + DEBUG_MUTATION("ChangeFeedWriteClear", version, mutation, server->thisServerID) + .detail("Range", it->range) + .detail("ChangeFeedID", it->id) + .detail("Source", "Adding"); } } } @@ -4972,7 +4982,8 @@ private: } data->keyChangeFeed.coalesce(feed->second->range.contents()); data->uidChangeFeed.erase(feed); - } else { + } else if (status != ChangeFeedStatus::CHANGE_FEED_CREATE) { + // Can be a change feed create from move, ignore // must be pop or stop if (status == ChangeFeedStatus::CHANGE_FEED_STOP) { TraceEvent(SevDebug, "StoppingChangeFeed", data->thisServerID)