tdesktop/Telegram/SourceFiles/media/streaming/media_streaming_reader.cpp

/*
This file is part of Telegram Desktop,
the official desktop application for the Telegram messaging service.

For license and copyright information please follow this link:
https://github.com/telegramdesktop/tdesktop/blob/master/LEGAL
*/
#include "media/streaming/media_streaming_reader.h"

#include "media/streaming/media_streaming_common.h"
#include "media/streaming/media_streaming_loader.h"
#include "storage/cache/storage_cache_database.h"

namespace Media {
namespace Streaming {
namespace {

constexpr auto kPartSize = Loader::kPartSize;
constexpr auto kPartsInSlice = 64;
constexpr auto kInSlice = kPartsInSlice * kPartSize;
constexpr auto kMaxPartsInHeader = 64;
constexpr auto kMaxOnlyInHeader = 80 * kPartSize;
constexpr auto kPartsOutsideFirstSliceGood = 8;
constexpr auto kSlicesInMemory = 2;

// 1 MB of parts are requested from cloud ahead of reading demand.
constexpr auto kPreloadPartsAhead = 8;
constexpr auto kDownloaderRequestsLimit = 4;

using PartsMap = base::flat_map<int, QByteArray>;

struct ParsedCacheEntry {
	PartsMap parts;
	std::optional<PartsMap> included;
};

bool IsContiguousSerialization(int serializedSize, int maxSliceSize) {
	return !(serializedSize % kPartSize) || (serializedSize == maxSliceSize);
}

bool IsFullInHeader(int size) {
	return (size <= kMaxOnlyInHeader);
}

bool ComputeIsGoodHeader(int size, const PartsMap &header) {
	if (IsFullInHeader(size)) {
		return false;
	}
	const auto outsideFirstSliceIt = ranges::lower_bound(
		header,
		kInSlice,
		ranges::less(),
		&PartsMap::value_type::first);
	const auto outsideFirstSlice = end(header) - outsideFirstSliceIt;
	return (outsideFirstSlice <= kPartsOutsideFirstSliceGood);
}

int SlicesCount(int size) {
	return (size + kInSlice - 1) / kInSlice;
}

int MaxSliceSize(int sliceNumber, int size) {
	return !sliceNumber
		? size
		: (sliceNumber == SlicesCount(size))
		? (size - (sliceNumber - 1) * kInSlice)
		: kInSlice;
}

bytes::const_span ParseComplexCachedMap(
		PartsMap &result,
		bytes::const_span data,
		int maxSize) {
	const auto takeInt = [&]() -> std::optional<int> {
		if (data.size() < sizeof(int32)) {
			return std::nullopt;
		}
		const auto bytes = data.data();
		const auto result = *reinterpret_cast<const int32*>(bytes);
		data = data.subspan(sizeof(int32));
		return result;
	};
	const auto takeBytes = [&](int count) {
		if (count <= 0 || data.size() < count) {
			return bytes::const_span();
		}
		const auto result = data.subspan(0, count);
		data = data.subspan(count);
		return result;
	};
	const auto maybeCount = takeInt();
	if (!maybeCount) {
		return {};
	}
	const auto count = *maybeCount;
	if (count < 0) {
		return {};
	} else if (!count) {
		return data;
	}
	for (auto i = 0; i != count; ++i) {
		const auto offset = takeInt().value_or(0);
		const auto size = takeInt().value_or(0);
		const auto bytes = takeBytes(size);
		if (offset < 0
			|| offset >= maxSize
			|| size <= 0
			|| size > maxSize
			|| offset + size > maxSize
			|| bytes.size() != size) {
			return {};
		}
		result.try_emplace(
			offset,
			reinterpret_cast<const char*>(bytes.data()),
			bytes.size());
	}
	return data;
}

bytes::const_span ParseCachedMap(
		PartsMap &result,
		bytes::const_span data,
		int maxSize) {
	const auto size = int(data.size());
	if (IsContiguousSerialization(size, maxSize)) {
		if (size > maxSize) {
			return {};
		}
		for (auto offset = 0; offset < size; offset += kPartSize) {
			const auto part = data.subspan(
				offset,
				std::min(kPartSize, size - offset));
			result.try_emplace(
				offset,
				reinterpret_cast<const char*>(part.data()),
				part.size());
		}
		return {};
	}
	return ParseComplexCachedMap(result, data, maxSize);
}

ParsedCacheEntry ParseCacheEntry(
		bytes::const_span data,
		int sliceNumber,
		int size) {
	auto result = ParsedCacheEntry();
	const auto remaining = ParseCachedMap(
		result.parts,
		data,
		MaxSliceSize(sliceNumber, size));
	if (!sliceNumber && ComputeIsGoodHeader(size, result.parts)) {
		result.included = PartsMap();
		ParseCachedMap(*result.included, remaining, MaxSliceSize(1, size));
	}
	return result;
}

template <typename Range> // Range::value_type is Pair<int, QByteArray>
int FindNotLoadedStart(Range &&parts, int offset) {
	auto result = offset;
	for (const auto &part : parts) {
		const auto partStart = part.first;
		const auto partEnd = partStart + part.second.size();
		if (partStart <= result && partEnd >= result) {
			result = partEnd;
		} else {
			break;
		}
	}
	return result;
}

template <typename Range> // Range::value_type is Pair<int, QByteArray>
void CopyLoaded(bytes::span buffer, Range &&parts, int offset, int till) {
	auto filled = offset;
	for (const auto &part : parts) {
		const auto bytes = bytes::make_span(part.second);
		const auto partStart = part.first;
		const auto partEnd = int(partStart + bytes.size());
		const auto copyTill = std::min(partEnd, till);
		Assert(partStart <= filled && filled < copyTill);

		const auto from = filled - partStart;
		const auto copy = copyTill - filled;
		bytes::copy(buffer, bytes.subspan(from, copy));
		buffer = buffer.subspan(copy);
		filled += copy;
	}
}

} // namespace

template <int Size>
bool Reader::StackIntVector<Size>::add(int value) {
	using namespace rpl::mappers;

	const auto i = ranges::find_if(_storage, _1 < 0);
	if (i == end(_storage)) {
		return false;
	}
	*i = value;
	const auto next = i + 1;
	if (next != end(_storage)) {
		*next = -1;
	}
	return true;
}

template <int Size>
auto Reader::StackIntVector<Size>::values() const {
	using namespace rpl::mappers;

	return ranges::view::all(_storage) | ranges::view::take_while(_1 >= 0);
}

struct Reader::CacheHelper {
	explicit CacheHelper(Storage::Cache::Key baseKey);

	Storage::Cache::Key key(int sliceNumber) const;

	const Storage::Cache::Key baseKey;

	QMutex mutex;
	base::flat_map<int, PartsMap> results;
	std::vector<int> sizes;
	std::atomic<crl::semaphore*> waiting = nullptr;
};

Reader::CacheHelper::CacheHelper(Storage::Cache::Key baseKey)
: baseKey(baseKey) {
}

Storage::Cache::Key Reader::CacheHelper::key(int sliceNumber) const {
	return Storage::Cache::Key{ baseKey.high, baseKey.low + sliceNumber };
}

void Reader::Slice::processCacheData(PartsMap &&data) {
	Expects((flags & Flag::LoadingFromCache) != 0);
	Expects(!(flags & Flag::LoadedFromCache));

	const auto guard = gsl::finally([&] {
		flags |= Flag::LoadedFromCache;
		flags &= ~Flag::LoadingFromCache;
	});
	if (parts.empty()) {
		parts = std::move(data);
	} else {
		for (auto &[offset, bytes] : data) {
			parts.emplace(offset, std::move(bytes));
		}
	}
}

void Reader::Slice::addPart(int offset, QByteArray bytes) {
	Expects(!parts.contains(offset));

	parts.emplace(offset, std::move(bytes));
	if (flags & Flag::LoadedFromCache) {
		flags |= Flag::ChangedSinceCache;
	}
}

auto Reader::Slice::prepareFill(int from, int till) -> PrepareFillResult {
	auto result = PrepareFillResult();

	result.ready = false;
	const auto fromOffset = (from / kPartSize) * kPartSize;
	const auto tillPart = (till + kPartSize - 1) / kPartSize;
	const auto preloadTillOffset = (tillPart + kPreloadPartsAhead)
		* kPartSize;

	const auto after = ranges::upper_bound(
		parts,
		from,
		ranges::less(),
		&PartsMap::value_type::first);
	if (after == begin(parts)) {
		result.offsetsFromLoader = offsetsFromLoader(
			fromOffset,
			preloadTillOffset);
		return result;
	}

	const auto start = after - 1;
	const auto finish = ranges::lower_bound(
		start,
		end(parts),
		till,
		ranges::less(),
		&PartsMap::value_type::first);
	const auto haveTill = FindNotLoadedStart(
		ranges::make_subrange(start, finish),
		fromOffset);
	if (haveTill < till) {
		result.offsetsFromLoader = offsetsFromLoader(
			haveTill,
			preloadTillOffset);
		return result;
	}
	result.ready = true;
	result.start = start;
	result.finish = finish;
	result.offsetsFromLoader = offsetsFromLoader(
		tillPart * kPartSize,
		preloadTillOffset);
	return result;
}

auto Reader::Slice::offsetsFromLoader(int from, int till) const
-> StackIntVector<Reader::kLoadFromRemoteMax> {
	auto result = StackIntVector<kLoadFromRemoteMax>();

	const auto after = ranges::upper_bound(
		parts,
		from,
		ranges::less(),
		&PartsMap::value_type::first);
	auto check = (after == begin(parts)) ? after : (after - 1);
	const auto end = parts.end();
	for (auto offset = from; offset != till; offset += kPartSize) {
		while (check != end && check->first < offset) {
			++check;
		}
		if (check != end && check->first == offset) {
			continue;
		} else if (!result.add(offset)) {
			break;
		}
	}
	return result;
}

Reader::Slices::Slices(int size, bool useCache)
: _size(size) {
	Expects(size > 0);

	if (useCache) {
		_header.flags |= Slice::Flag::LoadingFromCache;
	} else {
		_headerMode = HeaderMode::NoCache;
	}
	if (!isFullInHeader()) {
		_data.resize(SlicesCount(_size));
	}
}

bool Reader::Slices::headerModeUnknown() const {
	return (_headerMode == HeaderMode::Unknown);
}

bool Reader::Slices::isFullInHeader() const {
	return IsFullInHeader(_size);
}

bool Reader::Slices::isGoodHeader() const {
	return (_headerMode == HeaderMode::Good);
}

bool Reader::Slices::computeIsGoodHeader() const {
	return ComputeIsGoodHeader(_size, _header.parts);
}

void Reader::Slices::headerDone(bool fromCache) {
	if (_headerMode != HeaderMode::Unknown) {
		return;
	}
	_headerMode = isFullInHeader()
		? HeaderMode::Full
		: computeIsGoodHeader()
		? HeaderMode::Good
		: HeaderMode::Small;
	if (!fromCache) {
		for (auto &slice : _data) {
			using Flag = Slice::Flag;
			Assert(!(slice.flags
				& (Flag::LoadingFromCache | Flag::LoadedFromCache)));
			slice.flags |= Slice::Flag::LoadedFromCache;
		}
	}
}

int Reader::Slices::headerSize() const {
	return _header.parts.size() * kPartSize;
}

bool Reader::Slices::fullInCache() const {
	return _fullInCache;
}

int Reader::Slices::requestSliceSizesCount() const {
	if (!headerModeUnknown() || isFullInHeader()) {
		return 0;
	}
	return _data.size();
}

bool Reader::Slices::headerWontBeFilled() const {
	return headerModeUnknown()
		&& (_header.parts.size() >= kMaxPartsInHeader);
}

void Reader::Slices::applyHeaderCacheData() {
	using namespace rpl::mappers;

	const auto applyWhile = [&](auto &&predicate) {
		for (const auto &[offset, part] : _header.parts) {
			const auto index = offset / kInSlice;
			if (!predicate(index)) {
				break;
			}
			_data[index].addPart(
				offset - index * kInSlice,
				base::duplicate(part));
		}
	};
	if (_header.parts.empty()) {
		return;
	} else if (_headerMode == HeaderMode::Good) {
		// Always apply data to first block if it is cached in the header.
		applyWhile(_1 == 0);
	} else if (_headerMode != HeaderMode::Unknown) {
		return;
	} else if (isFullInHeader()) {
		headerDone(true);
	} else {
		applyWhile(_1 < int(_data.size()));
		headerDone(true);
	}
}

void Reader::Slices::processCacheResult(int sliceNumber, PartsMap &&result) {
	Expects(sliceNumber >= 0 && sliceNumber <= _data.size());

	auto &slice = (sliceNumber ? _data[sliceNumber - 1] : _header);
	if (!sliceNumber && isGoodHeader()) {
		// We've loaded header slice because really we wanted first slice.
		if (!(_data[0].flags & Slice::Flag::LoadingFromCache)) {
			// We could've already unloaded this slice using LRU _usedSlices.
			return;
		}
		// So just process whole result even if we didn't want header really.
		slice.flags |= Slice::Flag::LoadingFromCache;
		slice.flags &= ~Slice::Flag::LoadedFromCache;
	}
	if (!(slice.flags & Slice::Flag::LoadingFromCache)) {
		// We could've already unloaded this slice using LRU _usedSlices.
		return;
	}
	slice.processCacheData(std::move(result));
	checkSliceFullLoaded(sliceNumber);
	if (!sliceNumber) {
		applyHeaderCacheData();
		if (isGoodHeader()) {
			// When we first read header we don't request the first slice.
			// But we get it, so let's apply it anyway.
			_data[0].flags |= Slice::Flag::LoadingFromCache;
		}
	}
}

void Reader::Slices::processCachedSizes(const std::vector<int> &sizes) {
	Expects(sizes.size() == _data.size());

	using Flag = Slice::Flag;
	const auto count = int(sizes.size());
	auto loadedCount = 0;
	for (auto i = 0; i != count; ++i) {
		const auto sliceNumber = (i + 1);
		const auto sliceSize = (sliceNumber < _data.size())
			? kInSlice
			: (_size - (sliceNumber - 1) * kInSlice);
		const auto loaded = (sizes[i] == sliceSize);

		if (_data[i].flags & Flag::FullInCache) {
			++loadedCount;
		} else if (loaded) {
			_data[i].flags |= Flag::FullInCache;
			++loadedCount;
		}
	}
	_fullInCache = (loadedCount == count);
}

void Reader::Slices::checkSliceFullLoaded(int sliceNumber) {
	if (!sliceNumber && !isFullInHeader()) {
		return;
	}
	const auto partsCount = [&] {
		if (!sliceNumber) {
			return (_size + kPartSize - 1) / kPartSize;
		}
		return (sliceNumber < _data.size())
			? kPartsInSlice
			: ((_size - (sliceNumber - 1) * kInSlice + kPartSize - 1)
				/ kPartSize);
	}();
	auto &slice = (sliceNumber ? _data[sliceNumber - 1] : _header);
	const auto loaded = (slice.parts.size() == partsCount);

	using Flag = Slice::Flag;
	if ((slice.flags & Flag::FullInCache) && !loaded) {
		slice.flags &= ~Flag::FullInCache;
		_fullInCache = false;
	} else if (!(slice.flags & Flag::FullInCache) && loaded) {
		slice.flags |= Flag::FullInCache;
		_fullInCache = checkFullInCache();
	}
}

bool Reader::Slices::checkFullInCache() const {
	using Flag = Slice::Flag;
	if (isFullInHeader()) {
		return (_header.flags & Flag::FullInCache);
	}
	const auto i = ranges::find_if(_data, [](const Slice &slice) {
		return !(slice.flags & Flag::FullInCache);
	});
	return (i == end(_data));
}

void Reader::Slices::processPart(
		int offset,
		QByteArray &&bytes) {
	Expects(isFullInHeader() || (offset / kInSlice < _data.size()));

	if (isFullInHeader()) {
		_header.addPart(offset, bytes);
		checkSliceFullLoaded(0);
		return;
	} else if (_headerMode == HeaderMode::Unknown) {
		if (_header.parts.contains(offset)) {
			return;
		} else if (_header.parts.size() < kMaxPartsInHeader) {
			_header.addPart(offset, bytes);
		}
	}
	const auto index = offset / kInSlice;
	_data[index].addPart(offset - index * kInSlice, std::move(bytes));
	checkSliceFullLoaded(index + 1);
}

auto Reader::Slices::fill(int offset, bytes::span buffer) -> FillResult {
	Expects(!buffer.empty());
	Expects(offset >= 0 && offset < _size);
	Expects(offset + buffer.size() <= _size);
	Expects(buffer.size() <= kInSlice);

	using Flag = Slice::Flag;

	if (_headerMode != HeaderMode::NoCache
		&& !(_header.flags & Flag::LoadedFromCache)) {
		// Waiting for initial cache query.
		Assert(waitingForHeaderCache());
		return {};
	} else if (isFullInHeader()) {
		return fillFromHeader(offset, buffer);
	}

	auto result = FillResult();
	const auto till = int(offset + buffer.size());
	const auto fromSlice = offset / kInSlice;
	const auto tillSlice = (till + kInSlice - 1) / kInSlice;
	Assert(fromSlice >= 0
		&& (fromSlice + 1 == tillSlice || fromSlice + 2 == tillSlice)
		&& tillSlice <= _data.size());

	const auto cacheNotLoaded = [&](int sliceIndex) {
		return (_headerMode != HeaderMode::NoCache)
			&& (_headerMode != HeaderMode::Unknown)
			&& !(_data[sliceIndex].flags & Flag::LoadedFromCache);
	};
	const auto handlePrepareResult = [&](
			int sliceIndex,
			const Slice::PrepareFillResult &prepared) {
		if (cacheNotLoaded(sliceIndex)) {
			return;
		}
		for (const auto offset : prepared.offsetsFromLoader.values()) {
			const auto full = offset + sliceIndex * kInSlice;
			if (offset < kInSlice && full < _size) {
				result.offsetsFromLoader.add(full);
			}
		}
	};
	const auto handleReadFromCache = [&](int sliceIndex) {
		if (cacheNotLoaded(sliceIndex)
			&& !(_data[sliceIndex].flags & Flag::LoadingFromCache)) {
			_data[sliceIndex].flags |= Flag::LoadingFromCache;
			result.sliceNumbersFromCache.add(sliceIndex + 1);
		}
	};
	const auto firstFrom = offset - fromSlice * kInSlice;
	const auto firstTill = std::min(kInSlice, till - fromSlice * kInSlice);
	const auto secondFrom = 0;
	const auto secondTill = till - (fromSlice + 1) * kInSlice;
	const auto first = _data[fromSlice].prepareFill(firstFrom, firstTill);
	const auto second = (fromSlice + 1 < tillSlice)
		? _data[fromSlice + 1].prepareFill(secondFrom, secondTill)
		: Slice::PrepareFillResult();
	handlePrepareResult(fromSlice, first);
	if (fromSlice + 1 < tillSlice) {
		handlePrepareResult(fromSlice + 1, second);
	}
	if (first.ready && second.ready) {
		markSliceUsed(fromSlice);
		CopyLoaded(
			buffer,
			ranges::make_subrange(first.start, first.finish),
			firstFrom,
			firstTill);
		if (fromSlice + 1 < tillSlice) {
			markSliceUsed(fromSlice + 1);
			CopyLoaded(
				buffer.subspan(firstTill - firstFrom),
				ranges::make_subrange(second.start, second.finish),
				secondFrom,
				secondTill);
		}
		result.toCache = serializeAndUnloadUnused();
		result.filled = true;
	} else {
		handleReadFromCache(fromSlice);
		if (fromSlice + 1 < tillSlice) {
			handleReadFromCache(fromSlice + 1);
		}
	}
	return result;
}

auto Reader::Slices::fillFromHeader(int offset, bytes::span buffer)
-> FillResult {
	auto result = FillResult();
	const auto from = offset;
	const auto till = int(offset + buffer.size());

	const auto prepared = _header.prepareFill(from, till);
	for (const auto full : prepared.offsetsFromLoader.values()) {
		if (full < _size) {
			result.offsetsFromLoader.add(full);
		}
	}
	if (prepared.ready) {
		CopyLoaded(
			buffer,
			ranges::make_subrange(prepared.start, prepared.finish),
			from,
			till);
		result.filled = true;
	}
	return result;
}

QByteArray Reader::Slices::partForDownloader(int offset) const {
	Expects(offset < _size);

	if (const auto i = _header.parts.find(offset); i != end(_header.parts)) {
		return i->second;
	} else if (isFullInHeader()) {
		return QByteArray();
	}
	const auto index = offset / kInSlice;
	const auto &slice = _data[index];
	const auto i = slice.parts.find(offset - index * kInSlice);
	return (i != end(slice.parts)) ? i->second : QByteArray();
}

bool Reader::Slices::waitingForHeaderCache() const {
	return (_header.flags & Slice::Flag::LoadingFromCache);
}

bool Reader::Slices::readCacheForDownloaderRequired(int offset) {
	Expects(offset < _size);
	Expects(!waitingForHeaderCache());

	if (isFullInHeader()) {
		return false;
	}
	const auto index = offset / kInSlice;
	auto &slice = _data[index];
	return !(slice.flags & Slice::Flag::LoadedFromCache);
}

void Reader::Slices::markSliceUsed(int sliceIndex) {
	const auto i = ranges::find(_usedSlices, sliceIndex);
	const auto end = _usedSlices.end();
	if (i == end) {
		_usedSlices.push_back(sliceIndex);
	} else {
		const auto next = i + 1;
		if (next != end) {
			std::rotate(i, next, end);
		}
	}
}

int Reader::Slices::maxSliceSize(int sliceNumber) const {
	return MaxSliceSize(sliceNumber, _size);
}

Reader::SerializedSlice Reader::Slices::serializeAndUnloadUnused() {
	using Flag = Slice::Flag;

	if (_headerMode == HeaderMode::Unknown
		|| _usedSlices.size() <= kSlicesInMemory) {
		return {};
	}
	const auto purgeSlice = _usedSlices.front();
	_usedSlices.pop_front();
	if (!(_data[purgeSlice].flags & Flag::LoadedFromCache)) {
		// If the only data in this slice was from _header, just leave it.
		return {};
	}
	const auto noNeedToSaveToCache = [&] {
		if (_headerMode == HeaderMode::NoCache) {
			// Cache is not used.
			return true;
		} else if (!(_data[purgeSlice].flags & Flag::ChangedSinceCache)) {
			// If no data was changed we should still save first slice,
			// if header data was changed since loading from cache.
			// Otherwise in destructor we won't be able to unload header.
			if (!isGoodHeader()
				|| (purgeSlice > 0)
				|| (!(_header.flags & Flag::ChangedSinceCache))) {
				return true;
			}
		}
		return false;
	}();
	if (noNeedToSaveToCache) {
		unloadSlice(_data[purgeSlice]);
		return {};
	}
	return serializeAndUnloadSlice(purgeSlice + 1);
}

Reader::SerializedSlice Reader::Slices::serializeAndUnloadSlice(
		int sliceNumber) {
	Expects(_headerMode != HeaderMode::Unknown);
	Expects(_headerMode != HeaderMode::NoCache);
	Expects(sliceNumber >= 0 && sliceNumber <= _data.size());

	if (isGoodHeader() && (sliceNumber == 1)) {
		return serializeAndUnloadSlice(0);
	}
	const auto writeHeaderAndSlice = isGoodHeader() && !sliceNumber;

	auto &slice = sliceNumber ? _data[sliceNumber - 1] : _header;
	const auto count = slice.parts.size();
	Assert(count > 0);

	auto result = SerializedSlice();
	result.number = sliceNumber;

	// We always use complex serialization for header + first slice.
	const auto continuousTill = writeHeaderAndSlice
		? 0
		: FindNotLoadedStart(slice.parts, 0);
	const auto continuous = (continuousTill > slice.parts.back().first);
	if (continuous) {
		// All data is continuous.
		result.data.reserve(count * kPartSize);
		for (const auto &[offset, part] : slice.parts) {
			result.data.append(part);
		}
	} else {
		result.data = serializeComplexSlice(slice);
		if (writeHeaderAndSlice) {
			result.data.append(serializeAndUnloadFirstSliceNoHeader());
		}

		// Make sure this data won't be taken for full continuous data.
		const auto maxSize = maxSliceSize(sliceNumber);
		while (IsContiguousSerialization(result.data.size(), maxSize)) {
			result.data.push_back(char(0));
		}
	}

	// We may serialize header in the middle of streaming, if we use
	// HeaderMode::Good and we unload first slice. We still require
	// header data to continue working, so don't really unload the header.
	if (sliceNumber) {
		unloadSlice(slice);
	} else {
		slice.flags &= ~Slice::Flag::ChangedSinceCache;
	}
	return result;
}

void Reader::Slices::unloadSlice(Slice &slice) const {
	const auto full = (slice.flags & Slice::Flag::FullInCache);
	slice = Slice();
	if (full) {
		slice.flags |= Slice::Flag::FullInCache;
	}
}

QByteArray Reader::Slices::serializeComplexSlice(const Slice &slice) const {
	auto result = QByteArray();
	const auto count = slice.parts.size();
	const auto intSize = sizeof(int32);
	result.reserve(count * kPartSize + 2 * intSize * (count + 1));
	const auto appendInt = [&](int value) {
		auto serialized = int32(value);
		result.append(
			reinterpret_cast<const char*>(&serialized),
			intSize);
	};
	appendInt(count);
	for (const auto &[offset, part] : slice.parts) {
		appendInt(offset);
		appendInt(part.size());
		result.append(part);
	}
	return result;
}

QByteArray Reader::Slices::serializeAndUnloadFirstSliceNoHeader() {
	Expects(_data[0].flags & Slice::Flag::LoadedFromCache);

	auto &slice = _data[0];
	for (const auto &[offset, part] : _header.parts) {
		slice.parts.erase(offset);
	}
	auto result = serializeComplexSlice(slice);
	unloadSlice(slice);
	return result;
}

Reader::SerializedSlice Reader::Slices::unloadToCache() {
	if (_headerMode == HeaderMode::Unknown
		|| _headerMode == HeaderMode::NoCache) {
		return {};
	}
	if (_header.flags & Slice::Flag::ChangedSinceCache) {
		return serializeAndUnloadSlice(0);
	}
	for (auto i = 0, count = int(_data.size()); i != count; ++i) {
		if (_data[i].flags & Slice::Flag::ChangedSinceCache) {
			return serializeAndUnloadSlice(i + 1);
		}
	}
	return {};
}

Reader::Reader(
	std::unique_ptr<Loader> loader,
	Storage::Cache::Database *cache)
: _loader(std::move(loader))
, _cache(cache)
, _cacheHelper(cache ? InitCacheHelper(_loader->baseCacheKey()) : nullptr)
, _slices(_loader->size(), _cacheHelper != nullptr) {
	_loader->parts(
	) | rpl::start_with_next([=](LoadedPart &&part) {
		if (_attachedDownloader) {
			_partsForDownloader.fire_copy(part);
		}
		if (_streamingActive) {
			_loadedParts.emplace(std::move(part));
		}
		if (const auto waiting = _waiting.load(std::memory_order_acquire)) {
			_waiting.store(nullptr, std::memory_order_release);
			waiting->release();
		}
	}, _lifetime);

	if (_cacheHelper) {
		readFromCache(0);
	}
}

void Reader::startSleep(not_null<crl::semaphore*> wake) {
	_sleeping.store(wake, std::memory_order_release);
	processDownloaderRequests();
}

void Reader::wakeFromSleep() {
	if (const auto sleeping = _sleeping.load(std::memory_order_acquire)) {
		_sleeping.store(nullptr, std::memory_order_release);
		sleeping->release();
	}
}

void Reader::stopSleep() {
	_sleeping.store(nullptr, std::memory_order_release);
}

void Reader::stopStreamingAsync() {
	_stopStreamingAsync = true;
	crl::on_main(this, [=] {
		if (_stopStreamingAsync) {
			stopStreaming(false);
		}
	});
}

void Reader::tryRemoveLoaderAsync() {
	_loader->tryRemoveFromQueue();
}

void Reader::startStreaming() {
	_streamingActive = true;
	refreshLoaderPriority();
}

void Reader::stopStreaming(bool stillActive) {
	Expects(_sleeping == nullptr);

	_stopStreamingAsync = false;
	_waiting.store(nullptr, std::memory_order_release);
	if (!stillActive) {
		_streamingActive = false;
		refreshLoaderPriority();
		_loadingOffsets.clear();
		processDownloaderRequests();
	}
}

rpl::producer<LoadedPart> Reader::partsForDownloader() const {
	return _partsForDownloader.events();
}

void Reader::loadForDownloader(
		not_null<Storage::StreamedFileDownloader*> downloader,
		int offset) {
	if (_attachedDownloader != downloader) {
		if (_attachedDownloader) {
			cancelForDownloader(_attachedDownloader);
		}
		_attachedDownloader = downloader;
		_loader->attachDownloader(downloader);
	}
	_downloaderOffsetRequests.emplace(offset);
	if (_streamingActive) {
		wakeFromSleep();
	} else {
		processDownloaderRequests();
	}
}

void Reader::doneForDownloader(int offset) {
	_downloaderOffsetAcks.emplace(offset);
	if (!_streamingActive) {
		processDownloaderRequests();
	}
}

void Reader::cancelForDownloader(
		not_null<Storage::StreamedFileDownloader*> downloader) {
	if (_attachedDownloader == downloader) {
		_downloaderOffsetRequests.take();
		_attachedDownloader = nullptr;
		_loader->clearAttachedDownloader();
	}
}

void Reader::enqueueDownloaderOffsets() {
	auto offsets = _downloaderOffsetRequests.take();
	if (!empty(offsets)) {
		if (!empty(_offsetsForDownloader)) {
			_offsetsForDownloader.insert(
				end(_offsetsForDownloader),
				std::make_move_iterator(begin(offsets)),
				std::make_move_iterator(end(offsets)));
			checkForDownloaderChange(offsets.size() + 1);
		} else {
			_offsetsForDownloader = std::move(offsets);
			checkForDownloaderChange(offsets.size());
		}
	}
}

void Reader::checkForDownloaderChange(int checkItemsCount) {
	Expects(checkItemsCount <= _offsetsForDownloader.size());

	// If a requested offset is less-or-equal of some previously requested
	// offset, it means that the downloader was changed, ignore old offsets.
	const auto end = _offsetsForDownloader.end();
	const auto changed = std::adjacent_find(
		end - checkItemsCount,
		end,
		[](int first, int second) { return (second <= first); });
	if (changed != end) {
		_offsetsForDownloader.erase(
			begin(_offsetsForDownloader),
			changed + 1);
		_downloaderReadCache.clear();
		_downloaderOffsetAcks.take();
	}
}

void Reader::checkForDownloaderReadyOffsets() {
	// If a requested part is available right now we simply fire it on the
	// main thread, until the first not-available-right-now offset is found.
	const auto unavailableInBytes = [&](int offset, QByteArray &&bytes) {
		if (bytes.isEmpty()) {
			return true;
		}
		crl::on_main(this, [=, bytes = std::move(bytes)]() mutable {
			_partsForDownloader.fire({ offset, std::move(bytes) });
		});
		return false;
	};
	const auto unavailableInCache = [&](int offset) {
		const auto index = (offset / kInSlice);
		const auto sliceNumber = index + 1;
		const auto i = _downloaderReadCache.find(sliceNumber);
		if (i == end(_downloaderReadCache) || !i->second) {
			return true;
		}
		const auto j = i->second->find(offset - index * kInSlice);
		if (j == end(*i->second)) {
			return true;
		}
		return unavailableInBytes(offset, std::move(j->second));
	};
	const auto unavailable = [&](int offset) {
		return unavailableInBytes(offset, _slices.partForDownloader(offset))
			&& unavailableInCache(offset);
	};
	_offsetsForDownloader.erase(
		begin(_offsetsForDownloader),
		ranges::find_if(_offsetsForDownloader, unavailable));
}

void Reader::processDownloaderRequests() {
	processCacheResults();
	enqueueDownloaderOffsets();
	checkForDownloaderReadyOffsets();
	pruneDoneDownloaderRequests();
	if (!empty(_offsetsForDownloader)) {
		pruneDownloaderCache(_offsetsForDownloader.front());
		sendDownloaderRequests();
	}
}

void Reader::pruneDownloaderCache(int minimalOffset) {
	const auto minimalSliceNumber = (minimalOffset / kInSlice) + 1;
	const auto removeTill = ranges::lower_bound(
		_downloaderReadCache,
		minimalSliceNumber,
		ranges::less(),
		&base::flat_map<int, std::optional<PartsMap>>::value_type::first);
	_downloaderReadCache.erase(_downloaderReadCache.begin(), removeTill);
}

void Reader::pruneDoneDownloaderRequests() {
	for (const auto done : _downloaderOffsetAcks.take()) {
		_downloaderOffsetsRequested.remove(done);
		const auto i = ranges::find(_offsetsForDownloader, done);
		if (i != end(_offsetsForDownloader)) {
			_offsetsForDownloader.erase(i);
		}
	}
}

void Reader::sendDownloaderRequests() {
	auto &&offsets = ranges::view::all(
		_offsetsForDownloader
	) | ranges::view::take(kDownloaderRequestsLimit);
	for (const auto offset : offsets) {
		if ((!_cacheHelper || !downloaderWaitForCachedSlice(offset))
			&& _downloaderOffsetsRequested.emplace(offset).second) {
			_loader->load(offset);
		}
	}
}

bool Reader::downloaderWaitForCachedSlice(int offset) {
	if (_slices.waitingForHeaderCache()) {
		return true;
	}
	if (!_slices.readCacheForDownloaderRequired(offset)) {
		return false;
	}
	const auto sliceNumber = (offset / kInSlice) + 1;
	auto i = _downloaderReadCache.find(sliceNumber);
	if (i == _downloaderReadCache.end()) {
		// If we didn't request that slice yet, try requesting it.
		// If there is no need to (header mode is unknown) - place empty map.
		// Otherwise place std::nullopt and wait for the cache result.
		i = _downloaderReadCache.emplace(
			sliceNumber,
			(readFromCacheForDownloader(sliceNumber)
				? std::nullopt
				: std::make_optional(PartsMap()))).first;
	}
	return !i->second;
}

void Reader::checkCacheResultsForDownloader() {
	if (_streamingActive) {
		return;
	}
	processDownloaderRequests();
}

void Reader::setLoaderPriority(int priority) {
	if (_realPriority == priority) {
		return;
	}
	_realPriority = priority;
	refreshLoaderPriority();
}

void Reader::refreshLoaderPriority() {
	_loader->setPriority(_streamingActive ? _realPriority : 0);
}

bool Reader::isRemoteLoader() const {
	return _loader->baseCacheKey().valid();
}

std::shared_ptr<Reader::CacheHelper> Reader::InitCacheHelper(
		Storage::Cache::Key baseKey) {
	if (!baseKey) {
		return nullptr;
	}
	return std::make_shared<Reader::CacheHelper>(baseKey);
}

// 0 is for headerData, slice index = sliceNumber - 1.
void Reader::readFromCache(int sliceNumber) {
	Expects(_cache != nullptr);
	Expects(_cacheHelper != nullptr);
	Expects(!sliceNumber || !_slices.headerModeUnknown());

	if (sliceNumber == 1 && _slices.isGoodHeader()) {
		return readFromCache(0);
	}
	const auto size = _loader->size();
	const auto key = _cacheHelper->key(sliceNumber);
	const auto cache = std::weak_ptr<CacheHelper>(_cacheHelper);
	const auto weak = base::make_weak(this);
	const auto ready = [=](
			QByteArray &&result,
			std::vector<int> &&sizes = {}) {
		crl::async([
			=,
			result = std::move(result),
			sizes = std::move(sizes)
		]() mutable{
			auto entry = ParseCacheEntry(
				bytes::make_span(result),
				sliceNumber,
				size);
			if (const auto strong = cache.lock()) {
				QMutexLocker lock(&strong->mutex);
				strong->results.emplace(sliceNumber, std::move(entry.parts));
				if (!sliceNumber && entry.included) {
					strong->results.emplace(1, std::move(*entry.included));
				}
				strong->sizes = std::move(sizes);
				if (const auto waiting = strong->waiting.load()) {
					strong->waiting.store(nullptr, std::memory_order_release);
					waiting->release();
				} else {
					crl::on_main(weak, [=] {
						checkCacheResultsForDownloader();
					});
				}
			}
		});
	};
	auto keys = std::vector<Storage::Cache::Key>();
	const auto count = _slices.requestSliceSizesCount();
	for (auto i = 0; i != count; ++i) {
		keys.push_back(_cacheHelper->key(i + 1));
	}
	_cache->getWithSizes(key, std::move(keys), ready);
}

bool Reader::readFromCacheForDownloader(int sliceNumber) {
	Expects(_cacheHelper != nullptr);
	Expects(sliceNumber > 0);

	if (_slices.headerModeUnknown()) {
		return false;
	}
	readFromCache(sliceNumber);
	return true;
}

void Reader::putToCache(SerializedSlice &&slice) {
	Expects(_cache != nullptr);
	Expects(_cacheHelper != nullptr);
	Expects(slice.number >= 0);

	_cache->put(_cacheHelper->key(slice.number), std::move(slice.data));
}

int Reader::size() const {
	return _loader->size();
}

std::optional<Error> Reader::streamingError() const {
	return _streamingError;
}

void Reader::headerDone() {
	_slices.headerDone(false);
}

int Reader::headerSize() const {
	return _slices.headerSize();
}

bool Reader::fullInCache() const {
	return _slices.fullInCache();
}

bool Reader::fill(
		int offset,
		bytes::span buffer,
		not_null<crl::semaphore*> notify) {
	Expects(offset + buffer.size() <= size());

	const auto startWaiting = [&] {
		if (_cacheHelper) {
			_cacheHelper->waiting = notify.get();
		}
		_waiting.store(notify.get(), std::memory_order_release);
	};
	const auto clearWaiting = [&] {
		_waiting.store(nullptr, std::memory_order_release);
		if (_cacheHelper) {
			_cacheHelper->waiting.store(nullptr, std::memory_order_release);
		}
	};
	const auto done = [&] {
		clearWaiting();
		return true;
	};
	const auto failed = [&] {
		clearWaiting();
		notify->release();
		return false;
	};

	checkForSomethingMoreReceived();
	if (_streamingError) {
		return failed();
	}

	do {
		if (fillFromSlices(offset, buffer)) {
			clearWaiting();
			return true;
		}
		startWaiting();
	} while (checkForSomethingMoreReceived());

	return _streamingError ? failed() : false;
}

bool Reader::fillFromSlices(int offset, bytes::span buffer) {
	using namespace rpl::mappers;

	auto result = _slices.fill(offset, buffer);
	if (!result.filled && _slices.headerWontBeFilled()) {
		_streamingError = Error::NotStreamable;
		return false;
	}

	for (const auto sliceNumber : result.sliceNumbersFromCache.values()) {
		readFromCache(sliceNumber);
	}

	if (_cacheHelper && result.toCache.number >= 0) {
		// If we put to cache the header (number == 0) that means we're in
		// HeaderMode::Good and really are putting the first slice to cache.
		Assert(result.toCache.number > 0 || _slices.isGoodHeader());

		const auto index = std::max(result.toCache.number, 1) - 1;
		cancelLoadInRange(index * kInSlice, (index + 1) * kInSlice);
		putToCache(std::move(result.toCache));
	}
	auto checkPriority = true;
	for (const auto offset : result.offsetsFromLoader.values()) {
		if (checkPriority) {
			checkLoadWillBeFirst(offset);
			checkPriority = false;
		}
		loadAtOffset(offset);
	}
	return result.filled;
}

void Reader::cancelLoadInRange(int from, int till) {
	Expects(from < till);

	for (const auto offset : _loadingOffsets.takeInRange(from, till)) {
		if (!_downloaderOffsetsRequested.contains(offset)) {
			_loader->cancel(offset);
		}
	}
}

void Reader::checkLoadWillBeFirst(int offset) {
	if (_loadingOffsets.front().value_or(offset) != offset) {
		_loadingOffsets.resetPriorities();
		_loader->resetPriorities();
	}
}

bool Reader::processCacheResults() {
	if (!_cacheHelper) {
		return false;
	}

	QMutexLocker lock(&_cacheHelper->mutex);
	auto loaded = base::take(_cacheHelper->results);
	auto sizes = base::take(_cacheHelper->sizes);
	lock.unlock();

	for (auto &[sliceNumber, cachedParts] : _downloaderReadCache) {
		if (!cachedParts) {
			const auto i = loaded.find(sliceNumber);
			if (i != end(loaded)) {
				cachedParts = i->second;
			}
		}
	}

	if (_streamingError) {
		return false;
	}
	for (auto &[sliceNumber, result] : loaded) {
		_slices.processCacheResult(sliceNumber, std::move(result));
	}
	if (!sizes.empty()) {
		_slices.processCachedSizes(sizes);
	}
	if (!loaded.empty()
		&& (loaded.front().first == 0)
		&& _slices.isGoodHeader()) {
		Assert(loaded.size() > 1);
		Assert((loaded.begin() + 1)->first == 1);
	}
	return !loaded.empty();
}

bool Reader::processLoadedParts() {
	if (_streamingError) {
		return false;
	}

	auto loaded = _loadedParts.take();
	for (auto &part : loaded) {
		if (!part.valid(size())) {
			_streamingError = Error::LoadFailed;
			return false;
		} else if (!_loadingOffsets.remove(part.offset)) {
			continue;
		}
		_slices.processPart(
			part.offset,
			std::move(part.bytes));
	}
	return !loaded.empty();
}

bool Reader::checkForSomethingMoreReceived() {
	const auto result1 = processCacheResults();
	const auto result2 = processLoadedParts();
	return result1 || result2;
}

void Reader::loadAtOffset(int offset) {
	if (_loadingOffsets.add(offset)) {
		_loader->load(offset);
	}
}

void Reader::finalizeCache() {
	if (!_cacheHelper) {
		return;
	}
	Assert(_cache != nullptr);
	if (_cacheHelper->waiting != nullptr) {
		QMutexLocker lock(&_cacheHelper->mutex);
		_cacheHelper->waiting.store(nullptr, std::memory_order_release);
	}
	auto toCache = _slices.unloadToCache();
	while (toCache.number >= 0) {
		putToCache(std::move(toCache));
		toCache = _slices.unloadToCache();
	}
	_cache->sync();
}

Reader::~Reader() {
	finalizeCache();
}

} // namespace Streaming
} // namespace Media