Method dumpData

fdbclient/DatabaseBackupAgent.actor.cpp:730–828

	// store mutation data from results until the end of stream or the timeout. If breaks on timeout returns the first
	// uncopied version
	ACTOR static Future<Optional<Version>> dumpData(Database cx,
	                                                Reference<Task> task,
	                                                PromiseStream<RCGroup> results,
	                                                FlowLock* lock,
	                                                Reference<TaskBucket> tb,
	                                                double breakTime) {
		state bool endOfStream = false;
		state Subspace conf = Subspace(databaseBackupPrefixRange.begin)
		                          .get(BackupAgentBase::keyConfig)
		                          .get(task->params[BackupAgentBase::keyConfigLogUid]);
		state std::vector<RangeResult> nextMutations;
		state bool isTimeoutOccured = false;
		state Optional<KeyRef> lastKey;
		state Version lastVersion;
		state int64_t nextMutationSize = 0;
		loop {
			try {
				if (endOfStream && !nextMutationSize) {
					return Optional<Version>();
				}

				state std::vector<RangeResult> mutations = std::move(nextMutations);
				state int64_t mutationSize = nextMutationSize;
				nextMutations = std::vector<RangeResult>();
				nextMutationSize = 0;

				if (!endOfStream) {
					loop {
						try {
							RCGroup group = waitNext(results.getFuture());
							lock->release(group.items.expectedSize());

							int vecSize = group.items.expectedSize();
							if (mutationSize + vecSize >= CLIENT_KNOBS->BACKUP_LOG_WRITE_BATCH_MAX_SIZE) {

								nextMutations.push_back(group.items);
								nextMutationSize = vecSize;
								break;
							}

							mutations.push_back(group.items);
							mutationSize += vecSize;
						} catch (Error& e) {
							state Error error = e;
							if (e.code() == error_code_end_of_stream) {
								endOfStream = true;
								break;
							}

							throw error;
						}
					}
				}

				state Optional<Version> nextVersionAfterBreak;
				state Transaction tr(cx);

				loop {