Method getNext_impl

fdbclient/MutationLogReader.actor.cpp:66–114

}

ACTOR Future<Void> PipelinedReader::getNext_impl(PipelinedReader* self, Database cx) {
	state Transaction tr(cx);

	state GetRangeLimits limits(GetRangeLimits::ROW_LIMIT_UNLIMITED,
	                            (g_network->isSimulated() && !g_simulator.speedUpSimulation)
	                                ? CLIENT_KNOBS->BACKUP_SIMULATED_LIMIT_BYTES
	                                : CLIENT_KNOBS->BACKUP_GET_RANGE_LIMIT_BYTES);

	state Key begin = versionToKey(self->currentBeginVersion, self->prefix);
	state Key end = versionToKey(self->endVersion, self->prefix);

	loop {
		// Get the lock
		wait(self->readerLimit.take());

		// Read begin to end forever until successful
		loop {
			try {
				tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
				tr.setOption(FDBTransactionOptions::LOCK_AWARE);

				RangeResult kvs = wait(tr.getRange(KeyRangeRef(begin, end), limits));

				// No more results, send end of stream
				if (!kvs.empty()) {
					// Send results to the reads stream
					self->reads.send(
					    RangeResultBlock{ .result = kvs,
					                      .firstVersion = keyRefToVersion(kvs.front().key, self->prefix.size()),
					                      .lastVersion = keyRefToVersion(kvs.back().key, self->prefix.size()),
					                      .hash = self->hash,
					                      .prefixLen = self->prefix.size(),
					                      .indexToRead = 0 });
				}

				if (!kvs.more) {
					self->reads.sendError(end_of_stream());
					return Void();
				}

				begin = kvs.readThrough.present() ? kvs.readThrough.get() : keyAfter(kvs.back().key);

				break;
			} catch (Error& e) {
				if (e.code() == error_code_transaction_too_old) {
					// We are using this transaction until it's too old and then resetting to a fresh one,
					// so we don't need to delay.
					tr.fullReset();
				} else {
					wait(tr.onError(e));
				}
			}