Skip to content
New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

DataMove Should Decide BulkLoading After Old DataMove Actor Has Been Cleared #11947

Merged
merged 2 commits into from
Feb 13, 2025
+85 −5
Merged

DataMove Should Decide BulkLoading After Old DataMove Actor Has Been Cleared #11947

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
cbcc3af
fix bulkload bug
kakaiu Feb 13, 2025
0d36d2a
fix CI
kakaiu Feb 13, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 25 additions & 0 deletions fdbclient/ManagementAPI.actor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2935,24 +2935,49 @@ ACTOR Future<BulkLoadTaskState> getBulkLoadTask(Transaction* tr,
tr->setOption(FDBTransactionOptions::READ_SYSTEM_KEYS);
RangeResult result = wait(krmGetRanges(tr, bulkLoadTaskPrefix, range));
if (result.size() > 2) {
TraceEvent(SevInfo, "GetBulkLoadTaskError")
.detail("Reason", "TooManyRanges")
.detail("Range", printable(range))
.detail("Size", result.size())
.detail("TaskId", taskId.toString());
throw bulkload_task_outdated();
} else if (result[0].value.empty()) {
TraceEvent(SevInfo, "GetBulkLoadTaskError")
.detail("Reason", "EmptyValue")
.detail("Range", printable(range))
.detail("TaskId", taskId.toString());
throw bulkload_task_outdated();
}
ASSERT(result.size() == 2);
bulkLoadTaskState = decodeBulkLoadTaskState(result[0].value);
ASSERT(bulkLoadTaskState.getTaskId().isValid());
if (taskId != bulkLoadTaskState.getTaskId()) {
// This task is overwritten by a newer task
TraceEvent(SevInfo, "GetBulkLoadTaskError")
.detail("Reason", "TaskIdMismatch")
.detail("Range", printable(range))
.detail("TaskId", taskId.toString())
.detail("TaskIdInDB", bulkLoadTaskState.getTaskId().toString());
throw bulkload_task_outdated();
}
KeyRange currentRange = KeyRangeRef(result[0].key, result[1].key);
if (bulkLoadTaskState.getRange() != currentRange) {
// This task is partially overwritten by a newer task
ASSERT(bulkLoadTaskState.getRange().contains(currentRange));
TraceEvent(SevInfo, "GetBulkLoadTaskError")
.detail("Reason", "RangeMismatch")
.detail("Range", printable(range))
.detail("TaskId", taskId.toString())
.detail("RangeInDB", printable(currentRange))
.detail("RangeInTask", printable(bulkLoadTaskState.getRange()));
throw bulkload_task_outdated();
}
if (phases.size() > 0 && !bulkLoadTaskState.onAnyPhase(phases)) {
TraceEvent(SevInfo, "GetBulkLoadTaskError")
.detail("Reason", "PhaseMismatch")
.detail("Range", printable(range))
.detail("TaskId", taskId.toString())
.detail("Phase", bulkLoadTaskState.phase);
throw bulkload_task_outdated();
}
return bulkLoadTaskState;
Expand Down
65 changes: 60 additions & 5 deletions fdbserver/DDRelocationQueue.actor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
#include "flow/Trace.h"
#include "flow/Util.h"
#include "fdbrpc/sim_validation.h"
#include "fdbclient/ManagementAPI.actor.h"
#include "fdbclient/SystemData.h"
#include "fdbserver/DataDistribution.actor.h"
#include "fdbserver/MoveKeys.actor.h"
Expand Down Expand Up @@ -1174,6 +1175,10 @@ void DDQueue::launchQueuedWork(std::set<RelocateData, std::greater<RelocateData>
if (SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) {
if (SERVER_KNOBS->ENABLE_DD_PHYSICAL_SHARD) {
rrs.dataMoveId = UID();
} else if (rrs.bulkLoadTask.present()) {
// We have to decide this after prevCleanup completes.
// For details, see the comment in dataDistributionRelocator.
rrs.dataMoveId = UID();
} else {
DataMoveType dataMoveType = newDataMoveType(rrs.bulkLoadTask.present());
rrs.dataMoveId = newDataMoveId(
Expand All @@ -1183,7 +1188,6 @@ void DDQueue::launchQueuedWork(std::set<RelocateData, std::greater<RelocateData>
.detail("TrackID", rrs.randomId)
.detail("Range", rrs.keys)
.detail("Reason", rrs.reason.toString())
.detail("DoBulkLoading", rrs.bulkLoadTask.present())
.detail("DataMoveType", dataMoveType)
.detail("DataMoveReason", static_cast<int>(rrs.dmReason));
}
Expand Down Expand Up @@ -1472,6 +1476,9 @@ ACTOR Future<Void> dataDistributionRelocator(DDQueue* self,
state WantTrueBestIfMoveout wantTrueBestIfMoveout(getWantTrueBestIfMoveout(rd.priority));
state uint64_t debugID = deterministicRandom()->randomUInt64();
state bool enableShardMove = SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA && SERVER_KNOBS->ENABLE_DD_PHYSICAL_SHARD;

// We will decide doBulkLoading after prevCleanup completes.
// rd.bulkLoadTask.present() is just the default value.
state bool doBulkLoading = rd.bulkLoadTask.present();

try {
Expand Down Expand Up @@ -1501,6 +1508,54 @@ ACTOR Future<Void> dataDistributionRelocator(DDQueue* self,

wait(prevCleanup);

if (doBulkLoading) {
// Wait until the overlapped old bulkload relocator get cancelled.
// At this point, the existing metadata is updated by the old relocator.
// No more change will be made to the metadata.
// At this point, we check if the current rd.bulkLoadTask is outdated.
// If yes, do not do bulkload.
state Transaction tr(self->txnProcessor->context());
loop {
try {
BulkLoadTaskState currentBulkLoadTaskState =
wait(getBulkLoadTask(&tr,
rd.bulkLoadTask.get().coreState.getRange(),
rd.bulkLoadTask.get().coreState.getTaskId(),
{ BulkLoadPhase::Triggered, BulkLoadPhase::Running }));
break;
} catch (Error& e) {
if (e.code() == error_code_bulkload_task_outdated) {
// Notify the bulkload task actor to exit. This avoid bulkload task engine
// gets stuck in case a new task overwrite this task on metadata. To schedule
// the new task, the old task actor has to exit.
if (rd.bulkLoadTask.get().completeAck.canBeSet()) {
rd.bulkLoadTask.get().completeAck.sendError(bulkload_task_outdated());
}
doBulkLoading = false;
// we do not want to reset rd.bulkLoadTask here because
// the busyness calculation is based on whether rd.bulkLoadTask is set.
// When calculating the busyness, we do not count the work for any
// rd with bulkLoadTask set.
// TODO(BulkLoad): reset rd.bulkLoadTask here for the risk of overloading the source
// servers.
break;
}
wait(tr.onError(e));
}
}
DataMoveType dataMoveType = newDataMoveType(doBulkLoading);
rd.dataMoveId = newDataMoveId(
deterministicRandom()->randomUInt64(), AssignEmptyRange::False, dataMoveType, rd.dmReason);
TraceEvent(SevInfo, "DDBulkLoadNewDataMoveID", self->distributorId)
.detail("DataMoveID", rd.dataMoveId.toString())
.detail("TrackID", rd.randomId)
.detail("Range", rd.keys)
.detail("Reason", rd.reason.toString())
.detail("DataMoveType", dataMoveType)
.detail("DoBulkLoading", doBulkLoading)
.detail("DataMoveReason", static_cast<int>(rd.dmReason));
}

auto f = self->dataMoves.intersectingRanges(rd.keys);
for (auto it = f.begin(); it != f.end(); ++it) {
KeyRangeRef kr(it->range().begin, it->range().end);
Expand Down Expand Up @@ -1991,8 +2046,8 @@ ACTOR Future<Void> dataDistributionRelocator(DDQueue* self,
relocateShardInterval.pairID,
ddEnabledState,
CancelConflictingDataMoves::False,
rd.bulkLoadTask.present() ? rd.bulkLoadTask.get().coreState
: Optional<BulkLoadTaskState>());
doBulkLoading ? rd.bulkLoadTask.get().coreState
: Optional<BulkLoadTaskState>());
} else {
params = std::make_unique<MoveKeysParams>(rd.dataMoveId,
rd.keys,
Expand All @@ -2006,8 +2061,8 @@ ACTOR Future<Void> dataDistributionRelocator(DDQueue* self,
relocateShardInterval.pairID,
ddEnabledState,
CancelConflictingDataMoves::False,
rd.bulkLoadTask.present() ? rd.bulkLoadTask.get().coreState
: Optional<BulkLoadTaskState>());
doBulkLoading ? rd.bulkLoadTask.get().coreState
: Optional<BulkLoadTaskState>());
}
state Future<Void> doMoveKeys = self->txnProcessor->moveKeys(*params);
state Future<Void> pollHealth =
Expand Down