RepoMirrors/tdesktop
1
0
Fork 0
mirror of https://github.com/telegramdesktop/tdesktop synced 2025-04-01 23:00:58 +00:00
Code Issues Projects Releases Wiki Activity

Allow variable count of download sessions.

Browse Source
This commit is contained in:
John Preston 2019-12-04 09:51:21 +03:00
parent cb2c6e4b89
commit 3ae2986c25
11 changed files with 32 additions and 25 deletions

2
Telegram/SourceFiles/media/streaming/media_streaming_loader.h
View File

@ -40,7 +40,7 @@ public:
[[nodiscard]] virtual rpl::producer<LoadedPart> parts() const = 0; [[nodiscard]] virtual rpl::producer<LoadedPart> parts() const = 0;
virtual void attachDownloader( virtual void attachDownloader(
Storage::StreamedFileDownloader *downloader) = 0; not_null<Storage::StreamedFileDownloader*> downloader) = 0;
virtual void clearAttachedDownloader() = 0; virtual void clearAttachedDownloader() = 0;
virtual ~Loader() = default; virtual ~Loader() = default;

2
Telegram/SourceFiles/media/streaming/media_streaming_loader_local.cpp
View File

@ -79,7 +79,7 @@ rpl::producer<LoadedPart> LoaderLocal::parts() const {
} }
void LoaderLocal::attachDownloader( void LoaderLocal::attachDownloader(
Storage::StreamedFileDownloader *downloader) { not_null<Storage::StreamedFileDownloader*> downloader) {
Unexpected("Downloader attached to a local streaming loader."); Unexpected("Downloader attached to a local streaming loader.");
} }

2
Telegram/SourceFiles/media/streaming/media_streaming_loader_local.h
View File

@ -33,7 +33,7 @@ public:
[[nodiscard]] rpl::producer<LoadedPart> parts() const override; [[nodiscard]] rpl::producer<LoadedPart> parts() const override;
void attachDownloader( void attachDownloader(
Storage::StreamedFileDownloader *downloader) override; not_null<Storage::StreamedFileDownloader*> downloader) override;
void clearAttachedDownloader() override; void clearAttachedDownloader() override;
private: private:

2
Telegram/SourceFiles/media/streaming/media_streaming_loader_mtproto.cpp
View File

@ -91,7 +91,7 @@ void LoaderMtproto::cancelForOffset(int offset) {
} }
void LoaderMtproto::attachDownloader( void LoaderMtproto::attachDownloader(
Storage::StreamedFileDownloader *downloader) { not_null<Storage::StreamedFileDownloader*> downloader) {
_downloader = downloader; _downloader = downloader;
} }

2
Telegram/SourceFiles/media/streaming/media_streaming_loader_mtproto.h
View File

@ -40,7 +40,7 @@ public:
[[nodiscard]] rpl::producer<LoadedPart> parts() const override; [[nodiscard]] rpl::producer<LoadedPart> parts() const override;
void attachDownloader( void attachDownloader(
Storage::StreamedFileDownloader *downloader) override; not_null<Storage::StreamedFileDownloader*> downloader) override;
void clearAttachedDownloader() override; void clearAttachedDownloader() override;
private: private:

4
Telegram/SourceFiles/media/streaming/media_streaming_reader.cpp
View File

@ -906,7 +906,7 @@ rpl::producer<LoadedPart> Reader::partsForDownloader() const {
} }
void Reader::loadForDownloader( void Reader::loadForDownloader(
Storage::StreamedFileDownloader *downloader, not_null<Storage::StreamedFileDownloader*> downloader,
int offset) { int offset) {
if (_attachedDownloader != downloader) { if (_attachedDownloader != downloader) {
if (_attachedDownloader) { if (_attachedDownloader) {
@ -931,7 +931,7 @@ void Reader::doneForDownloader(int offset) {
} }
void Reader::cancelForDownloader( void Reader::cancelForDownloader(
Storage::StreamedFileDownloader *downloader) { not_null<Storage::StreamedFileDownloader*> downloader) {
if (_attachedDownloader == downloader) { if (_attachedDownloader == downloader) {
_downloaderOffsetRequests.take(); _downloaderOffsetRequests.take();
_attachedDownloader = nullptr; _attachedDownloader = nullptr;

5
Telegram/SourceFiles/media/streaming/media_streaming_reader.h
View File

@ -61,10 +61,11 @@ public:
void stopStreaming(bool stillActive = false); void stopStreaming(bool stillActive = false);
[[nodiscard]] rpl::producer<LoadedPart> partsForDownloader() const; [[nodiscard]] rpl::producer<LoadedPart> partsForDownloader() const;
void loadForDownloader( void loadForDownloader(
Storage::StreamedFileDownloader *downloader, not_null<Storage::StreamedFileDownloader*> downloader,
int offset); int offset);
void doneForDownloader(int offset); void doneForDownloader(int offset);
void cancelForDownloader(Storage::StreamedFileDownloader *downloader); void cancelForDownloader(
not_null<Storage::StreamedFileDownloader*> downloader);
~Reader(); ~Reader();

7
Telegram/SourceFiles/mtproto/facade.h
View File

@ -34,13 +34,13 @@ constexpr ShiftedDcId updaterDcId(DcId dcId) {
return ShiftDcId(dcId, kUpdaterDcShift); return ShiftDcId(dcId, kUpdaterDcShift);
} }
constexpr auto kDownloadSessionsCount = 2;
constexpr auto kUploadSessionsCount = 2; constexpr auto kUploadSessionsCount = 2;
namespace details { namespace details {
constexpr ShiftedDcId downloadDcId(DcId dcId, int index) { constexpr ShiftedDcId downloadDcId(DcId dcId, int index) {
static_assert(kDownloadSessionsCount < kMaxMediaDcCount, "Too large MTPDownloadSessionsCount!"); Expects(index < kMaxMediaDcCount);
return ShiftDcId(dcId, kBaseDownloadDcShift + index); return ShiftDcId(dcId, kBaseDownloadDcShift + index);
}; };
@ -48,13 +48,12 @@ constexpr ShiftedDcId downloadDcId(DcId dcId, int index) {
// send(req, callbacks, MTP::downloadDcId(dc, index)) - for download shifted dc id // send(req, callbacks, MTP::downloadDcId(dc, index)) - for download shifted dc id
inline ShiftedDcId downloadDcId(DcId dcId, int index) { inline ShiftedDcId downloadDcId(DcId dcId, int index) {
Expects(index >= 0 && index < kDownloadSessionsCount);
return details::downloadDcId(dcId, index); return details::downloadDcId(dcId, index);
} }
inline constexpr bool isDownloadDcId(ShiftedDcId shiftedDcId) { inline constexpr bool isDownloadDcId(ShiftedDcId shiftedDcId) {
return (shiftedDcId >= details::downloadDcId(0, 0)) return (shiftedDcId >= details::downloadDcId(0, 0))
&& (shiftedDcId < details::downloadDcId(0, kDownloadSessionsCount - 1) + kDcShift); && (shiftedDcId < details::downloadDcId(0, kMaxMediaDcCount - 1) + kDcShift);
} }
inline bool isCdnDc(MTPDdcOption::Flags flags) { inline bool isCdnDc(MTPDdcOption::Flags flags) {

2
Telegram/SourceFiles/mtproto/session_private.cpp
View File

@ -1033,8 +1033,6 @@ void SessionPrivate::onSentSome(uint64 size) {
} }
if (isUploadDcId(_shiftedDcId)) { if (isUploadDcId(_shiftedDcId)) {
remain *= kUploadSessionsCount; remain *= kUploadSessionsCount;
} else if (isDownloadDcId(_shiftedDcId)) {
remain *= kDownloadSessionsCount;
} }
_waitForReceivedTimer.callOnce(remain); _waitForReceivedTimer.callOnce(remain);
} }

24
Telegram/SourceFiles/storage/file_download.cpp
View File

@ -43,6 +43,9 @@ constexpr auto kMaxWebFileQueries = 8;
// fixed part size download for hash checking. // fixed part size download for hash checking.
constexpr auto kPartSize = 128 * 1024; constexpr auto kPartSize = 128 * 1024;
constexpr auto kStartSessionsCount = 1;
constexpr auto kMaxSessionsCount = 8;
} // namespace } // namespace
void DownloadManager::Queue::enqueue(not_null<Downloader*> loader) { void DownloadManager::Queue::enqueue(not_null<Downloader*> loader) {
@ -121,6 +124,7 @@ void DownloadManager::checkSendNext() {
const auto bestIndex = [&] { const auto bestIndex = [&] {
const auto i = _requestedBytesAmount.find(dcId); const auto i = _requestedBytesAmount.find(dcId);
if (i == end(_requestedBytesAmount)) { if (i == end(_requestedBytesAmount)) {
_requestedBytesAmount[dcId].resize(kStartSessionsCount);
return 0; return 0;
} }
const auto j = ranges::min_element(i->second); const auto j = ranges::min_element(i->second);
@ -136,6 +140,9 @@ void DownloadManager::checkSendNext() {
loader->loadPart(bestIndex); loader->loadPart(bestIndex);
} }
} }
if (_requestedBytesAmount[0].empty()) {
_requestedBytesAmount[0] = std::vector<int>(1, 0);
}
if (_requestedBytesAmount[0][0] < kMaxWebFileQueries) { if (_requestedBytesAmount[0][0] < kMaxWebFileQueries) {
if (const auto loader = _webLoaders.nextLoader()) { if (const auto loader = _webLoaders.nextLoader()) {
loader->loadPart(0); loader->loadPart(0);
@ -147,13 +154,14 @@ void DownloadManager::requestedAmountIncrement(
MTP::DcId dcId, MTP::DcId dcId,
int index, int index,
int amount) { int amount) {
Expects(index >= 0 && index < MTP::kDownloadSessionsCount);
using namespace rpl::mappers; using namespace rpl::mappers;
auto it = _requestedBytesAmount.find(dcId); auto it = _requestedBytesAmount.find(dcId);
if (it == _requestedBytesAmount.end()) { if (it == _requestedBytesAmount.end()) {
it = _requestedBytesAmount.emplace(dcId, RequestedInDc { { 0 } }).first; it = _requestedBytesAmount.emplace(
dcId,
std::vector<int>(dcId ? kStartSessionsCount : 1, 0)
).first;
} }
it->second[index] += amount; it->second[index] += amount;
if (!dcId) { if (!dcId) {
@ -163,6 +171,7 @@ void DownloadManager::requestedAmountIncrement(
killDownloadSessionsStop(dcId); killDownloadSessionsStop(dcId);
} else if (ranges::find_if(it->second, _1 > 0) == end(it->second)) { } else if (ranges::find_if(it->second, _1 > 0) == end(it->second)) {
killDownloadSessionsStart(dcId); killDownloadSessionsStart(dcId);
checkSendNext();
} }
} }
@ -197,8 +206,11 @@ void DownloadManager::killDownloadSessions() {
auto left = kKillSessionTimeout; auto left = kKillSessionTimeout;
for (auto i = _killDownloadSessionTimes.begin(); i != _killDownloadSessionTimes.end(); ) { for (auto i = _killDownloadSessionTimes.begin(); i != _killDownloadSessionTimes.end(); ) {
if (i->second <= now) { if (i->second <= now) {
for (int j = 0; j < MTP::kDownloadSessionsCount; ++j) { const auto j = _requestedBytesAmount.find(i->first);
MTP::stopSession(MTP::downloadDcId(i->first, j)); if (j != end(_requestedBytesAmount)) {
for (auto index = 0; index != int(j->second.size()); ++index) {
MTP::stopSession(MTP::downloadDcId(i->first, index));
}
} }
i = _killDownloadSessionTimes.erase(i); i = _killDownloadSessionTimes.erase(i);
} else { } else {
@ -458,8 +470,6 @@ void FileLoader::cancel(bool fail) {
} }
_data = QByteArray(); _data = QByteArray();
const auto downloader = _downloader;
const auto sessionGuard = &session();
const auto weak = QPointer<FileLoader>(this); const auto weak = QPointer<FileLoader>(this);
if (fail) { if (fail) {
emit failed(this, started); emit failed(this, started);

5
Telegram/SourceFiles/storage/file_download.h
View File

@ -41,7 +41,7 @@ public:
[[nodiscard]] virtual MTP::DcId dcId() const = 0; [[nodiscard]] virtual MTP::DcId dcId() const = 0;
[[nodiscard]] virtual bool readyToRequest() const = 0; [[nodiscard]] virtual bool readyToRequest() const = 0;
[[nodiscard]] virtual void loadPart(int dcIndex) = 0; virtual void loadPart(int dcIndex) = 0;
}; };
@ -91,8 +91,7 @@ private:
base::Observable<void> _taskFinishedObservable; base::Observable<void> _taskFinishedObservable;
using RequestedInDc = std::array<int64, MTP::kDownloadSessionsCount>; base::flat_map<MTP::DcId, std::vector<int>> _requestedBytesAmount;
base::flat_map<MTP::DcId, RequestedInDc> _requestedBytesAmount;
base::flat_map<MTP::DcId, crl::time> _killDownloadSessionTimes; base::flat_map<MTP::DcId, crl::time> _killDownloadSessionTimes;
base::Timer _killDownloadSessionsTimer; base::Timer _killDownloadSessionsTimer;