Function createCheckpointImpl

fdbclient/NativeAPI.actor.cpp:8441–8475

ACTOR template <class T>
static Future<Void> createCheckpointImpl(T tr, KeyRangeRef range, CheckpointFormat format) {
	ASSERT(!tr->getTenant().present());
	TraceEvent("CreateCheckpointTransactionBegin").detail("Range", range);

	state RangeResult keyServers = wait(krmGetRanges(tr, keyServersPrefix, range));
	ASSERT(!keyServers.more);

	state RangeResult UIDtoTagMap = wait(tr->getRange(serverTagKeys, CLIENT_KNOBS->TOO_MANY));
	ASSERT(!UIDtoTagMap.more && UIDtoTagMap.size() < CLIENT_KNOBS->TOO_MANY);

	for (int i = 0; i < keyServers.size() - 1; ++i) {
		KeyRangeRef shard(keyServers[i].key, keyServers[i + 1].key);
		std::vector<UID> src;
		std::vector<UID> dest;
		decodeKeyServersValue(UIDtoTagMap, keyServers[i].value, src, dest);

		// The checkpoint request is sent to all replicas, in case any of them is unhealthy.
		// An alternative is to choose a healthy replica.
		const UID checkpointID = deterministicRandom()->randomUniqueID();
		for (int idx = 0; idx < src.size(); ++idx) {
			CheckpointMetaData checkpoint(shard & range, format, src[idx], checkpointID);
			checkpoint.setState(CheckpointMetaData::Pending);
			tr->set(checkpointKeyFor(checkpointID), checkpointValue(checkpoint));
		}

		TraceEvent("CreateCheckpointTransactionShard")
		    .detail("Shard", shard)
		    .detail("SrcServers", describe(src))
		    .detail("ServerSelected", describe(src))
		    .detail("CheckpointKey", checkpointKeyFor(checkpointID))
		    .detail("ReadVersion", tr->getReadVersion().get());
	}

	return Void();
}

Future<Void> createCheckpoint(Reference<ReadYourWritesTransaction> tr, KeyRangeRef range, CheckpointFormat format) {
	return holdWhile(tr, createCheckpointImpl(tr, range, format));