From dfb9655c579e7ef58b0a91445e5275300046c5a0 Mon Sep 17 00:00:00 2001 From: Suraj Gupta Date: Fri, 1 Oct 2021 11:08:00 -0400 Subject: [PATCH 1/7] Handle blob work failure --- fdbclient/BlobWorkerInterface.h | 19 ++++ fdbserver/BlobManager.actor.cpp | 150 ++++++++++++++++++++++---- fdbserver/BlobWorker.actor.cpp | 184 +++++++++++++++++++++----------- 3 files changed, 272 insertions(+), 81 deletions(-) diff --git a/fdbclient/BlobWorkerInterface.h b/fdbclient/BlobWorkerInterface.h index eb2be3623e..eafa8eb338 100644 --- a/fdbclient/BlobWorkerInterface.h +++ b/fdbclient/BlobWorkerInterface.h @@ -35,6 +35,8 @@ struct BlobWorkerInterface { RequestStream assignBlobRangeRequest; RequestStream revokeBlobRangeRequest; RequestStream granuleStatusStreamRequest; + RequestStream haltBlobWorker; + struct LocalityData locality; UID myId; @@ -57,6 +59,7 @@ struct BlobWorkerInterface { assignBlobRangeRequest, revokeBlobRangeRequest, granuleStatusStreamRequest, + haltBlobWorker, locality, myId); } @@ -182,4 +185,20 @@ struct GranuleStatusStreamRequest { } }; +struct HaltBlobWorkerRequest { + constexpr static FileIdentifier file_identifier = 1985879; + UID requesterID; + ReplyPromise reply; + + int64_t managerEpoch; + + HaltBlobWorkerRequest() {} + explicit HaltBlobWorkerRequest(int64_t managerEpoch, UID uid) : requesterID(uid), managerEpoch(managerEpoch) {} + + template + void serialize(Ar& ar) { + serializer(ar, managerEpoch, requesterID, reply); + } +}; + #endif diff --git a/fdbserver/BlobManager.actor.cpp b/fdbserver/BlobManager.actor.cpp index f1ad4ffc3f..4da0fde2b9 100644 --- a/fdbserver/BlobManager.actor.cpp +++ b/fdbserver/BlobManager.actor.cpp @@ -208,7 +208,7 @@ struct BlobManagerData { KeyRangeMap workerAssignments; KeyRangeMap knownBlobRanges; - Debouncer restartRecruiting; + AsyncVar restartRecruiting; std::set recruitingLocalities; // the addrs of the workers being recruited on int64_t epoch = -1; @@ -221,8 +221,7 @@ struct BlobManagerData { PromiseStream rangesToAssign; BlobManagerData(UID id, Database db) - : id(id), db(db), knownBlobRanges(false, normalKeys.end), - restartRecruiting(SERVER_KNOBS->DEBOUNCE_RECRUITING_DELAY) {} + : id(id), db(db), knownBlobRanges(false, normalKeys.end), restartRecruiting() {} ~BlobManagerData() { printf("Destroying blob manager data for %s\n", id.toString().c_str()); } }; @@ -283,6 +282,10 @@ static UID pickWorkerForAssign(BlobManagerData* bmData) { } // pick a random worker out of the eligible workers + if (eligibleWorkers.size() == 0) { + printf("%d eligible workers\n", bmData->workerStats.size()); + } + ASSERT(eligibleWorkers.size() > 0); int idx = deterministicRandom()->randomInt(0, eligibleWorkers.size()); if (BM_DEBUG) { @@ -298,7 +301,7 @@ ACTOR Future doRangeAssignment(BlobManagerData* bmData, RangeAssignment as if (BM_DEBUG) { printf("BM %s %s range [%s - %s) @ (%lld, %lld)\n", - workerID.toString().c_str(), + bmData->id.toString().c_str(), assignment.isAssign ? "assigning" : "revoking", assignment.keyRange.begin.printable().c_str(), assignment.keyRange.end.printable().c_str(), @@ -318,6 +321,11 @@ ACTOR Future doRangeAssignment(BlobManagerData* bmData, RangeAssignment as req.managerEpoch = bmData->epoch; req.managerSeqno = seqNo; req.continueAssignment = assignment.assign.get().continueAssignment; + + // if that worker isn't alive anymore, add the range back into the stream + if (bmData->workersById.count(workerID) == 0) { + throw granule_assignment_conflict(); // TODO: find a better error to throw + } AssignBlobRangeReply _rep = wait(bmData->workersById[workerID].assignBlobRangeRequest.getReply(req)); rep = _rep; } else { @@ -331,8 +339,13 @@ ACTOR Future doRangeAssignment(BlobManagerData* bmData, RangeAssignment as req.managerSeqno = seqNo; req.dispose = assignment.revoke.get().dispose; - AssignBlobRangeReply _rep = wait(bmData->workersById[workerID].revokeBlobRangeRequest.getReply(req)); - rep = _rep; + // if that worker isn't alive anymore, this is a noop + if (bmData->workersById.count(workerID)) { + AssignBlobRangeReply _rep = wait(bmData->workersById[workerID].revokeBlobRangeRequest.getReply(req)); + rep = _rep; + } else { + return Void(); + } } if (!rep.epochOk) { if (BM_DEBUG) { @@ -349,7 +362,8 @@ ACTOR Future doRangeAssignment(BlobManagerData* bmData, RangeAssignment as if (BM_DEBUG) { printf("BM got error assigning range [%s - %s) to worker %s, requeueing\n", assignment.keyRange.begin.printable().c_str(), - assignment.keyRange.end.printable().c_str()); + assignment.keyRange.end.printable().c_str(), + workerID.toString().c_str()); } // re-send revoke to queue to handle range being un-assigned from that worker before the new one RangeAssignment revokeOld; @@ -404,7 +418,9 @@ ACTOR Future rangeAssigner(BlobManagerData* bmData) { // Ensure range isn't currently assigned anywhere, and there is only 1 intersecting range auto currentAssignments = bmData->workerAssignments.intersectingRanges(assignment.keyRange); int count = 0; + printf("intersecting ranges in currentAssignments:\n"); for (auto& it : currentAssignments) { + printf("[%s - %s]\n", it.begin().printable().c_str(), it.end().printable().c_str()); if (assignment.assign.get().continueAssignment) { ASSERT(assignment.worker.present()); ASSERT(it.value() == assignment.worker.get()); @@ -419,6 +435,11 @@ ACTOR Future rangeAssigner(BlobManagerData* bmData) { bmData->workerAssignments.insert(assignment.keyRange, workerId); bmData->workerStats[workerId].numGranulesAssigned += 1; + printf("current ranges after inserting assign: \n"); + for (auto it : bmData->workerAssignments.ranges()) { + printf("[%s - %s]\n", it.begin().printable().c_str(), it.end().printable().c_str()); + } + // FIXME: if range is assign, have some sort of semaphore for outstanding assignments so we don't assign // a ton ranges at once and blow up FDB with reading initial snapshots. bmData->addActor.send(doRangeAssignment(bmData, assignment, workerId, seqNo)); @@ -432,12 +453,20 @@ ACTOR Future rangeAssigner(BlobManagerData* bmData) { // It is fine for multiple disjoint sub-ranges to have the same sequence number since they were part of // the same logical change - bmData->workerStats[it.value()].numGranulesAssigned -= 1; - if (!assignment.worker.present() || assignment.worker.get() == it.value()) - bmData->addActor.send(doRangeAssignment(bmData, assignment, it.value(), seqNo)); + + if (bmData->workerStats.count(it.value())) { + bmData->workerStats[it.value()].numGranulesAssigned -= 1; + } + + // revoke the range for the worker that owns it, not the worker specified in the revoke + bmData->addActor.send(doRangeAssignment(bmData, assignment, it.value(), seqNo)); } bmData->workerAssignments.insert(assignment.keyRange, UID()); + printf("current ranges after inserting revoke: \n"); + for (auto it : bmData->workerAssignments.ranges()) { + printf("[%s - %s]\n", it.begin().printable().c_str(), it.end().printable().c_str()); + } } } } @@ -701,6 +730,73 @@ ACTOR Future maybeSplitRange(BlobManagerData* bmData, UID currentWorkerId, return Void(); } +void reassignRanges(BlobManagerData* bmData, UID bwId) { + printf("taking back ranges for worker %s\n", bwId.toString().c_str()); + // for every range owned by this blob worker, we want to + // - send a revoke request for that range to the blob worker + // - add the range back to the stream of ranges to be assigned + for (auto& it : bmData->workerAssignments.ranges()) { + if (it.cvalue() == bwId) { + // Send revoke request to worker + RangeAssignment raRevoke; + raRevoke.isAssign = false; + raRevoke.worker = bwId; + raRevoke.keyRange = it.range(); + raRevoke.revoke = RangeRevokeData(false); + bmData->rangesToAssign.send(raRevoke); + + // Add range back into the stream of ranges to be assigned + RangeAssignment raAssign; + raAssign.isAssign = true; + raAssign.worker = Optional(); + raAssign.keyRange = it.range(); + raAssign.assign = RangeAssignmentData(false); // not a continue + bmData->rangesToAssign.send(raAssign); + } + } +} + +void killBlobWorker(BlobManagerData* bmData, BlobWorkerInterface bwInterf) { + UID bwId = bwInterf.id(); + + // Remove blob worker from stats map so that when we try to find a worker to takeover the range, + // the one we just killed isn't considered. + // Remove it from workersById also since otherwise that addr will remain excluded + // when we try to recruit new blob workers. + printf("removing bw %s from BM workerStats\n", bwId.toString().c_str()); + bmData->workerStats.erase(bwId); + bmData->workersById.erase(bwId); + + // for every range owned by this blob worker, we want to + // - send a revoke request for that range to the blob worker + // - add the range back to the stream of ranges to be assigned + printf("taking back ranges from bw %s\n", bwId.toString().c_str()); + for (auto& it : bmData->workerAssignments.ranges()) { + if (it.cvalue() == bwId) { + // Send revoke request to worker + RangeAssignment raRevoke; + raRevoke.isAssign = false; + raRevoke.worker = bwId; + raRevoke.keyRange = it.range(); + raRevoke.revoke = RangeRevokeData(false); + bmData->rangesToAssign.send(raRevoke); + + // Add range back into the stream of ranges to be assigned + RangeAssignment raAssign; + raAssign.isAssign = true; + raAssign.worker = Optional(); + raAssign.keyRange = it.range(); + raAssign.assign = RangeAssignmentData(false); // not a continue + bmData->rangesToAssign.send(raAssign); + } + } + + // Send halt to blob worker, with no expectation of hearing back + printf("sending halt to bw %s\n", bwId.toString().c_str()); + bmData->addActor.send( + brokenPromiseToNever(bwInterf.haltBlobWorker.getReply(HaltBlobWorkerRequest(bmData->epoch, bmData->id)))); +} + ACTOR Future monitorBlobWorkerStatus(BlobManagerData* bmData, BlobWorkerInterface bwInterf) { state KeyRangeMap> lastSeenSeqno; // outer loop handles reconstructing stream if it got a retryable error @@ -711,6 +807,7 @@ ACTOR Future monitorBlobWorkerStatus(BlobManagerData* bmData, BlobWorkerIn // read from stream until worker fails (should never get explicit end_of_stream) loop { GranuleStatusReply rep = waitNext(statusStream.getFuture()); + if (BM_DEBUG) { printf("BM %lld got status of [%s - %s) @ (%lld, %lld) from BW %s: %s\n", bmData->epoch, @@ -723,7 +820,8 @@ ACTOR Future monitorBlobWorkerStatus(BlobManagerData* bmData, BlobWorkerIn } if (rep.epoch > bmData->epoch) { if (BM_DEBUG) { - printf("BM heard from BW that there is a new manager with higher epoch\n"); + printf("BM heard from BW %s that there is a new manager with higher epoch\n", + bwInterf.id().toString().c_str()); } if (bmData->iAmReplaced.canBeSet()) { bmData->iAmReplaced.send(Void()); @@ -734,8 +832,12 @@ ACTOR Future monitorBlobWorkerStatus(BlobManagerData* bmData, BlobWorkerIn // to split the range. ASSERT(rep.doSplit); - // FIXME: only evaluate for split if this worker currently owns the granule in this blob manager's - // mapping + auto currGranuleAssignment = bmData->workerAssignments.rangeContaining(rep.granuleRange.begin); + if (!(currGranuleAssignment.begin() == rep.granuleRange.begin && + currGranuleAssignment.end() == rep.granuleRange.end && + currGranuleAssignment.cvalue() == bwInterf.id())) { + continue; + } auto lastReqForGranule = lastSeenSeqno.rangeContaining(rep.granuleRange.begin); if (rep.granuleRange.begin == lastReqForGranule.begin() && @@ -797,10 +899,7 @@ ACTOR Future monitorBlobWorker(BlobManagerData* bmData, BlobWorkerInterfac printf("BM %lld detected BW %s is dead\n", bmData->epoch, bwInterf.id().toString().c_str()); } TraceEvent("BlobWorkerFailed", bmData->id).detail("BlobWorkerID", bwInterf.id()); - // get all of its ranges - // send revoke request to get back all its ranges - // send halt (look at rangeMover) - // send all its ranges to assignranges stream + killBlobWorker(bmData, bwInterf); return Void(); } when(wait(monitorStatus)) { @@ -820,6 +919,13 @@ ACTOR Future monitorBlobWorker(BlobManagerData* bmData, BlobWorkerInterfac TraceEvent(SevError, "BWMonitoringFailed", bmData->id).detail("BlobWorkerID", bwInterf.id()).error(e); throw e; } + + // Trigger recruitment for a new blob worker + printf("restarting recruitment in monitorblobworker\n"); + bmData->restartRecruiting.trigger(); + + printf("about to stop monitoring %s\n", bwInterf.id().toString().c_str()); + return Void(); } // TODO this is only for chaos testing right now!! REMOVE LATER @@ -937,8 +1043,8 @@ ACTOR Future initializeBlobWorker(BlobManagerData* self, RecruitBlobWorker if (newBlobWorker.present()) { BlobWorkerInterface bwi = newBlobWorker.get().interf; - self->workersById.insert({ bwi.id(), bwi }); - self->workerStats.insert({ bwi.id(), BlobWorkerStats() }); + self->workersById[bwi.id()] = bwi; + self->workerStats[bwi.id()] = BlobWorkerStats(); self->addActor.send(monitorBlobWorker(self, bwi)); TraceEvent("BMRecruiting") @@ -984,6 +1090,10 @@ ACTOR Future blobWorkerRecruiter( } TraceEvent("BMRecruiting").detail("State", "Sending request to CC"); + printf("EXCLUDING THE FOLLOWING IN REQ:\n"); + for (auto addr : recruitReq.excludeAddresses) { + printf("- %s\n", addr.toString().c_str()); + } if (!fCandidateWorker.isValid() || fCandidateWorker.isReady() || recruitReq.excludeAddresses != lastRequest.excludeAddresses) { @@ -1003,7 +1113,7 @@ ACTOR Future blobWorkerRecruiter( when(wait(recruitBlobWorker->onChange())) { fCandidateWorker = Future(); } // signal used to restart the loop and try to recruit the next blob worker - when(wait(self->restartRecruiting.onTrigger())) {} + when(wait(self->restartRecruiting.onChange())) { printf("RESTARTED RECRUITING. BACK TO TOP\n"); } } wait(delay(FLOW_KNOBS->PREVENT_FAST_SPIN_DELAY, TaskPriority::BlobManager)); } catch (Error& e) { diff --git a/fdbserver/BlobWorker.actor.cpp b/fdbserver/BlobWorker.actor.cpp index 0782cbddab..f4f648c493 100644 --- a/fdbserver/BlobWorker.actor.cpp +++ b/fdbserver/BlobWorker.actor.cpp @@ -34,6 +34,7 @@ #include "fdbserver/MutationTracking.h" #include "fdbserver/WaitFailure.h" #include "flow/Arena.h" +#include "flow/Error.h" #include "flow/IRandom.h" #include "flow/actorcompiler.h" // has to be last include #include "flow/flow.h" @@ -41,8 +42,6 @@ #define BW_DEBUG true #define BW_REQUEST_DEBUG false -// FIXME: change all BlobWorkerData* to Reference to avoid segfaults if core loop gets error - // TODO add comments + documentation struct BlobFileIndex { Version version; @@ -76,10 +75,12 @@ struct GranuleChangeFeedInfo { // FIXME: the circular dependencies here are getting kind of gross struct GranuleMetadata; struct BlobWorkerData; -ACTOR Future persistAssignWorkerRange(BlobWorkerData* bwData, AssignBlobRangeRequest req); -ACTOR Future blobGranuleUpdateFiles(BlobWorkerData* bwData, Reference metadata); +ACTOR Future persistAssignWorkerRange(Reference bwData, + AssignBlobRangeRequest req); +ACTOR Future blobGranuleUpdateFiles(Reference bwData, Reference metadata); + +// for a range that may or may not be set -// for a range that is active struct GranuleMetadata : NonCopyable, ReferenceCounted { KeyRange keyRange; @@ -112,12 +113,17 @@ struct GranuleMetadata : NonCopyable, ReferenceCounted { AssignBlobRangeRequest originalReq; - Future start(BlobWorkerData* bwData, AssignBlobRangeRequest req) { + Future start(Reference bwData, AssignBlobRangeRequest req) { originalReq = req; assignFuture = persistAssignWorkerRange(bwData, req); fileUpdaterFuture = blobGranuleUpdateFiles(bwData, Reference::addRef(this)); + // bwData->actors.add(blobGranuleUpdateFiles(bwData, Reference::addRef(this))); + // this could be the cause of the seg fault. since this is not being waited on, + // when start get cancelled, blobGranuleUpdateFiles won't get cancelled. so instead I added it to actors, so + // that it is explicitly cancelled. maybe this fixes it? return success(assignFuture); + // return Void(); } void resume() { @@ -145,7 +151,6 @@ struct GranuleMetadata : NonCopyable, ReferenceCounted { } }; -// for a range that may or may not be set struct GranuleRangeMetadata { int64_t lastEpoch; int64_t lastSeqno; @@ -154,14 +159,25 @@ struct GranuleRangeMetadata { GranuleRangeMetadata() : lastEpoch(0), lastSeqno(0) {} GranuleRangeMetadata(int64_t epoch, int64_t seqno, Reference activeMetadata) : lastEpoch(epoch), lastSeqno(seqno), activeMetadata(activeMetadata) {} + /* + ~GranuleRangeMetadata() { + if (activeMetadata.isValid()) { + activeMetadata->cancel(false); + } + } + */ }; -struct BlobWorkerData { +struct BlobWorkerData : NonCopyable, ReferenceCounted { UID id; Database db; + AsyncVar dead; BlobWorkerStats stats; + PromiseStream> addActor; + ActorCollection actors{ false }; + LocalityData locality; int64_t currentManagerEpoch = -1; @@ -178,7 +194,8 @@ struct BlobWorkerData { PromiseStream granuleUpdateErrors; - BlobWorkerData(UID id, Database db) : id(id), db(db), stats(id, SERVER_KNOBS->WORKER_LOGGING_INTERVAL) {} + BlobWorkerData(UID id, Database db) + : id(id), db(db), stats(id, SERVER_KNOBS->WORKER_LOGGING_INTERVAL), actors(false), dead(false) {} ~BlobWorkerData() { printf("Destroying blob worker data for %s\n", id.toString().c_str()); } bool managerEpochOk(int64_t epoch) { @@ -481,7 +498,7 @@ static Value getFileValue(std::string fname, int64_t offset, int64_t length) { // the data in it may not yet be committed, and even though previous delta fiels with lower versioned data may still be // in flight. The synchronization happens after the s3 file is written, but before we update the FDB index of what files // exist. Before updating FDB, we ensure the version is committed and all previous delta files have updated FDB. -ACTOR Future writeDeltaFile(BlobWorkerData* bwData, +ACTOR Future writeDeltaFile(Reference bwData, KeyRange keyRange, int64_t epoch, int64_t seqno, @@ -589,7 +606,7 @@ ACTOR Future writeDeltaFile(BlobWorkerData* bwData, } } -ACTOR Future writeSnapshot(BlobWorkerData* bwData, +ACTOR Future writeSnapshot(Reference bwData, KeyRange keyRange, int64_t epoch, int64_t seqno, @@ -707,7 +724,8 @@ ACTOR Future writeSnapshot(BlobWorkerData* bwData, return BlobFileIndex(version, fname, 0, serialized.size()); } -ACTOR Future dumpInitialSnapshotFromFDB(BlobWorkerData* bwData, Reference metadata) { +ACTOR Future dumpInitialSnapshotFromFDB(Reference bwData, + Reference metadata) { if (BW_DEBUG) { printf("Dumping snapshot from FDB for [%s - %s)\n", metadata->keyRange.begin.printable().c_str(), @@ -755,7 +773,7 @@ ACTOR Future dumpInitialSnapshotFromFDB(BlobWorkerData* bwData, R } // files might not be the current set of files in metadata, in the case of doing the initial snapshot of a granule. -ACTOR Future compactFromBlob(BlobWorkerData* bwData, +ACTOR Future compactFromBlob(Reference bwData, Reference metadata, GranuleFiles files) { wait(delay(0, TaskPriority::BlobWorkerUpdateStorage)); @@ -874,7 +892,7 @@ static bool filterOldMutations(const KeyRange& range, return false; } -ACTOR Future handleCompletedDeltaFile(BlobWorkerData* bwData, +ACTOR Future handleCompletedDeltaFile(Reference bwData, Reference metadata, BlobFileIndex completedDeltaFile, Key cfKey, @@ -892,12 +910,15 @@ ACTOR Future handleCompletedDeltaFile(BlobWorkerData* bwData, // have completed // FIXME: also have these be async, have each pop change feed wait on the prior one, wait on them before // re-snapshotting + printf("in handleCompletedDeltaFile for BW %s\n", bwData->id.toString().c_str()); Future popFuture = bwData->db->popChangeFeedMutations(cfKey, completedDeltaFile.version); wait(popFuture); + printf("popChangeFeedMutations returned\n"); } while (!rollbacksInProgress.empty() && completedDeltaFile.version >= rollbacksInProgress.front().first) { rollbacksInProgress.pop_front(); } + printf("removed rollbacks\n"); return Void(); } @@ -1026,7 +1047,7 @@ static Version doGranuleRollback(Reference metadata, // 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) -ACTOR Future blobGranuleUpdateFiles(BlobWorkerData* bwData, Reference metadata) { +ACTOR Future blobGranuleUpdateFiles(Reference bwData, Reference metadata) { state PromiseStream>> oldChangeFeedStream; state PromiseStream>> changeFeedStream; state Future inFlightBlobSnapshot; @@ -1146,6 +1167,7 @@ ACTOR Future blobGranuleUpdateFiles(BlobWorkerData* bwData, ReferencekeyRange*/); } else { readOldChangeFeed = false; + printf("before getChangeFeedStream, my ID is %s\n", bwData->id.toString().c_str()); changeFeedFuture = bwData->db->getChangeFeedStream( changeFeedStream, cfKey, startVersion + 1, MAX_VERSION, metadata->keyRange); } @@ -1347,12 +1369,15 @@ ACTOR Future blobGranuleUpdateFiles(BlobWorkerData* bwData, Reference result = wait(timeout(metadata->resumeSnapshot.getFuture(), 1.0)); if (result.present()) { break; + } else if (bwData->dead.get()) { + throw actor_cancelled(); } // FIXME: re-trigger this loop if blob manager status stream changes if (BW_DEBUG) { - printf("Granule [%s - %s)\n, hasn't heard back from BM, re-sending status\n", + printf("Granule [%s - %s)\n, hasn't heard back from BM in BW %s, re-sending status\n", metadata->keyRange.begin.printable().c_str(), - metadata->keyRange.end.printable().c_str()); + metadata->keyRange.end.printable().c_str(), + bwData->id.toString().c_str()); } } @@ -1528,6 +1553,7 @@ ACTOR Future blobGranuleUpdateFiles(BlobWorkerData* bwData, Reference waitForVersion(Reference metadata, Version return waitForVersionActor(metadata, v); } -ACTOR Future handleBlobGranuleFileRequest(BlobWorkerData* bwData, BlobGranuleFileRequest req) { +ACTOR Future handleBlobGranuleFileRequest(Reference bwData, BlobGranuleFileRequest req) { try { // TODO REMOVE in api V2 ASSERT(req.beginVersion == 0); @@ -1713,7 +1739,10 @@ ACTOR Future handleBlobGranuleFileRequest(BlobWorkerData* bwData, BlobGran choose { when(wait(waitForVersionFuture)) {} when(wait(metadata->rollbackCount.onChange())) {} - when(wait(metadata->cancelled.getFuture())) { throw wrong_shard_server(); } + when(wait(metadata->cancelled.getFuture())) { + printf("metadata was cancelled\n"); + throw wrong_shard_server(); + } } if (rollbackCount == metadata->rollbackCount.get()) { @@ -1821,7 +1850,8 @@ ACTOR Future handleBlobGranuleFileRequest(BlobWorkerData* bwData, BlobGran return Void(); } -ACTOR Future persistAssignWorkerRange(BlobWorkerData* bwData, AssignBlobRangeRequest req) { +ACTOR Future persistAssignWorkerRange(Reference bwData, + AssignBlobRangeRequest req) { ASSERT(!req.continueAssignment); state Transaction tr(bwData->db); state Key lockKey = granuleLockKey(req.keyRange); @@ -1971,7 +2001,7 @@ ACTOR Future persistAssignWorkerRange(BlobWorkerData* bwD } } -static GranuleRangeMetadata constructActiveBlobRange(BlobWorkerData* bwData, +static GranuleRangeMetadata constructActiveBlobRange(Reference bwData, KeyRange keyRange, int64_t epoch, int64_t seqno) { @@ -2013,7 +2043,7 @@ static bool newerRangeAssignment(GranuleRangeMetadata oldMetadata, int64_t epoch // Returns future to wait on to ensure prior work of other granules is done before responding to the manager with a // successful assignment And if the change produced a new granule that needs to start doing work, returns the new // granule so that the caller can start() it with the appropriate starting state. -static std::pair, Reference> changeBlobRange(BlobWorkerData* bwData, +static std::pair, Reference> changeBlobRange(Reference bwData, KeyRange keyRange, int64_t epoch, int64_t seqno, @@ -2105,7 +2135,7 @@ static std::pair, Reference> changeBlobRange(BlobW return std::pair(waitForAll(futures), newMetadata.activeMetadata); } -static bool resumeBlobRange(BlobWorkerData* bwData, KeyRange keyRange, int64_t epoch, int64_t seqno) { +static bool resumeBlobRange(Reference bwData, KeyRange keyRange, int64_t epoch, int64_t seqno) { auto existingRange = bwData->granuleMetadata.rangeContaining(keyRange.begin); // if range boundaries don't match, or this (epoch, seqno) is old or the granule is inactive, ignore if (keyRange.begin != existingRange.begin() || keyRange.end != existingRange.end() || @@ -2142,7 +2172,7 @@ static bool resumeBlobRange(BlobWorkerData* bwData, KeyRange keyRange, int64_t e return true; } -ACTOR Future registerBlobWorker(BlobWorkerData* bwData, BlobWorkerInterface interf) { +ACTOR Future registerBlobWorker(Reference bwData, BlobWorkerInterface interf) { state Reference tr = makeReference(bwData->db); loop { tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS); @@ -2167,7 +2197,9 @@ ACTOR Future registerBlobWorker(BlobWorkerData* bwData, BlobWorkerInterfac } } -ACTOR Future handleRangeAssign(BlobWorkerData* bwData, AssignBlobRangeRequest req, bool isSelfReassign) { +ACTOR Future handleRangeAssign(Reference bwData, + AssignBlobRangeRequest req, + bool isSelfReassign) { try { if (req.continueAssignment) { resumeBlobRange(bwData, req.keyRange, req.managerEpoch, req.managerSeqno); @@ -2178,6 +2210,9 @@ ACTOR Future handleRangeAssign(BlobWorkerData* bwData, AssignBlobRangeRequ wait(futureAndNewGranule.first); if (futureAndNewGranule.second.isValid()) { + printf("BW %s ABOUT TO WAIT IN HANDLERANGEASSIGN\n", bwData->id.toString().c_str()); + // WAITING ON START BUT ITS NOT AN ACTOR!!!!!!! SO WHEN handlerangeassign gets operation_cancelled, it + // won't get propogated to start wait(futureAndNewGranule.second->start(bwData, req)); } } @@ -2187,6 +2222,8 @@ ACTOR Future handleRangeAssign(BlobWorkerData* bwData, AssignBlobRangeRequ } return Void(); } catch (Error& e) { + printf("BW %s GOT ERROR %s IN HANDLERANGEASSIGN\n", bwData->id.toString().c_str(), e.name()); + state Error eState = e; if (BW_DEBUG) { printf("AssignRange [%s - %s) got error %s\n", req.keyRange.begin.printable().c_str(), @@ -2194,16 +2231,23 @@ ACTOR Future handleRangeAssign(BlobWorkerData* bwData, AssignBlobRangeRequ e.name()); } + /* + if (futureAndNewGranule.get().second.isValid()) { + wait(futureAndNewGranule.get().second->cancel(false)); + } + */ + if (!isSelfReassign) { - if (canReplyWith(e)) { - req.reply.sendError(e); + if (canReplyWith(eState)) { + req.reply.sendError(eState); } } - throw; + + throw eState; } } -ACTOR Future handleRangeRevoke(BlobWorkerData* bwData, RevokeBlobRangeRequest req) { +ACTOR Future handleRangeRevoke(Reference bwData, RevokeBlobRangeRequest req) { try { wait( changeBlobRange(bwData, req.keyRange, req.managerEpoch, req.managerSeqno, false, req.dispose, false).first); @@ -2229,7 +2273,7 @@ ACTOR Future handleRangeRevoke(BlobWorkerData* bwData, RevokeBlobRangeRequ // 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) { +ACTOR Future runCommitVersionChecks(Reference bwData) { state Transaction tr(bwData->db); loop { // only do grvs to get committed version if we need it to persist delta files @@ -2262,9 +2306,12 @@ ACTOR Future runCommitVersionChecks(BlobWorkerData* bwData) { ACTOR Future blobWorker(BlobWorkerInterface bwInterf, ReplyPromise recruitReply, Reference const> dbInfo) { - state BlobWorkerData self(bwInterf.id(), openDBOnServer(dbInfo, TaskPriority::DefaultEndpoint, LockAware::True)); - self.id = bwInterf.id(); - self.locality = bwInterf.locality; + state Reference self( + new BlobWorkerData(bwInterf.id(), openDBOnServer(dbInfo, TaskPriority::DefaultEndpoint, LockAware::True))); + self->id = bwInterf.id(); + self->locality = bwInterf.locality; + + state Future collection = actorCollection(self->addActor.getFuture()); if (BW_DEBUG) { printf("Initializing blob worker s3 stuff\n"); @@ -2275,19 +2322,19 @@ ACTOR Future blobWorker(BlobWorkerInterface bwInterf, if (BW_DEBUG) { printf("BW constructing simulated backup container\n"); } - self.bstore = BackupContainerFileSystem::openContainerFS("file://fdbblob/"); + self->bstore = BackupContainerFileSystem::openContainerFS("file://fdbblob/"); } else { if (BW_DEBUG) { printf("BW constructing backup container from %s\n", SERVER_KNOBS->BG_URL.c_str()); } - self.bstore = BackupContainerFileSystem::openContainerFS(SERVER_KNOBS->BG_URL); + self->bstore = BackupContainerFileSystem::openContainerFS(SERVER_KNOBS->BG_URL); if (BW_DEBUG) { printf("BW constructed backup container\n"); } } // register the blob worker to the system keyspace - wait(registerBlobWorker(&self, bwInterf)); + wait(registerBlobWorker(self, bwInterf)); } catch (Error& e) { if (BW_DEBUG) { printf("BW got backup container init error %s\n", e.name()); @@ -2307,11 +2354,8 @@ ACTOR Future blobWorker(BlobWorkerInterface bwInterf, rep.interf = bwInterf; recruitReply.send(rep); - state PromiseStream> addActor; - state Future collection = actorCollection(addActor.getFuture()); - - addActor.send(waitFailureServer(bwInterf.waitFailure.getFuture())); - addActor.send(runCommitVersionChecks(&self)); + self->actors.add(waitFailureServer(bwInterf.waitFailure.getFuture())); + self->actors.add(runCommitVersionChecks(self)); try { loop choose { @@ -2319,25 +2363,25 @@ ACTOR Future blobWorker(BlobWorkerInterface bwInterf, /*printf("Got blob granule request [%s - %s)\n", req.keyRange.begin.printable().c_str(), req.keyRange.end.printable().c_str());*/ - ++self.stats.readRequests; - ++self.stats.activeReadRequests; - addActor.send(handleBlobGranuleFileRequest(&self, req)); + ++self->stats.readRequests; + ++self->stats.activeReadRequests; + self->actors.add(handleBlobGranuleFileRequest(self, req)); } when(GranuleStatusStreamRequest req = waitNext(bwInterf.granuleStatusStreamRequest.getFuture())) { - if (self.managerEpochOk(req.managerEpoch)) { + if (self->managerEpochOk(req.managerEpoch)) { if (BW_DEBUG) { - printf("Worker %s got new granule status endpoint\n", self.id.toString().c_str()); + printf("Worker %s got new granule status endpoint\n", self->id.toString().c_str()); } - self.currentManagerStatusStream = req.reply; + self->currentManagerStatusStream = req.reply; } } when(AssignBlobRangeRequest _req = waitNext(bwInterf.assignBlobRangeRequest.getFuture())) { - ++self.stats.rangeAssignmentRequests; - --self.stats.numRangesAssigned; + ++self->stats.rangeAssignmentRequests; + --self->stats.numRangesAssigned; state AssignBlobRangeRequest assignReq = _req; if (BW_DEBUG) { printf("Worker %s assigned range [%s - %s) @ (%lld, %lld):\n continue=%s\n", - self.id.toString().c_str(), + self->id.toString().c_str(), assignReq.keyRange.begin.printable().c_str(), assignReq.keyRange.end.printable().c_str(), assignReq.managerEpoch, @@ -2345,18 +2389,18 @@ ACTOR Future blobWorker(BlobWorkerInterface bwInterf, assignReq.continueAssignment ? "T" : "F"); } - if (self.managerEpochOk(assignReq.managerEpoch)) { - addActor.send(handleRangeAssign(&self, assignReq, false)); + if (self->managerEpochOk(assignReq.managerEpoch)) { + self->actors.add(handleRangeAssign(self, assignReq, false)); } else { assignReq.reply.send(AssignBlobRangeReply(false)); } } when(RevokeBlobRangeRequest _req = waitNext(bwInterf.revokeBlobRangeRequest.getFuture())) { state RevokeBlobRangeRequest revokeReq = _req; - --self.stats.numRangesAssigned; + --self->stats.numRangesAssigned; if (BW_DEBUG) { printf("Worker %s revoked range [%s - %s) @ (%lld, %lld):\n dispose=%s\n", - self.id.toString().c_str(), + self->id.toString().c_str(), revokeReq.keyRange.begin.printable().c_str(), revokeReq.keyRange.end.printable().c_str(), revokeReq.managerEpoch, @@ -2364,30 +2408,48 @@ ACTOR Future blobWorker(BlobWorkerInterface bwInterf, revokeReq.dispose ? "T" : "F"); } - if (self.managerEpochOk(revokeReq.managerEpoch)) { - addActor.send(handleRangeRevoke(&self, revokeReq)); + if (self->managerEpochOk(revokeReq.managerEpoch)) { + self->actors.add(handleRangeRevoke(self, revokeReq)); } else { revokeReq.reply.send(AssignBlobRangeReply(false)); } } - when(AssignBlobRangeRequest granuleToReassign = waitNext(self.granuleUpdateErrors.getFuture())) { - addActor.send(handleRangeAssign(&self, granuleToReassign, true)); + when(AssignBlobRangeRequest granuleToReassign = waitNext(self->granuleUpdateErrors.getFuture())) { + self->actors.add(handleRangeAssign(self, granuleToReassign, true)); } + when(HaltBlobWorkerRequest req = waitNext(bwInterf.haltBlobWorker.getFuture())) { + req.reply.send(Void()); + if (self->managerEpochOk(req.managerEpoch)) { + TraceEvent("BlobWorkerHalted", bwInterf.id()).detail("ReqID", req.requesterID); + printf("BW %s was halted\n", bwInterf.id().toString().c_str()); + break; + } + } + // when(wait(delay(10))) { throw granule_assignment_conflict(); } when(wait(collection)) { if (BW_DEBUG) { printf("BW actor collection returned, exiting\n"); } ASSERT(false); - throw internal_error(); + throw granule_assignment_conflict(); } } } catch (Error& e) { if (BW_DEBUG) { printf("Blob worker got error %s, exiting\n", e.name()); } - TraceEvent("BlobWorkerDied", self.id).error(e, true); - throw e; + TraceEvent("BlobWorkerDied", self->id).error(e, true); } + + printf("cancelling actors for BW %s\n", self->id.toString().c_str()); + self->actors.clear(false); + // self->addActor.sendError(granule_assignment_conflict()); + // + // self.addActor..clear(false); // tehcnically shouldn't need this since when self goes out of scope, so will + // self.actors + // at which point the cancels will be triggered? + self->dead.set(true); + return Void(); } // TODO add unit tests for assign/revoke range, especially version ordering From 423a67f44822b265d7e49506fc650dc3d2ba8598 Mon Sep 17 00:00:00 2001 From: Suraj Gupta Date: Fri, 8 Oct 2021 13:46:06 -0400 Subject: [PATCH 2/7] trying to fix infinite loop --- fdbserver/BlobManager.actor.cpp | 15 +- fdbserver/BlobWorker.actor.cpp | 554 ++++++++++++++++++-------------- 2 files changed, 315 insertions(+), 254 deletions(-) diff --git a/fdbserver/BlobManager.actor.cpp b/fdbserver/BlobManager.actor.cpp index 4da0fde2b9..e2d7f3d549 100644 --- a/fdbserver/BlobManager.actor.cpp +++ b/fdbserver/BlobManager.actor.cpp @@ -208,7 +208,7 @@ struct BlobManagerData { KeyRangeMap workerAssignments; KeyRangeMap knownBlobRanges; - AsyncVar restartRecruiting; + Debouncer restartRecruiting; std::set recruitingLocalities; // the addrs of the workers being recruited on int64_t epoch = -1; @@ -221,7 +221,8 @@ struct BlobManagerData { PromiseStream rangesToAssign; BlobManagerData(UID id, Database db) - : id(id), db(db), knownBlobRanges(false, normalKeys.end), restartRecruiting() {} + : id(id), db(db), knownBlobRanges(false, normalKeys.end), + restartRecruiting(SERVER_KNOBS->DEBOUNCE_RECRUITING_DELAY) {} ~BlobManagerData() { printf("Destroying blob manager data for %s\n", id.toString().c_str()); } }; @@ -433,7 +434,9 @@ ACTOR Future rangeAssigner(BlobManagerData* bmData) { workerId = assignment.worker.present() ? assignment.worker.get() : pickWorkerForAssign(bmData); bmData->workerAssignments.insert(assignment.keyRange, workerId); - bmData->workerStats[workerId].numGranulesAssigned += 1; + if (bmData->workerStats.count(workerId)) { + bmData->workerStats[workerId].numGranulesAssigned += 1; + } printf("current ranges after inserting assign: \n"); for (auto it : bmData->workerAssignments.ranges()) { @@ -1090,10 +1093,12 @@ ACTOR Future blobWorkerRecruiter( } TraceEvent("BMRecruiting").detail("State", "Sending request to CC"); + /* printf("EXCLUDING THE FOLLOWING IN REQ:\n"); for (auto addr : recruitReq.excludeAddresses) { - printf("- %s\n", addr.toString().c_str()); + printf("- %s\n", addr.toString().c_str()); } + */ if (!fCandidateWorker.isValid() || fCandidateWorker.isReady() || recruitReq.excludeAddresses != lastRequest.excludeAddresses) { @@ -1113,7 +1118,7 @@ ACTOR Future blobWorkerRecruiter( when(wait(recruitBlobWorker->onChange())) { fCandidateWorker = Future(); } // signal used to restart the loop and try to recruit the next blob worker - when(wait(self->restartRecruiting.onChange())) { printf("RESTARTED RECRUITING. BACK TO TOP\n"); } + when(wait(self->restartRecruiting.onTrigger())) { printf("RESTARTED RECRUITING. BACK TO TOP\n"); } } wait(delay(FLOW_KNOBS->PREVENT_FAST_SPIN_DELAY, TaskPriority::BlobManager)); } catch (Error& e) { diff --git a/fdbserver/BlobWorker.actor.cpp b/fdbserver/BlobWorker.actor.cpp index f4f648c493..b135702f02 100644 --- a/fdbserver/BlobWorker.actor.cpp +++ b/fdbserver/BlobWorker.actor.cpp @@ -72,30 +72,22 @@ struct GranuleChangeFeedInfo { Optional existingFiles; }; -// FIXME: the circular dependencies here are getting kind of gross -struct GranuleMetadata; -struct BlobWorkerData; -ACTOR Future persistAssignWorkerRange(Reference bwData, - AssignBlobRangeRequest req); -ACTOR Future blobGranuleUpdateFiles(Reference bwData, Reference metadata); - -// for a range that may or may not be set - struct GranuleMetadata : NonCopyable, ReferenceCounted { KeyRange keyRange; GranuleFiles files; - GranuleDeltas currentDeltas; + GranuleDeltas currentDeltas; // only contain deltas in pendingDeltaVersion + 1, bufferedDeltaVersion // TODO get rid of this and do Reference>? Arena deltaArena; uint64_t bytesInNewDeltaFiles = 0; uint64_t bufferedDeltaBytes = 0; - NotifiedVersion bufferedDeltaVersion; - Version pendingDeltaVersion = 0; - NotifiedVersion durableDeltaVersion; - NotifiedVersion durableSnapshotVersion; + // for client to know when it is safe to read a certain version and from where (check waitForVersion) + NotifiedVersion bufferedDeltaVersion; // largest delta version in currentDeltas (including empty versions) + Version pendingDeltaVersion = 0; // largest version in progress writing to s3/fdb + NotifiedVersion durableDeltaVersion; // largest version persisted in s3/fdb + NotifiedVersion durableSnapshotVersion; // same as delta vars, except for snapshots Version pendingSnapshotVersion = 0; AsyncVar rollbackCount; @@ -105,67 +97,51 @@ struct GranuleMetadata : NonCopyable, ReferenceCounted { int64_t continueEpoch; int64_t continueSeqno; - Future assignFuture; - Future fileUpdaterFuture; Promise resumeSnapshot; + + // used to coordinate granule file updater is done Promise cancelled; Promise readable; - AssignBlobRangeRequest originalReq; - - Future start(Reference bwData, AssignBlobRangeRequest req) { - originalReq = req; - assignFuture = persistAssignWorkerRange(bwData, req); - fileUpdaterFuture = blobGranuleUpdateFiles(bwData, Reference::addRef(this)); - // bwData->actors.add(blobGranuleUpdateFiles(bwData, Reference::addRef(this))); - // this could be the cause of the seg fault. since this is not being waited on, - // when start get cancelled, blobGranuleUpdateFiles won't get cancelled. so instead I added it to actors, so - // that it is explicitly cancelled. maybe this fixes it? - - return success(assignFuture); - // return Void(); - } - void resume() { ASSERT(resumeSnapshot.canBeSet()); resumeSnapshot.send(Void()); } - // FIXME: right now there is a dependency because this contains both the actual file/delta data as well as the - // metadata (worker futures), so removing this reference from the map doesn't actually cancel the workers. It'd be - // better to have this in 2 separate objects, where the granule metadata map has the futures, but the read - // queries/file updater/range feed only copy the reference to the file/delta data. - Future cancel(bool dispose) { + ~GranuleMetadata() { + printf("in dtor for GranuleMetadata\n"); if (cancelled.canBeSet()) { - // Could have been cancelled already by rollback or error in BGUpdateFiles cancelled.send(Void()); } - assignFuture.cancel(); - fileUpdaterFuture.cancel(); - - if (dispose) { - // FIXME: implement dispose! - return delay(0.1); - } - return Future(Void()); } }; struct GranuleRangeMetadata { + int id = 0; int64_t lastEpoch; int64_t lastSeqno; Reference activeMetadata; + Future assignFuture; + Future fileUpdaterFuture; + + AssignBlobRangeRequest originalReq; + GranuleRangeMetadata() : lastEpoch(0), lastSeqno(0) {} GranuleRangeMetadata(int64_t epoch, int64_t seqno, Reference activeMetadata) : lastEpoch(epoch), lastSeqno(seqno), activeMetadata(activeMetadata) {} - /* + ~GranuleRangeMetadata() { - if (activeMetadata.isValid()) { - activeMetadata->cancel(false); - } + assignFuture.cancel(); + fileUpdaterFuture.cancel(); + printf("GranuleRangeMetadata is being destroyed\n"); + /* + if (id == 42) { + printf("GranuleRangeMetadata with id %d is being destroyed\n", id); + sleep(10); + } + */ } - */ }; struct BlobWorkerData : NonCopyable, ReferenceCounted { @@ -195,7 +171,7 @@ struct BlobWorkerData : NonCopyable, ReferenceCounted { PromiseStream granuleUpdateErrors; BlobWorkerData(UID id, Database db) - : id(id), db(db), stats(id, SERVER_KNOBS->WORKER_LOGGING_INTERVAL), actors(false), dead(false) {} + : id(id), db(db), stats(id, SERVER_KNOBS->WORKER_LOGGING_INTERVAL), actors(false) {} ~BlobWorkerData() { printf("Destroying blob worker data for %s\n", id.toString().c_str()); } bool managerEpochOk(int64_t epoch) { @@ -238,7 +214,8 @@ static void checkGranuleLock(int64_t epoch, int64_t seqno, int64_t ownerEpoch, i // sanity check - lock value should never go backwards because of acquireGranuleLock /* printf( - "Checking granule lock: \n mine: (%lld, %lld)\n owner: (%lld, %lld)\n", epoch, seqno, ownerEpoch, ownerSeqno); + "Checking granule lock: \n mine: (%lld, %lld)\n owner: (%lld, %lld)\n", epoch, seqno, ownerEpoch, + ownerSeqno); */ ASSERT(epoch <= ownerEpoch); ASSERT(epoch < ownerEpoch || (epoch == ownerEpoch && seqno <= ownerSeqno)); @@ -339,21 +316,23 @@ ACTOR Future loadPreviousFiles(Transaction* tr, KeyRange keyRange) // update shared state to coordinate when it is safe to clean up the old change feed. // his goes through 3 phases for each new sub-granule: // 1. Starting - the blob manager writes all sub-granules with this state as a durable intent to split the range -// 2. Assigned - a worker that is assigned a sub-granule updates that granule's state here. This means that the worker +// 2. Assigned - a worker that is assigned a sub-granule updates that granule's state here. This means that the +// worker // has started a new change feed for the new sub-granule, but still needs to consume from the old change feed. -// 3. Done - the worker that is assigned this sub-granule has persisted all of the data from its part of the old change +// 3. Done - the worker that is assigned this sub-granule has persisted all of the data from its part of the old +// change // feed in delta files. From this granule's perspective, it is safe to clean up the old change feed. -// Once all sub-granules have reached step 2 (Assigned), the change feed can be safely "stopped" - it needs to continue -// to serve the mutations it has seen so far, but will not need any new mutations after this version. -// The last sub-granule to reach this step is responsible for commiting the change feed stop as part of its -// transaction. Because this change feed stops commits in the same transaction as the worker's new change feed start, -// it is guaranteed that no versions are missed between the old and new change feed. +// Once all sub-granules have reached step 2 (Assigned), the change feed can be safely "stopped" - it needs to +// continue to serve the mutations it has seen so far, but will not need any new mutations after this version. The +// last sub-granule to reach this step is responsible for commiting the change feed stop as part of its transaction. +// Because this change feed stops commits in the same transaction as the worker's new change feed start, it is +// guaranteed that no versions are missed between the old and new change feed. // -// Once all sub-granules have reached step 3 (Done), the change feed can be safely destroyed, as all of the mutations in -// the old change feed are guaranteed to be persisted in delta files. The last sub-granule to reach this step is -// responsible for committing the change feed destroy, and for cleaning up the split state for all sub-granules as part -// of its transaction. +// Once all sub-granules have reached step 3 (Done), the change feed can be safely destroyed, as all of the +// mutations in the old change feed are guaranteed to be persisted in delta files. The last sub-granule to reach +// this step is responsible for committing the change feed destroy, and for cleaning up the split state for all +// sub-granules as part of its transaction. ACTOR Future updateGranuleSplitState(Transaction* tr, KeyRange previousGranule, @@ -428,8 +407,8 @@ ACTOR Future updateGranuleSplitState(Transaction* tr, Key myStateKey = myStateTuple.getDataAsStandalone().withPrefix(blobGranuleSplitKeys.begin); if (newState == BlobGranuleSplitState::Done && currentState == BlobGranuleSplitState::Assigned && totalDone == total - 1) { - // we are the last one to change from Assigned -> Done, so everything can be cleaned up for the old change - // feed and splitting state + // we are the last one to change from Assigned -> Done, so everything can be cleaned up for the old + // change feed and splitting state if (BW_DEBUG) { printf("[%s - %s) destroying old change feed %s and granule lock + split state for [%s - %s)\n", currentGranule.begin.printable().c_str(), @@ -494,10 +473,11 @@ static Value getFileValue(std::string fname, int64_t offset, int64_t length) { return fileValue.getDataAsStandalone(); } -// writeDelta file writes speculatively in the common case to optimize throughput. It creates the s3 object even though -// the data in it may not yet be committed, and even though previous delta fiels with lower versioned data may still be -// in flight. The synchronization happens after the s3 file is written, but before we update the FDB index of what files -// exist. Before updating FDB, we ensure the version is committed and all previous delta files have updated FDB. +// writeDelta file writes speculatively in the common case to optimize throughput. It creates the s3 object even +// though the data in it may not yet be committed, and even though previous delta fiels with lower versioned data +// may still be in flight. The synchronization happens after the s3 file is written, but before we update the FDB +// index of what files exist. Before updating FDB, we ensure the version is committed and all previous delta files +// have updated FDB. ACTOR Future writeDeltaFile(Reference bwData, KeyRange keyRange, int64_t epoch, @@ -534,6 +514,7 @@ ACTOR Future writeDeltaFile(Reference bwData, wait(objectFile->append(serialized.begin(), serialized.size())); wait(objectFile->finish()); + state int numIterations = 0; try { // before updating FDB, wait for the delta file version to be committed and previous delta files to finish while (bwData->knownCommittedVersion.get() < currentDeltaVersion) { @@ -560,8 +541,8 @@ ACTOR Future writeDeltaFile(Reference bwData, Key dfKey = deltaFileKey.getDataAsStandalone().withPrefix(blobGranuleFileKeys.begin); tr->set(dfKey, getFileValue(fname, 0, serialized.size())); - // FIXME: if previous granule present and delta file version >= previous change feed version, update the - // state here + // FIXME: if previous granule present and delta file version >= previous change feed version, update + // the state here if (oldChangeFeedDataComplete.present()) { ASSERT(oldChangeFeedId.present()); wait(updateGranuleSplitState(&tr->getTransaction(), @@ -587,6 +568,8 @@ ACTOR Future writeDeltaFile(Reference bwData, } return BlobFileIndex(currentDeltaVersion, fname, 0, serialized.size()); } catch (Error& e) { + numIterations++; + printf("writeDeltaFile error: %s\n", e.name()); wait(tr->onError(e)); } } @@ -594,6 +577,10 @@ ACTOR Future writeDeltaFile(Reference bwData, if (e.code() == error_code_operation_cancelled) { throw e; } + // TODO: do this for writeSnapshot + if (numIterations != 1 || e.code() != error_code_granule_assignment_conflict) { + throw e; + } // FIXME: only delete if key doesn't exist if (BW_DEBUG) { @@ -680,6 +667,7 @@ ACTOR Future writeSnapshot(Reference bwData, snapshotFileKey.append(LiteralStringRef("S")).append(version); state Reference tr = makeReference(bwData->db); + state int numIterations = 0; try { loop { @@ -691,6 +679,7 @@ ACTOR Future writeSnapshot(Reference bwData, wait(tr->commit()); break; } catch (Error& e) { + numIterations++; wait(tr->onError(e)); } } @@ -699,6 +688,10 @@ ACTOR Future writeSnapshot(Reference bwData, throw e; } + if (numIterations != 1 || e.code() != error_code_granule_assignment_conflict) { + throw e; + } + // FIXME: only delete if key doesn't exist if (BW_DEBUG) { printf("deleting s3 snapshot file %s after error %s\n", fname.c_str(), e.name()); @@ -839,8 +832,8 @@ ACTOR Future compactFromBlob(Reference bwData, DEBUG_KEY_RANGE("BlobWorkerBlobSnapshot", version, metadata->keyRange, bwData->id); return f; } catch (Error& e) { - // TODO better error handling eventually - should retry unless the error is because another worker took over - // the range + // TODO better error handling eventually - should retry unless the error is because another worker took + // over the range if (BW_DEBUG) { printf("Compacting snapshot from blob for [%s - %s) got error %s\n", metadata->keyRange.begin.printable().c_str(), @@ -1019,8 +1012,8 @@ static Version doGranuleRollback(Reference metadata, metadata->currentDeltas.resize(metadata->deltaArena, mIdx); - // delete all deltas in rollback range, but we can optimize here to just skip the uncommitted mutations directly - // and immediately pop the rollback out of inProgress + // delete all deltas in rollback range, but we can optimize here to just skip the uncommitted mutations + // directly and immediately pop the rollback out of inProgress metadata->bufferedDeltaVersion.set(rollbackVersion); cfRollbackVersion = mutationVersion; } @@ -1047,7 +1040,7 @@ static Version doGranuleRollback(Reference metadata, // 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) -ACTOR Future blobGranuleUpdateFiles(Reference bwData, Reference metadata) { +ACTOR Future blobGranuleUpdateFiles(Reference bwData, GranuleRangeMetadata* rangeMetadata) { state PromiseStream>> oldChangeFeedStream; state PromiseStream>> changeFeedStream; state Future inFlightBlobSnapshot; @@ -1067,12 +1060,17 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, Refe state bool snapshotEligible; // just wrote a delta file or just took granule over from another worker state bool justDidRollback = false; + state Reference metadata = rangeMetadata->activeMetadata; + printf( + "metadata %s %s\n", metadata->keyRange.begin.printable().c_str(), metadata->keyRange.end.printable().c_str()); try { // set resume snapshot so it's not valid until we pause to ask the blob manager for a re-snapshot metadata->resumeSnapshot.send(Void()); // before starting, make sure worker persists range assignment and acquires the granule lock - GranuleChangeFeedInfo _info = wait(metadata->assignFuture); + printf("before wait on assignFuture\n"); + GranuleChangeFeedInfo _info = wait(rangeMetadata->assignFuture); + printf("after wait on assignFuture\n"); changeFeedInfo = _info; wait(delay(0, TaskPriority::BlobWorkerUpdateStorage)); @@ -1152,9 +1150,9 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, Refe metadata->readable.send(Void()); if (changeFeedInfo.prevChangeFeedId.present()) { - // FIXME: once we have empty versions, only include up to changeFeedInfo.changeFeedStartVersion in the read - // stream. Then we can just stop the old stream when we get end_of_stream from this and not handle the - // mutation version truncation stuff + // FIXME: once we have empty versions, only include up to changeFeedInfo.changeFeedStartVersion in the + // read stream. Then we can just stop the old stream when we get end_of_stream from this and not handle + // the mutation version truncation stuff // FIXME: filtering on key range != change feed range doesn't work ASSERT(changeFeedInfo.granuleSplitFrom.present()); @@ -1173,6 +1171,8 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, Refe } loop { + printf("bw %s\n", bwData->id.toString().c_str()); + // check outstanding snapshot/delta files for completion if (inFlightBlobSnapshot.isValid() && inFlightBlobSnapshot.isReady()) { BlobFileIndex completedSnapshot = wait(inFlightBlobSnapshot); @@ -1219,8 +1219,8 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, Refe // TODO filter old mutations won't be necessary, SS does it already if (filterOldMutations( metadata->keyRange, &oldMutations, &mutations, changeFeedInfo.changeFeedStartVersion)) { - // if old change feed has caught up with where new one would start, finish last one and start new - // one + // if old change feed has caught up with where new one would start, finish last one and start + // new one Key cfKey = StringRef(changeFeedInfo.changeFeedId.toString()); changeFeedFuture = bwData->db->getChangeFeedStream(changeFeedStream, @@ -1241,12 +1241,13 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, Refe } // process mutations + printf("mutations.size() == %d\n", mutations.size()); for (MutationsAndVersionRef d : mutations) { state MutationsAndVersionRef deltas = d; ASSERT(deltas.version >= metadata->bufferedDeltaVersion.get()); // Write a new delta file IF we have enough bytes, and we have all of the previous version's stuff - // there to ensure no versions span multiple delta files. Check this by ensuring the version of this new - // delta is larger than the previous largest seen version + // there to ensure no versions span multiple delta files. Check this by ensuring the version of this + // new delta is larger than the previous largest seen version if (metadata->bufferedDeltaBytes >= SERVER_KNOBS->BG_DELTA_FILE_TARGET_BYTES && deltas.version > metadata->bufferedDeltaVersion.get()) { if (BW_DEBUG) { @@ -1305,8 +1306,8 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, Refe metadata->bufferedDeltaBytes = 0; // if we just wrote a delta file, check if we need to compact here. - // exhaust old change feed before compacting - otherwise we could end up with an endlessly growing - // list of previous change feeds in the worst case. + // exhaust old change feed before compacting - otherwise we could end up with an endlessly + // growing list of previous change feeds in the worst case. snapshotEligible = true; } @@ -1369,16 +1370,19 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, Refe Optional result = wait(timeout(metadata->resumeSnapshot.getFuture(), 1.0)); if (result.present()) { break; - } else if (bwData->dead.get()) { + } + + if (bwData->dead.get()) { + std::cout << "bw detected dead in blobGranuleUpdateFiles" << std::endl; throw actor_cancelled(); } - // FIXME: re-trigger this loop if blob manager status stream changes if (BW_DEBUG) { printf("Granule [%s - %s)\n, hasn't heard back from BM in BW %s, re-sending status\n", metadata->keyRange.begin.printable().c_str(), metadata->keyRange.end.printable().c_str(), bwData->id.toString().c_str()); } + wait(yield()); } if (BW_DEBUG) { @@ -1404,8 +1408,8 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, Refe metadata->bytesInNewDeltaFiles = 0; } else if (snapshotEligible && metadata->bytesInNewDeltaFiles >= SERVER_KNOBS->BG_DELTA_BYTES_BEFORE_COMPACT) { - // if we're in the old change feed case and can't snapshot but we have enough data to, don't queue - // too many delta files in parallel + // if we're in the old change feed case and can't snapshot but we have enough data to, don't + // queue too many delta files in parallel while (inFlightDeltaFiles.size() > 10) { if (BW_DEBUG) { printf("[%s - %s) Waiting on delta file b/c old change feed\n", @@ -1550,9 +1554,10 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, Refe } } justDidRollback = false; + std::cout << "looping in blobGranuleUpdateFiles" << std::endl; } - } catch (Error& e) { + printf("error is %s\n", e.name()); printf("IN CATCH FOR blobGranuleUpdateFiles -----------------------------------\n "); if (e.code() == error_code_operation_cancelled) { throw; @@ -1575,14 +1580,14 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, Refe 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 + // 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.future.cancel(); } - bwData->granuleUpdateErrors.send(metadata->originalReq); + bwData->granuleUpdateErrors.send(rangeMetadata->originalReq); } } throw e; @@ -1669,6 +1674,7 @@ static Future waitForVersion(Reference metadata, Version } ACTOR Future handleBlobGranuleFileRequest(Reference bwData, BlobGranuleFileRequest req) { + // printf("In handleBlobGranuleFileRequest\n"); try { // TODO REMOVE in api V2 ASSERT(req.beginVersion == 0); @@ -1690,6 +1696,8 @@ ACTOR Future handleBlobGranuleFileRequest(Reference bwData req.keyRange.begin.printable().c_str(), req.keyRange.end.printable().c_str()); } + + printf("lastRangeEnd < r.begin() || !isValid\n"); throw wrong_shard_server(); } granules.push_back(r.value().activeMetadata); @@ -1704,6 +1712,7 @@ ACTOR Future handleBlobGranuleFileRequest(Reference bwData req.keyRange.end.printable().c_str()); } + printf("lastRangeEnd < req.keyRange.end\n"); throw wrong_shard_server(); } @@ -1725,6 +1734,7 @@ ACTOR Future handleBlobGranuleFileRequest(Reference bwData throw transaction_too_old(); } if (metadata->cancelled.isSet()) { + printf("metadata->cancelled.isSet()\n"); throw wrong_shard_server(); } @@ -1740,7 +1750,7 @@ ACTOR Future handleBlobGranuleFileRequest(Reference bwData when(wait(waitForVersionFuture)) {} when(wait(metadata->rollbackCount.onChange())) {} when(wait(metadata->cancelled.getFuture())) { - printf("metadata was cancelled\n"); + printf("metadata->cancelled.getFuture()\n"); throw wrong_shard_server(); } } @@ -1852,6 +1862,7 @@ ACTOR Future handleBlobGranuleFileRequest(Reference bwData ACTOR Future persistAssignWorkerRange(Reference bwData, AssignBlobRangeRequest req) { + printf("in persistAssignWorkerRange\n"); ASSERT(!req.continueAssignment); state Transaction tr(bwData->db); state Key lockKey = granuleLockKey(req.keyRange); @@ -1863,141 +1874,163 @@ ACTOR Future persistAssignWorkerRange(Reference prevLockValue = wait(tr.get(lockKey)); - 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); - - GranuleFiles granuleFiles = wait(loadPreviousFiles(&tr, req.keyRange)); - info.existingFiles = granuleFiles; - info.doSnapshot = false; - - if (info.existingFiles.get().snapshotFiles.empty()) { - ASSERT(info.existingFiles.get().deltaFiles.empty()); - info.previousDurableVersion = invalidVersion; - info.doSnapshot = true; - } else if (info.existingFiles.get().deltaFiles.empty()) { - info.previousDurableVersion = info.existingFiles.get().snapshotFiles.back().version; - } else { - info.previousDurableVersion = info.existingFiles.get().deltaFiles.back().version; - } - - // 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; - } - - 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.begin).append(req.keyRange.end); - 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 (parentGranulesValue.present()) { - state Standalone> 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(parentGranules.size() == 1); - - state std::pair granuleSplitState; + // 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)); + state bool hasPrevOwner = prevLockValue.present(); if (hasPrevOwner) { - std::pair _st = - wait(getGranuleSplitState(&tr, parentGranules[0], req.keyRange)); - granuleSplitState = _st; + 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); + + GranuleFiles granuleFiles = wait(loadPreviousFiles(&tr, req.keyRange)); + info.existingFiles = granuleFiles; + info.doSnapshot = false; + + if (info.existingFiles.get().snapshotFiles.empty()) { + ASSERT(info.existingFiles.get().deltaFiles.empty()); + info.previousDurableVersion = invalidVersion; + info.doSnapshot = true; + } else if (info.existingFiles.get().deltaFiles.empty()) { + info.previousDurableVersion = info.existingFiles.get().snapshotFiles.back().version; + } else { + info.previousDurableVersion = info.existingFiles.get().deltaFiles.back().version; + } + + // 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 { - granuleSplitState = std::pair(BlobGranuleSplitState::Started, invalidVersion); + // 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; } - ASSERT(!hasPrevOwner || granuleSplitState.first > BlobGranuleSplitState::Started); + tr.set(lockKey, blobGranuleLockValueFor(req.managerEpoch, req.managerSeqno, info.changeFeedId)); + wait(krmSetRange( + &tr, blobGranuleMappingKeys.begin, req.keyRange, blobGranuleMappingValueFor(bwData->id))); - // 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]; + Tuple historyKey; + historyKey.append(req.keyRange.begin).append(req.keyRange.end); + Optional parentGranulesValue = + wait(tr.get(historyKey.getDataAsStandalone().withPrefix(blobGranuleHistoryKeys.begin))); - if (granuleSplitState.first == BlobGranuleSplitState::Assigned) { - // was already assigned, use change feed start version - ASSERT(granuleSplitState.second != invalidVersion); - info.changeFeedStartVersion = granuleSplitState.second; - } else if (granuleSplitState.first == BlobGranuleSplitState::Started) { - wait(updateGranuleSplitState(&tr, - parentGranules[0], - req.keyRange, - info.prevChangeFeedId.get(), - BlobGranuleSplitState::Assigned)); - // change feed was created as part of this transaction, changeFeedStartVersion will be set - // later + // 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 (parentGranulesValue.present()) { + state Standalone> 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(parentGranules.size() == 1); + + state std::pair granuleSplitState; + if (hasPrevOwner) { + std::pair _st = + wait(getGranuleSplitState(&tr, parentGranules[0], req.keyRange)); + granuleSplitState = _st; } else { - ASSERT(false); + granuleSplitState = std::pair(BlobGranuleSplitState::Started, invalidVersion); + } + + ASSERT(!hasPrevOwner || granuleSplitState.first > BlobGranuleSplitState::Started); + + // 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; + } else if (granuleSplitState.first == BlobGranuleSplitState::Started) { + wait(updateGranuleSplitState(&tr, + parentGranules[0], + req.keyRange, + info.prevChangeFeedId.get(), + BlobGranuleSplitState::Assigned)); + // change feed was created as part of this transaction, changeFeedStartVersion will be + // set later + } else { + ASSERT(false); + } + } + + if (info.doSnapshot) { + // only need to do snapshot if no files exist yet for this granule. + ASSERT(info.previousDurableVersion == invalidVersion); + // 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; } } - if (info.doSnapshot) { - // only need to do snapshot if no files exist yet for this granule. - ASSERT(info.previousDurableVersion == invalidVersion); - // FIXME: store this somewhere useful for time travel reads - GranuleFiles prevFiles = wait(loadPreviousFiles(&tr, parentGranules[0])); - ASSERT(!prevFiles.snapshotFiles.empty() || !prevFiles.deltaFiles.empty()); + wait(tr.commit()); - info.blobFilesToSnapshot = prevFiles; - info.previousDurableVersion = info.blobFilesToSnapshot.get().deltaFiles.empty() - ? info.blobFilesToSnapshot.get().snapshotFiles.back().version - : info.blobFilesToSnapshot.get().deltaFiles.back().version; + if (info.changeFeedStartVersion == invalidVersion) { + 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) { + throw e; + } + wait(tr.onError(e)); } - - wait(tr.commit()); - - if (info.changeFeedStartVersion == invalidVersion) { - 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) { - throw e; - } - wait(tr.onError(e)); } + } catch (Error& e) { + printf("ERROR IN PERSIST: %s\n", e.name()); + throw; + } +} + +ACTOR Future start(Reference bwData, GranuleRangeMetadata* meta, AssignBlobRangeRequest req) { + try { + meta->originalReq = req; + meta->assignFuture = persistAssignWorkerRange(bwData, req); + meta->fileUpdaterFuture = blobGranuleUpdateFiles(bwData, meta); + bwData->actors.add(meta->fileUpdaterFuture); + wait(success(meta->assignFuture)); + return Void(); + } catch (Error& e) { + meta->assignFuture.cancel(); + meta->fileUpdaterFuture.cancel(); + throw; } } @@ -2043,13 +2076,14 @@ static bool newerRangeAssignment(GranuleRangeMetadata oldMetadata, int64_t epoch // Returns future to wait on to ensure prior work of other granules is done before responding to the manager with a // successful assignment And if the change produced a new granule that needs to start doing work, returns the new // granule so that the caller can start() it with the appropriate starting state. -static std::pair, Reference> changeBlobRange(Reference bwData, - KeyRange keyRange, - int64_t epoch, - int64_t seqno, - bool active, - bool disposeOnCleanup, - bool selfReassign) { +ACTOR Future changeBlobRange(Reference bwData, + KeyRange keyRange, + int64_t epoch, + int64_t seqno, + bool active, + bool disposeOnCleanup, + bool selfReassign) { + printf("changeBlobRange called\n"); if (BW_DEBUG) { printf("%s range for [%s - %s): %s @ (%lld, %lld)\n", selfReassign ? "Re-assigning" : "Changing", @@ -2066,11 +2100,12 @@ static std::pair, Reference> changeBlobRange(Refer // older range, cancel it if it is active. Insert the current range. Re-insert all newer ranges over the current // range. - std::vector> futures; + state std::vector> futures; - std::vector> newerRanges; + state std::vector> newerRanges; auto ranges = bwData->granuleMetadata.intersectingRanges(keyRange); + bool alreadyAssigned = false; for (auto& r : ranges) { bool thisAssignmentNewer = newerRangeAssignment(r.value(), epoch, seqno); if (r.value().lastEpoch == epoch && r.value().lastSeqno == seqno) { @@ -2080,12 +2115,13 @@ static std::pair, Reference> changeBlobRange(Refer if (selfReassign) { thisAssignmentNewer = true; } else { + printf("same assignment\n"); // applied the same assignment twice, make idempotent if (r.value().activeMetadata.isValid()) { - futures.push_back(success(r.value().activeMetadata->assignFuture)); + futures.push_back(success(r.value().assignFuture)); } - return std::pair(waitForAll(futures), - Reference()); // already applied, nothing to do + alreadyAssigned = true; + break; } } @@ -2098,7 +2134,6 @@ static std::pair, Reference> changeBlobRange(Refer r.value().lastEpoch, r.value().lastSeqno); } - futures.push_back(r.value().activeMetadata->cancel(disposeOnCleanup)); r.value().activeMetadata.clear(); } else if (!thisAssignmentNewer) { // this assignment is outdated, re-insert it over the current range @@ -2106,10 +2141,16 @@ static std::pair, Reference> changeBlobRange(Refer } } + if (alreadyAssigned) { + wait(waitForAll(futures)); // already applied, nothing to do + return false; + } + // if range is active, and isn't surpassed by a newer range already, insert an active range GranuleRangeMetadata newMetadata = (active && newerRanges.empty()) ? constructActiveBlobRange(bwData, keyRange, epoch, seqno) : constructInactiveBlobRange(epoch, seqno); + newMetadata.id = 42; bwData->granuleMetadata.insert(keyRange, newMetadata); if (BW_DEBUG) { printf("Inserting new range [%s - %s): %s @ (%lld, %lld)\n", @@ -2132,7 +2173,9 @@ static std::pair, Reference> changeBlobRange(Refer bwData->granuleMetadata.insert(it.first, it.second); } - return std::pair(waitForAll(futures), newMetadata.activeMetadata); + printf("returning from changeblobrange"); + wait(waitForAll(futures)); + return true; } static bool resumeBlobRange(Reference bwData, KeyRange keyRange, int64_t epoch, int64_t seqno) { @@ -2204,21 +2247,39 @@ ACTOR Future handleRangeAssign(Reference bwData, if (req.continueAssignment) { resumeBlobRange(bwData, req.keyRange, req.managerEpoch, req.managerSeqno); } else { - state std::pair, Reference> futureAndNewGranule = - changeBlobRange(bwData, req.keyRange, req.managerEpoch, req.managerSeqno, true, false, isSelfReassign); + bool shouldStart = wait( + changeBlobRange(bwData, req.keyRange, req.managerEpoch, req.managerSeqno, true, false, isSelfReassign)); - wait(futureAndNewGranule.first); + if (shouldStart) { + auto m = bwData->granuleMetadata.rangeContaining(req.keyRange.begin); + ASSERT(m.begin() == req.keyRange.begin && m.end() == req.keyRange.end); + printf("About to start for BW %s\n", bwData->id.toString().c_str()); + wait(start(bwData, &m.value(), req)); + /* + int count = 0; + // GranuleRangeMetadata& x; + for (auto& it : m) { + printf("BW %s ABOUT TO WAIT IN HANDLERANGEASSIGN\n", bwData->id.toString().c_str()); + wait(start(bwData, &it.value(), req)); + printf("done waiting in handleRangeAssign\n"); + count++; + } + ASSERT(count == 1); + // x.id = 42; - if (futureAndNewGranule.second.isValid()) { printf("BW %s ABOUT TO WAIT IN HANDLERANGEASSIGN\n", bwData->id.toString().c_str()); - // WAITING ON START BUT ITS NOT AN ACTOR!!!!!!! SO WHEN handlerangeassign gets operation_cancelled, it - // won't get propogated to start - wait(futureAndNewGranule.second->start(bwData, req)); + // WAITING ON START BUT ITS NOT AN ACTOR!!!!!!! SO WHEN handlerangeassign gets operation_cancelled, + // it won't get propogated to start + // wait(start(bwData, x, req)); + printf("done waiting in handleRangeAssign\n"); + */ } } if (!isSelfReassign) { ASSERT(!req.reply.isSet()); + printf("about to send reply\n"); req.reply.send(AssignBlobRangeReply(true)); + printf("done sending reply\n"); } return Void(); } catch (Error& e) { @@ -2231,11 +2292,11 @@ ACTOR Future handleRangeAssign(Reference bwData, e.name()); } - /* - if (futureAndNewGranule.get().second.isValid()) { - wait(futureAndNewGranule.get().second->cancel(false)); - } - */ + // + // if (futureAndNewGranule.get().second.isValid()) { + // wait(futureAndNewGranule.get().second->cancel(false)); + //} + // if (!isSelfReassign) { if (canReplyWith(eState)) { @@ -2249,8 +2310,8 @@ ACTOR Future handleRangeAssign(Reference bwData, ACTOR Future handleRangeRevoke(Reference bwData, RevokeBlobRangeRequest req) { try { - wait( - changeBlobRange(bwData, req.keyRange, req.managerEpoch, req.managerSeqno, false, req.dispose, false).first); + bool _ = + wait(changeBlobRange(bwData, req.keyRange, req.managerEpoch, req.managerSeqno, false, req.dispose, false)); req.reply.send(AssignBlobRangeReply(true)); return Void(); } catch (Error& e) { @@ -2443,12 +2504,7 @@ ACTOR Future blobWorker(BlobWorkerInterface bwInterf, printf("cancelling actors for BW %s\n", self->id.toString().c_str()); self->actors.clear(false); - // self->addActor.sendError(granule_assignment_conflict()); - // - // self.addActor..clear(false); // tehcnically shouldn't need this since when self goes out of scope, so will - // self.actors - // at which point the cancels will be triggered? - self->dead.set(true); + // self->dead = true; return Void(); } From 266a5b06fabe3174e3029bb1eab1c0a50dc6b599 Mon Sep 17 00:00:00 2001 From: Suraj Gupta Date: Tue, 12 Oct 2021 15:52:55 -0400 Subject: [PATCH 3/7] Fix infinite loop. --- fdbrpc/fdbrpc.h | 4 +- fdbserver/BlobManager.actor.cpp | 5 +- fdbserver/BlobWorker.actor.cpp | 160 ++++++++++++------ fdbserver/ClusterController.actor.cpp | 5 + .../workloads/BlobGranuleVerifier.actor.cpp | 3 +- tests/CMakeLists.txt | 3 +- tests/fast/BlobGranuleCorrectness.toml | 24 +++ tests/fast/BlobGranuleCorrectnessClean.toml | 10 ++ 8 files changed, 154 insertions(+), 60 deletions(-) create mode 100644 tests/fast/BlobGranuleCorrectnessClean.toml diff --git a/fdbrpc/fdbrpc.h b/fdbrpc/fdbrpc.h index c2641b95b2..b4a3606d94 100644 --- a/fdbrpc/fdbrpc.h +++ b/fdbrpc/fdbrpc.h @@ -481,12 +481,14 @@ public: const Endpoint& getEndpoint() const { return queue->getEndpoint(TaskPriority::ReadSocket); } bool operator==(const ReplyPromiseStream& rhs) const { return queue == rhs.queue; } + bool operator!=(const ReplyPromiseStream& rhs) const { return !(*this == rhs); } + bool isEmpty() const { return !queue->isReady(); } uint32_t size() const { return queue->size(); } // Must be called on the server before sending results on the stream to ratelimit the amount of data outstanding to // the client - Future onReady() { + Future onReady() const { ASSERT(queue->acknowledgements.bytesLimit > 0); if (queue->acknowledgements.failures.isError()) { return queue->acknowledgements.failures.getError(); diff --git a/fdbserver/BlobManager.actor.cpp b/fdbserver/BlobManager.actor.cpp index e2d7f3d549..208674970a 100644 --- a/fdbserver/BlobManager.actor.cpp +++ b/fdbserver/BlobManager.actor.cpp @@ -434,7 +434,7 @@ ACTOR Future rangeAssigner(BlobManagerData* bmData) { workerId = assignment.worker.present() ? assignment.worker.get() : pickWorkerForAssign(bmData); bmData->workerAssignments.insert(assignment.keyRange, workerId); - if (bmData->workerStats.count(workerId)) { + if (bmData->workerStats.count(workerId) && !assignment.assign.get().continueAssignment) { bmData->workerStats[workerId].numGranulesAssigned += 1; } @@ -897,13 +897,11 @@ ACTOR Future monitorBlobWorker(BlobManagerData* bmData, BlobWorkerInterfac choose { when(wait(waitFailure)) { - // FIXME: actually handle this!! if (BM_DEBUG) { printf("BM %lld detected BW %s is dead\n", bmData->epoch, bwInterf.id().toString().c_str()); } TraceEvent("BlobWorkerFailed", bmData->id).detail("BlobWorkerID", bwInterf.id()); killBlobWorker(bmData, bwInterf); - return Void(); } when(wait(monitorStatus)) { ASSERT(false); @@ -963,7 +961,6 @@ ACTOR Future rangeMover(BlobManagerData* bmData) { RangeAssignment revokeOld; revokeOld.isAssign = false; revokeOld.keyRange = randomRange.range(); - revokeOld.worker = randomRange.value(); revokeOld.revoke = RangeRevokeData(false); bmData->rangesToAssign.send(revokeOld); diff --git a/fdbserver/BlobWorker.actor.cpp b/fdbserver/BlobWorker.actor.cpp index b135702f02..109cfed2e4 100644 --- a/fdbserver/BlobWorker.actor.cpp +++ b/fdbserver/BlobWorker.actor.cpp @@ -40,7 +40,7 @@ #include "flow/flow.h" #define BW_DEBUG true -#define BW_REQUEST_DEBUG false +#define BW_REQUEST_DEBUG true // TODO add comments + documentation struct BlobFileIndex { @@ -103,19 +103,17 @@ struct GranuleMetadata : NonCopyable, ReferenceCounted { Promise cancelled; Promise readable; + AssignBlobRangeRequest originalReq; + void resume() { ASSERT(resumeSnapshot.canBeSet()); resumeSnapshot.send(Void()); } - ~GranuleMetadata() { - printf("in dtor for GranuleMetadata\n"); - if (cancelled.canBeSet()) { - cancelled.send(Void()); - } - } + ~GranuleMetadata() { printf("in dtor for GranuleMetadata\n"); } }; +// TODO: rename this struct struct GranuleRangeMetadata { int id = 0; int64_t lastEpoch; @@ -125,16 +123,26 @@ struct GranuleRangeMetadata { Future assignFuture; Future fileUpdaterFuture; - AssignBlobRangeRequest originalReq; - GranuleRangeMetadata() : lastEpoch(0), lastSeqno(0) {} GranuleRangeMetadata(int64_t epoch, int64_t seqno, Reference activeMetadata) - : lastEpoch(epoch), lastSeqno(seqno), activeMetadata(activeMetadata) {} + : lastEpoch(epoch), lastSeqno(seqno), activeMetadata(activeMetadata) { + /* + if (activeMetadata.isValid()) { + activeMetadata->cancelled.reset(); + } + */ + } ~GranuleRangeMetadata() { + printf("GranuleRangeMetadata is being destroyed\n"); + /* + if (activeMetadata.isValid() && activeMetadata->cancelled.canBeSet()) { + printf("Cancelling activeMetadata\n"); + activeMetadata->cancelled.send(Void()); + } assignFuture.cancel(); fileUpdaterFuture.cancel(); - printf("GranuleRangeMetadata is being destroyed\n"); + */ /* if (id == 42) { printf("GranuleRangeMetadata with id %d is being destroyed\n", id); @@ -157,7 +165,7 @@ struct BlobWorkerData : NonCopyable, ReferenceCounted { LocalityData locality; int64_t currentManagerEpoch = -1; - ReplyPromiseStream currentManagerStatusStream; + AsyncVar> currentManagerStatusStream; // FIXME: refactor out the parts of this that are just for interacting with blob stores from the backup business // logic @@ -221,6 +229,7 @@ static void checkGranuleLock(int64_t epoch, int64_t seqno, int64_t ownerEpoch, i ASSERT(epoch < ownerEpoch || (epoch == ownerEpoch && seqno <= ownerSeqno)); // returns true if we still own the lock, false if someone else does + printf("epoch: %lld, seqno: %lld, ownerEpoch: %lld, ownerSeqno: %lld\n", epoch, seqno, ownerEpoch, ownerSeqno); if (epoch != ownerEpoch || seqno != ownerSeqno) { if (BW_DEBUG) { printf("Lock assignment check failed. Expected (%lld, %lld), got (%lld, %lld)\n", @@ -1040,7 +1049,9 @@ static Version doGranuleRollback(Reference metadata, // 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) -ACTOR Future blobGranuleUpdateFiles(Reference bwData, GranuleRangeMetadata* rangeMetadata) { +ACTOR Future blobGranuleUpdateFiles(Reference bwData, + Reference metadata, + Future assignFuture) { state PromiseStream>> oldChangeFeedStream; state PromiseStream>> changeFeedStream; state Future inFlightBlobSnapshot; @@ -1060,7 +1071,6 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, Gran state bool snapshotEligible; // just wrote a delta file or just took granule over from another worker state bool justDidRollback = false; - state Reference metadata = rangeMetadata->activeMetadata; printf( "metadata %s %s\n", metadata->keyRange.begin.printable().c_str(), metadata->keyRange.end.printable().c_str()); try { @@ -1069,7 +1079,7 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, Gran // before starting, make sure worker persists range assignment and acquires the granule lock printf("before wait on assignFuture\n"); - GranuleChangeFeedInfo _info = wait(rangeMetadata->assignFuture); + GranuleChangeFeedInfo _info = wait(assignFuture); printf("after wait on assignFuture\n"); changeFeedInfo = _info; @@ -1188,6 +1198,8 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, Gran wait(yield(TaskPriority::BlobWorkerUpdateStorage)); } if (!inFlightBlobSnapshot.isValid()) { + printf("!inFlightBlobSnapshot.isValid\n"); + printf("inFlightDeltaFiles.size() = %d\n", inFlightDeltaFiles.size()); while (inFlightDeltaFiles.size() > 0) { if (inFlightDeltaFiles.front().future.isReady()) { BlobFileIndex completedDeltaFile = wait(inFlightDeltaFiles.front().future); @@ -1215,6 +1227,7 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, Gran state Standalone> mutations; if (readOldChangeFeed) { + printf("readOldChangeFeed\n"); Standalone> oldMutations = waitNext(oldChangeFeedStream.getFuture()); // TODO filter old mutations won't be necessary, SS does it already if (filterOldMutations( @@ -1244,6 +1257,9 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, Gran printf("mutations.size() == %d\n", mutations.size()); for (MutationsAndVersionRef d : mutations) { state MutationsAndVersionRef deltas = d; + if (deltas.version >= 158685394) { + printf("deltas.version=%lld\n", deltas.version); + } ASSERT(deltas.version >= metadata->bufferedDeltaVersion.get()); // Write a new delta file IF we have enough bytes, and we have all of the previous version's stuff // there to ensure no versions span multiple delta files. Check this by ensuring the version of this @@ -1315,7 +1331,8 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, Gran // bunch of extra delta files at some point, even if we don't consider it for a split yet if (snapshotEligible && metadata->bytesInNewDeltaFiles >= SERVER_KNOBS->BG_DELTA_BYTES_BEFORE_COMPACT && !readOldChangeFeed) { - + printf("snapshotEligible && metadata->bytesInNewDeltaFiles >= " + "SERVER_KNOBS->BG_DELTA_BYTES_BEFORE_COMPACT && !readOldChangeFeed\n"); if (BW_DEBUG && (inFlightBlobSnapshot.isValid() || !inFlightDeltaFiles.empty())) { printf("Granule [%s - %s) ready to re-snapshot, waiting for outstanding %d snapshot and %d " "deltas to " @@ -1364,25 +1381,37 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, Gran 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; + loop { + try { + wait(bwData->currentManagerStatusStream.get().onReady()); + bwData->currentManagerStatusStream.get().send( + GranuleStatusReply(metadata->keyRange, true, statusEpoch, statusSeqno)); + break; + } catch (Error& e) { + printf("manager stream was changed\n"); + wait(bwData->currentManagerStatusStream.onChange()); + } } + choose { + when(wait(metadata->resumeSnapshot.getFuture())) { break; } + when(wait(delay(1.0))) {} + when(wait(bwData->currentManagerStatusStream.onChange())) {} + } + + /* if (bwData->dead.get()) { - std::cout << "bw detected dead in blobGranuleUpdateFiles" << std::endl; - throw actor_cancelled(); + std::cout << "bw detected dead in blobGranuleUpdateFiles" << std::endl; + throw actor_cancelled(); } + */ if (BW_DEBUG) { printf("Granule [%s - %s)\n, hasn't heard back from BM in BW %s, re-sending status\n", metadata->keyRange.begin.printable().c_str(), metadata->keyRange.end.printable().c_str(), bwData->id.toString().c_str()); } - wait(yield()); + // wait(yield()); } if (BW_DEBUG) { @@ -1408,6 +1437,8 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, Gran metadata->bytesInNewDeltaFiles = 0; } else if (snapshotEligible && metadata->bytesInNewDeltaFiles >= SERVER_KNOBS->BG_DELTA_BYTES_BEFORE_COMPACT) { + printf("snapshotEligible && metadata->bytesInNewDeltaFiles >= " + "SERVER_KNOBS->BG_DELTA_BYTES_BEFORE_COMPACT\n"); // if we're in the old change feed case and can't snapshot but we have enough data to, don't // queue too many delta files in parallel while (inFlightDeltaFiles.size() > 10) { @@ -1438,6 +1469,7 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, Gran // finally, after we optionally write delta and snapshot files, add new mutations to buffer if (!deltas.mutations.empty()) { + printf("!deltas.mutations.empty()\n"); if (deltas.mutations.size() == 1 && deltas.mutations.back().param1 == lastEpochEndPrivateKey) { // Note rollbackVerision is durable, [rollbackVersion+1 - deltas.version] needs to be tossed // For correctness right now, there can be no waits and yields either in rollback handling @@ -1557,8 +1589,8 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, Gran std::cout << "looping in blobGranuleUpdateFiles" << std::endl; } } catch (Error& e) { + printf("IN CATCH FOR %s blobGranuleUpdateFiles -----\n", bwData->id.toString().c_str()); printf("error is %s\n", e.name()); - printf("IN CATCH FOR blobGranuleUpdateFiles -----------------------------------\n "); if (e.code() == error_code_operation_cancelled) { throw; } @@ -1587,7 +1619,7 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, Gran f.future.cancel(); } - bwData->granuleUpdateErrors.send(rangeMetadata->originalReq); + bwData->granuleUpdateErrors.send(metadata->originalReq); } } throw e; @@ -1650,17 +1682,19 @@ static Future waitForVersion(Reference metadata, Version // if we don't have to wait for change feed version to catch up or wait for any pending file writes to complete, // nothing to do - /*printf(" [%s - %s) waiting for %lld\n readable:%s\n bufferedDelta=%lld\n pendingDelta=%lld\n " - "durableDelta=%lld\n pendingSnapshot=%lld\n durableSnapshot=%lld\n", - metadata->keyRange.begin.printable().c_str(), - metadata->keyRange.end.printable().c_str(), - v, - metadata->readable.isSet() ? "T" : "F", - metadata->bufferedDeltaVersion.get(), - metadata->pendingDeltaVersion, - metadata->durableDeltaVersion.get(), - metadata->pendingSnapshotVersion, - metadata->durableSnapshotVersion.get());*/ + if (v >= 162692991) { + printf(" [%s - %s) waiting for %lld\n readable:%s\n bufferedDelta=%lld\n pendingDelta=%lld\n " + "durableDelta=%lld\n pendingSnapshot=%lld\n durableSnapshot=%lld\n", + metadata->keyRange.begin.printable().c_str(), + metadata->keyRange.end.printable().c_str(), + v, + metadata->readable.isSet() ? "T" : "F", + metadata->bufferedDeltaVersion.get(), + metadata->pendingDeltaVersion, + metadata->durableDeltaVersion.get(), + metadata->pendingSnapshotVersion, + metadata->durableSnapshotVersion.get()); + } if (metadata->readable.isSet() && v <= metadata->bufferedDeltaVersion.get() && (v <= metadata->durableDeltaVersion.get() || @@ -1674,7 +1708,8 @@ static Future waitForVersion(Reference metadata, Version } ACTOR Future handleBlobGranuleFileRequest(Reference bwData, BlobGranuleFileRequest req) { - // printf("In handleBlobGranuleFileRequest\n"); + printf( + "In handleBlobGranuleFileRequest for BW %s @ version %lld\n", bwData->id.toString().c_str(), req.readVersion); try { // TODO REMOVE in api V2 ASSERT(req.beginVersion == 0); @@ -1718,6 +1753,10 @@ ACTOR Future handleBlobGranuleFileRequest(Reference bwData // do work for each range for (auto m : granules) { + if (req.readVersion >= 162692991) { + printf( + "For BW %s, granule: %s\n", bwData->id.toString().c_str(), m->keyRange.begin.printable().c_str()); + } state Reference metadata = m; // try to check version_too_old, cancelled, waitForVersion without yielding first if (metadata->readable.isSet() && @@ -1746,6 +1785,9 @@ ACTOR Future handleBlobGranuleFileRequest(Reference bwData } // rollback resets all of the version information, so we have to redo wait for version on rollback state int rollbackCount = metadata->rollbackCount.get(); + if (req.readVersion >= 162692991) { + printf("For BW %s, before choose\n", bwData->id.toString().c_str()); + } choose { when(wait(waitForVersionFuture)) {} when(wait(metadata->rollbackCount.onChange())) {} @@ -1754,6 +1796,9 @@ ACTOR Future handleBlobGranuleFileRequest(Reference bwData throw wrong_shard_server(); } } + if (req.readVersion >= 162692991) { + printf("For BW %s, after choose\n", bwData->id.toString().c_str()); + } if (rollbackCount == metadata->rollbackCount.get()) { break; @@ -1765,6 +1810,10 @@ ACTOR Future handleBlobGranuleFileRequest(Reference bwData } } + if (req.readVersion >= 162692991) { + printf("For BW %s, after loop\n", bwData->id.toString().c_str()); + } + // granule is up to date, do read BlobGranuleChunkRef chunk; @@ -2019,19 +2068,15 @@ ACTOR Future persistAssignWorkerRange(Reference start(Reference bwData, GranuleRangeMetadata* meta, AssignBlobRangeRequest req) { - try { - meta->originalReq = req; - meta->assignFuture = persistAssignWorkerRange(bwData, req); - meta->fileUpdaterFuture = blobGranuleUpdateFiles(bwData, meta); - bwData->actors.add(meta->fileUpdaterFuture); - wait(success(meta->assignFuture)); - return Void(); - } catch (Error& e) { - meta->assignFuture.cancel(); - meta->fileUpdaterFuture.cancel(); - throw; - } + ASSERT(meta->activeMetadata.isValid()); + meta->activeMetadata->originalReq = req; + meta->assignFuture = persistAssignWorkerRange(bwData, req); + meta->fileUpdaterFuture = blobGranuleUpdateFiles(bwData, meta->activeMetadata, meta->assignFuture); + // bwData->actors.add(meta->fileUpdaterFuture); + wait(success(meta->assignFuture)); + return Void(); } static GranuleRangeMetadata constructActiveBlobRange(Reference bwData, @@ -2107,6 +2152,12 @@ ACTOR Future changeBlobRange(Reference bwData, auto ranges = bwData->granuleMetadata.intersectingRanges(keyRange); bool alreadyAssigned = false; for (auto& r : ranges) { + if (!active) { + if (r.value().activeMetadata.isValid() && r.value().activeMetadata->cancelled.canBeSet()) { + printf("Cancelling activeMetadata\n"); + r.value().activeMetadata->cancelled.send(Void()); + } + } bool thisAssignmentNewer = newerRangeAssignment(r.value(), epoch, seqno); if (r.value().lastEpoch == epoch && r.value().lastSeqno == seqno) { ASSERT(r.begin() == keyRange.begin); @@ -2428,12 +2479,15 @@ ACTOR Future blobWorker(BlobWorkerInterface bwInterf, ++self->stats.activeReadRequests; self->actors.add(handleBlobGranuleFileRequest(self, req)); } - when(GranuleStatusStreamRequest req = waitNext(bwInterf.granuleStatusStreamRequest.getFuture())) { + when(state GranuleStatusStreamRequest req = waitNext(bwInterf.granuleStatusStreamRequest.getFuture())) { if (self->managerEpochOk(req.managerEpoch)) { if (BW_DEBUG) { printf("Worker %s got new granule status endpoint\n", self->id.toString().c_str()); } - self->currentManagerStatusStream = req.reply; + // req.reply is marked const unless you mark req as `state`?!?!? + // TODO: pick a reasonable byte limit instead of just piggy-backing + req.reply.setByteLimit(SERVER_KNOBS->RANGESTREAM_LIMIT_BYTES); + self->currentManagerStatusStream.set(req.reply); } } when(AssignBlobRangeRequest _req = waitNext(bwInterf.assignBlobRangeRequest.getFuture())) { diff --git a/fdbserver/ClusterController.actor.cpp b/fdbserver/ClusterController.actor.cpp index c44ac69deb..a41376938e 100644 --- a/fdbserver/ClusterController.actor.cpp +++ b/fdbserver/ClusterController.actor.cpp @@ -317,6 +317,11 @@ public: } } + /* printf("No blob workers found bc we excluded the following\n"); + for (auto addr : req.excludeAddresses) { + printf("- %s\n", addr.toString().c_str()); + } + */ throw no_more_servers(); } diff --git a/fdbserver/workloads/BlobGranuleVerifier.actor.cpp b/fdbserver/workloads/BlobGranuleVerifier.actor.cpp index d1b361ac51..a8c943fc19 100644 --- a/fdbserver/workloads/BlobGranuleVerifier.actor.cpp +++ b/fdbserver/workloads/BlobGranuleVerifier.actor.cpp @@ -29,7 +29,7 @@ #include "flow/actorcompiler.h" // This must be the last #include. -#define BGV_DEBUG false +#define BGV_DEBUG true /* * This workload is designed to verify the correctness of the blob data produced by the blob workers. @@ -209,6 +209,7 @@ struct BlobGranuleVerifierWorkload : TestWorkload { out.append(out.arena(), chunkRows.begin(), chunkRows.size()); } catch (Error& e) { if (e.code() == error_code_end_of_stream) { + printf("got end of stream\n"); break; } throw e; diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 4ca14d230e..91bc2fd423 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -251,7 +251,8 @@ if(WITH_PYTHON) add_fdb_test(TEST_FILES slow/ApiCorrectness.toml) add_fdb_test(TEST_FILES slow/ApiCorrectnessAtomicRestore.toml) add_fdb_test(TEST_FILES slow/ApiCorrectnessSwitchover.toml) - add_fdb_test(TEST_FILES fast/BlobGranuleCorrectness.toml) + add_fdb_test(TEST_FILES fast/BlobGranuleCorrectness.toml IGNORE) + add_fdb_test(TEST_FILES fast/BlobGranuleCorrectnessClean.toml) add_fdb_test(TEST_FILES slow/BlobGranuleCorrectnessLarge.toml) add_fdb_test(TEST_FILES slow/ClogWithRollbacks.toml) add_fdb_test(TEST_FILES slow/CloggedCycleTest.toml) diff --git a/tests/fast/BlobGranuleCorrectness.toml b/tests/fast/BlobGranuleCorrectness.toml index 59d6be0364..20446ec66c 100644 --- a/tests/fast/BlobGranuleCorrectness.toml +++ b/tests/fast/BlobGranuleCorrectness.toml @@ -8,3 +8,27 @@ testTitle = 'BlobGranuleCorrectnessTest' [[test.workload]] testName = 'BlobGranuleVerifier' testDuration = 120.0 + + [[test.workload]] + testName = 'RandomClogging' + testDuration = 120.0 + + [[test.workload]] + testName = 'Rollback' + meanDelay = 30.0 + testDuration = 120.0 + + [[test.workload]] + testName = 'Attrition' + machinesToKill = 10 + machinesToLeave = 3 + reboot = true + testDuration = 120.0 + + [[test.workload]] + testName = 'Attrition' + machinesToKill = 10 + machinesToLeave = 3 + reboot = true + testDuration = 120.0 + diff --git a/tests/fast/BlobGranuleCorrectnessClean.toml b/tests/fast/BlobGranuleCorrectnessClean.toml new file mode 100644 index 0000000000..59d6be0364 --- /dev/null +++ b/tests/fast/BlobGranuleCorrectnessClean.toml @@ -0,0 +1,10 @@ +[[test]] +testTitle = 'BlobGranuleCorrectnessTest' + + [[test.workload]] + testName = 'WriteDuringRead' + testDuration = 120.0 + + [[test.workload]] + testName = 'BlobGranuleVerifier' + testDuration = 120.0 From ef67feed67d5ceea776bddbb907ad2e4c841e854 Mon Sep 17 00:00:00 2001 From: Suraj Gupta Date: Tue, 12 Oct 2021 16:36:05 -0400 Subject: [PATCH 4/7] Clean up blob manager changes. --- fdbserver/BlobManager.actor.cpp | 84 ++++++------------- fdbserver/ClusterController.actor.cpp | 5 -- .../workloads/BlobGranuleVerifier.actor.cpp | 3 +- flow/error_definitions.h | 1 + 4 files changed, 26 insertions(+), 67 deletions(-) diff --git a/fdbserver/BlobManager.actor.cpp b/fdbserver/BlobManager.actor.cpp index 208674970a..ec720989b7 100644 --- a/fdbserver/BlobManager.actor.cpp +++ b/fdbserver/BlobManager.actor.cpp @@ -283,10 +283,6 @@ static UID pickWorkerForAssign(BlobManagerData* bmData) { } // pick a random worker out of the eligible workers - if (eligibleWorkers.size() == 0) { - printf("%d eligible workers\n", bmData->workerStats.size()); - } - ASSERT(eligibleWorkers.size() > 0); int idx = deterministicRandom()->randomInt(0, eligibleWorkers.size()); if (BM_DEBUG) { @@ -325,7 +321,7 @@ ACTOR Future doRangeAssignment(BlobManagerData* bmData, RangeAssignment as // if that worker isn't alive anymore, add the range back into the stream if (bmData->workersById.count(workerID) == 0) { - throw granule_assignment_conflict(); // TODO: find a better error to throw + throw worker_for_granule_not_found(); } AssignBlobRangeReply _rep = wait(bmData->workersById[workerID].assignBlobRangeRequest.getReply(req)); rep = _rep; @@ -419,9 +415,7 @@ ACTOR Future rangeAssigner(BlobManagerData* bmData) { // Ensure range isn't currently assigned anywhere, and there is only 1 intersecting range auto currentAssignments = bmData->workerAssignments.intersectingRanges(assignment.keyRange); int count = 0; - printf("intersecting ranges in currentAssignments:\n"); for (auto& it : currentAssignments) { - printf("[%s - %s]\n", it.begin().printable().c_str(), it.end().printable().c_str()); if (assignment.assign.get().continueAssignment) { ASSERT(assignment.worker.present()); ASSERT(it.value() == assignment.worker.get()); @@ -434,13 +428,10 @@ ACTOR Future rangeAssigner(BlobManagerData* bmData) { workerId = assignment.worker.present() ? assignment.worker.get() : pickWorkerForAssign(bmData); bmData->workerAssignments.insert(assignment.keyRange, workerId); - if (bmData->workerStats.count(workerId) && !assignment.assign.get().continueAssignment) { - bmData->workerStats[workerId].numGranulesAssigned += 1; - } - printf("current ranges after inserting assign: \n"); - for (auto it : bmData->workerAssignments.ranges()) { - printf("[%s - %s]\n", it.begin().printable().c_str(), it.end().printable().c_str()); + ASSERT(bmData->workerStats.count(workerId)); + if (!assignment.assign.get().continueAssignment) { + bmData->workerStats[workerId].numGranulesAssigned += 1; } // FIXME: if range is assign, have some sort of semaphore for outstanding assignments so we don't assign @@ -466,10 +457,6 @@ ACTOR Future rangeAssigner(BlobManagerData* bmData) { } bmData->workerAssignments.insert(assignment.keyRange, UID()); - printf("current ranges after inserting revoke: \n"); - for (auto it : bmData->workerAssignments.ranges()) { - printf("[%s - %s]\n", it.begin().printable().c_str(), it.end().printable().c_str()); - } } } } @@ -733,53 +720,27 @@ ACTOR Future maybeSplitRange(BlobManagerData* bmData, UID currentWorkerId, return Void(); } -void reassignRanges(BlobManagerData* bmData, UID bwId) { - printf("taking back ranges for worker %s\n", bwId.toString().c_str()); - // for every range owned by this blob worker, we want to - // - send a revoke request for that range to the blob worker - // - add the range back to the stream of ranges to be assigned - for (auto& it : bmData->workerAssignments.ranges()) { - if (it.cvalue() == bwId) { - // Send revoke request to worker - RangeAssignment raRevoke; - raRevoke.isAssign = false; - raRevoke.worker = bwId; - raRevoke.keyRange = it.range(); - raRevoke.revoke = RangeRevokeData(false); - bmData->rangesToAssign.send(raRevoke); - - // Add range back into the stream of ranges to be assigned - RangeAssignment raAssign; - raAssign.isAssign = true; - raAssign.worker = Optional(); - raAssign.keyRange = it.range(); - raAssign.assign = RangeAssignmentData(false); // not a continue - bmData->rangesToAssign.send(raAssign); - } - } -} - void killBlobWorker(BlobManagerData* bmData, BlobWorkerInterface bwInterf) { UID bwId = bwInterf.id(); // Remove blob worker from stats map so that when we try to find a worker to takeover the range, // the one we just killed isn't considered. - // Remove it from workersById also since otherwise that addr will remain excluded + // Remove it from workersById also since otherwise that worker addr will remain excluded // when we try to recruit new blob workers. - printf("removing bw %s from BM workerStats\n", bwId.toString().c_str()); bmData->workerStats.erase(bwId); bmData->workersById.erase(bwId); // for every range owned by this blob worker, we want to - // - send a revoke request for that range to the blob worker + // - send a revoke request for that range // - add the range back to the stream of ranges to be assigned - printf("taking back ranges from bw %s\n", bwId.toString().c_str()); + if (BM_DEBUG) { + printf("Taking back ranges from BW %s\n", bwId.toString().c_str()); + } for (auto& it : bmData->workerAssignments.ranges()) { if (it.cvalue() == bwId) { - // Send revoke request to worker + // Send revoke request RangeAssignment raRevoke; raRevoke.isAssign = false; - raRevoke.worker = bwId; raRevoke.keyRange = it.range(); raRevoke.revoke = RangeRevokeData(false); bmData->rangesToAssign.send(raRevoke); @@ -795,7 +756,9 @@ void killBlobWorker(BlobManagerData* bmData, BlobWorkerInterface bwInterf) { } // Send halt to blob worker, with no expectation of hearing back - printf("sending halt to bw %s\n", bwId.toString().c_str()); + if (BM_DEBUG) { + printf("Sending halt to BW %s\n", bwId.toString().c_str()); + } bmData->addActor.send( brokenPromiseToNever(bwInterf.haltBlobWorker.getReply(HaltBlobWorkerRequest(bmData->epoch, bmData->id)))); } @@ -835,6 +798,7 @@ ACTOR Future monitorBlobWorkerStatus(BlobManagerData* bmData, BlobWorkerIn // to split the range. ASSERT(rep.doSplit); + // only evaluate for split if this worker currently owns the granule in this blob manager's mapping auto currGranuleAssignment = bmData->workerAssignments.rangeContaining(rep.granuleRange.begin); if (!(currGranuleAssignment.begin() == rep.granuleRange.begin && currGranuleAssignment.end() == rep.granuleRange.end && @@ -901,7 +865,6 @@ ACTOR Future monitorBlobWorker(BlobManagerData* bmData, BlobWorkerInterfac printf("BM %lld detected BW %s is dead\n", bmData->epoch, bwInterf.id().toString().c_str()); } TraceEvent("BlobWorkerFailed", bmData->id).detail("BlobWorkerID", bwInterf.id()); - killBlobWorker(bmData, bwInterf); } when(wait(monitorStatus)) { ASSERT(false); @@ -921,11 +884,18 @@ ACTOR Future monitorBlobWorker(BlobManagerData* bmData, BlobWorkerInterfac throw e; } + // kill the blob worker + killBlobWorker(bmData, bwInterf); + // Trigger recruitment for a new blob worker - printf("restarting recruitment in monitorblobworker\n"); + if (BM_DEBUG) { + printf("Restarting recruitment to replace dead BW %s\n", bwInterf.id().toString().c_str()); + } bmData->restartRecruiting.trigger(); - printf("about to stop monitoring %s\n", bwInterf.id().toString().c_str()); + if (BM_DEBUG) { + printf("No longer monitoring BW %s\n", bwInterf.id().toString().c_str()); + } return Void(); } @@ -1090,12 +1060,6 @@ ACTOR Future blobWorkerRecruiter( } TraceEvent("BMRecruiting").detail("State", "Sending request to CC"); - /* - printf("EXCLUDING THE FOLLOWING IN REQ:\n"); - for (auto addr : recruitReq.excludeAddresses) { - printf("- %s\n", addr.toString().c_str()); - } - */ if (!fCandidateWorker.isValid() || fCandidateWorker.isReady() || recruitReq.excludeAddresses != lastRequest.excludeAddresses) { @@ -1115,7 +1079,7 @@ ACTOR Future blobWorkerRecruiter( when(wait(recruitBlobWorker->onChange())) { fCandidateWorker = Future(); } // signal used to restart the loop and try to recruit the next blob worker - when(wait(self->restartRecruiting.onTrigger())) { printf("RESTARTED RECRUITING. BACK TO TOP\n"); } + when(wait(self->restartRecruiting.onTrigger())) {} } wait(delay(FLOW_KNOBS->PREVENT_FAST_SPIN_DELAY, TaskPriority::BlobManager)); } catch (Error& e) { diff --git a/fdbserver/ClusterController.actor.cpp b/fdbserver/ClusterController.actor.cpp index a41376938e..c44ac69deb 100644 --- a/fdbserver/ClusterController.actor.cpp +++ b/fdbserver/ClusterController.actor.cpp @@ -317,11 +317,6 @@ public: } } - /* printf("No blob workers found bc we excluded the following\n"); - for (auto addr : req.excludeAddresses) { - printf("- %s\n", addr.toString().c_str()); - } - */ throw no_more_servers(); } diff --git a/fdbserver/workloads/BlobGranuleVerifier.actor.cpp b/fdbserver/workloads/BlobGranuleVerifier.actor.cpp index a8c943fc19..d1b361ac51 100644 --- a/fdbserver/workloads/BlobGranuleVerifier.actor.cpp +++ b/fdbserver/workloads/BlobGranuleVerifier.actor.cpp @@ -29,7 +29,7 @@ #include "flow/actorcompiler.h" // This must be the last #include. -#define BGV_DEBUG true +#define BGV_DEBUG false /* * This workload is designed to verify the correctness of the blob data produced by the blob workers. @@ -209,7 +209,6 @@ struct BlobGranuleVerifierWorkload : TestWorkload { out.append(out.arena(), chunkRows.begin(), chunkRows.size()); } catch (Error& e) { if (e.code() == error_code_end_of_stream) { - printf("got end of stream\n"); break; } throw e; diff --git a/flow/error_definitions.h b/flow/error_definitions.h index d1a48f5499..fc11b70ec0 100755 --- a/flow/error_definitions.h +++ b/flow/error_definitions.h @@ -80,6 +80,7 @@ ERROR( local_config_changed, 1056, "Local configuration file has changed. Restar ERROR( failed_to_reach_quorum, 1057, "Failed to reach quorum from configuration database nodes. Retry sending these requests" ) ERROR( unknown_change_feed, 1058, "Change feed not found" ) ERROR( granule_assignment_conflict, 1059, "Conflicting attempts to assign blob granules" ) +ERROR( worker_for_granule_not_found, 1060, "The chosen worker to assign the granule to was not found" ) ERROR( broken_promise, 1100, "Broken promise" ) ERROR( operation_cancelled, 1101, "Asynchronous operation cancelled" ) From d002df3b21353c4cae3ec59bfa0d4c094c79ade1 Mon Sep 17 00:00:00 2001 From: Suraj Gupta Date: Tue, 12 Oct 2021 19:54:22 -0400 Subject: [PATCH 5/7] Clean up blob worker changes. --- fdbserver/BlobWorker.actor.cpp | 452 ++++++++++++--------------------- 1 file changed, 161 insertions(+), 291 deletions(-) diff --git a/fdbserver/BlobWorker.actor.cpp b/fdbserver/BlobWorker.actor.cpp index 109cfed2e4..aa06bc2da9 100644 --- a/fdbserver/BlobWorker.actor.cpp +++ b/fdbserver/BlobWorker.actor.cpp @@ -40,7 +40,7 @@ #include "flow/flow.h" #define BW_DEBUG true -#define BW_REQUEST_DEBUG true +#define BW_REQUEST_DEBUG false // TODO add comments + documentation struct BlobFileIndex { @@ -109,13 +109,10 @@ struct GranuleMetadata : NonCopyable, ReferenceCounted { ASSERT(resumeSnapshot.canBeSet()); resumeSnapshot.send(Void()); } - - ~GranuleMetadata() { printf("in dtor for GranuleMetadata\n"); } }; // TODO: rename this struct struct GranuleRangeMetadata { - int id = 0; int64_t lastEpoch; int64_t lastSeqno; Reference activeMetadata; @@ -125,42 +122,16 @@ struct GranuleRangeMetadata { GranuleRangeMetadata() : lastEpoch(0), lastSeqno(0) {} GranuleRangeMetadata(int64_t epoch, int64_t seqno, Reference activeMetadata) - : lastEpoch(epoch), lastSeqno(seqno), activeMetadata(activeMetadata) { - /* - if (activeMetadata.isValid()) { - activeMetadata->cancelled.reset(); - } - */ - } - - ~GranuleRangeMetadata() { - printf("GranuleRangeMetadata is being destroyed\n"); - /* - if (activeMetadata.isValid() && activeMetadata->cancelled.canBeSet()) { - printf("Cancelling activeMetadata\n"); - activeMetadata->cancelled.send(Void()); - } - assignFuture.cancel(); - fileUpdaterFuture.cancel(); - */ - /* - if (id == 42) { - printf("GranuleRangeMetadata with id %d is being destroyed\n", id); - sleep(10); - } - */ - } + : lastEpoch(epoch), lastSeqno(seqno), activeMetadata(activeMetadata) {} }; struct BlobWorkerData : NonCopyable, ReferenceCounted { UID id; Database db; - AsyncVar dead; BlobWorkerStats stats; PromiseStream> addActor; - ActorCollection actors{ false }; LocalityData locality; int64_t currentManagerEpoch = -1; @@ -178,8 +149,7 @@ struct BlobWorkerData : NonCopyable, ReferenceCounted { PromiseStream granuleUpdateErrors; - BlobWorkerData(UID id, Database db) - : id(id), db(db), stats(id, SERVER_KNOBS->WORKER_LOGGING_INTERVAL), actors(false) {} + 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()); } bool managerEpochOk(int64_t epoch) { @@ -229,7 +199,6 @@ static void checkGranuleLock(int64_t epoch, int64_t seqno, int64_t ownerEpoch, i ASSERT(epoch < ownerEpoch || (epoch == ownerEpoch && seqno <= ownerSeqno)); // returns true if we still own the lock, false if someone else does - printf("epoch: %lld, seqno: %lld, ownerEpoch: %lld, ownerSeqno: %lld\n", epoch, seqno, ownerEpoch, ownerSeqno); if (epoch != ownerEpoch || seqno != ownerSeqno) { if (BW_DEBUG) { printf("Lock assignment check failed. Expected (%lld, %lld), got (%lld, %lld)\n", @@ -578,7 +547,6 @@ ACTOR Future writeDeltaFile(Reference bwData, return BlobFileIndex(currentDeltaVersion, fname, 0, serialized.size()); } catch (Error& e) { numIterations++; - printf("writeDeltaFile error: %s\n", e.name()); wait(tr->onError(e)); } } @@ -586,7 +554,7 @@ ACTOR Future writeDeltaFile(Reference bwData, if (e.code() == error_code_operation_cancelled) { throw e; } - // TODO: do this for writeSnapshot + if (numIterations != 1 || e.code() != error_code_granule_assignment_conflict) { throw e; } @@ -912,15 +880,12 @@ ACTOR Future handleCompletedDeltaFile(Reference bwData, // have completed // FIXME: also have these be async, have each pop change feed wait on the prior one, wait on them before // re-snapshotting - printf("in handleCompletedDeltaFile for BW %s\n", bwData->id.toString().c_str()); Future popFuture = bwData->db->popChangeFeedMutations(cfKey, completedDeltaFile.version); wait(popFuture); - printf("popChangeFeedMutations returned\n"); } while (!rollbacksInProgress.empty() && completedDeltaFile.version >= rollbacksInProgress.front().first) { rollbacksInProgress.pop_front(); } - printf("removed rollbacks\n"); return Void(); } @@ -1071,16 +1036,12 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, state bool snapshotEligible; // just wrote a delta file or just took granule over from another worker state bool justDidRollback = false; - printf( - "metadata %s %s\n", metadata->keyRange.begin.printable().c_str(), metadata->keyRange.end.printable().c_str()); try { // set resume snapshot so it's not valid until we pause to ask the blob manager for a re-snapshot metadata->resumeSnapshot.send(Void()); // before starting, make sure worker persists range assignment and acquires the granule lock - printf("before wait on assignFuture\n"); GranuleChangeFeedInfo _info = wait(assignFuture); - printf("after wait on assignFuture\n"); changeFeedInfo = _info; wait(delay(0, TaskPriority::BlobWorkerUpdateStorage)); @@ -1175,14 +1136,11 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, changeFeedInfo.granuleSplitFrom.get() /*metadata->keyRange*/); } else { readOldChangeFeed = false; - printf("before getChangeFeedStream, my ID is %s\n", bwData->id.toString().c_str()); changeFeedFuture = bwData->db->getChangeFeedStream( changeFeedStream, cfKey, startVersion + 1, MAX_VERSION, metadata->keyRange); } loop { - printf("bw %s\n", bwData->id.toString().c_str()); - // check outstanding snapshot/delta files for completion if (inFlightBlobSnapshot.isValid() && inFlightBlobSnapshot.isReady()) { BlobFileIndex completedSnapshot = wait(inFlightBlobSnapshot); @@ -1198,8 +1156,6 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, wait(yield(TaskPriority::BlobWorkerUpdateStorage)); } if (!inFlightBlobSnapshot.isValid()) { - printf("!inFlightBlobSnapshot.isValid\n"); - printf("inFlightDeltaFiles.size() = %d\n", inFlightDeltaFiles.size()); while (inFlightDeltaFiles.size() > 0) { if (inFlightDeltaFiles.front().future.isReady()) { BlobFileIndex completedDeltaFile = wait(inFlightDeltaFiles.front().future); @@ -1227,7 +1183,6 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, state Standalone> mutations; if (readOldChangeFeed) { - printf("readOldChangeFeed\n"); Standalone> oldMutations = waitNext(oldChangeFeedStream.getFuture()); // TODO filter old mutations won't be necessary, SS does it already if (filterOldMutations( @@ -1254,12 +1209,8 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, } // process mutations - printf("mutations.size() == %d\n", mutations.size()); for (MutationsAndVersionRef d : mutations) { state MutationsAndVersionRef deltas = d; - if (deltas.version >= 158685394) { - printf("deltas.version=%lld\n", deltas.version); - } ASSERT(deltas.version >= metadata->bufferedDeltaVersion.get()); // Write a new delta file IF we have enough bytes, and we have all of the previous version's stuff // there to ensure no versions span multiple delta files. Check this by ensuring the version of this @@ -1331,8 +1282,6 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, // bunch of extra delta files at some point, even if we don't consider it for a split yet if (snapshotEligible && metadata->bytesInNewDeltaFiles >= SERVER_KNOBS->BG_DELTA_BYTES_BEFORE_COMPACT && !readOldChangeFeed) { - printf("snapshotEligible && metadata->bytesInNewDeltaFiles >= " - "SERVER_KNOBS->BG_DELTA_BYTES_BEFORE_COMPACT && !readOldChangeFeed\n"); if (BW_DEBUG && (inFlightBlobSnapshot.isValid() || !inFlightDeltaFiles.empty())) { printf("Granule [%s - %s) ready to re-snapshot, waiting for outstanding %d snapshot and %d " "deltas to " @@ -1399,19 +1348,12 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, when(wait(bwData->currentManagerStatusStream.onChange())) {} } - /* - if (bwData->dead.get()) { - std::cout << "bw detected dead in blobGranuleUpdateFiles" << std::endl; - throw actor_cancelled(); - } - */ if (BW_DEBUG) { printf("Granule [%s - %s)\n, hasn't heard back from BM in BW %s, re-sending status\n", metadata->keyRange.begin.printable().c_str(), metadata->keyRange.end.printable().c_str(), bwData->id.toString().c_str()); } - // wait(yield()); } if (BW_DEBUG) { @@ -1437,8 +1379,6 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, metadata->bytesInNewDeltaFiles = 0; } else if (snapshotEligible && metadata->bytesInNewDeltaFiles >= SERVER_KNOBS->BG_DELTA_BYTES_BEFORE_COMPACT) { - printf("snapshotEligible && metadata->bytesInNewDeltaFiles >= " - "SERVER_KNOBS->BG_DELTA_BYTES_BEFORE_COMPACT\n"); // if we're in the old change feed case and can't snapshot but we have enough data to, don't // queue too many delta files in parallel while (inFlightDeltaFiles.size() > 10) { @@ -1469,7 +1409,6 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, // finally, after we optionally write delta and snapshot files, add new mutations to buffer if (!deltas.mutations.empty()) { - printf("!deltas.mutations.empty()\n"); if (deltas.mutations.size() == 1 && deltas.mutations.back().param1 == lastEpochEndPrivateKey) { // Note rollbackVerision is durable, [rollbackVersion+1 - deltas.version] needs to be tossed // For correctness right now, there can be no waits and yields either in rollback handling @@ -1586,11 +1525,8 @@ ACTOR Future blobGranuleUpdateFiles(Reference bwData, } } justDidRollback = false; - std::cout << "looping in blobGranuleUpdateFiles" << std::endl; } } catch (Error& e) { - printf("IN CATCH FOR %s blobGranuleUpdateFiles -----\n", bwData->id.toString().c_str()); - printf("error is %s\n", e.name()); if (e.code() == error_code_operation_cancelled) { throw; } @@ -1682,19 +1618,19 @@ static Future waitForVersion(Reference metadata, Version // if we don't have to wait for change feed version to catch up or wait for any pending file writes to complete, // nothing to do - if (v >= 162692991) { - printf(" [%s - %s) waiting for %lld\n readable:%s\n bufferedDelta=%lld\n pendingDelta=%lld\n " - "durableDelta=%lld\n pendingSnapshot=%lld\n durableSnapshot=%lld\n", - metadata->keyRange.begin.printable().c_str(), - metadata->keyRange.end.printable().c_str(), - v, - metadata->readable.isSet() ? "T" : "F", - metadata->bufferedDeltaVersion.get(), - metadata->pendingDeltaVersion, - metadata->durableDeltaVersion.get(), - metadata->pendingSnapshotVersion, - metadata->durableSnapshotVersion.get()); - } + /* + printf(" [%s - %s) waiting for %lld\n readable:%s\n bufferedDelta=%lld\n pendingDelta=%lld\n " + "durableDelta=%lld\n pendingSnapshot=%lld\n durableSnapshot=%lld\n", + metadata->keyRange.begin.printable().c_str(), + metadata->keyRange.end.printable().c_str(), + v, + metadata->readable.isSet() ? "T" : "F", + metadata->bufferedDeltaVersion.get(), + metadata->pendingDeltaVersion, + metadata->durableDeltaVersion.get(), + metadata->pendingSnapshotVersion, + metadata->durableSnapshotVersion.get()); + */ if (metadata->readable.isSet() && v <= metadata->bufferedDeltaVersion.get() && (v <= metadata->durableDeltaVersion.get() || @@ -1708,8 +1644,6 @@ static Future waitForVersion(Reference metadata, Version } ACTOR Future handleBlobGranuleFileRequest(Reference bwData, BlobGranuleFileRequest req) { - printf( - "In handleBlobGranuleFileRequest for BW %s @ version %lld\n", bwData->id.toString().c_str(), req.readVersion); try { // TODO REMOVE in api V2 ASSERT(req.beginVersion == 0); @@ -1732,7 +1666,6 @@ ACTOR Future handleBlobGranuleFileRequest(Reference bwData req.keyRange.end.printable().c_str()); } - printf("lastRangeEnd < r.begin() || !isValid\n"); throw wrong_shard_server(); } granules.push_back(r.value().activeMetadata); @@ -1747,16 +1680,11 @@ ACTOR Future handleBlobGranuleFileRequest(Reference bwData req.keyRange.end.printable().c_str()); } - printf("lastRangeEnd < req.keyRange.end\n"); throw wrong_shard_server(); } // do work for each range for (auto m : granules) { - if (req.readVersion >= 162692991) { - printf( - "For BW %s, granule: %s\n", bwData->id.toString().c_str(), m->keyRange.begin.printable().c_str()); - } state Reference metadata = m; // try to check version_too_old, cancelled, waitForVersion without yielding first if (metadata->readable.isSet() && @@ -1773,7 +1701,6 @@ ACTOR Future handleBlobGranuleFileRequest(Reference bwData throw transaction_too_old(); } if (metadata->cancelled.isSet()) { - printf("metadata->cancelled.isSet()\n"); throw wrong_shard_server(); } @@ -1785,19 +1712,10 @@ ACTOR Future handleBlobGranuleFileRequest(Reference bwData } // rollback resets all of the version information, so we have to redo wait for version on rollback state int rollbackCount = metadata->rollbackCount.get(); - if (req.readVersion >= 162692991) { - printf("For BW %s, before choose\n", bwData->id.toString().c_str()); - } choose { when(wait(waitForVersionFuture)) {} when(wait(metadata->rollbackCount.onChange())) {} - when(wait(metadata->cancelled.getFuture())) { - printf("metadata->cancelled.getFuture()\n"); - throw wrong_shard_server(); - } - } - if (req.readVersion >= 162692991) { - printf("For BW %s, after choose\n", bwData->id.toString().c_str()); + when(wait(metadata->cancelled.getFuture())) { throw wrong_shard_server(); } } if (rollbackCount == metadata->rollbackCount.get()) { @@ -1810,10 +1728,6 @@ ACTOR Future handleBlobGranuleFileRequest(Reference bwData } } - if (req.readVersion >= 162692991) { - printf("For BW %s, after loop\n", bwData->id.toString().c_str()); - } - // granule is up to date, do read BlobGranuleChunkRef chunk; @@ -1911,7 +1825,6 @@ ACTOR Future handleBlobGranuleFileRequest(Reference bwData ACTOR Future persistAssignWorkerRange(Reference bwData, AssignBlobRangeRequest req) { - printf("in persistAssignWorkerRange\n"); ASSERT(!req.continueAssignment); state Transaction tr(bwData->db); state Key lockKey = granuleLockKey(req.keyRange); @@ -1923,158 +1836,149 @@ ACTOR Future persistAssignWorkerRange(Reference prevLockValue = wait(tr.get(lockKey)); - 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); + // 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)); + 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); - GranuleFiles granuleFiles = wait(loadPreviousFiles(&tr, req.keyRange)); - info.existingFiles = granuleFiles; - info.doSnapshot = false; + GranuleFiles granuleFiles = wait(loadPreviousFiles(&tr, req.keyRange)); + info.existingFiles = granuleFiles; + info.doSnapshot = false; - if (info.existingFiles.get().snapshotFiles.empty()) { - ASSERT(info.existingFiles.get().deltaFiles.empty()); - info.previousDurableVersion = invalidVersion; - info.doSnapshot = true; - } else if (info.existingFiles.get().deltaFiles.empty()) { - info.previousDurableVersion = info.existingFiles.get().snapshotFiles.back().version; - } else { - info.previousDurableVersion = info.existingFiles.get().deltaFiles.back().version; - } - - // 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; + if (info.existingFiles.get().snapshotFiles.empty()) { + ASSERT(info.existingFiles.get().deltaFiles.empty()); info.previousDurableVersion = invalidVersion; - } - - 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.begin).append(req.keyRange.end); - 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 (parentGranulesValue.present()) { - state Standalone> 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(parentGranules.size() == 1); - - 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); - } - - ASSERT(!hasPrevOwner || granuleSplitState.first > BlobGranuleSplitState::Started); - - // 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; - } else if (granuleSplitState.first == BlobGranuleSplitState::Started) { - wait(updateGranuleSplitState(&tr, - parentGranules[0], - req.keyRange, - info.prevChangeFeedId.get(), - BlobGranuleSplitState::Assigned)); - // change feed was created as part of this transaction, changeFeedStartVersion will be - // set later - } else { - ASSERT(false); - } - } - - if (info.doSnapshot) { - // only need to do snapshot if no files exist yet for this granule. - ASSERT(info.previousDurableVersion == invalidVersion); - // 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; - } - } - - wait(tr.commit()); - - if (info.changeFeedStartVersion == invalidVersion) { - info.changeFeedStartVersion = tr.getCommittedVersion(); + info.doSnapshot = true; + } else if (info.existingFiles.get().deltaFiles.empty()) { + info.previousDurableVersion = info.existingFiles.get().snapshotFiles.back().version; } else { - ASSERT(info.changeFeedStartVersion != invalidVersion); + info.previousDurableVersion = info.existingFiles.get().deltaFiles.back().version; } - TraceEvent("BlobWorkerPersistedAssignment", bwData->id).detail("Granule", req.keyRange); - - return info; - } catch (Error& e) { - if (e.code() == error_code_granule_assignment_conflict) { - throw e; - } - wait(tr.onError(e)); + // 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; } + + 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.begin).append(req.keyRange.end); + 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 (parentGranulesValue.present()) { + state Standalone> 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(parentGranules.size() == 1); + + 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); + } + + ASSERT(!hasPrevOwner || granuleSplitState.first > BlobGranuleSplitState::Started); + + // 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; + } else if (granuleSplitState.first == BlobGranuleSplitState::Started) { + wait(updateGranuleSplitState(&tr, + parentGranules[0], + req.keyRange, + info.prevChangeFeedId.get(), + BlobGranuleSplitState::Assigned)); + // change feed was created as part of this transaction, changeFeedStartVersion will be + // set later + } else { + ASSERT(false); + } + } + + if (info.doSnapshot) { + // only need to do snapshot if no files exist yet for this granule. + ASSERT(info.previousDurableVersion == invalidVersion); + // 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; + } + } + + wait(tr.commit()); + + if (info.changeFeedStartVersion == invalidVersion) { + 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) { + throw e; + } + wait(tr.onError(e)); } - } catch (Error& e) { - printf("ERROR IN PERSIST: %s\n", e.name()); - throw; } } -// try to use GranuleMetadata ACTOR Future start(Reference bwData, GranuleRangeMetadata* meta, AssignBlobRangeRequest req) { ASSERT(meta->activeMetadata.isValid()); meta->activeMetadata->originalReq = req; meta->assignFuture = persistAssignWorkerRange(bwData, req); meta->fileUpdaterFuture = blobGranuleUpdateFiles(bwData, meta->activeMetadata, meta->assignFuture); - // bwData->actors.add(meta->fileUpdaterFuture); wait(success(meta->assignFuture)); return Void(); } @@ -2128,7 +2032,6 @@ ACTOR Future changeBlobRange(Reference bwData, bool active, bool disposeOnCleanup, bool selfReassign) { - printf("changeBlobRange called\n"); if (BW_DEBUG) { printf("%s range for [%s - %s): %s @ (%lld, %lld)\n", selfReassign ? "Re-assigning" : "Changing", @@ -2154,7 +2057,9 @@ ACTOR Future changeBlobRange(Reference bwData, for (auto& r : ranges) { if (!active) { if (r.value().activeMetadata.isValid() && r.value().activeMetadata->cancelled.canBeSet()) { - printf("Cancelling activeMetadata\n"); + if (BW_DEBUG) { + printf("Cancelling activeMetadata\n"); + } r.value().activeMetadata->cancelled.send(Void()); } } @@ -2201,7 +2106,7 @@ ACTOR Future changeBlobRange(Reference bwData, GranuleRangeMetadata newMetadata = (active && newerRanges.empty()) ? constructActiveBlobRange(bwData, keyRange, epoch, seqno) : constructInactiveBlobRange(epoch, seqno); - newMetadata.id = 42; + bwData->granuleMetadata.insert(keyRange, newMetadata); if (BW_DEBUG) { printf("Inserting new range [%s - %s): %s @ (%lld, %lld)\n", @@ -2304,38 +2209,15 @@ ACTOR Future handleRangeAssign(Reference bwData, if (shouldStart) { auto m = bwData->granuleMetadata.rangeContaining(req.keyRange.begin); ASSERT(m.begin() == req.keyRange.begin && m.end() == req.keyRange.end); - printf("About to start for BW %s\n", bwData->id.toString().c_str()); wait(start(bwData, &m.value(), req)); - /* - int count = 0; - // GranuleRangeMetadata& x; - for (auto& it : m) { - printf("BW %s ABOUT TO WAIT IN HANDLERANGEASSIGN\n", bwData->id.toString().c_str()); - wait(start(bwData, &it.value(), req)); - printf("done waiting in handleRangeAssign\n"); - count++; - } - ASSERT(count == 1); - // x.id = 42; - - printf("BW %s ABOUT TO WAIT IN HANDLERANGEASSIGN\n", bwData->id.toString().c_str()); - // WAITING ON START BUT ITS NOT AN ACTOR!!!!!!! SO WHEN handlerangeassign gets operation_cancelled, - // it won't get propogated to start - // wait(start(bwData, x, req)); - printf("done waiting in handleRangeAssign\n"); - */ } } if (!isSelfReassign) { ASSERT(!req.reply.isSet()); - printf("about to send reply\n"); req.reply.send(AssignBlobRangeReply(true)); - printf("done sending reply\n"); } return Void(); } catch (Error& e) { - printf("BW %s GOT ERROR %s IN HANDLERANGEASSIGN\n", bwData->id.toString().c_str(), e.name()); - state Error eState = e; if (BW_DEBUG) { printf("AssignRange [%s - %s) got error %s\n", req.keyRange.begin.printable().c_str(), @@ -2343,25 +2225,19 @@ ACTOR Future handleRangeAssign(Reference bwData, e.name()); } - // - // if (futureAndNewGranule.get().second.isValid()) { - // wait(futureAndNewGranule.get().second->cancel(false)); - //} - // - if (!isSelfReassign) { - if (canReplyWith(eState)) { - req.reply.sendError(eState); + if (canReplyWith(e)) { + req.reply.sendError(e); } } - throw eState; + throw; } } ACTOR Future handleRangeRevoke(Reference bwData, RevokeBlobRangeRequest req) { try { - bool _ = + bool _shouldStart = wait(changeBlobRange(bwData, req.keyRange, req.managerEpoch, req.managerSeqno, false, req.dispose, false)); req.reply.send(AssignBlobRangeReply(true)); return Void(); @@ -2466,8 +2342,8 @@ ACTOR Future blobWorker(BlobWorkerInterface bwInterf, rep.interf = bwInterf; recruitReply.send(rep); - self->actors.add(waitFailureServer(bwInterf.waitFailure.getFuture())); - self->actors.add(runCommitVersionChecks(self)); + self->addActor.send(waitFailureServer(bwInterf.waitFailure.getFuture())); + self->addActor.send(runCommitVersionChecks(self)); try { loop choose { @@ -2477,7 +2353,7 @@ ACTOR Future blobWorker(BlobWorkerInterface bwInterf, req.keyRange.end.printable().c_str());*/ ++self->stats.readRequests; ++self->stats.activeReadRequests; - self->actors.add(handleBlobGranuleFileRequest(self, req)); + self->addActor.send(handleBlobGranuleFileRequest(self, req)); } when(state GranuleStatusStreamRequest req = waitNext(bwInterf.granuleStatusStreamRequest.getFuture())) { if (self->managerEpochOk(req.managerEpoch)) { @@ -2505,7 +2381,7 @@ ACTOR Future blobWorker(BlobWorkerInterface bwInterf, } if (self->managerEpochOk(assignReq.managerEpoch)) { - self->actors.add(handleRangeAssign(self, assignReq, false)); + self->addActor.send(handleRangeAssign(self, assignReq, false)); } else { assignReq.reply.send(AssignBlobRangeReply(false)); } @@ -2524,13 +2400,13 @@ ACTOR Future blobWorker(BlobWorkerInterface bwInterf, } if (self->managerEpochOk(revokeReq.managerEpoch)) { - self->actors.add(handleRangeRevoke(self, revokeReq)); + self->addActor.send(handleRangeRevoke(self, revokeReq)); } else { revokeReq.reply.send(AssignBlobRangeReply(false)); } } when(AssignBlobRangeRequest granuleToReassign = waitNext(self->granuleUpdateErrors.getFuture())) { - self->actors.add(handleRangeAssign(self, granuleToReassign, true)); + self->addActor.send(handleRangeAssign(self, granuleToReassign, true)); } when(HaltBlobWorkerRequest req = waitNext(bwInterf.haltBlobWorker.getFuture())) { req.reply.send(Void()); @@ -2540,25 +2416,19 @@ ACTOR Future blobWorker(BlobWorkerInterface bwInterf, break; } } - // when(wait(delay(10))) { throw granule_assignment_conflict(); } when(wait(collection)) { - if (BW_DEBUG) { - printf("BW actor collection returned, exiting\n"); - } + TraceEvent("BlobWorkerActorCollectionError"); ASSERT(false); - throw granule_assignment_conflict(); + throw internal_error(); } } } catch (Error& e) { if (BW_DEBUG) { - printf("Blob worker got error %s, exiting\n", e.name()); + printf("Blob worker got error %s. Exiting...\n", e.name()); } TraceEvent("BlobWorkerDied", self->id).error(e, true); } - printf("cancelling actors for BW %s\n", self->id.toString().c_str()); - self->actors.clear(false); - // self->dead = true; return Void(); } From cfb8368da651ac39eab877bc3a7b097f02125179 Mon Sep 17 00:00:00 2001 From: Suraj Gupta Date: Wed, 13 Oct 2021 14:56:17 -0400 Subject: [PATCH 6/7] Address PR comments. --- fdbserver/BlobManager.actor.cpp | 2 +- fdbserver/BlobWorker.actor.cpp | 10 +++++++ flow/error_definitions.h | 1 - tests/fast/BlobGranuleCorrectnessClean.toml | 2 +- tests/slow/BlobGranuleCorrectnessLarge.toml | 26 ++++++++++++++++++- .../BlobGranuleCorrectnessLargeClean.toml | 21 +++++++++++++++ 6 files changed, 58 insertions(+), 4 deletions(-) create mode 100644 tests/slow/BlobGranuleCorrectnessLargeClean.toml diff --git a/fdbserver/BlobManager.actor.cpp b/fdbserver/BlobManager.actor.cpp index ec720989b7..a524a6a4e2 100644 --- a/fdbserver/BlobManager.actor.cpp +++ b/fdbserver/BlobManager.actor.cpp @@ -321,7 +321,7 @@ ACTOR Future doRangeAssignment(BlobManagerData* bmData, RangeAssignment as // if that worker isn't alive anymore, add the range back into the stream if (bmData->workersById.count(workerID) == 0) { - throw worker_for_granule_not_found(); + throw no_more_servers(); } AssignBlobRangeReply _rep = wait(bmData->workersById[workerID].assignBlobRangeRequest.getReply(req)); rep = _rep; diff --git a/fdbserver/BlobWorker.actor.cpp b/fdbserver/BlobWorker.actor.cpp index aa06bc2da9..45605e8934 100644 --- a/fdbserver/BlobWorker.actor.cpp +++ b/fdbserver/BlobWorker.actor.cpp @@ -125,6 +125,10 @@ struct GranuleRangeMetadata { : lastEpoch(epoch), lastSeqno(seqno), activeMetadata(activeMetadata) {} }; +// FIXME: there is a reference cycle here. BWData has GranuleRangeMetadata objects in a map, +// but each of those has a future to a forever-running actor which has a reference to BWData. +// To fix this, we should only pass the necessary, specfic fields of BWData to those actors +// rather than the reference to BWData itself. struct BlobWorkerData : NonCopyable, ReferenceCounted { UID id; Database db; @@ -555,6 +559,9 @@ ACTOR Future writeDeltaFile(Reference bwData, throw e; } + // if commit failed the first time due to granule assignment conflict (which is non-retryable), + // then the file key was persisted and we should delete it. Otherwise, the commit failed + // for some other reason and the key wasn't persisted, so we should just propogate the error if (numIterations != 1 || e.code() != error_code_granule_assignment_conflict) { throw e; } @@ -665,6 +672,9 @@ ACTOR Future writeSnapshot(Reference bwData, throw e; } + // if commit failed the first time due to granule assignment conflict (which is non-retryable), + // then the file key was persisted and we should delete it. Otherwise, the commit failed + // for some other reason and the key wasn't persisted, so we should just propogate the error if (numIterations != 1 || e.code() != error_code_granule_assignment_conflict) { throw e; } diff --git a/flow/error_definitions.h b/flow/error_definitions.h index fc11b70ec0..d1a48f5499 100755 --- a/flow/error_definitions.h +++ b/flow/error_definitions.h @@ -80,7 +80,6 @@ ERROR( local_config_changed, 1056, "Local configuration file has changed. Restar ERROR( failed_to_reach_quorum, 1057, "Failed to reach quorum from configuration database nodes. Retry sending these requests" ) ERROR( unknown_change_feed, 1058, "Change feed not found" ) ERROR( granule_assignment_conflict, 1059, "Conflicting attempts to assign blob granules" ) -ERROR( worker_for_granule_not_found, 1060, "The chosen worker to assign the granule to was not found" ) ERROR( broken_promise, 1100, "Broken promise" ) ERROR( operation_cancelled, 1101, "Asynchronous operation cancelled" ) diff --git a/tests/fast/BlobGranuleCorrectnessClean.toml b/tests/fast/BlobGranuleCorrectnessClean.toml index 59d6be0364..168790dd9d 100644 --- a/tests/fast/BlobGranuleCorrectnessClean.toml +++ b/tests/fast/BlobGranuleCorrectnessClean.toml @@ -1,5 +1,5 @@ [[test]] -testTitle = 'BlobGranuleCorrectnessTest' +testTitle = 'BlobGranuleCorrectnessCleanTest' [[test.workload]] testName = 'WriteDuringRead' diff --git a/tests/slow/BlobGranuleCorrectnessLarge.toml b/tests/slow/BlobGranuleCorrectnessLarge.toml index e88f315225..edf400b6d9 100644 --- a/tests/slow/BlobGranuleCorrectnessLarge.toml +++ b/tests/slow/BlobGranuleCorrectnessLarge.toml @@ -1,5 +1,5 @@ [[test]] -testTitle = 'BlobGranuleCorrectnessTestLarge' +testTitle = 'BlobGranuleCorrectnessLargeTest' [[test.workload]] testName = 'ReadWrite' @@ -19,3 +19,27 @@ testTitle = 'BlobGranuleCorrectnessTestLarge' [[test.workload]] testName = 'BlobGranuleVerifier' testDuration = 200.0 + + [[test.workload]] + testName = 'RandomClogging' + testDuration = 200.0 + + [[test.workload]] + testName = 'Rollback' + meanDelay = 30.0 + testDuration = 200.0 + + [[test.workload]] + testName = 'Attrition' + machinesToKill = 10 + machinesToLeave = 3 + reboot = true + testDuration = 200.0 + + [[test.workload]] + testName = 'Attrition' + machinesToKill = 10 + machinesToLeave = 3 + reboot = true + testDuration = 200.0 + diff --git a/tests/slow/BlobGranuleCorrectnessLargeClean.toml b/tests/slow/BlobGranuleCorrectnessLargeClean.toml new file mode 100644 index 0000000000..1a12e6f47f --- /dev/null +++ b/tests/slow/BlobGranuleCorrectnessLargeClean.toml @@ -0,0 +1,21 @@ +[[test]] +testTitle = 'BlobGranuleCorrectnessLargeCleanTest' + + [[test.workload]] + testName = 'ReadWrite' + testDuration = 200.0 + transactionsPerSecond = 200 + writesPerTransactionA = 0 + readsPerTransactionA = 10 + writesPerTransactionB = 10 + readsPerTransactionB = 1 + alpha = 0.5 + nodeCount = 2000000 + valueBytes = 128 + discardEdgeMeasurements = false + warmingDelay = 10.0 + setup = false + + [[test.workload]] + testName = 'BlobGranuleVerifier' + testDuration = 200.0 From 180e806086063916a044a114e97ad73749d90a51 Mon Sep 17 00:00:00 2001 From: Suraj Gupta Date: Wed, 13 Oct 2021 15:10:52 -0400 Subject: [PATCH 7/7] Add new test to cmakelists --- tests/CMakeLists.txt | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 91bc2fd423..c1f911a6d0 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -252,8 +252,9 @@ if(WITH_PYTHON) add_fdb_test(TEST_FILES slow/ApiCorrectnessAtomicRestore.toml) add_fdb_test(TEST_FILES slow/ApiCorrectnessSwitchover.toml) add_fdb_test(TEST_FILES fast/BlobGranuleCorrectness.toml IGNORE) + add_fdb_test(TEST_FILES slow/BlobGranuleCorrectnessLarge.toml IGNORE) add_fdb_test(TEST_FILES fast/BlobGranuleCorrectnessClean.toml) - add_fdb_test(TEST_FILES slow/BlobGranuleCorrectnessLarge.toml) + add_fdb_test(TEST_FILES slow/BlobGranuleCorrectnessLargeClean.toml) add_fdb_test(TEST_FILES slow/ClogWithRollbacks.toml) add_fdb_test(TEST_FILES slow/CloggedCycleTest.toml) add_fdb_test(TEST_FILES slow/CloggedStorefront.toml)