Function readRange

fdbserver/StorageCache.actor.cpp:555–624

};

GetKeyValuesReply readRange(StorageCacheData* data, Version version, KeyRangeRef range, int limit, int* pLimitBytes) {
	GetKeyValuesReply result;
	StorageCacheData::VersionedData::ViewAtVersion view = data->data().at(version);
	StorageCacheData::VersionedData::iterator vCurrent = view.end();
	KeyRef readBegin;
	KeyRef readEnd;
	KeyRef rangeBegin = range.begin;
	KeyRef rangeEnd = range.end;
	int accumulatedBytes = 0;
	// printf("\nSCReadRange\n");

	// if (limit >= 0) we are reading forward, else backward
	if (limit >= 0) {
		// We might care about a clear beginning before start that runs into range
		vCurrent = view.lastLessOrEqual(rangeBegin);
		if (vCurrent && vCurrent->isClearTo() && vCurrent->getEndKey() > rangeBegin)
			readBegin = vCurrent->getEndKey();
		else
			readBegin = rangeBegin;

		vCurrent = view.lower_bound(readBegin);
		ASSERT(!vCurrent || vCurrent.key() >= readBegin);
		if (vCurrent) {
			auto b = vCurrent;
			--b;
			ASSERT(!b || b.key() < readBegin);
		}
		accumulatedBytes = 0;
		while (vCurrent && vCurrent.key() < rangeEnd && limit > 0 && accumulatedBytes < *pLimitBytes) {
			if (!vCurrent->isClearTo()) {
				result.data.push_back_deep(result.arena, KeyValueRef(vCurrent.key(), vCurrent->getValue()));
				accumulatedBytes += sizeof(KeyValueRef) + result.data.end()[-1].expectedSize();
				--limit;
			}
			++vCurrent;
		}
	} else { // reverse readRange
		vCurrent = view.lastLess(rangeEnd);

		// A clear might extend all the way to range.end
		if (vCurrent && vCurrent->isClearTo() && vCurrent->getEndKey() >= rangeEnd) {
			readEnd = vCurrent.key();
			--vCurrent;
		} else {
			readEnd = rangeEnd;
		}
		ASSERT(!vCurrent || vCurrent.key() < readEnd);
		if (vCurrent) {
			auto b = vCurrent;
			--b;
			ASSERT(!b || b.key() >= readEnd);
		}
		accumulatedBytes = 0;
		while (vCurrent && vCurrent.key() >= rangeEnd && limit > 0 && accumulatedBytes < *pLimitBytes) {
			if (!vCurrent->isClearTo()) {
				result.data.push_back_deep(result.arena, KeyValueRef(vCurrent.key(), vCurrent->getValue()));
				accumulatedBytes += sizeof(KeyValueRef) + result.data.end()[-1].expectedSize();
				--limit;