From 5ddf08dfe5a62cb588d0eda19cba51e8749bdcbc Mon Sep 17 00:00:00 2001 From: Josh Slocum Date: Wed, 22 Sep 2021 12:46:20 -0500 Subject: [PATCH] Got basic range reassignment working --- fdbclient/BlobWorkerCommon.h | 5 +- fdbclient/BlobWorkerInterface.h | 14 +- fdbclient/SystemData.cpp | 27 +- fdbclient/SystemData.h | 17 +- fdbserver/BlobManager.actor.cpp | 44 +-- fdbserver/BlobWorker.actor.cpp | 526 ++++++++++++++++++++------------ 6 files changed, 389 insertions(+), 244 deletions(-) diff --git a/fdbclient/BlobWorkerCommon.h b/fdbclient/BlobWorkerCommon.h index 9216c1d300..bc4c2a687c 100644 --- a/fdbclient/BlobWorkerCommon.h +++ b/fdbclient/BlobWorkerCommon.h @@ -35,6 +35,8 @@ struct BlobWorkerStats { Counter changeFeedInputBytes; Counter readReqTotalFilesReturned; Counter readReqDeltaBytesReturned; + Counter commitVersionChecks; + Counter granuleUpdateErrors; int numRangesAssigned; int mutationBytesBuffered; @@ -54,7 +56,8 @@ struct BlobWorkerStats { rangeAssignmentRequests("RangeAssignmentRequests", cc), readRequests("ReadRequests", cc), wrongShardServer("WrongShardServer", cc), changeFeedInputBytes("RangeFeedInputBytes", cc), readReqTotalFilesReturned("ReadReqTotalFilesReturned", cc), - readReqDeltaBytesReturned("ReadReqDeltaBytesReturned", cc), numRangesAssigned(0), mutationBytesBuffered(0) { + readReqDeltaBytesReturned("ReadReqDeltaBytesReturned", cc), commitVersionChecks("CommitVersionChecks", cc), + granuleUpdateErrors("GranuleUpdateErrors", cc), numRangesAssigned(0), mutationBytesBuffered(0) { specialCounter(cc, "NumRangesAssigned", [this]() { return this->numRangesAssigned; }); specialCounter(cc, "MutationBytesBuffered", [this]() { return this->mutationBytesBuffered; }); specialCounter(cc, "ActiveReadRequests", [this]() { return this->activeReadRequests; }); diff --git a/fdbclient/BlobWorkerInterface.h b/fdbclient/BlobWorkerInterface.h index 11d75a48fa..e122d732e3 100644 --- a/fdbclient/BlobWorkerInterface.h +++ b/fdbclient/BlobWorkerInterface.h @@ -120,17 +120,11 @@ struct AssignBlobRangeRequest { KeyRangeRef keyRange; int64_t managerEpoch; int64_t managerSeqno; - // If continueAssignment is true, this is just to instruct the worker that it still owns the range, so it should - // re-snapshot it and continue. If continueAssignment is false and previousGranules is empty, this is either the - // initial assignment to construct a previously non-existent granule, or a reassignment. Depending on what state - // exists for the granule currently, the worker will either start a new granule, or just pick up from where the - // previous worker left off. + // If continueAssignment is true, this is just to instruct the worker that it *still* owns the range, so it should + // re-snapshot it and continue. - // For a split or merge, continueAssignment==false. - // For a split, previousGranules will contain one granule that contains keyRange. For a merge, previousGranules will - // contain two or more granules, the union of which will be keyRange. + // For an initial assignment, reassignent, split, or merge, continueAssignment==false. bool continueAssignment; - VectorRef previousGranules; // only set if there is a granule boundary change ReplyPromise reply; @@ -138,7 +132,7 @@ struct AssignBlobRangeRequest { template void serialize(Ar& ar) { - serializer(ar, keyRange, managerEpoch, managerSeqno, continueAssignment, previousGranules, reply, arena); + serializer(ar, keyRange, managerEpoch, managerSeqno, continueAssignment, reply, arena); } }; diff --git a/fdbclient/SystemData.cpp b/fdbclient/SystemData.cpp index cb45fc408f..4b07e44232 100644 --- a/fdbclient/SystemData.cpp +++ b/fdbclient/SystemData.cpp @@ -1112,6 +1112,7 @@ const KeyRangeRef blobGranuleFileKeys(LiteralStringRef("\xff\x02/bgf/"), Literal const KeyRangeRef blobGranuleMappingKeys(LiteralStringRef("\xff\x02/bgm/"), LiteralStringRef("\xff\x02/bgm0")); const KeyRangeRef blobGranuleLockKeys(LiteralStringRef("\xff\x02/bgl/"), LiteralStringRef("\xff\x02/bgl0")); const KeyRangeRef blobGranuleSplitKeys(LiteralStringRef("\xff\x02/bgs/"), LiteralStringRef("\xff\x02/bgs0")); +const KeyRangeRef blobGranuleHistoryKeys(LiteralStringRef("\xff\x02/bgh/"), LiteralStringRef("\xff\x02/bgh0")); const Value blobGranuleMappingValueFor(UID const& workerID) { BinaryWriter wr(Unversioned()); @@ -1147,14 +1148,32 @@ std::tuple decodeBlobGranuleLockValue(const ValueRef& val const Value blobGranuleSplitValueFor(BlobGranuleSplitState st) { BinaryWriter wr(Unversioned()); wr << st; + return addVersionStampAtEnd(wr.toValue()); +} + +std::pair decodeBlobGranuleSplitValue(const ValueRef& value) { + BlobGranuleSplitState st; + Version v; + BinaryReader reader(value, Unversioned()); + reader >> st; + reader >> v; + return std::pair(st, v); +} + +// const Value blobGranuleHistoryValueFor(VectorRef const& parentGranules); +// VectorRef decodeBlobGranuleHistoryValue(ValueRef const& value); + +const Value blobGranuleHistoryValueFor(VectorRef const& parentGranules) { + BinaryWriter wr(Unversioned()); + wr << parentGranules; return wr.toValue(); } -BlobGranuleSplitState decodeBlobGranuleSplitValue(const ValueRef& value) { - BlobGranuleSplitState st; +VectorRef decodeBlobGranuleHistoryValue(const ValueRef& value) { + VectorRef parentGranules; BinaryReader reader(value, Unversioned()); - reader >> st; - return st; + reader >> parentGranules; + return parentGranules; } const KeyRangeRef blobWorkerListKeys(LiteralStringRef("\xff\x02/bwList/"), LiteralStringRef("\xff\x02/bwList0")); diff --git a/fdbclient/SystemData.h b/fdbclient/SystemData.h index 02eb4dbe18..8dc5a2fd5c 100644 --- a/fdbclient/SystemData.h +++ b/fdbclient/SystemData.h @@ -526,21 +526,24 @@ int64_t decodeBlobManagerEpochValue(ValueRef const& value); // blob granule keys -// \xff/bgf/(startKey, endKey, {snapshot|delta}, version) = [[filename]] +// \xff\x02/bgf/(startKey, endKey, {snapshot|delta}, version) = [[filename]] extern const KeyRangeRef blobGranuleFileKeys; // TODO could shrink the size of the mapping keyspace by using something similar to tags instead of UIDs. We'd probably // want to do that in V1 or it'd be a big migration. -// \xff/bgm/[[begin]] = [[BlobWorkerUID]] +// \xff\x02/bgm/[[begin]] = [[BlobWorkerUID]] extern const KeyRangeRef blobGranuleMappingKeys; -// \xff/bgl/(begin,end) = (epoch, seqno, changefeed id) +// \xff\x02/bgl/(begin,end) = (epoch, seqno, changefeed id) extern const KeyRangeRef blobGranuleLockKeys; -// \xff/bgs/(oldbegin,oldend,newbegin) = state +// \xff\x02/bgs/(oldbegin,oldend,newbegin) = state extern const KeyRangeRef blobGranuleSplitKeys; +// \xff\x02/bgh/(start,end) = [(oldbegin, oldend)] +extern const KeyRangeRef blobGranuleHistoryKeys; + const Value blobGranuleMappingValueFor(UID const& workerID); UID decodeBlobGranuleMappingValue(ValueRef const& value); @@ -548,8 +551,12 @@ const Value blobGranuleLockValueFor(int64_t epochNum, int64_t sequenceNum, UID c // FIXME: maybe just define a struct? std::tuple decodeBlobGranuleLockValue(ValueRef const& value); +// these are versionstamped const Value blobGranuleSplitValueFor(BlobGranuleSplitState st); -BlobGranuleSplitState decodeBlobGranuleSplitValue(ValueRef const& value); +std::pair decodeBlobGranuleSplitValue(ValueRef const& value); + +const Value blobGranuleHistoryValueFor(VectorRef const& parentGranules); +VectorRef decodeBlobGranuleHistoryValue(ValueRef const& value); // \xff/bwl/[[BlobWorkerID]] = [[BlobWorkerInterface]] extern const KeyRangeRef blobWorkerListKeys; diff --git a/fdbserver/BlobManager.actor.cpp b/fdbserver/BlobManager.actor.cpp index 8e655b08d9..7d12c7c681 100644 --- a/fdbserver/BlobManager.actor.cpp +++ b/fdbserver/BlobManager.actor.cpp @@ -165,11 +165,9 @@ void getRanges(std::vector>& results, KeyRangeMap previousRanges; RangeAssignmentData() : continueAssignment(false) {} - RangeAssignmentData(bool continueAssignment, std::vector previousRanges) - : continueAssignment(continueAssignment), previousRanges(previousRanges) {} + RangeAssignmentData(bool continueAssignment) : continueAssignment(continueAssignment) {} }; struct RangeRevokeData { @@ -332,9 +330,6 @@ ACTOR Future doRangeAssignment(BlobManagerData* bmData, RangeAssignment as req.managerEpoch = bmData->epoch; req.managerSeqno = seqNo; req.continueAssignment = assignment.assign.get().continueAssignment; - for (auto& it : assignment.assign.get().previousRanges) { - req.previousGranules.push_back_deep(req.arena, it); - } AssignBlobRangeReply _rep = wait(bmData->workersById[workerID].assignBlobRangeRequest.getReply(req)); rep = _rep; } else { @@ -580,7 +575,7 @@ ACTOR Future monitorClientRanges(BlobManagerData* bmData) { RangeAssignment ra; ra.isAssign = true; ra.keyRange = range; - ra.assign = RangeAssignmentData(); // continue=false, no previous granules + ra.assign = RangeAssignmentData(false); // continue=false bmData->rangesToAssign.send(ra); } } @@ -640,8 +635,7 @@ ACTOR Future maybeSplitRange(BlobManagerData* bmData, UID currentWorkerId, raContinue.isAssign = true; raContinue.worker = currentWorkerId; raContinue.keyRange = range; - raContinue.assign = - RangeAssignmentData(true, std::vector()); // continue, no "previous" range to do handover + raContinue.assign = RangeAssignmentData(true); // continue assignment and re-snapshot bmData->rangesToAssign.send(raContinue); return Void(); } @@ -686,17 +680,25 @@ ACTOR Future maybeSplitRange(BlobManagerData* bmData, UID currentWorkerId, ASSERT(newLockSeqno >= std::get<1>(prevGranuleLock)); } + // acquire granule lock so nobody else can make changes to this granule. tr->set(lockKey, blobGranuleLockValueFor(bmData->epoch, newLockSeqno, std::get<2>(prevGranuleLock))); + Standalone> history; + history.push_back(history.arena(), range); + Value historyValue = blobGranuleHistoryValueFor(history); // set up split metadata for (int i = 0; i < newRanges.size() - 1; i++) { - Tuple key; - key.append(range.begin).append(range.end).append(newRanges[i]); - tr->set(key.getDataAsStandalone().withPrefix(blobGranuleSplitKeys.begin), - blobGranuleSplitValueFor(BlobGranuleSplitState::Started)); + Tuple splitKey; + splitKey.append(range.begin).append(range.end).append(newRanges[i]); + tr->atomicOp(splitKey.getDataAsStandalone().withPrefix(blobGranuleSplitKeys.begin), + blobGranuleSplitValueFor(BlobGranuleSplitState::Started), + MutationRef::SetVersionstampedValue); - // acquire granule lock so nobody else can make changes to this granule. + Tuple historyKey; + historyKey.append(newRanges[i]).append(newRanges[i + 1]); + tr->set(historyKey.getDataAsStandalone().withPrefix(blobGranuleHistoryKeys.begin), historyValue); } + wait(tr->commit()); break; } catch (Error& e) { @@ -720,14 +722,12 @@ ACTOR Future maybeSplitRange(BlobManagerData* bmData, UID currentWorkerId, raRevoke.revoke = RangeRevokeData(false); // not a dispose bmData->rangesToAssign.send(raRevoke); - std::vector originalRange; - originalRange.push_back(range); for (int i = 0; i < newRanges.size() - 1; i++) { // reassign new range and do handover of previous range RangeAssignment raAssignSplit; raAssignSplit.isAssign = true; raAssignSplit.keyRange = KeyRangeRef(newRanges[i], newRanges[i + 1]); - raAssignSplit.assign = RangeAssignmentData(false, originalRange); + raAssignSplit.assign = RangeAssignmentData(false); // don't care who this range gets assigned to bmData->rangesToAssign.send(raAssignSplit); } @@ -839,14 +839,13 @@ ACTOR Future rangeMover(BlobManagerData* bmData) { RangeAssignment assignNew; assignNew.isAssign = true; assignNew.keyRange = randomRange.range(); - assignNew.assign = - RangeAssignmentData(false, std::vector()); // not a continue, no boundary change + assignNew.assign = RangeAssignmentData(false); // not a continue bmData->rangesToAssign.send(assignNew); break; } } if (tries == 0 && BM_DEBUG) { - printf("Range mover couldn't find range to move, skipping\n"); + printf("Range mover couldn't find random range to move, skipping\n"); } } else if (BM_DEBUG) { printf("Range mover found %d workers, skipping\n", bmData->workerAssignments.size()); @@ -893,8 +892,9 @@ ACTOR Future blobManager(LocalityData locality, Reference granuleSplitFrom; Optional blobFilesToSnapshot; + Optional existingFiles; }; // FIXME: the circular dependencies here are getting kind of gross @@ -105,8 +106,10 @@ struct GranuleMetadata : NonCopyable, ReferenceCounted { Promise cancelled; Promise readable; - Future start(BlobWorkerData* bwData, AssignBlobRangeRequest req) { + AssignBlobRangeRequest originalReq; + Future start(BlobWorkerData* bwData, AssignBlobRangeRequest req) { + originalReq = req; assignFuture = persistAssignWorkerRange(bwData, req); fileUpdaterFuture = blobGranuleUpdateFiles(bwData, Reference::addRef(this)); @@ -169,6 +172,8 @@ struct BlobWorkerData { AsyncVar knownCommittedVersion; uint64_t knownCommittedCheckCount = 0; + PromiseStream granuleUpdateErrors; + BlobWorkerData(UID id, Database db) : id(id), db(db), stats(id, SERVER_KNOBS->WORKER_LOGGING_INTERVAL) {} ~BlobWorkerData() { printf("Destroying blob worker data for %s\n", id.toString().c_str()); } @@ -371,7 +376,7 @@ ACTOR Future updateGranuleSplitState(Transaction* tr, ASSERT(key.getString(0) == previousGranule.begin); ASSERT(key.getString(1) == previousGranule.end); - BlobGranuleSplitState st = decodeBlobGranuleSplitValue(it.value); + BlobGranuleSplitState st = decodeBlobGranuleSplitValue(it.value).first; ASSERT(st != BlobGranuleSplitState::Unknown); if (st == BlobGranuleSplitState::Started) { totalStarted++; @@ -430,7 +435,8 @@ ACTOR Future updateGranuleSplitState(Transaction* tr, // FIXME: enable once implemented // tr.stopChangeFeed(KeyRef(prevChangeFeedId.toString())); } - tr->set(myStateKey, blobGranuleSplitValueFor(newState)); + // TODO also add versionstamp + tr->atomicOp(myStateKey, blobGranuleSplitValueFor(newState), MutationRef::SetVersionstampedValue); } } else if (BW_DEBUG) { printf("Ignoring granule [%s - %s) split state from [%s - %s) %d -> %d\n", @@ -445,6 +451,22 @@ ACTOR Future updateGranuleSplitState(Transaction* tr, return Void(); } +// returns the split state for a given granule on granule reassignment +ACTOR Future> getGranuleSplitState(Transaction* tr, + KeyRange previousGranule, + KeyRange currentGranule) { + Tuple myStateTuple; + myStateTuple.append(previousGranule.begin).append(previousGranule.end).append(currentGranule.begin); + Key myStateKey = myStateTuple.getDataAsStandalone().withPrefix(blobGranuleSplitKeys.begin); + + Optional st = wait(tr->get(myStateKey)); + if (!st.present()) { + // must have been that all granules reached done and state was cleaned up + return std::pair(BlobGranuleSplitState::Done, invalidVersion); + } + return decodeBlobGranuleSplitValue(st.get()); +} + static Value getFileValue(std::string fname, int64_t offset, int64_t length) { Tuple fileValue; fileValue.append(fname).append(offset).append(length); @@ -860,6 +882,20 @@ static Future handleCompletedDeltaFile(BlobWorkerData* bwData, return Future(Void()); } +// if we get an i/o error updating files, or a rollback, reassign the granule to ourselves and start fresh +// FIXME: is this the correct set of errors? +static bool granuleCanRetry(const Error& e) { + switch (e.code()) { + case error_code_please_reboot: + case error_code_io_error: + case error_code_io_timeout: + case error_code_http_request_failed: + return true; + default: + return false; + }; +} + // updater for a single granule // TODO: this is getting kind of large. Should try to split out this actor if it continues to grow? // FIXME: handle errors here (forward errors) @@ -876,6 +912,7 @@ ACTOR Future blobGranuleUpdateFiles(BlobWorkerData* bwData, Reference oldChangeFeedDataComplete; state Key cfKey; state Optional oldCFKey; + state bool snapshotEligible; // just wrote a delta file or just took granule over from another worker try { // set resume snapshot so it's not valid until we pause to ask the blob manager for a re-snapshot @@ -910,11 +947,24 @@ ACTOR Future blobGranuleUpdateFiles(BlobWorkerData* bwData, Reference(); // not valid! + // if this is a reassign, calculate how close to a snapshot the previous owner was + if (changeFeedInfo.existingFiles.present()) { + GranuleFiles files = changeFeedInfo.existingFiles.get(); + if (!files.snapshotFiles.empty() && !files.deltaFiles.empty()) { + Version snapshotVersion = files.snapshotFiles.back().version; + for (int i = files.deltaFiles.size() - 1; i >= 0; i--) { + if (files.deltaFiles[i].version > snapshotVersion) { + metadata->bytesInNewDeltaFiles += files.deltaFiles[i].length; + } + } + } + metadata->files = changeFeedInfo.existingFiles.get(); + snapshotEligible = true; + } + // FIXME: not true for reassigns - ASSERT(changeFeedInfo.doSnapshot); if (!changeFeedInfo.doSnapshot) { startVersion = changeFeedInfo.previousDurableVersion; - // TODO metadata.files = } else { if (changeFeedInfo.blobFilesToSnapshot.present()) { inFlightBlobSnapshot = compactFromBlob(bwData, metadata, changeFeedInfo.blobFilesToSnapshot.get()); @@ -1039,8 +1089,7 @@ ACTOR Future blobGranuleUpdateFiles(BlobWorkerData* bwData, Referenceid) - .detail("GranuleStart", metadata->keyRange.begin) - .detail("GranuleEnd", metadata->keyRange.end) + .detail("Granule", metadata->keyRange) .detail("Version", metadata->bufferedDeltaVersion.get()); // launch pipelined, but wait for previous operation to complete before persisting to FDB @@ -1079,101 +1128,108 @@ ACTOR Future blobGranuleUpdateFiles(BlobWorkerData* bwData, ReferencebytesInNewDeltaFiles >= SERVER_KNOBS->BG_DELTA_BYTES_BEFORE_COMPACT && - !readOldChangeFeed && !lastFromOldChangeFeed) { - - if (BW_DEBUG && (inFlightBlobSnapshot.isValid() || !inFlightDeltaFiles.empty())) { - printf("Granule [%s - %s) ready to re-snapshot, waiting for outstanding %d snapshot and %d " - "deltas to " - "finish\n", - metadata->keyRange.begin.printable().c_str(), - metadata->keyRange.end.printable().c_str(), - inFlightBlobSnapshot.isValid() ? 1 : 0, - inFlightDeltaFiles.size()); - } - // wait for all in flight snapshot/delta files - if (inFlightBlobSnapshot.isValid()) { - BlobFileIndex completedSnapshot = wait(inFlightBlobSnapshot); - metadata->files.snapshotFiles.push_back(completedSnapshot); - metadata->durableSnapshotVersion.set(completedSnapshot.version); - inFlightBlobSnapshot = Future(); // not valid! - } - for (auto& it : inFlightDeltaFiles) { - BlobFileIndex completedDeltaFile = wait(it); - wait(handleCompletedDeltaFile( - bwData, metadata, completedDeltaFile, cfKey, changeFeedInfo.changeFeedStartVersion)); - } - inFlightDeltaFiles.clear(); - - if (BW_DEBUG) { - printf("Granule [%s - %s) checking with BM for re-snapshot after %d bytes\n", - metadata->keyRange.begin.printable().c_str(), - metadata->keyRange.end.printable().c_str(), - metadata->bytesInNewDeltaFiles); - } - - TraceEvent("BlobGranuleSnapshotCheck", bwData->id) - .detail("GranuleStart", metadata->keyRange.begin) - .detail("GranuleEnd", metadata->keyRange.end) - .detail("Version", metadata->durableDeltaVersion.get()); - - // Save these from the start so repeated requests are idempotent - // Need to retry in case response is dropped or manager changes. Eventually, a manager will - // either reassign the range with continue=true, or will revoke the range. But, we will keep the - // range open at this version for reads until that assignment change happens - metadata->resumeSnapshot.reset(); - state int64_t statusEpoch = metadata->continueEpoch; - state int64_t statusSeqno = metadata->continueSeqno; - loop { - bwData->currentManagerStatusStream.send( - GranuleStatusReply(metadata->keyRange, true, statusEpoch, statusSeqno)); - - Optional result = wait(timeout(metadata->resumeSnapshot.getFuture(), 1.0)); - if (result.present()) { - break; - } - if (BW_DEBUG) { - printf("Granule [%s - %s)\n, hasn't heard back from BM, re-sending status\n", - metadata->keyRange.begin.printable().c_str(), - metadata->keyRange.end.printable().c_str()); - } - } - - if (BW_DEBUG) { - printf("Granule [%s - %s) re-snapshotting after %d bytes\n", - metadata->keyRange.begin.printable().c_str(), - metadata->keyRange.end.printable().c_str(), - metadata->bytesInNewDeltaFiles); - } - TraceEvent("BlobGranuleSnapshotFile", bwData->id) - .detail("GranuleStart", metadata->keyRange.begin) - .detail("GranuleEnd", metadata->keyRange.end) - .detail("Version", metadata->durableDeltaVersion.get()); - // TODO: this could read from FDB instead if it knew there was a large range clear at the end or - // it knew the granule was small, or something - // BlobFileIndex newSnapshotFile = wait(compactFromBlob(bwData, metadata, metadata->files)); - - // Have to copy files object so that adding to it as we start writing new delta files in - // parallel doesn't conflict. We could also pass the snapshot version and ignore any snapshot - // files >= version and any delta files > version, but that's more complicated - inFlightBlobSnapshot = compactFromBlob(bwData, metadata, metadata->files); - metadata->pendingSnapshotVersion = metadata->durableDeltaVersion.get(); - - // reset metadata - metadata->bytesInNewDeltaFiles = 0; - } + snapshotEligible = true; } + if (snapshotEligible && metadata->bytesInNewDeltaFiles >= SERVER_KNOBS->BG_DELTA_BYTES_BEFORE_COMPACT && + !readOldChangeFeed && !lastFromOldChangeFeed) { + + if (BW_DEBUG && (inFlightBlobSnapshot.isValid() || !inFlightDeltaFiles.empty())) { + printf("Granule [%s - %s) ready to re-snapshot, waiting for outstanding %d snapshot and %d " + "deltas to " + "finish\n", + metadata->keyRange.begin.printable().c_str(), + metadata->keyRange.end.printable().c_str(), + inFlightBlobSnapshot.isValid() ? 1 : 0, + inFlightDeltaFiles.size()); + } + // wait for all in flight snapshot/delta files + if (inFlightBlobSnapshot.isValid()) { + BlobFileIndex completedSnapshot = wait(inFlightBlobSnapshot); + metadata->files.snapshotFiles.push_back(completedSnapshot); + metadata->durableSnapshotVersion.set(completedSnapshot.version); + inFlightBlobSnapshot = Future(); // not valid! + } + for (auto& it : inFlightDeltaFiles) { + BlobFileIndex completedDeltaFile = wait(it); + wait(handleCompletedDeltaFile( + bwData, metadata, completedDeltaFile, cfKey, changeFeedInfo.changeFeedStartVersion)); + } + inFlightDeltaFiles.clear(); + + if (BW_DEBUG) { + printf("Granule [%s - %s) checking with BM for re-snapshot after %d bytes\n", + metadata->keyRange.begin.printable().c_str(), + metadata->keyRange.end.printable().c_str(), + metadata->bytesInNewDeltaFiles); + } + + TraceEvent("BlobGranuleSnapshotCheck", bwData->id) + .detail("Granule", metadata->keyRange) + .detail("Version", metadata->durableDeltaVersion.get()); + + // Save these from the start so repeated requests are idempotent + // Need to retry in case response is dropped or manager changes. Eventually, a manager will + // either reassign the range with continue=true, or will revoke the range. But, we will keep the + // range open at this version for reads until that assignment change happens + metadata->resumeSnapshot.reset(); + state int64_t statusEpoch = metadata->continueEpoch; + state int64_t statusSeqno = metadata->continueSeqno; + loop { + bwData->currentManagerStatusStream.send( + GranuleStatusReply(metadata->keyRange, true, statusEpoch, statusSeqno)); + + Optional result = wait(timeout(metadata->resumeSnapshot.getFuture(), 1.0)); + if (result.present()) { + break; + } + if (BW_DEBUG) { + printf("Granule [%s - %s)\n, hasn't heard back from BM, re-sending status\n", + metadata->keyRange.begin.printable().c_str(), + metadata->keyRange.end.printable().c_str()); + } + } + + if (BW_DEBUG) { + printf("Granule [%s - %s) re-snapshotting after %d bytes\n", + metadata->keyRange.begin.printable().c_str(), + metadata->keyRange.end.printable().c_str(), + metadata->bytesInNewDeltaFiles); + } + TraceEvent("BlobGranuleSnapshotFile", bwData->id) + .detail("Granule", metadata->keyRange) + .detail("Version", metadata->durableDeltaVersion.get()); + // TODO: this could read from FDB instead if it knew there was a large range clear at the end or + // it knew the granule was small, or something + // BlobFileIndex newSnapshotFile = wait(compactFromBlob(bwData, metadata, metadata->files)); + + // Have to copy files object so that adding to it as we start writing new delta files in + // parallel doesn't conflict. We could also pass the snapshot version and ignore any snapshot + // files >= version and any delta files > version, but that's more complicated + inFlightBlobSnapshot = compactFromBlob(bwData, metadata, metadata->files); + metadata->pendingSnapshotVersion = metadata->durableDeltaVersion.get(); + + // reset metadata + metadata->bytesInNewDeltaFiles = 0; + } + snapshotEligible = false; + // finally, after we optionally write delta and snapshot files, add new mutations to buffer if (!deltas.mutations.empty()) { if (deltas.mutations.size() == 1 && deltas.mutations.back().param1 == lastEpochEndPrivateKey) { // FIXME: do rollback here!!! look at ChangeFeedRollback trace event if (BW_DEBUG) { - printf("BW [%s - %s) NEEDS TO ROLLBACK @ %lld\n", + printf("BW [%s - %s) ROLLBACK @ %lld\n", metadata->keyRange.begin.printable().c_str(), metadata->keyRange.end.printable().c_str(), deltas.version); + TraceEvent(SevWarn, "GranuleRollback", bwData->id) + .detail("Granule", metadata->keyRange) + .detail("Version", deltas.version); } + // FIXME: handle this better! If rollback version is after pendingDurableVersion, don't need to + // relinquish whole granule, just need to discard in-memory deltas and buffered delta version + throw please_reboot(); } else { for (auto& delta : deltas.mutations) { // 8 for version, 1 for type, 4 for each param length then actual param size @@ -1214,23 +1270,44 @@ ACTOR Future blobGranuleUpdateFiles(BlobWorkerData* bwData, ReferencekeyRange.begin.printable().c_str(), - metadata->keyRange.end.printable().c_str(), - e.name()); + + if (metadata->cancelled.canBeSet()) { + metadata->cancelled.send(Void()); + } + + if (e.code() == error_code_granule_assignment_conflict) { + TraceEvent(SevInfo, "GranuleAssignmentConflict", bwData->id).detail("Granule", metadata->keyRange); + } else { + if (e.code() != error_code_please_reboot) { + ++bwData->stats.granuleUpdateErrors; + if (BW_DEBUG) { + printf("Granule file updater for [%s - %s) got error %s, exiting\n", + metadata->keyRange.begin.printable().c_str(), + metadata->keyRange.end.printable().c_str(), + e.name()); + } + TraceEvent(SevWarn, "GranuleFileUpdaterError", bwData->id) + .detail("Granule", metadata->keyRange) + .error(e); + } + + if (granuleCanRetry(e)) { + // explicitly cancel all outstanding write futures BEFORE updating promise stream, to ensure they can't + // update files after the re-assigned granule acquires the lock + inFlightBlobSnapshot.cancel(); + for (auto& f : inFlightDeltaFiles) { + f.cancel(); + } + + bwData->granuleUpdateErrors.send(metadata->originalReq); + } } - TraceEvent(SevError, "GranuleFileUpdaterError", bwData->id) - .detail("GranuleStart", metadata->keyRange.begin) - .detail("GranuleEnd", metadata->keyRange.end) - .error(e); - // TODO in this case, need to update range mapping that it doesn't have the range, and/or try to re-"open" the - // range if someone else doesn't have it throw e; } } -// TODO might want to separate this out for valid values for range assignments vs read requests +// TODO might want to separate this out for valid values for range assignments vs read requests. Assignment conflict +// isn't valid for read requests but is for assignments namespace { bool canReplyWith(Error e) { switch (e.code()) { @@ -1238,7 +1315,6 @@ bool canReplyWith(Error e) { case error_code_future_version: // not thrown yet case error_code_wrong_shard_server: case error_code_process_behind: // not thrown yet - // TODO should we reply with granule_assignment_conflict? return true; default: return false; @@ -1353,13 +1429,13 @@ ACTOR Future handleBlobGranuleFileRequest(BlobWorkerData* bwData, BlobGran throw transaction_too_old(); } if (metadata->cancelled.isSet()) { - throw transaction_too_old(); + throw wrong_shard_server(); } Future waitForVersionFuture = waitForVersion(metadata, req.readVersion); if (!waitForVersionFuture.isReady()) { choose { when(wait(waitForVersionFuture)) {} - when(wait(metadata->cancelled.getFuture())) { throw transaction_too_old(); } + when(wait(metadata->cancelled.getFuture())) { throw wrong_shard_server(); } } } @@ -1458,18 +1534,11 @@ ACTOR Future handleBlobGranuleFileRequest(BlobWorkerData* bwData, BlobGran return Void(); } -// FIXME: in split, need to persist version of created change feed so if worker immediately fails afterwards, new worker -// picking up the splitting shard knows where the change feed handoff point is. OR need to have change feed return -// end_of_stream when it knows it has nothing up to the specified end version, and use the commit takeover version as -// the end version. If it sealed successfully there would trivially be nothing between the seal version and the new -// commit takeover version. You'd need to start the new change feed at the seal version though, not the commit takeover -// version. ACTOR Future persistAssignWorkerRange(BlobWorkerData* bwData, AssignBlobRangeRequest req) { ASSERT(!req.continueAssignment); state Transaction tr(bwData->db); state Key lockKey = granuleLockKey(req.keyRange); state GranuleChangeFeedInfo info; - info.changeFeedId = deterministicRandom()->randomUniqueID(); if (BW_DEBUG) { printf("%s persisting assignment [%s - %s)\n", bwData->id.toString().c_str(), @@ -1485,25 +1554,27 @@ ACTOR Future persistAssignWorkerRange(BlobWorkerData* bwD // FIXME: could add list of futures and do the different parts that are disjoint in parallel? info.changeFeedStartVersion = invalidVersion; Optional prevLockValue = wait(tr.get(lockKey)); - if (prevLockValue.present()) { + state bool hasPrevOwner = prevLockValue.present(); + if (hasPrevOwner) { std::tuple prevOwner = decodeBlobGranuleLockValue(prevLockValue.get()); acquireGranuleLock(req.managerEpoch, req.managerSeqno, std::get<0>(prevOwner), std::get<1>(prevOwner)); info.changeFeedId = std::get<2>(prevOwner); - info.doSnapshot = false; - ASSERT(info.changeFeedId == UID()); - - /*info.existingFiles = wait(loadPreviousFiles(&tr, req.keyRange)); + GranuleFiles granuleFiles = wait(loadPreviousFiles(&tr, req.keyRange)); + info.existingFiles = granuleFiles; info.previousDurableVersion = info.existingFiles.get().deltaFiles.empty() ? info.existingFiles.get().snapshotFiles.back().version - : info.existingFiles.get().deltaFiles.back().version;*/ - // FIXME: Handle granule reassignments! - ASSERT(false); + : info.existingFiles.get().deltaFiles.back().version; + info.doSnapshot = info.existingFiles.get().snapshotFiles.empty(); + // for the non-splitting cases, this doesn't need to be 100% accurate, it just needs to be smaller than + // the next delta file write. + info.changeFeedStartVersion = info.previousDurableVersion; } else { // else we are first, no need to check for owner conflict // FIXME: use actual 16 bytes of UID instead of converting it to 32 character string and then that to // bytes + info.changeFeedId = deterministicRandom()->randomUniqueID(); wait(tr.registerChangeFeed(StringRef(info.changeFeedId.toString()), req.keyRange)); info.doSnapshot = true; info.previousDurableVersion = invalidVersion; @@ -1512,49 +1583,85 @@ ACTOR Future persistAssignWorkerRange(BlobWorkerData* bwD tr.set(lockKey, blobGranuleLockValueFor(req.managerEpoch, req.managerSeqno, info.changeFeedId)); wait(krmSetRange(&tr, blobGranuleMappingKeys.begin, req.keyRange, blobGranuleMappingValueFor(bwData->id))); + Tuple historyKey; + historyKey.append(req.keyRange.end).append(req.keyRange.end); + state Optional parentGranulesValue = + wait(tr.get(historyKey.getDataAsStandalone().withPrefix(blobGranuleHistoryKeys.begin))); + // If anything in previousGranules, need to do the handoff logic and set ret.previousChangeFeedId, and the // previous durable version will come from the previous granules - if (!req.previousGranules.empty()) { + if (parentGranulesValue.present()) { + // references memory in parentGranulesValue standalone + state VectorRef parentGranules = decodeBlobGranuleHistoryValue(parentGranulesValue.get()); + // TODO REMOVE + if (BW_DEBUG) { + printf("Decoded parent granules for [%s - %s)\n", + req.keyRange.begin.printable().c_str(), + req.keyRange.end.printable().c_str()); + for (auto& pg : parentGranules) { + printf(" [%s - %s)\n", pg.begin.printable().c_str(), pg.end.printable().c_str()); + } + } + // TODO change this for merge - ASSERT(req.previousGranules.size() == 1); - Optional prevGranuleLockValue = wait(tr.get(granuleLockKey(req.previousGranules[0]))); + ASSERT(parentGranules.size() == 1); - ASSERT(prevGranuleLockValue.present()); + state std::pair granuleSplitState; + if (hasPrevOwner) { + std::pair _st = + wait(getGranuleSplitState(&tr, parentGranules[0], req.keyRange)); + granuleSplitState = _st; + } else { + granuleSplitState = std::pair(BlobGranuleSplitState::Started, invalidVersion); + } - std::tuple prevGranuleLock = - decodeBlobGranuleLockValue(prevGranuleLockValue.get()); - info.prevChangeFeedId = std::get<2>(prevGranuleLock); + ASSERT(!hasPrevOwner || granuleSplitState.first > BlobGranuleSplitState::Started); - wait(updateGranuleSplitState(&tr, - req.previousGranules[0], - req.keyRange, - info.prevChangeFeedId.get(), - BlobGranuleSplitState::Assigned)); + if (granuleSplitState.first == BlobGranuleSplitState::Started) { + wait(updateGranuleSplitState(&tr, + parentGranules[0], + req.keyRange, + info.prevChangeFeedId.get(), + BlobGranuleSplitState::Assigned)); + } - // FIXME: store this somewhere useful for time travel reads - GranuleFiles prevFiles = wait(loadPreviousFiles(&tr, req.previousGranules[0])); - ASSERT(!prevFiles.snapshotFiles.empty() || !prevFiles.deltaFiles.empty()); - info.granuleSplitFrom = req.previousGranules[0]; - info.blobFilesToSnapshot = prevFiles; - info.previousDurableVersion = info.blobFilesToSnapshot.get().deltaFiles.empty() - ? info.blobFilesToSnapshot.get().snapshotFiles.back().version - : info.blobFilesToSnapshot.get().deltaFiles.back().version; + // if granule wasn't done with old change feed, load it + if (granuleSplitState.first < BlobGranuleSplitState::Done) { + Optional prevGranuleLockValue = wait(tr.get(granuleLockKey(parentGranules[0]))); + ASSERT(prevGranuleLockValue.present()); + std::tuple prevGranuleLock = + decodeBlobGranuleLockValue(prevGranuleLockValue.get()); + info.prevChangeFeedId = std::get<2>(prevGranuleLock); + info.granuleSplitFrom = parentGranules[0]; + if (granuleSplitState.first == BlobGranuleSplitState::Assigned) { + // was already assigned, use change feed start version + ASSERT(granuleSplitState.second != invalidVersion); + info.changeFeedStartVersion = granuleSplitState.second; + } + } - // FIXME: need to handle takeover of a splitting range! If snapshot and/or deltas found for new range, - // don't snapshot + if (info.doSnapshot) { + // FIXME: store this somewhere useful for time travel reads + GranuleFiles prevFiles = wait(loadPreviousFiles(&tr, parentGranules[0])); + ASSERT(!prevFiles.snapshotFiles.empty() || !prevFiles.deltaFiles.empty()); + + info.blobFilesToSnapshot = prevFiles; + info.previousDurableVersion = info.blobFilesToSnapshot.get().deltaFiles.empty() + ? info.blobFilesToSnapshot.get().snapshotFiles.back().version + : info.blobFilesToSnapshot.get().deltaFiles.back().version; + } } - // else: FIXME: If nothing in previousGranules, previous durable version is max of previous snapshot version - // and previous delta version. If neither present, need to do a snapshot at the start. - // Assumes for now that this isn't a takeover, so nothing to do here + wait(tr.commit()); - TraceEvent("BlobWorkerPersistedAssignment", bwData->id) - .detail("GranuleStart", req.keyRange.begin) - .detail("GranuleEnd", req.keyRange.end); - - if (info.changeFeedStartVersion == invalidVersion) { + if (!hasPrevOwner) { info.changeFeedStartVersion = tr.getCommittedVersion(); + } else { + ASSERT(info.changeFeedStartVersion != invalidVersion); } + + TraceEvent("BlobWorkerPersistedAssignment", bwData->id).detail("Granule", req.keyRange); + return info; } catch (Error& e) { if (e.code() == error_code_granule_assignment_conflict) { @@ -1572,6 +1679,7 @@ static GranuleRangeMetadata constructActiveBlobRange(BlobWorkerData* bwData, Reference newMetadata = makeReference(); newMetadata->keyRange = keyRange; + // FIXME: original Epoch/Seqno is now not necessary with originalReq newMetadata->originalEpoch = epoch; newMetadata->originalSeqno = seqno; newMetadata->continueEpoch = epoch; @@ -1611,9 +1719,11 @@ static std::pair, Reference> changeBlobRange(BlobW int64_t epoch, int64_t seqno, bool active, - bool disposeOnCleanup) { + bool disposeOnCleanup, + bool selfReassign) { if (BW_DEBUG) { - printf("Changing range for [%s - %s): %s @ (%lld, %lld)\n", + printf("%s range for [%s - %s): %s @ (%lld, %lld)\n", + selfReassign ? "Re-assigning" : "Changing", keyRange.begin.printable().c_str(), keyRange.end.printable().c_str(), active ? "T" : "F", @@ -1622,11 +1732,9 @@ static std::pair, Reference> changeBlobRange(BlobW } // For each range that intersects this update: - // If the identical range already exists at the same assignment sequence nunmber, this is a noop - // Otherwise, this will consist of a series of ranges that are either older, or newer. - // For each older range, cancel it if it is active. - // Insert the current range. - // Re-insert all newer ranges over the current range. + // If the identical range already exists at the same assignment sequence number and it is not a self-reassign, this + // is a noop. Otherwise, this will consist of a series of ranges that are either older, or newer. For each older + // range, cancel it if it is active. Insert the current range. Re-insert all newer ranges over the current range. std::vector> futures; @@ -1634,16 +1742,22 @@ static std::pair, Reference> changeBlobRange(BlobW auto ranges = bwData->granuleMetadata.intersectingRanges(keyRange); for (auto& r : ranges) { + bool thisAssignmentNewer = newerRangeAssignment(r.value(), epoch, seqno); if (r.value().lastEpoch == epoch && r.value().lastSeqno == seqno) { - // applied the same assignment twice, make idempotent ASSERT(r.begin() == keyRange.begin); ASSERT(r.end() == keyRange.end); - if (r.value().activeMetadata.isValid()) { - futures.push_back(success(r.value().activeMetadata->assignFuture)); + + if (selfReassign) { + thisAssignmentNewer = true; + } else { + // applied the same assignment twice, make idempotent + if (r.value().activeMetadata.isValid()) { + futures.push_back(success(r.value().activeMetadata->assignFuture)); + } + return std::pair(waitForAll(futures), Reference()); // already applied, nothing to do } - return std::pair(waitForAll(futures), Reference()); // already applied, nothing to do } - bool thisAssignmentNewer = newerRangeAssignment(r.value(), epoch, seqno); + if (r.value().activeMetadata.isValid() && thisAssignmentNewer) { // cancel actors for old range and clear reference if (BW_DEBUG) { @@ -1752,14 +1866,13 @@ ACTOR Future registerBlobWorker(BlobWorkerData* bwData, BlobWorkerInterfac } } -ACTOR Future handleRangeAssign(BlobWorkerData* bwData, AssignBlobRangeRequest req) { +ACTOR Future handleRangeAssign(BlobWorkerData* bwData, AssignBlobRangeRequest req, bool isSelfReassign) { try { if (req.continueAssignment) { resumeBlobRange(bwData, req.keyRange, req.managerEpoch, req.managerSeqno); } else { - // FIXME: wait to reply unless worker confirms it should own range and takes out lock? state std::pair, Reference> futureAndNewGranule = - changeBlobRange(bwData, req.keyRange, req.managerEpoch, req.managerSeqno, true, false); + changeBlobRange(bwData, req.keyRange, req.managerEpoch, req.managerSeqno, true, false, isSelfReassign); wait(futureAndNewGranule.first); @@ -1767,14 +1880,23 @@ ACTOR Future handleRangeAssign(BlobWorkerData* bwData, AssignBlobRangeRequ wait(futureAndNewGranule.second->start(bwData, req)); } } - req.reply.send(AssignBlobRangeReply(true)); + if (!isSelfReassign) { + ASSERT(!req.reply.isSet()); + req.reply.send(AssignBlobRangeReply(true)); + } return Void(); } catch (Error& e) { if (BW_DEBUG) { - printf("AssignRange got error %s\n", e.name()); + printf("AssignRange [%s - %s) got error %s\n", + req.keyRange.begin.printable().c_str(), + req.keyRange.end.printable().c_str(), + e.name()); } - if (canReplyWith(e)) { - req.reply.sendError(e); + + if (!isSelfReassign) { + if (canReplyWith(e)) { + req.reply.sendError(e); + } } throw; } @@ -1782,12 +1904,17 @@ ACTOR Future handleRangeAssign(BlobWorkerData* bwData, AssignBlobRangeRequ ACTOR Future handleRangeRevoke(BlobWorkerData* bwData, RevokeBlobRangeRequest req) { try { - wait(changeBlobRange(bwData, req.keyRange, req.managerEpoch, req.managerSeqno, false, req.dispose).first); + wait( + changeBlobRange(bwData, req.keyRange, req.managerEpoch, req.managerSeqno, false, req.dispose, false).first); req.reply.send(AssignBlobRangeReply(true)); return Void(); } catch (Error& e) { + // FIXME: retry on error if dispose fails? if (BW_DEBUG) { - printf("RevokeRange got error %s\n", e.name()); + printf("RevokeRange [%s - %s) got error %s\n", + req.keyRange.begin.printable().c_str(), + req.keyRange.end.printable().c_str(), + e.name()); } if (canReplyWith(e)) { req.reply.sendError(e); @@ -1797,11 +1924,12 @@ ACTOR Future handleRangeRevoke(BlobWorkerData* bwData, RevokeBlobRangeRequ } // FIXME: handle errors -// Because change feeds send uncommitted data and explicit rollback messages, we speculatively buffer/write uncommitted -// data. This means we must ensure the data is actually committed before "committing" those writes in the blob granule. -// The simplest way to do this is to have the blob worker do a periodic GRV, which is guaranteed to be an earlier -// committed version. -ACTOR Future runGrvChecks(BlobWorkerData* bwData) { +// Because change feeds send uncommitted data and explicit rollback messages, we speculatively buffer/write +// uncommitted data. This means we must ensure the data is actually committed before "committing" those writes in +// the blob granule. The simplest way to do this is to have the blob worker do a periodic GRV, which is guaranteed +// to be an earlier committed version. +ACTOR Future runCommitVersionChecks(BlobWorkerData* bwData) { + state Transaction tr(bwData->db); loop { // only do grvs to get committed version if we need it to persist delta files while (bwData->pendingDeltaFileCommitChecks.get() == 0) { @@ -1813,14 +1941,19 @@ ACTOR Future runGrvChecks(BlobWorkerData* bwData) { state int checksToResolve = bwData->pendingDeltaFileCommitChecks.get(); - Transaction tr(bwData->db); - Version readVersion = wait(tr.getReadVersion()); + tr.reset(); + try { + Version readVersion = wait(tr.getReadVersion()); - ASSERT(readVersion >= bwData->knownCommittedVersion.get()); - if (readVersion > bwData->knownCommittedVersion.get()) { - ++bwData->knownCommittedCheckCount; - bwData->knownCommittedVersion.set(readVersion); - bwData->pendingDeltaFileCommitChecks.set(bwData->pendingDeltaFileCommitChecks.get() - checksToResolve); + ASSERT(readVersion >= bwData->knownCommittedVersion.get()); + if (readVersion > bwData->knownCommittedVersion.get()) { + ++bwData->knownCommittedCheckCount; + bwData->knownCommittedVersion.set(readVersion); + bwData->pendingDeltaFileCommitChecks.set(bwData->pendingDeltaFileCommitChecks.get() - checksToResolve); + } + ++bwData->stats.commitVersionChecks; + } catch (Error& e) { + wait(tr.onError(e)); } } } @@ -1861,7 +1994,7 @@ ACTOR Future blobWorker(BlobWorkerInterface bwInterf, Reference collection = actorCollection(addActor.getFuture()); addActor.send(waitFailureServer(bwInterf.waitFailure.getFuture())); - addActor.send(runGrvChecks(&self)); + addActor.send(runCommitVersionChecks(&self)); try { loop choose { @@ -1885,31 +2018,18 @@ ACTOR Future blobWorker(BlobWorkerInterface bwInterf, Reference blobWorker(BlobWorkerInterface bwInterf, Reference blobWorker(BlobWorkerInterface bwInterf, Reference