From d89da64b1d21b7faf107d26104f3210ea01be7a1 Mon Sep 17 00:00:00 2001 From: pukkandan Date: Sun, 30 May 2021 22:04:31 +0530 Subject: [PATCH] [hls,dash] When using `concurrent_fragment_downloads`, do not keep the fragment content in memory Partial fix for #359 This is a temporary solution until #364 can be implemented --- yt_dlp/downloader/dash.py | 11 ++++++++--- yt_dlp/downloader/hls.py | 11 ++++++++--- 2 files changed, 16 insertions(+), 6 deletions(-) diff --git a/yt_dlp/downloader/dash.py b/yt_dlp/downloader/dash.py index b3ebfb057..398294176 100644 --- a/yt_dlp/downloader/dash.py +++ b/yt_dlp/downloader/dash.py @@ -154,8 +154,9 @@ def append_fragment(frag_content, frag_index): max_workers = self.params.get('concurrent_fragment_downloads', 1) if can_threaded_download and max_workers > 1: self.report_warning('The download speed shown is only of one thread. This is a known issue') + _download_fragment = lambda f: (f, download_fragment(f)[1]) with concurrent.futures.ThreadPoolExecutor(max_workers) as pool: - futures = [pool.submit(download_fragment, fragment) for fragment in fragments_to_download] + futures = [pool.submit(_download_fragment, fragment) for fragment in fragments_to_download] # timeout must be 0 to return instantly done, not_done = concurrent.futures.wait(futures, timeout=0) try: @@ -169,9 +170,13 @@ def append_fragment(frag_content, frag_index): # timeout must be none to cancel concurrent.futures.wait(not_done, timeout=None) raise KeyboardInterrupt - results = [future.result() for future in futures] - for frag_content, frag_index in results: + for fragment, frag_index in map(lambda x: x.result(), futures): + fragment_filename = '%s-Frag%d' % (ctx['tmpfilename'], frag_index) + down, frag_sanitized = sanitize_open(fragment_filename, 'rb') + fragment['fragment_filename_sanitized'] = frag_sanitized + frag_content = down.read() + down.close() result = append_fragment(frag_content, frag_index) if not result: return False diff --git a/yt_dlp/downloader/hls.py b/yt_dlp/downloader/hls.py index b3afb9e9b..19def6693 100644 --- a/yt_dlp/downloader/hls.py +++ b/yt_dlp/downloader/hls.py @@ -424,8 +424,9 @@ def append_fragment(frag_content, frag_index): max_workers = self.params.get('concurrent_fragment_downloads', 1) if can_threaded_download and max_workers > 1: self.report_warning('The download speed shown is only of one thread. This is a known issue') + _download_fragment = lambda f: (f, download_fragment(f)[1]) with concurrent.futures.ThreadPoolExecutor(max_workers) as pool: - futures = [pool.submit(download_fragment, fragment) for fragment in fragments] + futures = [pool.submit(_download_fragment, fragment) for fragment in fragments] # timeout must be 0 to return instantly done, not_done = concurrent.futures.wait(futures, timeout=0) try: @@ -439,9 +440,13 @@ def append_fragment(frag_content, frag_index): # timeout must be none to cancel concurrent.futures.wait(not_done, timeout=None) raise KeyboardInterrupt - results = [future.result() for future in futures] - for frag_content, frag_index in results: + for fragment, frag_index in map(lambda x: x.result(), futures): + fragment_filename = '%s-Frag%d' % (ctx['tmpfilename'], frag_index) + down, frag_sanitized = sanitize_open(fragment_filename, 'rb') + fragment['fragment_filename_sanitized'] = frag_sanitized + frag_content = down.read() + down.close() result = append_fragment(frag_content, frag_index) if not result: return False