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

1399 lines
36 KiB
C++

/*
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