| 1292 | static constexpr uint32_t version = 1; |
| 1293 | |
| 1294 | ACTOR static Future<Void> _finish(Reference<ReadYourWritesTransaction> tr, |
| 1295 | Reference<TaskBucket> taskBucket, |
| 1296 | Reference<FutureBucket> futureBucket, |
| 1297 | Reference<Task> task) { |
| 1298 | state Subspace conf = Subspace(databaseBackupPrefixRange.begin) |
| 1299 | .get(BackupAgentBase::keyConfig) |
| 1300 | .get(task->params[BackupAgentBase::keyConfigLogUid]); |
| 1301 | state Subspace states = Subspace(databaseBackupPrefixRange.begin) |
| 1302 | .get(BackupAgentBase::keyStates) |
| 1303 | .get(task->params[BackupAgentBase::keyConfigLogUid]); |
| 1304 | wait(checkTaskVersion(tr, task, CopyDiffLogsTaskFunc::name, CopyDiffLogsTaskFunc::version)); |
| 1305 | |
| 1306 | state Version beginVersion = |
| 1307 | BinaryReader::fromStringRef<Version>(task->params[DatabaseBackupAgent::keyBeginVersion], Unversioned()); |
| 1308 | state Version prevBeginVersion = |
| 1309 | BinaryReader::fromStringRef<Version>(task->params[DatabaseBackupAgent::keyPrevBeginVersion], Unversioned()); |
| 1310 | state Future<Optional<Value>> fStopWhenDone = tr->get(conf.pack(DatabaseBackupAgent::keyConfigStopWhenDoneKey)); |
| 1311 | |
| 1312 | Transaction srcTr(taskBucket->src); |
| 1313 | srcTr.setOption(FDBTransactionOptions::LOCK_AWARE); |
| 1314 | state Version endVersion = wait(srcTr.getReadVersion()); |
| 1315 | |
| 1316 | state Reference<TaskFuture> onDone = futureBucket->unpack(task->params[Task::reservedTaskParamKeyDone]); |
| 1317 | |
| 1318 | if (endVersion <= beginVersion) { |
| 1319 | wait(delay(FLOW_KNOBS->PREVENT_FAST_SPIN_DELAY)); |
| 1320 | wait(success(CopyDiffLogsTaskFunc::addTask( |
| 1321 | tr, taskBucket, task, prevBeginVersion, beginVersion, TaskCompletionKey::signal(onDone)))); |
| 1322 | wait(taskBucket->finish(tr, task)); |
| 1323 | return Void(); |
| 1324 | } |
| 1325 | |
| 1326 | tr->set(task->params[BackupAgentBase::keyConfigLogUid].withPrefix(applyMutationsEndRange.begin), |
| 1327 | BinaryWriter::toValue(beginVersion, Unversioned())); |
| 1328 | Optional<Value> stopWhenDone = wait(fStopWhenDone); |
| 1329 | |
| 1330 | if (endVersion - beginVersion > deterministicRandom()->randomInt64(0, CLIENT_KNOBS->BACKUP_VERSION_DELAY)) { |
| 1331 | TraceEvent("DBA_CopyDiffLogs") |
| 1332 | .detail("BeginVersion", beginVersion) |
| 1333 | .detail("EndVersion", endVersion) |
| 1334 | .detail("LogUID", task->params[BackupAgentBase::keyConfigLogUid]); |
| 1335 | } |
| 1336 | |
| 1337 | // set the log version to the state |
| 1338 | tr->set(StringRef(states.pack(DatabaseBackupAgent::keyStateLogBeginVersion)), |
| 1339 | BinaryWriter::toValue(beginVersion, Unversioned())); |
| 1340 | |
| 1341 | if (!stopWhenDone.present()) { |
| 1342 | state Reference<TaskFuture> allPartsDone = futureBucket->future(tr); |
| 1343 | std::vector<Future<Key>> addTaskVector; |
| 1344 | addTaskVector.push_back(CopyDiffLogsTaskFunc::addTask( |
| 1345 | tr, taskBucket, task, beginVersion, endVersion, TaskCompletionKey::signal(onDone), allPartsDone)); |
| 1346 | int blockSize = std::max<int>( |
| 1347 | 1, ((endVersion - beginVersion) / CLIENT_KNOBS->BACKUP_COPY_TASKS) / CLIENT_KNOBS->BACKUP_BLOCK_SIZE); |
| 1348 | for (int64_t vblock = beginVersion / CLIENT_KNOBS->BACKUP_BLOCK_SIZE; |
| 1349 | vblock < (endVersion + CLIENT_KNOBS->BACKUP_BLOCK_SIZE - 1) / CLIENT_KNOBS->BACKUP_BLOCK_SIZE; |
| 1350 | vblock += blockSize) { |
| 1351 | addTaskVector.push_back(CopyLogRangeTaskFunc::addTask( |
nothing calls this directly
no test coverage detected