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

1447 lines
54 KiB
C++
Raw Permalink Blame History

/*
* RangePartitionedBackupWorker.cpp
*
* This source file is part of the FoundationDB open source project
*
* Copyright 2013-2026 Apple Inc. and the FoundationDB project authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include "fdbclient/BackupAgent.h"
#include "fdbclient/BackupFileFormat.h"
#include "fdbclient/BackupContainer.h"
#include "fdbclient/DatabaseContext.h"
#include "fdbclient/SystemData.h"
#include "fdbserver/core/BackupPartitionMap.h"
#include "fdbserver/core/BackupProgress.h"
#include "fdbserver/core/Knobs.h"
#include "PartitionMapMessage.h"
#include "fdbserver/core/WaitFailure.h"
#include "fdbserver/logsystem/LogSystem.h"
#include "fdbserver/logsystem/LogSystemConsumer.h"
#include "fdbserver/logsystem/LogSystemFactory.h"
#include "flow/CoroUtils.h"
#define SevDebugMemory SevVerbose
struct RangePartitionedVersionedMessage {
LogMessageVersion version;
StringRef message;
VectorRef<Tag> tags;
Arena arena;
RangePartitionedVersionedMessage(LogMessageVersion v, StringRef m, const VectorRef<Tag>& t, const Arena& a)
: version(v), message(m), tags(t), arena(a) {}
Version getVersion() const { return version.version; }
size_t getEstimatedSize() const { return message.size() + TagsAndMessage::getHeaderSize(6); }
// Returns true if the message is a mutation that could be backed up (normal keys, system key backup ranges, or the
// metadata version key).
bool isCandidateBackupMessage(MutationRef* m) {
// TODO akanksha: Implement this function to filter out messages that are not mutations or not relevant to
// backup. Need to figure out the what those message can be.
return true;
}
};
struct RangePartitionedLogFileInfo {
UID backupUid;
int32_t partitionId;
KeyRange fileKeyRange;
Version beginVersion;
Reference<IBackupFile> file;
int64_t blockEnd = 0;
};
struct RangePartitionedBackupData {
const UID myId;
const Tag tag; // tag for this backup worker
const int totalTags; // Total backup worker tags
const Version startVersion; // This worker's start version
const Optional<Version> endVersion; // old epoch's end version (inclusive), or empty for current epoch
const LogEpoch recruitedEpoch; // current epoch whose tLogs are receiving mutations
const LogEpoch backupEpoch; // the epoch workers should pull mutations
LogEpoch oldestBackupEpoch = 0; // oldest epoch that still has data on tLogs for backup to pull
// Minimumum known committed version in StorageServers.
Version minKnownCommittedVersion;
Version savedVersion; // Largest version saved to blob storage
NotifiedVersion pulledVersion;
Version logFolderBaseVersion;
AsyncVar<Reference<LogSystemConsumer>> logSystem;
AsyncVar<bool> paused; // Track if "backupPausedKey" is set.
Reference<FlowLock> lock;
AsyncTrigger doneTrigger;
AsyncTrigger changedTrigger;
// Set to true when the worker is shutting down (e.g., worker_removed). Used by uploadData to exit gracefully via
// allMessageSaved(), letting in-flight file writes and progress commits finish first.
bool stopped = false;
Database cx;
std::vector<RangePartitionedVersionedMessage> messages;
// Key range to partition ID map, used to determine which partition a mutation belongs to based on its key.
KeyRangeMap<int> keyRangeToPartitionId;
// Partition ID to key range map for easy lookup of partition's key range.
std::unordered_map<int, KeyRange> partitionToKeyRange;
// KeyRange to backup UID and partition id map needed to create log files for the right backup and partition.
KeyRangeMap<std::vector<std::pair<UID, int32_t>>> keyRangeToBackupAssignment;
struct PerBackupInfo {
PerBackupInfo() = default;
PerBackupInfo(RangePartitionedBackupData* data, UID uid, Version v) : self(data), startVersion(v) {
// Open the container and get the key ranges.
BackupConfig config(uid);
container = config.backupContainer().get(data->cx.getReference());
ranges = config.backupRanges().get(data->cx.getReference());
TraceEvent("RangePartitionedBWAddBackup", data->myId).detail("BackupID", uid).detail("Version", v);
}
RangePartitionedBackupData* self = nullptr;
Future<Optional<std::vector<KeyRange>>> ranges; // Key ranges of this backup
Future<Optional<Reference<IBackupContainer>>> container;
// Backup request's commit version. Mutations are logged at some version after this.
Version startVersion = invalidVersion;
// The next log's begin version.
Version nextFileBeginVersion = invalidVersion;
bool isBackupReady() const { return container.isReady() && ranges.isReady(); }
Future<Void> waitBackupReady() { co_await (success(container) && success(ranges)); }
};
// TODO akanksha: Add backups in this map when backup worker receives backup request.
std::unordered_map<UID, PerBackupInfo> backups; // Backup UID to infos
explicit RangePartitionedBackupData(UID id,
Reference<AsyncVar<ServerDBInfo> const> db,
const InitializeRangePartitionedBackupRequest& req)
: myId(id), tag(req.tag), totalTags(req.totalTags), startVersion(req.startVersion), endVersion(req.endVersion),
recruitedEpoch(req.recruitedEpoch), backupEpoch(req.backupEpoch), minKnownCommittedVersion(invalidVersion),
savedVersion(req.startVersion - 1), pulledVersion(0), logFolderBaseVersion(invalidVersion), paused(false),
lock(new FlowLock(SERVER_KNOBS->BACKUP_WORKER_LOCK_BYTES)) {
cx = openDBOnServer(db, TaskPriority::DefaultEndpoint, LockAware::True);
}
bool pullFinished() const { return endVersion.present() && pulledVersion.get() > endVersion.get(); }
Version maxPopVersion() const { return endVersion.present() ? endVersion.get() : minKnownCommittedVersion; }
bool allMessageSaved() const { return (endVersion.present() && savedVersion >= endVersion.get()) || stopped; }
// Tells uploadData to exit: sets stopped (read by allMessageSaved) and wakes it up via doneTrigger.
void stop() {
stopped = true;
doneTrigger.trigger();
}
// Erases messages and updates lock with memory released.
void eraseMessages(int num) {
ASSERT(num <= messages.size());
if (num == 0)
return;
// Accumulate erased message sizes
int64_t bytes = 0;
for (int i = 0; i < num; i++) {
bytes += messages[i].getEstimatedSize();
}
TraceEvent(SevDebugMemory, "RangePartitionedBWMemory", myId)
.detail("Release", bytes)
.detail("Total", lock->activePermits());
lock->release(bytes);
messages.erase(messages.begin(), messages.begin() + num);
}
void eraseMessagesAfterEndVersion() {
ASSERT(endVersion.present());
const Version ver = endVersion.get();
while (!messages.empty()) {
if (messages.back().getVersion() > ver) {
size_t bytes = messages.back().getEstimatedSize();
TraceEvent(SevDebugMemory, "RangePartitionedBWMemory", myId).detail("Release", bytes);
lock->release(bytes);
messages.pop_back();
} else {
break;
}
}
}
// Inserts a backup's single range into rangeMap.
template <class T>
void insertRange(KeyRangeMap<std::set<T>>& keyRangeMap, KeyRangeRef range, T value) {
for (auto& logRange : keyRangeMap.modify(range)) {
logRange->value().insert(value);
}
for (auto& logRange : keyRangeMap.modify(singleKeyRange(metadataVersionKey))) {
logRange->value().insert(value);
}
TraceEvent("BackupWorkerInsertRange", myId)
.detail("Value", value)
.detail("Begin", range.begin)
.detail("End", range.end);
}
// Finds the intersection between a vector of ranges and a target range.
// Returns the union of all intersecting portions as a single range.
// Returns empty Optional if there are no intersections.
Optional<KeyRange> getKeyRangeIntersection(const std::vector<KeyRange>& ranges, const KeyRange& target) {
Optional<KeyRange> result;
for (const auto& range : ranges) {
KeyRange intersection = range & target;
if (intersection.empty()) {
continue;
}
if (!result.present()) {
result = intersection;
} else {
KeyRef newBegin = std::min(result.get().begin, intersection.begin);
KeyRef newEnd = std::max(result.get().end, intersection.end);
result = KeyRange(KeyRangeRef(newBegin, newEnd));
}
}
return result;
}
void pop() {
if (!logSystem.get()) {
return;
}
// Defer the pop in two cases, both to avoid losing mutations a future worker might still need:
// 1. An older epoch still has work to finish (backupEpoch > oldestBackupEpoch). Wait for that
// older epoch to catch up before popping from this epoch.
// 2. We're shutting down (stopped) — our saved progress may not be visible to the next master
// in time, so let the next worker pop after it re-reads progress safely.
if (backupEpoch > oldestBackupEpoch || stopped) {
TraceEvent("RangePartitionedBWPopDeferred", myId)
.suppressFor(1.0)
.detail("BackupEpoch", backupEpoch)
.detail("OldestEpoch", oldestBackupEpoch)
.detail("Stopped", stopped)
.detail("Version", savedVersion);
return;
}
ASSERT_WE_THINK(backupEpoch == oldestBackupEpoch);
logSystem.get()->pop(savedVersion, tag);
}
Future<Void> waitAllBackupsReady() {
std::vector<Future<Void>> all;
for (auto& [uid, info] : backups) {
all.push_back(info.waitBackupReady());
}
co_await waitForAll(all);
}
bool isAllBackupsReady() const {
for (const auto& [uid, info] : backups) {
if (!info.isBackupReady())
return false;
}
return true;
}
};
static Future<Void> computeKeyRangeToBackupAssignment(RangePartitionedBackupData* self) {
self->keyRangeToBackupAssignment = KeyRangeMap<std::vector<std::pair<UID, int32_t>>>();
while (!self->isAllBackupsReady()) {
co_await self->waitAllBackupsReady();
}
for (auto& [uid, info] : self->backups) {
const auto& backupRanges = info.ranges.get().get();
for (auto iter : self->keyRangeToPartitionId.ranges()) {
int32_t partitionId = iter.value();
KeyRange partitionRange = iter.range();
Optional<KeyRange> intersection = self->getKeyRangeIntersection(backupRanges, partitionRange);
if (!intersection.present())
continue;
std::pair<UID, int32_t> bk{ uid, partitionId };
for (auto& range : self->keyRangeToBackupAssignment.modify(intersection.get())) {
range->value().push_back(bk);
}
}
}
self->keyRangeToBackupAssignment.coalesce(allKeys);
}
static Future<Void> onBackupChanges(RangePartitionedBackupData* self,
std::vector<std::pair<UID, Version>> uidVersions) {
std::unordered_set<UID> activeUids;
for (const auto& [uid, version] : uidVersions) {
activeUids.insert(uid);
}
bool modified = false;
bool hasNewBackup = false;
Version newBackupsMinVersion = std::numeric_limits<Version>::max();
// Add any new backups.
for (const auto& [uid, version] : uidVersions) {
if (!self->backups.contains(uid)) {
self->backups.emplace(uid, RangePartitionedBackupData::PerBackupInfo(self, uid, version));
modified = true;
newBackupsMinVersion = std::min(newBackupsMinVersion, version);
hasNewBackup = true;
}
}
// Remove backups that are no longer active.
for (auto it = self->backups.begin(); it != self->backups.end();) {
if (!activeUids.contains(it->first)) {
it = self->backups.erase(it);
modified = true;
} else {
++it;
}
}
if (hasNewBackup && self->backupEpoch < self->recruitedEpoch && self->savedVersion + 1 == self->startVersion) {
// Advance savedVersion to minimize version ranges in case backupEpoch's progress is not saved. Master may set a
// very low startVersion that is already popped. Advance the version is safe because these versions are not
// popped -- if they are popped, their progress should be already recorded and Master would use a higher version
// than minVersion.
self->savedVersion = std::max(newBackupsMinVersion, self->savedVersion);
}
if (modified) {
self->changedTrigger.trigger();
co_await computeKeyRangeToBackupAssignment(self);
}
}
Future<Void> checkRemoved(Reference<AsyncVar<ServerDBInfo> const> db,
LogEpoch recoveryCount,
RangePartitionedBackupData* self) {
while (true) {
bool isDisplaced =
db->get().recoveryCount > recoveryCount && db->get().recoveryState != RecoveryState::UNINITIALIZED;
if (isDisplaced) {
TraceEvent("RangePartitionedBWDisplaced", self->myId)
.detail("RecoveryCount", recoveryCount)
.detail("RecoveryState", (int)db->get().recoveryState);
throw worker_removed();
}
co_await db->onChange();
}
}
Future<Version> pullPartitionMapFromTLog(RangePartitionedBackupData* self, PartitionMap* outPartitionMap) {
Reference<IPeekCursor> cursor;
Version partitionMapVersion = invalidVersion;
Future<Void> logSystemChange = Void();
while (true) {
while (true) {
auto res = co_await race(cursor ? cursor->getMore() : Never(), logSystemChange);
if (res.index() == 0) {
break;
} else {
if (self->logSystem.get()) {
cursor = self->logSystem.get()->peekSingle(self->myId, self->startVersion, self->tag);
} else {
cursor = Reference<IPeekCursor>();
}
logSystemChange = self->logSystem.onChange();
}
}
if (!cursor->hasMessage()) {
continue;
}
for (; cursor->hasMessage(); cursor->nextMessage()) {
Version msgVersion = cursor->version().version;
StringRef message = cursor->getMessage();
Arena arena = cursor->arena();
ArenaReader reader(arena, message, AssumeVersion(g_network->protocolVersion()));
if (reader.protocolVersion().hasSpanContext() && SpanContextMessage::isNextIn(reader)) {
continue;
}
if (reader.protocolVersion().hasOTELSpanContext() && OTELSpanContextMessage::isNextIn(reader)) {
continue;
}
bool isPartitionMap = PartitionMapMessage::isNextIn(reader);
ASSERT(isPartitionMap);
PartitionMapMessage pmMsg;
reader >> pmMsg;
*outPartitionMap = std::move(pmMsg.partitionMap);
partitionMapVersion = msgVersion;
co_return partitionMapVersion;
}
}
}
// Persist the (epoch, version) -> PartitionMap row to SS so older epoch backup workers can read it during
// recovery. Multiple workers may call this concurrently for the same (epoch, version) but only one succeed in writing
// to SS.
Future<Void> persistPartitionMapToSS(RangePartitionedBackupData* self,
Version partitionMapVersion,
PartitionMap const& partitionMap) {
Reference<ReadYourWritesTransaction> tr(new ReadYourWritesTransaction(self->cx));
Key key = backupPartitionMapHistoryKeyFor(self->backupEpoch, partitionMapVersion);
BinaryWriter valueWriter(IncludeVersion());
valueWriter << partitionMap;
Standalone<StringRef> serialized = valueWriter.toValue();
while (true) {
Error err;
try {
tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
tr->setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE);
tr->setOption(FDBTransactionOptions::LOCK_AWARE);
Optional<Value> existing = co_await tr->get(key);
if (existing.present()) {
co_return;
}
tr->set(key, serialized);
co_await tr->commit();
TraceEvent("RangePartitionedBWPMHistoryWritten", self->myId)
.detail("Epoch", self->backupEpoch)
.detail("Version", partitionMapVersion)
.detail("Size", serialized.size());
co_return;
} catch (Error& e) {
err = e;
}
co_await tr->onError(err);
}
}
// Reads the partition map active at `startVersion` for `epoch` from system keys. Any later re-partitions
// in this epoch arrive via the TLog cursor like a current-epoch worker, so we only need this one entry.
// Returns empty if no entry exists for this epoch (e.g., recovery happened before persistPartitionMapToSS).
Future<Optional<std::pair<Version, PartitionMap>>> loadActivePartitionMapFromSS(RangePartitionedBackupData* self,
LogEpoch epoch,
Version startVersion) {
Reference<ReadYourWritesTransaction> tr(new ReadYourWritesTransaction(self->cx));
KeyRange range = backupPartitionMapHistoryRangeFor(epoch);
while (true) {
Error err;
try {
tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
tr->setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE);
tr->setOption(FDBTransactionOptions::LOCK_AWARE);
// Goal: find the partition map that was active at this worker's startVersion within `epoch`.
//
// Example: epoch=5, startVersion=80. SS has entries at:
// [epoch=5, v=1], [epoch=5, v=50], [epoch=5, v=90], [epoch=5, v=95]
// We want the entry at v=50 (largest version <= 80 in epoch 5).
//
// getRange is used here so we get both the key and the value back in a single round trip
// - begin = lastLessOrEqual([epoch=5, v=80])
// "find the largest actual key in the DB <= this target".
// For our example, FDB resolves it to [epoch=5, v=50].
// - end = firstGreaterOrEqual(range.end)
// range.end = [epoch=6, v=0] (one past this epoch's keys), so the search
// never reads beyond this epoch.
// - limit = 1
// We only need that one entry.
RangeResult rows =
co_await tr->getRange(lastLessOrEqual(backupPartitionMapHistoryKeyFor(epoch, startVersion)),
firstGreaterOrEqual(range.end),
/*limit=*/1);
if (rows.empty()) {
TraceEvent("RangePartitionedBWActivePMNotFound", self->myId)
.detail("Epoch", epoch)
.detail("StartVersion", startVersion);
co_return Optional<std::pair<Version, PartitionMap>>();
}
// If no entry exists in our epoch at or before startVersion, lastLessOrEqual lands on a key from a previous
// epoch. We detect that by checking the decoded epoch on the result and treat it as "no entry". The caller
// falls back to pulling the partition map from TLog.
auto [decodedEpoch, version] = decodeBackupPartitionMapHistoryKey(rows[0].key);
if (decodedEpoch != epoch) {
TraceEvent("RangePartitionedBWActivePMNotFound", self->myId)
.detail("Epoch", epoch)
.detail("DecodedEpoch", decodedEpoch)
.detail("DecodedVersion", version)
.detail("StartVersion", startVersion);
co_return Optional<std::pair<Version, PartitionMap>>();
}
PartitionMap pm;
BinaryReader reader(rows[0].value, IncludeVersion());
reader >> pm;
TraceEvent("RangePartitionedBWActivePMRead", self->myId)
.detail("Epoch", epoch)
.detail("StartVersion", startVersion)
.detail("PMVersion", version);
co_return std::make_pair(version, std::move(pm));
} catch (Error& e) {
err = e;
}
co_await tr->onError(err);
}
}
// TODO akanksha:
// 1. Test if concurrent uploads of identical content to the same path in blob storage is safe or not.
// 2. When folder is advanced to next version, do we upload the partition map again to that version.
Future<Void> uploadPartitionList(RangePartitionedBackupData* self, PartitionMap partitionMap) {
std::vector<Future<Void>> fileFutures;
auto it = self->backups.begin();
std::string jsonContent = serializePartitionListJSON(partitionMap);
for (; it != self->backups.end();) {
if (!it->second.container.get().present()) {
TraceEvent("RangePartitionedBWRemoveContainer", self->myId).detail("BackupId", it->first);
it = self->backups.erase(it);
continue;
}
Reference<IBackupContainer> container = it->second.container.get().get();
fileFutures.push_back(container->writePartitionListFile(self->logFolderBaseVersion, jsonContent));
it++;
}
if (fileFutures.empty()) {
TraceEvent("RangePartitionedBWNoContainers", self->myId);
co_return;
}
co_await waitForAll(fileFutures);
}
// Persists partitionMap to SS history (so that catch-up backup workers can find it during recovery) and writes the
// partitionId_keyRange_Map file for every active backup container.
Future<Void> persistAndUploadPartitionMap(RangePartitionedBackupData* self,
Version pmVersion,
PartitionMap const& partitionMap) {
co_await persistPartitionMapToSS(self, pmVersion, partitionMap);
co_await uploadPartitionList(self, partitionMap);
}
// Updates local routing state to use the new partition map.
Future<Void> setActivePartitionMap(RangePartitionedBackupData* self,
Version pmVersion,
PartitionMap const& partitionMap) {
self->logFolderBaseVersion = pmVersion;
ASSERT(partitionMap.contains(self->tag));
const auto& tagPartitions = partitionMap.at(self->tag);
ASSERT_GT(tagPartitions.size(), 0);
self->keyRangeToPartitionId.clear();
for (const auto& partition : tagPartitions) {
self->keyRangeToPartitionId.insert(partition.ranges, partition.partitionId);
}
co_await computeKeyRangeToBackupAssignment(self);
}
Future<Void> processPartitionMap(RangePartitionedBackupData* self) {
TraceEvent("RangePartitionedBWWaitingForPartitionMap", self->myId)
.detail("Tag", self->tag.toString())
.detail("StartVersion", self->startVersion)
.detail("BackupEpoch", self->backupEpoch)
.detail("RecruitedEpoch", self->recruitedEpoch);
PartitionMap partitionMap;
Version partitionMapVersion;
Optional<std::pair<Version, PartitionMap>> startPMFromHistory;
if (self->backupEpoch != self->recruitedEpoch) {
// Old epoch worker: the partition map active at our startVersion was persisted by the previous
// epoch's workers. Read just that one from system keys; any later re-partitions in this epoch
// will arrive via the TLog cursor like a current-epoch worker.
startPMFromHistory = co_await loadActivePartitionMapFromSS(self, self->backupEpoch, self->startVersion);
if (startPMFromHistory.present()) {
partitionMapVersion = startPMFromHistory.get().first;
partitionMap = std::move(startPMFromHistory.get().second);
auto it = partitionMap.find(self->tag);
ASSERT(it != partitionMap.end() && !it->second.empty());
TraceEvent("RangePartitionedBWLoadedPartitionMap", self->myId)
.detail("Epoch", self->backupEpoch)
.detail("Version", partitionMapVersion)
.detail("NumTags", partitionMap.size())
.detail("Tag", self->tag.toString())
.detail("NumPartitions", it->second.size());
}
}
if (self->backupEpoch == self->recruitedEpoch || !startPMFromHistory.present()) {
// Current-epoch worker, or old epoch worker with no SS history (recovery happened before the
// previous epoch's persistPartitionMapToSS). Receive the partition map via TLog as the first
// message, then persist it so the next old epoch worker doesn't hit this case.
partitionMapVersion = co_await pullPartitionMapFromTLog(self, &partitionMap);
auto it = partitionMap.find(self->tag);
ASSERT(it != partitionMap.end() && !it->second.empty());
TraceEvent("RangePartitionedBWPulledPartitionMap", self->myId)
.detail("Version", partitionMapVersion)
.detail("NumTags", partitionMap.size())
.detail("Tag", self->tag.toString())
.detail("NumPartitions", it->second.size());
// Persist the partition map to system key so that catch-up backup workers can read it during recovery.
// Every BW also writes the partitionId_keyRange_Map file. Content is deterministic across workers
// through serializePartitionListJSON, so concurrent PUTs of identical bytes are safe.
co_await persistAndUploadPartitionMap(self, partitionMapVersion, partitionMap);
TraceEvent("RangePartitionedBWPartitionMapUploaded", self->myId)
.detail("Version", partitionMapVersion)
.detail("NumBackups", self->backups.size());
self->pulledVersion.set(partitionMapVersion);
self->savedVersion = partitionMapVersion;
self->pop();
}
co_await setActivePartitionMap(self, partitionMapVersion, partitionMap);
}
// Pulls mutations from TLog servers.
Future<Void> pullAsyncData(RangePartitionedBackupData* self) {
Future<Void> logSystemChange = Void();
Reference<IPeekCursor> cursor;
Version tagAt = std::max({ self->pulledVersion.get(), self->startVersion, self->savedVersion });
TraceEvent("RangePartitionedBWPull", self->myId)
.detail("Tag", self->tag)
.detail("Version", tagAt)
.detail("StartVersion", self->startVersion)
.detail("SavedVersion", self->savedVersion);
while (true) {
while (self->paused.get()) {
co_await self->paused.onChange();
}
while (true) {
auto res = co_await race(cursor ? cursor->getMore(TaskPriority::TLogCommit) : Never(), logSystemChange);
if (res.index() == 0) {
DisabledTraceEvent("RangePartitionedBWGotMore", self->myId)
.detail("Tag", self->tag)
.detail("CursorVersion", cursor->version().version);
break;
} else {
if (self->logSystem.get()) {
cursor = self->logSystem.get()->peekSingle(self->myId, tagAt, self->tag);
} else {
cursor = Reference<IPeekCursor>();
}
logSystemChange = self->logSystem.onChange();
}
}
if (cursor->popped() > 0) {
TraceEvent(SevError, "RangePartitionedBWDataPopped", self->myId)
.detail("Popped", cursor->popped())
.detail("Expected", tagAt);
throw worker_removed();
}
self->minKnownCommittedVersion =
std::max(self->minKnownCommittedVersion, cursor->getMinKnownCommittedVersion());
int64_t peekedBytes = 0;
// Hold messages until we know how many we can take, self->messages always
// contains messages that we have reserved memory for. Therefore, lock->release()
// will always encounter message with reserved memory.
std::vector<RangePartitionedVersionedMessage> tmpMessages;
// Messages may be prefetched in peek here, but uncommitted messages should not be uploaded in uploadData().
while (cursor->hasMessage()) {
StringRef rawMessage = cursor->getMessage();
Arena msgArena = cursor->arena();
ArenaReader reader(msgArena, rawMessage, AssumeVersion(g_network->protocolVersion()));
// Skip metadata-only messages so they don't reach uploadData.
if (reader.protocolVersion().hasSpanContext() && SpanContextMessage::isNextIn(reader)) {
cursor->nextMessage();
continue;
}
if (reader.protocolVersion().hasOTELSpanContext() && OTELSpanContextMessage::isNextIn(reader)) {
cursor->nextMessage();
continue;
}
// Mid-stream PartitionMap update (re-partition). Persist + upload the new map immediately so
// future old-epoch workers can find it. The raw message bytes fall through to be buffered
// into self->messages like any other cursor message; uploadData detects it during its batch
// and calls setActivePartitionMap at the right version boundary.
// Do not `cursor->nextMessage()` or `continue` — fall through to buffer this message.
if (PartitionMapMessage::isNextIn(reader)) {
Version pmVersion = cursor->version().version;
PartitionMapMessage pmMsg;
reader >> pmMsg;
co_await persistAndUploadPartitionMap(self, pmVersion, pmMsg.partitionMap);
TraceEvent("RangePartitionedBWReceivedMidStreamPM", self->myId)
.detail("Version", pmVersion)
.detail("NumPartitions", pmMsg.partitionMap[self->tag].size());
}
auto msg = RangePartitionedVersionedMessage(cursor->version(), rawMessage, cursor->getTags(), msgArena);
tmpMessages.emplace_back(std::move(msg));
peekedBytes += tmpMessages.back().getEstimatedSize();
cursor->nextMessage();
}
if (peekedBytes > 0) {
TraceEvent(SevDebugMemory, "RangePartitionedBWMemory", self->myId)
.detail("Take", peekedBytes)
.detail("Current", self->lock->activePermits());
co_await self->lock->take(TaskPriority::DefaultYield, peekedBytes);
self->messages.insert(self->messages.end(),
std::make_move_iterator(tmpMessages.begin()),
std::make_move_iterator(tmpMessages.end()));
}
tagAt = cursor->version().version;
self->pulledVersion.set(tagAt);
TraceEvent("RangePartitionedBWGot", self->myId).suppressFor(1.0).detail("LatestPulledVersion", tagAt);
// For older epochs, we may have an end version to stop at.
if (self->pullFinished()) {
self->eraseMessagesAfterEndVersion();
self->doneTrigger.trigger();
TraceEvent("RangePartitionedBWFinishPull", self->myId)
.detail("Tag", self->tag.toString())
.detail("VersionGot", tagAt)
.detail("EndVersion", self->endVersion.get())
.detail("LogEpoch", self->recruitedEpoch)
.detail("BackupEpoch", self->backupEpoch);
co_return;
}
co_await yield();
}
}
Future<Void> writeFileHeader(Reference<IBackupFile> logFile, int32_t partitionId, KeyRange range) {
co_await logFile->append((uint8_t*)&RANGE_PARTITIONED_MLOG_VERSION, sizeof(RANGE_PARTITIONED_MLOG_VERSION));
BinaryWriter wr(Unversioned());
wr << partitionId << range.begin << range.end;
Standalone<StringRef> header = wr.toValue();
co_await logFile->append(header.begin(), header.size());
}
Future<Void> addMutation(Reference<IBackupFile> logFile,
RangePartitionedVersionedMessage message,
StringRef mutation,
int64_t* blockEnd,
int blockSize) {
// Format: version, subversion, messageSize, message
int bytes = sizeof(Version) + sizeof(uint32_t) + sizeof(int) + mutation.size();
// Convert to big Endianness for version.version, version.sub, and msgSize
// The decoder assumes 0xFF is the end, so little endian can easily be
// mistaken as the end. In contrast, big endian for version almost guarantee
// the first byte is not 0xFF (should always be 0x00).
BinaryWriter wr(Unversioned());
wr << bigEndian64(message.version.version) << bigEndian32(message.version.sub) << bigEndian32(mutation.size());
Standalone<StringRef> mutationHeader = wr.toValue();
// Start a new block if needed
if (logFile->size() + bytes > *blockEnd) {
const int bytesLeft = *blockEnd - logFile->size();
if (bytesLeft > 0) {
Value paddingFFs = fileBackup::makePadding(bytesLeft);
co_await logFile->append(paddingFFs.begin(), bytesLeft);
}
*blockEnd += blockSize;
// Block header.
co_await logFile->append((uint8_t*)&RANGE_PARTITIONED_MLOG_VERSION, sizeof(RANGE_PARTITIONED_MLOG_VERSION));
}
co_await logFile->append((void*)mutationHeader.begin(), mutationHeader.size());
co_await logFile->append(mutation.begin(), mutation.size());
}
static Future<Void> updateLogBytesWritten(RangePartitionedBackupData* self, std::map<UID, int64_t> bytesPerBackup) {
Reference<ReadYourWritesTransaction> tr(new ReadYourWritesTransaction(self->cx));
while (true) {
Error err;
try {
tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
tr->setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE);
tr->setOption(FDBTransactionOptions::LOCK_AWARE);
for (const auto& [uid, bytes] : bytesPerBackup) {
BackupConfig config(uid);
config.logBytesWritten().atomicOp(tr, bytes, MutationRef::AddValue);
}
co_await tr->commit();
co_return;
} catch (Error& e) {
err = e;
}
co_await tr->onError(err);
}
}
Future<Void> saveMutationsToFile(RangePartitionedBackupData* self, Version lastVersionInFile, int numMsg) {
// Make sure all backups are ready, otherwise mutations will be lost.
while (!self->isAllBackupsReady()) {
co_await self->waitAllBackupsReady();
}
std::vector<RangePartitionedLogFileInfo> activeFiles;
// Map of (backupUid, partitionId) -> index into activeFiles.
std::map<std::pair<UID, int32_t>, int> fileIndexByBackupPartition;
int blockSize = SERVER_KNOBS->BACKUP_FILE_BLOCK_BYTES;
std::vector<Future<Reference<IBackupFile>>> fileFutures;
for (auto entry = self->keyRangeToBackupAssignment.ranges().begin();
entry != self->keyRangeToBackupAssignment.ranges().end();
++entry) {
for (const auto& [backupUid, partitionId] : entry->value()) {
auto it = self->backups.find(backupUid);
if (it == self->backups.end() || !it->second.container.get().present()) {
TraceEvent("RangePartitionedBWRemoveContainerInFileCreation", self->myId).detail("BackupId", backupUid);
continue;
}
std::pair<UID, int32_t> bpKey(backupUid, partitionId);
if (fileIndexByBackupPartition.contains(bpKey)) {
continue;
}
Version fileEndVersion = lastVersionInFile + 1;
if (it->second.nextFileBeginVersion == invalidVersion) {
it->second.nextFileBeginVersion = self->savedVersion + 1;
}
Version beginVersion = it->second.nextFileBeginVersion;
RangePartitionedLogFileInfo lf;
lf.backupUid = it->first;
lf.partitionId = partitionId;
lf.fileKeyRange = entry->range();
lf.beginVersion = beginVersion;
lf.blockEnd = 0;
activeFiles.push_back(lf);
fileIndexByBackupPartition[bpKey] = activeFiles.size() - 1;
fileFutures.push_back(it->second.container.get().get()->writeRangePartitionedLogFile(
beginVersion, fileEndVersion, self->logFolderBaseVersion, partitionId, blockSize));
}
}
if (fileFutures.empty()) {
co_return;
}
co_await waitForAll(fileFutures);
std::vector<Future<Void>> headerWrites;
int i;
for (i = 0; i < activeFiles.size(); i++) {
activeFiles[i].file = fileFutures[i].get();
headerWrites.push_back(
writeFileHeader(activeFiles[i].file, activeFiles[i].partitionId, activeFiles[i].fileKeyRange));
}
co_await waitForAll(headerWrites);
if (activeFiles.empty()) {
co_return;
}
// Process mutations
int idx;
for (idx = 0; idx < numMsg; idx++) {
auto& message = self->messages[idx];
MutationRef m;
if (!message.isCandidateBackupMessage(&m)) {
continue;
}
DEBUG_MUTATION("RangePartitionedBWAddMutation", message.version.version, m, self->myId)
.detail("KCV", self->minKnownCommittedVersion)
.detail("SavedVersion", self->savedVersion);
std::vector<Future<Void>> adds;
if (m.type != MutationRef::Type::ClearRange) {
for (const auto& entry : self->keyRangeToBackupAssignment[m.param1]) {
auto it = fileIndexByBackupPartition.find(entry);
ASSERT(it != fileIndexByBackupPartition.end());
int fileIdx = it->second;
auto& lf = activeFiles[fileIdx];
// Different backups may have different start version so need this check before writing.
if (message.getVersion() >= lf.beginVersion) {
adds.push_back(addMutation(lf.file, message, message.message, &lf.blockEnd, blockSize));
}
}
} else {
KeyRangeRef mutationRange(m.param1, m.param2);
std::unordered_set<int> writtenFiles;
for (auto range : self->keyRangeToBackupAssignment.intersectingRanges(mutationRange)) {
for (const auto& entry : range.value()) {
auto it = fileIndexByBackupPartition.find(entry);
ASSERT(it != fileIndexByBackupPartition.end());
int fileIdx = it->second;
// For ClearRange, we only need to write the full mutation once for each file.
if (writtenFiles.contains(fileIdx)) {
continue;
}
auto& lf = activeFiles[fileIdx];
if (message.getVersion() >= lf.beginVersion) {
adds.push_back(addMutation(lf.file, message, message.message, &lf.blockEnd, blockSize));
}
writtenFiles.insert(fileIdx);
}
}
}
if (!adds.empty()) {
co_await waitForAll(adds);
}
}
// Finish files
// TODO akanksha: Add FileLevel checksum.
std::vector<Future<Void>> finished;
for (auto& lf : activeFiles) {
finished.push_back(lf.file->finish());
}
co_await waitForAll(finished);
std::map<UID, int64_t> bytesPerBackup;
for (auto& lf : activeFiles) {
auto it = self->backups.find(lf.backupUid);
if (it != self->backups.end()) {
it->second.nextFileBeginVersion = lastVersionInFile + 1;
}
bytesPerBackup[lf.backupUid] += lf.file->size();
}
co_await updateLogBytesWritten(self, std::move(bytesPerBackup));
}
// It closes the race between getMinBackupVersion's snapshot at master-recruit time and the actual state of
// backupStartedKey when the old epoch backup worker comes up — specifically the case where backup configuration changed
// during that window so the backup worker is no longer needed.
static Future<bool> shouldBackupWorkerExitEarly(RangePartitionedBackupData* self) {
while (true) {
ReadYourWritesTransaction tr(self->cx);
while (true) {
Error err;
try {
tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
tr.setOption(FDBTransactionOptions::LOCK_AWARE);
Optional<Value> value = co_await tr.get(backupStartedKey);
std::vector<std::pair<UID, Version>> uidVersions;
if (value.present()) {
bool shouldExit = self->endVersion.present();
uidVersions = decodeBackupStartedValue(value.get());
TraceEvent e("RangePartitionedBWGotStartKey", self->myId);
int i = 1;
for (auto [uid, version] : uidVersions) {
e.detail(format("BackupID%d", i), uid).detail(format("Version%d", i), version);
i++;
if (shouldExit && version < self->endVersion.get()) {
shouldExit = false;
}
}
co_await onBackupChanges(self, uidVersions);
co_return shouldExit;
}
TraceEvent("RangePartitionedBWEmptyStartKey", self->myId);
Future<Void> watchFuture = tr.watch(backupStartedKey);
co_await tr.commit();
co_await watchFuture;
break;
} catch (Error& e) {
err = e;
}
co_await tr.onError(err);
}
}
}
static Future<Void> monitorBackupStartedKeyChanges(RangePartitionedBackupData* self) {
while (true) {
ReadYourWritesTransaction tr(self->cx);
while (true) {
Error err;
try {
tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
tr.setOption(FDBTransactionOptions::LOCK_AWARE);
Optional<Value> value = co_await tr.get(backupStartedKey);
std::vector<std::pair<UID, Version>> uidVersions;
if (value.present()) {
uidVersions = decodeBackupStartedValue(value.get());
TraceEvent e("RangePartitionedBWGotStartKey", self->myId);
int i = 1;
for (auto [uid, version] : uidVersions) {
e.detail(format("BackupID%d", i), uid).detail(format("Version%d", i), version);
i++;
}
}
co_await onBackupChanges(self, uidVersions);
Future<Void> watchFuture = tr.watch(backupStartedKey);
co_await tr.commit();
co_await watchFuture;
break;
} catch (Error& e) {
err = e;
}
co_await tr.onError(err);
}
}
}
// This function is used to set backup worker's saved version latestBackupWorkerSavedVersion in BackupConfig.
Future<Void> setBackupKeys(RangePartitionedBackupData* self, std::map<UID, Version> savedLogVersions) {
Reference<ReadYourWritesTransaction> tr(new ReadYourWritesTransaction(self->cx));
while (true) {
Error err;
try {
tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
tr->setOption(FDBTransactionOptions::LOCK_AWARE);
tr->setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE);
std::vector<Future<Optional<Version>>> prevBackupWorkerSavedVersions;
std::vector<BackupConfig> versionConfigs;
for (const auto& [uid, version] : savedLogVersions) {
BackupConfig config(uid);
versionConfigs.emplace_back(config);
prevBackupWorkerSavedVersions.push_back(config.latestBackupWorkerSavedVersion().get(tr));
}
co_await waitForAll(prevBackupWorkerSavedVersions);
for (int i = 0; i < prevBackupWorkerSavedVersions.size(); i++) {
const Version current = savedLogVersions[versionConfigs[i].getUid()];
if (prevBackupWorkerSavedVersions[i].get().present()) {
const Version prev = prevBackupWorkerSavedVersions[i].get().get();
if (prev > current) {
TraceEvent(SevWarn, "RangePartitionedBWVersionInverse", self->myId)
.detail("Prev", prev)
.detail("Current", current);
}
}
if (self->backupEpoch == self->oldestBackupEpoch &&
(!prevBackupWorkerSavedVersions[i].get().present() ||
prevBackupWorkerSavedVersions[i].get().get() < current)) {
TraceEvent("RangePartitionedBWSetVersion", self->myId)
.detail("BackupID", versionConfigs[i].getUid())
.detail("Version", current);
versionConfigs[i].latestBackupWorkerSavedVersion().set(tr, current);
}
}
co_await tr->commit();
co_return;
} catch (Error& e) {
err = e;
}
co_await tr->onError(err);
}
}
static Future<Void> monitorWorkerPause(RangePartitionedBackupData* self) {
Reference<ReadYourWritesTransaction> tr(new ReadYourWritesTransaction(self->cx));
Future<Void> watch;
while (true) {
Error err;
try {
tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
tr->setOption(FDBTransactionOptions::LOCK_AWARE);
tr->setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE);
Optional<Value> value = co_await tr->get(backupPausedKey);
bool paused = value.present() && value.get() == "1"_sr;
if (self->paused.get() != paused) {
TraceEvent(paused ? "RangePartitionedBWPaused" : "RangePartitionedBWResumed", self->myId).log();
self->paused.set(paused);
}
watch = tr->watch(backupPausedKey);
co_await tr->commit();
co_await watch;
tr->reset();
continue;
} catch (Error& e) {
err = e;
}
co_await tr->onError(err);
}
}
Future<Void> monitorRangePartitionedBackupProgress(RangePartitionedBackupData* self) {
Future<Void> interval;
while (true) {
interval = delay(SERVER_KNOBS->WORKER_LOGGING_INTERVAL / 2.0);
while (self->backups.empty() || !self->logSystem.get()) {
co_await (self->changedTrigger.onTrigger() || self->logSystem.onChange());
}
// Check all workers have started by checking their progress is larger than the backup's start version.
Reference<BackupProgress> progress(new BackupProgress(self->myId, {}));
co_await getBackupProgress(self->cx, self->myId, progress, SevDebug);
std::map<Tag, Version> tagVersions = progress->getEpochStatus(self->recruitedEpoch);
if (tagVersions.size() != self->totalTags) {
co_await interval;
continue;
}
std::map<UID, Version> savedLogVersions;
// update progress so far if previous epochs are done.
if (self->recruitedEpoch == self->oldestBackupEpoch) {
Version v = std::numeric_limits<Version>::max();
// Find the version we can gurantee is fully backed up for all backup workers.
for (const auto& [tag, version] : tagVersions) {
v = std::min(v, version);
}
for (auto& [uid, info] : self->backups) {
savedLogVersions.emplace(uid, v);
TraceEvent("RangePartitionedBWSavedBackupVersion", self->myId)
.detail("BackupID", uid)
.detail("Version", v);
}
}
Future<Void> setKeys = savedLogVersions.empty() ? Void() : setBackupKeys(self, savedLogVersions);
co_await (interval && setKeys);
}
}
Future<Void> saveProgress(RangePartitionedBackupData* self, Version backupVersion) {
Transaction tr(self->cx);
Key key = backupProgressKeyFor(self->myId);
while (true) {
Error err;
try {
// It's critical to save progress immediately so that after a master
// recovery, the new master can know the progress so far.
tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
tr.setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE);
tr.setOption(FDBTransactionOptions::LOCK_AWARE);
// CHECK: Don't save progress if backup workers are disabled
Optional<Value> backupWorkerEnabled = co_await tr.get(rangePartitionedBackupWorkerEnabledKey);
if (!backupWorkerEnabled.present() || backupWorkerEnabled.get() == "0"_sr) {
TraceEvent("RangePartitionedBWProgressSkipped", self->myId).detail("Reason", "BackupWorkersDisabled");
co_return;
}
WorkerBackupStatus status(self->backupEpoch, backupVersion, self->tag, self->totalTags);
tr.set(key, backupProgressValue(status));
tr.addReadConflictRange(singleKeyRange(key));
co_await tr.commit();
co_return;
} catch (Error& e) {
err = e;
}
co_await tr.onError(err);
}
}
// Uploads self->messages to storage and updates savedVersion.
Future<Void> uploadData(RangePartitionedBackupData* self) {
// Version up to which messages will be popped from tlog.
Version popVersion = invalidVersion;
while (true) {
// Too large uploadDelay will delay popping tLog data for too long.
Future<Void> uploadDelay = delay(SERVER_KNOBS->BACKUP_UPLOAD_DELAY);
int numMsg = 0;
// Last Version that we popped from tlog.
Version lastPopVersion = popVersion;
// index of last version's end position in self->messages. i.e. the first index of next version???
int lastVersionIndex = 0;
// Version just before current popVersion - to find version boundaries.
Version lastVersion = invalidVersion;
for (auto& message : self->messages) {
// message may be prefetched in peek; uncommitted message should not be uploaded.
const Version msgVersion = message.getVersion();
if (msgVersion > self->maxPopVersion()) {
break;
}
if (msgVersion > popVersion) {
lastVersionIndex = numMsg;
lastVersion = popVersion;
popVersion = msgVersion;
}
numMsg++;
}
if (self->pullFinished()) {
popVersion = self->endVersion.get();
} else {
// make sure file is saved on version boundary
popVersion = lastVersion;
numMsg = lastVersionIndex;
// If we aren't able to process any messages and the lock is blocking us from
// queuing more, then we are stuck. This could suggest the lock capacity is too small.
ASSERT(numMsg > 0 || self->lock->waiters() == 0);
}
// TODO akanksha: Removed redundant check popVersion > lastPopVersion. Remove todo after testing completes.
if (numMsg > 0 || self->pullFinished()) {
TraceEvent("RangePartitionedBWSave", self->myId)
.detail("Version", popVersion)
.detail("LastPopVersion", lastPopVersion)
.detail("SavedVersion", self->savedVersion)
.detail("NumMsg", numMsg)
.detail("MsgQ", self->messages.size());
// Walk the batch and look for any PartitionMapMessage. Whenever we hit one, flush all the
// mutations before it under the current map, switch to the new map, and drop the message.
int idx = 0;
while (idx < numMsg) {
ArenaReader reader(self->messages[idx].arena,
self->messages[idx].message,
AssumeVersion(g_network->protocolVersion()));
if (!PartitionMapMessage::isNextIn(reader)) {
idx++;
continue;
}
Version pmV = self->messages[idx].getVersion();
if (idx > 0) {
co_await saveMutationsToFile(self, pmV - 1, idx);
self->eraseMessages(idx);
numMsg -= idx;
}
// PartitionMapMessage is now at index 0. Decode, apply, drop.
ArenaReader pmReader(
self->messages[0].arena, self->messages[0].message, AssumeVersion(g_network->protocolVersion()));
PartitionMapMessage pmMsg;
pmReader >> pmMsg;
co_await setActivePartitionMap(self, pmV, pmMsg.partitionMap);
self->eraseMessages(1);
numMsg -= 1;
TraceEvent("RangePartitionedBWAppliedMidStreamPM", self->myId)
.detail("Version", pmV)
.detail("NumPartitions", pmMsg.partitionMap[self->tag].size());
idx = 0;
}
// Even if numMsg is 0 (pullFinished with nothing left to save), write a final file at endVersion. Restore
// checks each tag's log files form a continuous range; without this marker, restore would stop at the last
// real file and miss the tail of this epoch. In-between empty ranges (gaps within [startVersion,
// endVersion)) are fine — restore only needs the worker's continuous range to extend through endVersion.
co_await saveMutationsToFile(self, popVersion, numMsg);
self->eraseMessages(numMsg);
}
if (popVersion > self->savedVersion) {
co_await saveProgress(self, popVersion);
TraceEvent("RangePartitionedBWSavedProgress", self->myId)
.detail("Tag", self->tag.toString())
.detail("Version", popVersion)
.detail("MsgQ", self->messages.size());
self->savedVersion = popVersion;
self->pop();
}
if (self->allMessageSaved()) {
co_return;
}
if (!self->pullFinished()) {
co_await (uploadDelay || self->doneTrigger.onTrigger());
}
}
}
// Keeps `self->logSystem` and `self->oldestBackupEpoch` in sync with the latest ServerDBInfo.
static Future<Void> monitorLogSystemFromDbInfo(Reference<AsyncVar<ServerDBInfo> const> db,
RangePartitionedBackupData* self) {
while (true) {
Reference<LogSystem> ls = makeLogSystemFromServerDBInfo(self->myId, db->get(), true);
if (ls.isValid()) {
self->logSystem.set(ls->makeConsumer());
self->oldestBackupEpoch = std::max(self->oldestBackupEpoch, ls->getOldestBackupEpoch());
TraceEvent("RangePartitionedBWLogSystemUpdate", self->myId)
.detail("Tag", self->tag.toString())
.detail("TagLocality", self->tag.locality)
.detail("OldestEpoch", self->oldestBackupEpoch);
} else {
TraceEvent("RangePartitionedBWNoLogSystem", self->myId);
}
co_await db->onChange();
}
}
Future<Void> rangePartitionedBackupWorker(BackupInterface interf,
InitializeRangePartitionedBackupRequest req,
Reference<AsyncVar<ServerDBInfo> const> db) {
RangePartitionedBackupData self(interf.id(), db, req);
PromiseStream<Future<Void>> addActor;
Future<Void> error = actorCollection(addActor.getFuture());
Future<Void> pull;
Future<Void> done;
Error err;
TraceEvent("RangePartitionedBWStart", self.myId)
.detail("Tag", req.tag.toString())
.detail("TotalTags", req.totalTags)
.detail("StartVersion", req.startVersion)
.detail("EndVersion", req.endVersion.present() ? req.endVersion.get() : -1)
.detail("LogEpoch", req.recruitedEpoch)
.detail("BackupEpoch", req.backupEpoch);
try {
addActor.send(checkRemoved(db, req.recruitedEpoch, &self));
addActor.send(waitFailureServer(interf.waitFailure.getFuture()));
if (req.recruitedEpoch == req.backupEpoch && req.tag.id == 0) {
addActor.send(monitorRangePartitionedBackupProgress(&self));
}
addActor.send(monitorWorkerPause(&self));
// Must be sent before processPartitionMap so logSystem is populated before the partition-map peek.
addActor.send(monitorLogSystemFromDbInfo(db, &self));
// First need to call processPartitionMap before starting to pull data, because we need to know the
// partition assignment.
co_await processPartitionMap(&self);
// If the worker is on an old epoch and all backups starts a version >= the endVersion
bool exitEarly = co_await shouldBackupWorkerExitEarly(&self);
TraceEvent("RangePartitionedBWExitEarly", self.myId).detail("ExitEarly", exitEarly);
if (!exitEarly) {
addActor.send(monitorBackupStartedKeyChanges(&self));
}
pull = exitEarly ? Void() : pullAsyncData(&self);
addActor.send(pull);
done = exitEarly ? Void() : uploadData(&self);
while (true) {
auto res = co_await race(done, error);
if (res.index() == 0) {
TraceEvent("RangePartitionedBWDone", self.myId).detail("BackupEpoch", self.backupEpoch);
// Notify master so that this worker can be removed from log system, then this
// worker (for an old epoch's unfinished work) can safely exit.
co_await brokenPromiseToNever(db->get().clusterInterface.notifyBackupWorkerDone.getReply(
BackupWorkerDoneRequest(self.myId, self.backupEpoch)));
break;
}
}
co_return;
} catch (Error& e) {
err = e;
}
if (err.code() == error_code_worker_removed) {
pull = Void(); // cancels pulling
self.stop(); // lets uploadData finish its current upload and exit
try {
co_await done;
} catch (Error& shutdownErr) {
TraceEvent("RangePartitionedBWShutdownError", self.myId).errorUnsuppressed(shutdownErr);
}
}
TraceEvent("RangePartitionedBWTerminated", self.myId).errorUnsuppressed(err);
if (err.code() != error_code_actor_cancelled && err.code() != error_code_worker_removed) {
throw err;
}
}
namespace {
PartitionMap makeSamplePartitionMap() {
PartitionMap pm;
pm[Tag(tagLocalityRangePartitionedBackup, 0)] = {
Partition(0, KeyRangeRef("a"_sr, "c"_sr)),
Partition(1, KeyRangeRef("c"_sr, "f"_sr)),
};
pm[Tag(tagLocalityRangePartitionedBackup, 1)] = {
Partition(2, KeyRangeRef("f"_sr, "m"_sr)),
Partition(3, KeyRangeRef("m"_sr, "z"_sr)),
};
return pm;
}
void assertPartitionMapsEqual(PartitionMap const& a, PartitionMap const& b) {
ASSERT_EQ(a.size(), b.size());
for (auto const& [tag, list] : a) {
auto it = b.find(tag);
ASSERT(it != b.end());
ASSERT_EQ(list.size(), it->second.size());
for (size_t i = 0; i < list.size(); ++i) {
ASSERT_EQ(list[i].partitionId, it->second[i].partitionId);
ASSERT(list[i].ranges == it->second[i].ranges);
}
}
}
} // namespace
TEST_CASE("/RangePartitionedBackupWorker/PartitionMapMessage/RoundTrip") {
PartitionMap original = makeSamplePartitionMap();
PartitionMapMessage outgoing(original);
BinaryWriter wr(AssumeVersion(g_network->protocolVersion()));
wr << outgoing;
Standalone<StringRef> bytes = wr.toValue();
ArenaReader reader(bytes.arena(), bytes, AssumeVersion(g_network->protocolVersion()));
ASSERT(PartitionMapMessage::isNextIn(reader));
PartitionMapMessage incoming;
reader >> incoming;
assertPartitionMapsEqual(original, incoming.partitionMap);
return Void();
}
TEST_CASE("/RangePartitionedBackupWorker/PartitionMapMessage/RoundTripEmpty") {
PartitionMapMessage outgoing(PartitionMap{});
BinaryWriter wr(AssumeVersion(g_network->protocolVersion()));
wr << outgoing;
Standalone<StringRef> bytes = wr.toValue();
ArenaReader reader(bytes.arena(), bytes, AssumeVersion(g_network->protocolVersion()));
ASSERT(PartitionMapMessage::isNextIn(reader));
PartitionMapMessage incoming;
reader >> incoming;
ASSERT(incoming.partitionMap.empty());
return Void();
}
TEST_CASE("/RangePartitionedBackupWorker/PartitionMapMessage/IsNextInLeadingByte") {
BinaryWriter wr(AssumeVersion(g_network->protocolVersion()));
PartitionMapMessage outgoing(makeSamplePartitionMap());
wr << outgoing;
Standalone<StringRef> bytes = wr.toValue();
ASSERT(!bytes.empty());
ASSERT(PartitionMapMessage::startsPartitionMapMessage(bytes[0]));
ArenaReader pmmReader(bytes.arena(), bytes, AssumeVersion(g_network->protocolVersion()));
ASSERT(PartitionMapMessage::isNextIn(pmmReader));
uint8_t notPmm = MutationRef::SetValue;
StringRef otherBytes(&notPmm, 1);
Arena arena;
ArenaReader otherReader(arena, otherBytes, AssumeVersion(g_network->protocolVersion()));
ASSERT(!PartitionMapMessage::isNextIn(otherReader));
return Void();
}
// TODO akanksha: Remove once a production caller of rangePartitionedBackupWorker() is wired up;
// this only exists to keep TEST_CASEs in this file from being dead-stripped from the static lib.
void forceLinkRangePartitionedBackupWorkerTests() {}
Reference in New Issue View Git Blame Copy Permalink