Working code for streaming downloader.

Telegram/SourceFiles/data/data_document.cpp

@@ -871,9 +871,8 @@ void DocumentData::save(
 				id,
 				_dc,
 				origin,
-				(saveToCache()
-					? std::make_optional(Data::DocumentCacheKey(_dc, id))
-					: std::nullopt),
+				Data::DocumentCacheKey(_dc, id),
+				mediaKey(),
 				std::move(reader),
 				toFile,
 				size,

Telegram/SourceFiles/media/streaming/media_streaming_file.cpp

@@ -55,7 +55,7 @@ int File::Context::read(bytes::span buffer) {
 		_semaphore.acquire();
 		if (_interrupted) {
 			return -1;
-		} else if (const auto error = _reader->failed()) {
+		} else if (const auto error = _reader->streamingError()) {
 			fail(*error);
 			return -1;
 		}

Telegram/SourceFiles/media/streaming/media_streaming_reader.cpp

@@ -596,25 +596,9 @@ std::optional<int> Reader::Slices::readCacheRequiredFor(int offset) {
 	}
 	const auto index = offset / kInSlice;
 	auto &slice = _data[index];
-	if (slice.flags & (Flag::LoadedFromCache | Flag::LoadingFromCache)) {
-		return std::nullopt;
-	}
-	slice.flags |= Flag::LoadingFromCache;
-	return (index + 1);
-}
-
-bool Reader::Slices::waitForCacheRequiredFor(int offset) const {
-	Expects(offset < _size);
-
-	using Flag = Slice::Flag;
-	if (_header.flags & Flag::LoadingFromCache) {
-		return true;
-	} else if (isFullInHeader()) {
-		return false;
-	}
-	const auto index = offset / kInSlice;
-	const auto &slice = _data[index];
-	return (slice.flags & Flag::LoadingFromCache);
+	return (slice.flags & Flag::LoadedFromCache)
+		? std::nullopt
+		: std::make_optional(index + 1);
 }
 
 void Reader::Slices::markSliceUsed(int sliceIndex) {
@@ -841,6 +825,12 @@ void Reader::loadForDownloader(int offset) {
 	}
 }
 
+void Reader::doneForDownloader(int offset) {
+	if (_downloaderOffsetsRequested.remove(offset) && !_streamingActive) {
+		processDownloaderRequests();
+	}
+}
+
 void Reader::cancelForDownloader() {
 	if (_downloaderAttached.load(std::memory_order_acquire)) {
 		_downloaderOffsetRequests.take();
@@ -878,19 +868,33 @@ void Reader::checkForDownloaderChange(int checkItemsCount) {
 		_offsetsForDownloader.erase(
 			begin(_offsetsForDownloader),
 			changed + 1);
+		_downloaderSliceNumber = 0;
+		_downloaderSliceCache = std::nullopt;
 	}
 }
 
 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 ready = [&](int offset, QByteArray &&bytes) {
+		crl::on_main(this, [=, bytes = std::move(bytes)]() mutable {
+			_partsForDownloader.fire({ offset, std::move(bytes) });
+		});
+		return true;
+	};
 	const auto unavailable = [&](int offset) {
 		auto bytes = _slices.partForDownloader(offset);
 		if (!bytes.isEmpty()) {
-			crl::on_main(this, [=, bytes = std::move(bytes)]() mutable {
-				_partsForDownloader.fire({ offset, std::move(bytes) });
-			});
-			return false;
+			return !ready(offset, std::move(bytes));
+		}
+		const auto sliceIndex = (offset / kInSlice);
+		if ((sliceIndex + 1 == _downloaderSliceNumber)
+			&& _downloaderSliceCache) {
+			const auto i = _downloaderSliceCache->find(
+				offset - sliceIndex * kInSlice);
+			if (i != _downloaderSliceCache->end()) {
+				return !ready(offset, std::move(i->second));
+			}
 		}
 		return true;
 	};
@@ -900,7 +904,7 @@ void Reader::checkForDownloaderReadyOffsets() {
 }
 
 void Reader::processDownloaderRequests() {
-	checkForSomethingMoreReceived();
+	processCacheResults();
 	enqueueDownloaderOffsets();
 	checkForDownloaderReadyOffsets();
 	if (empty(_offsetsForDownloader)) {
@@ -908,17 +912,31 @@ void Reader::processDownloaderRequests() {
 	}
 
 	const auto offset = _offsetsForDownloader.front();
-	if (_cacheHelper) {
-		if (const auto sliceNumber = _slices.readCacheRequiredFor(offset)) {
-			readFromCache(*sliceNumber);
-			return;
-		} else if (_slices.waitForCacheRequiredFor(offset)) {
-			return;
-		}
+	if (_cacheHelper && downloaderWaitForCachedSlice(offset)) {
+		return;
 	}
 
 	_offsetsForDownloader.pop_front();
-	loadAtOffset(offset);
+	if (_downloaderOffsetsRequested.emplace(offset).second) {
+		_loader->load(offset);
+	}
+}
+
+bool Reader::downloaderWaitForCachedSlice(int offset) {
+	const auto sliceNumber = _slices.readCacheRequiredFor(offset);
+	if (sliceNumber.value_or(0) != _downloaderSliceNumber) {
+		_downloaderSliceNumber = sliceNumber.value_or(0);
+		_downloaderSliceCache = std::nullopt;
+		if (_downloaderSliceNumber) {
+			if (readFromCacheForDownloader()) {
+				return true;
+			}
+			_downloaderSliceCache = PartsMap();
+		}
+	} else if (_downloaderSliceNumber && !_downloaderSliceCache) {
+		return true;
+	}
+	return false;
 }
 
 void Reader::checkCacheResultsForDownloader() {
@@ -977,6 +995,17 @@ void Reader::readFromCache(int sliceNumber) {
 	});
 }
 
+bool Reader::readFromCacheForDownloader() {
+	Expects(_cacheHelper != nullptr);
+	Expects(_downloaderSliceNumber > 0);
+
+	if (_slices.headerModeUnknown()) {
+		return false;
+	}
+	readFromCache(_downloaderSliceNumber);
+	return true;
+}
+
 void Reader::putToCache(SerializedSlice &&slice) {
 	Expects(_cacheHelper != nullptr);
 	Expects(slice.number >= 0);
@@ -990,8 +1019,8 @@ int Reader::size() const {
 	return _loader->size();
 }
 
-std::optional<Error> Reader::failed() const {
-	return _failed;
+std::optional<Error> Reader::streamingError() const {
+	return _streamingError;
 }
 
 void Reader::headerDone() {
@@ -1027,7 +1056,7 @@ bool Reader::fill(
 	};
 
 	checkForSomethingMoreReceived();
-	if (_failed) {
+	if (_streamingError) {
 		return failed();
 	}
 
@@ -1039,7 +1068,7 @@ bool Reader::fill(
 		startWaiting();
 	} while (checkForSomethingMoreReceived());
 
-	return _failed ? failed() : false;
+	return _streamingError ? failed() : false;
 }
 
 bool Reader::fillFromSlices(int offset, bytes::span buffer) {
@@ -1047,7 +1076,7 @@ bool Reader::fillFromSlices(int offset, bytes::span buffer) {
 
 	auto result = _slices.fill(offset, buffer);
 	if (!result.filled && _slices.headerWontBeFilled()) {
-		_failed = Error::NotStreamable;
+		_streamingError = Error::NotStreamable;
 		return false;
 	}
 
@@ -1079,7 +1108,9 @@ void Reader::cancelLoadInRange(int from, int till) {
 	Expects(from < till);
 
 	for (const auto offset : _loadingOffsets.takeInRange(from, till)) {
-		_loader->cancel(offset);
+		if (!_downloaderOffsetsRequested.contains(offset)) {
+			_loader->cancel(offset);
+		}
 	}
 }
 
@@ -1093,14 +1124,22 @@ void Reader::checkLoadWillBeFirst(int offset) {
 bool Reader::processCacheResults() {
 	if (!_cacheHelper) {
 		return false;
-	} else if (_failed) {
-		return false;
 	}
 
 	QMutexLocker lock(&_cacheHelper->mutex);
 	auto loaded = base::take(_cacheHelper->results);
 	lock.unlock();
 
+	if (_downloaderSliceNumber) {
+		const auto i = loaded.find(_downloaderSliceNumber);
+		if (i != end(loaded)) {
+			_downloaderSliceCache = i->second;
+		}
+	}
+
+	if (_streamingError) {
+		return false;
+	}
 	for (auto &[sliceNumber, result] : loaded) {
 		_slices.processCacheResult(sliceNumber, std::move(result));
 	}
@@ -1114,14 +1153,14 @@ bool Reader::processCacheResults() {
 }
 
 bool Reader::processLoadedParts() {
-	if (_failed) {
+	if (_streamingError) {
 		return false;
 	}
 
 	auto loaded = _loadedParts.take();
 	for (auto &part : loaded) {
 		if (!part.valid(size())) {
-			_failed = Error::LoadFailed;
+			_streamingError = Error::LoadFailed;
 			return false;
 		} else if (!_loadingOffsets.remove(part.offset)) {
 			continue;

Telegram/SourceFiles/media/streaming/media_streaming_reader.h

@@ -43,7 +43,7 @@ public:
 		int offset,
 		bytes::span buffer,
 		not_null<crl::semaphore*> notify);
-	[[nodiscard]] std::optional<Error> failed() const;
+	[[nodiscard]] std::optional<Error> streamingError() const;
 	void headerDone();
 
 	// Thread safe.
@@ -56,6 +56,7 @@ public:
 	void stopStreaming(bool stillActive = false);
 	[[nodiscard]] rpl::producer<LoadedPart> partsForDownloader() const;
 	void loadForDownloader(int offset);
+	void doneForDownloader(int offset);
 	void cancelForDownloader();
 
 	~Reader();
@@ -139,7 +140,6 @@ private:
 
 		[[nodiscard]] QByteArray partForDownloader(int offset) const;
 		[[nodiscard]] std::optional<int> readCacheRequiredFor(int offset);
-		[[nodiscard]] bool waitForCacheRequiredFor(int offset) const;
 
 	private:
 		enum class HeaderMode {
@@ -173,7 +173,9 @@ private:
 	};
 
 	// 0 is for headerData, slice index = sliceNumber - 1.
+	// returns false if asked for a known-empty downloader slice cache.
 	void readFromCache(int sliceNumber);
+	[[nodiscard]] bool readFromCacheForDownloader();
 	bool processCacheResults();
 	void putToCache(SerializedSlice &&data);
 
@@ -190,6 +192,7 @@ private:
 
 	void processDownloaderRequests();
 	void checkCacheResultsForDownloader();
+	[[nodiscard]] bool downloaderWaitForCachedSlice(int offset);
 	void enqueueDownloaderOffsets();
 	void checkForDownloaderChange(int checkItemsCount);
 	void checkForDownloaderReadyOffsets();
@@ -207,11 +210,16 @@ private:
 	PriorityQueue _loadingOffsets;
 
 	Slices _slices;
-	std::optional<Error> _failed;
+
+	// Even if streaming had failed, the Reader can work for the downloader.
+	std::optional<Error> _streamingError;
 
 	std::atomic<bool> _downloaderAttached = false;
 	base::thread_safe_queue<int> _downloaderOffsetRequests;
 	std::deque<int> _offsetsForDownloader;
+	base::flat_set<int> _downloaderOffsetsRequested;
+	int _downloaderSliceNumber = 0; // > 0 means we want it from cache.
+	std::optional<PartsMap> _downloaderSliceCache;
 
 	// Main thread.
 	rpl::event_stream<LoadedPart> _partsForDownloader;

Telegram/SourceFiles/storage/file_download.cpp

@@ -473,8 +473,8 @@ bool FileLoader::tryLoadLocal() {
 	}
 
 	const auto weak = make_weak(this);
-	if (const auto key = cacheKey()) {
-		loadLocal(*key);
+	if (_toCache == LoadToCacheAsWell) {
+		loadLocal(cacheKey());
 		emit progress(this);
 	}
 	if (!weak) {
@@ -594,17 +594,18 @@ bool FileLoader::finalizeResult() {
 
 	if (_localStatus == LocalStatus::NotFound) {
 		if (const auto key = fileLocationKey()) {
-			Local::writeFileLocation(*key, FileLocation(_filename));
-		}
-		if (const auto key = cacheKey()) {
-			if (_data.size() <= Storage::kMaxFileInMemory) {
-				Auth().data().cache().put(
-					*key,
-					Storage::Cache::Database::TaggedValue(
-						base::duplicate(_data),
-						_cacheTag));
+			if (!_filename.isEmpty()) {
+				Local::writeFileLocation(*key, FileLocation(_filename));
 			}
 		}
+		if ((_toCache == LoadToCacheAsWell)
+			&& (_data.size() <= Storage::kMaxFileInMemory)) {
+			Auth().data().cache().put(
+				cacheKey(),
+				Storage::Cache::Database::TaggedValue(
+					base::duplicate(_data),
+					_cacheTag));
+		}
 	}
 	_downloader->taskFinished().notify();
 	return true;
@@ -1161,20 +1162,18 @@ void mtpFileLoader::changeCDNParams(
 	makeRequest(offset);
 }
 
-std::optional<Storage::Cache::Key> mtpFileLoader::cacheKey() const {
+Storage::Cache::Key mtpFileLoader::cacheKey() const {
 	return _location.match([&](const WebFileLocation &location) {
-		return std::make_optional(Data::WebDocumentCacheKey(location));
+		return Data::WebDocumentCacheKey(location);
 	}, [&](const GeoPointLocation &location) {
-		return std::make_optional(Data::GeoPointCacheKey(location));
+		return Data::GeoPointCacheKey(location);
 	}, [&](const StorageFileLocation &location) {
-		return (_toCache == LoadToCacheAsWell)
-			? std::make_optional(location.cacheKey())
-			: std::nullopt;
+		return location.cacheKey();
 	});
 }
 
 std::optional<MediaKey> mtpFileLoader::fileLocationKey() const {
-	if (_locationType != UnknownFileLocation && !_filename.isEmpty()) {
+	if (_locationType != UnknownFileLocation) {
 		return mediaKey(_locationType, dcId(), objId());
 	}
 	return std::nullopt;
@@ -1246,7 +1245,7 @@ void webFileLoader::loadError() {
 	cancel(true);
 }
 
-std::optional<Storage::Cache::Key> webFileLoader::cacheKey() const {
+Storage::Cache::Key webFileLoader::cacheKey() const {
 	return Data::UrlCacheKey(_url);
 }
 

Telegram/SourceFiles/storage/file_download.h

@@ -172,7 +172,7 @@ protected:
 
 	bool tryLoadLocal();
 	void loadLocal(const Storage::Cache::Key &key);
-	virtual std::optional<Storage::Cache::Key> cacheKey() const = 0;
+	virtual Storage::Cache::Key cacheKey() const = 0;
 	virtual std::optional<MediaKey> fileLocationKey() const = 0;
 	virtual void cancelRequests() = 0;
 
@@ -273,7 +273,7 @@ private:
 		int limit = 0;
 		QByteArray hash;
 	};
-	std::optional<Storage::Cache::Key> cacheKey() const override;
+	Storage::Cache::Key cacheKey() const override;
 	std::optional<MediaKey> fileLocationKey() const override;
 	void cancelRequests() override;
 
@@ -358,7 +358,7 @@ public:
 
 protected:
 	void cancelRequests() override;
-	std::optional<Storage::Cache::Key> cacheKey() const override;
+	Storage::Cache::Key cacheKey() const override;
 	std::optional<MediaKey> fileLocationKey() const override;
 	bool loadPart() override;
 

Telegram/SourceFiles/storage/streamed_file_downloader.cpp

@@ -23,7 +23,8 @@ StreamedFileDownloader::StreamedFileDownloader(
 	uint64 objectId,
 	MTP::DcId dcId,
 	Data::FileOrigin origin,
-	std::optional<Cache::Key> cacheKey,
+	Cache::Key cacheKey,
+	MediaKey fileLocationKey,
 	std::shared_ptr<Reader> reader,
 
 	// For FileLoader
@@ -45,6 +46,7 @@ StreamedFileDownloader::StreamedFileDownloader(
 , _objectId(objectId)
 , _origin(origin)
 , _cacheKey(cacheKey)
+, _fileLocationKey(fileLocationKey)
 , _reader(std::move(reader)) {
 	_partIsSaved.resize((size + kPartSize - 1) / kPartSize, false);
 
@@ -76,12 +78,12 @@ void StreamedFileDownloader::stop() {
 	cancelRequests();
 }
 
-std::optional<Storage::Cache::Key> StreamedFileDownloader::cacheKey() const {
+Storage::Cache::Key StreamedFileDownloader::cacheKey() const {
 	return _cacheKey;
 }
 
 std::optional<MediaKey> StreamedFileDownloader::fileLocationKey() const {
-	return std::nullopt; AssertIsDebug();
+	return _fileLocationKey;
 }
 
 void StreamedFileDownloader::cancelRequests() {
@@ -109,7 +111,7 @@ bool StreamedFileDownloader::loadPart() {
 		return false;
 	}
 	_nextPartIndex = index + 1;
-	_reader->loadForDownloader(index);
+	_reader->loadForDownloader(index * kPartSize);
 	AssertIsDebug();
 	//_downloader->requestedAmountIncrement(
 	//	requestData.dcId,
@@ -123,11 +125,13 @@ bool StreamedFileDownloader::loadPart() {
 void StreamedFileDownloader::savePart(const LoadedPart &part) {
 	Expects(part.offset >= 0 && part.offset < _reader->size());
 	Expects(part.offset % kPartSize == 0);
+
 	if (_finished || _cancelled) {
 		return;
 	}
 
-	const auto index = part.offset / kPartSize;
+	const auto offset = part.offset;
+	const auto index = offset / kPartSize;
 	Assert(index >= 0 && index < _partIsSaved.size());
 	if (_partIsSaved[index]) {
 		return;
@@ -142,7 +146,7 @@ void StreamedFileDownloader::savePart(const LoadedPart &part) {
 		//	-kPartSize);
 		--_queue->queriesCount;
 	}
-	if (!writeResultPart(part.offset, bytes::make_span(part.bytes))) {
+	if (!writeResultPart(offset, bytes::make_span(part.bytes))) {
 		return;
 	}
 	if (ranges::find(_partIsSaved, false) == end(_partIsSaved)) {
@@ -150,6 +154,7 @@ void StreamedFileDownloader::savePart(const LoadedPart &part) {
 			return;
 		}
 	}
+	_reader->doneForDownloader(offset);
 	notifyAboutProgress();
 }
 

Telegram/SourceFiles/storage/streamed_file_downloader.h

@@ -25,7 +25,8 @@ public:
 		uint64 objectId,
 		MTP::DcId dcId,
 		Data::FileOrigin origin,
-		std::optional<Cache::Key> cacheKey,
+		Cache::Key cacheKey,
+		MediaKey fileLocationKey,
 		std::shared_ptr<Media::Streaming::Reader> reader,
 
 		// For FileLoader
@@ -43,7 +44,7 @@ public:
 	void stop() override;
 
 private:
-	std::optional<Storage::Cache::Key> cacheKey() const override;
+	Cache::Key cacheKey() const override;
 	std::optional<MediaKey> fileLocationKey() const override;
 	void cancelRequests() override;
 	bool loadPart() override;
@@ -52,7 +53,8 @@ private:
 
 	uint64 _objectId = 0;
 	Data::FileOrigin _origin;
-	std::optional<Cache::Key> _cacheKey;
+	Cache::Key _cacheKey;
+	MediaKey _fileLocationKey;
 	std::shared_ptr<Media::Streaming::Reader> _reader;
 
 	std::vector<bool> _partIsSaved; // vector<bool> :D