diff --git a/app/Download.py b/app/Download.py index b8610622..4d1633f0 100644 --- a/app/Download.py +++ b/app/Download.py @@ -7,7 +7,7 @@ import re import shutil import yt_dlp import hashlib - +from AsyncPool import Terminator from Utils import Notifier, get_format, get_opts, jsonCookie, mergeConfig from ItemDTO import ItemDTO from Config import Config @@ -32,10 +32,12 @@ class Download: debug: bool = False tempPath: str = None notifier: Notifier = None + manager: multiprocessing.Manager = None canceled: bool = False is_live: bool = False info_dict: dict = None "yt-dlp metadata dict." + update_task = None bad_live_options: list = [ "concurrent_fragment_downloads", @@ -73,12 +75,10 @@ class Download: self.id = info._id self.default_ytdl_opts = config.ytdl_options self.debug = debug - self.canceled = False self.tmpfilename = None self.status_queue = None self.proc = None - self.loop = None self.notifier = None self.max_workers = int(config.max_workers) self.tempKeep = bool(config.temp_keep) @@ -183,9 +183,9 @@ class Download: LOG.info(f'Finished {os.getpid()=} id="{self.info.id}" title="{self.info.title}".') async def start(self, notifier: Notifier): - self.status_queue = multiprocessing.Manager().Queue() - self.loop = asyncio.get_running_loop() + self.manager = multiprocessing.Manager() self.notifier = notifier + self.status_queue = self.manager.Queue() # Create temp dir for each download. self.tempPath = os.path.join(self.temp_dir, hashlib.shake_256(self.info.id.encode('utf-8')).hexdigest(5)) @@ -193,14 +193,14 @@ class Download: if not os.path.exists(self.tempPath): os.makedirs(self.tempPath, exist_ok=True) - self.proc = multiprocessing.Process(target=self._download) + self.proc = multiprocessing.Process(name=f"download-{self.id}", target=self._download) self.proc.start() self.info.status = 'preparing' - asyncio.create_task(self.notifier.updated(self.info)) - asyncio.create_task(self.progress_update()) + asyncio.create_task(self.notifier.updated(self.info), name=f"notifier-{self.id}") + asyncio.create_task(self.progress_update(), name=f"update-{self.id}") - return await self.loop.run_in_executor(None, self.proc.join) + return await asyncio.get_running_loop().run_in_executor(None, self.proc.join) def started(self) -> bool: return self.proc is not None @@ -209,22 +209,54 @@ class Download: if not self.started(): return False - if self.kill(): - self.canceled = True + self.canceled = True return True - def close(self) -> bool: + async def close(self) -> bool: + """ + Close download process. + This method MUST be called to clear resources. + """ if not self.started(): return False + procId = self.proc.ident try: - LOG.info(f"Closing download process: '{self.proc.ident}'.") - self.proc.close() + LOG.info(f"Closing download process: '{procId}'.") + try: + if 'update_task' in self.__dict__ and self.update_task.done() is False: + self.status_queue.put(Terminator(), timeout=3) + LOG.debug(f"Closed status queue: '{procId}'.") + except Exception as e: + LOG.error(f"Failed to close status queue: '{procId}'. {e}") + pass + + self.kill() + + loop = asyncio.get_running_loop() + tasks = [] + + if self.proc.is_alive(): + tasks.append(loop.run_in_executor(None, self.proc.join)) + + if self.manager: + tasks.append(loop.run_in_executor(None, self.manager.shutdown)) + + if len(tasks) > 0: + await asyncio.gather(*tasks) + + if self.proc: + self.proc.close() + self.proc = None + self.delete_temp() + + LOG.debug(f"Closed download process: '{procId}'.") + return True except Exception as e: - LOG.error(f"Failed to close process: '{self.proc.ident}'. {e}") + LOG.error(f"Failed to close process: '{procId}'. {e}") return False @@ -238,7 +270,7 @@ class Download: return self.canceled def kill(self) -> bool: - if not self.started(): + if not self.running(): return False try: @@ -275,16 +307,23 @@ class Download: Update status of download task and notify the client. """ while True: - status = await self.loop.run_in_executor(None, self.status_queue.get) - if status.get('id') != self.id or len(status) < 2: + self.update_task = asyncio.get_running_loop().run_in_executor(None, self.status_queue.get) + + status = await self.update_task + + if status is None or status.__class__ is Terminator: + LOG.debug(f'Closing progress update for: {self.info._id=}.') return + if status.get('id') != self.id or len(status) < 2: + continue + if self.debug: LOG.debug(f'Status Update: {self.info._id=} {status=}') if isinstance(status, str): - asyncio.create_task(self.notifier.updated(self.info)) - return + asyncio.create_task(self.notifier.updated(self.info), name=f"notifier-u-{self.id}") + continue self.tmpfilename = status.get('tmpfilename') @@ -303,7 +342,7 @@ class Download: if self.info.status == 'error' and 'error' in status: self.info.error = status.get('error') - asyncio.create_task(self.notifier.error(self.info, self.info.error)) + asyncio.create_task(self.notifier.error(self.info, self.info.error), name=f"notifier-e-{self.id}") if 'downloaded_bytes' in status: total = status.get('total_bytes') or status.get('total_bytes_estimate') @@ -322,4 +361,4 @@ class Download: self.info.file_size = None pass - asyncio.create_task(self.notifier.updated(self.info)) + asyncio.create_task(self.notifier.updated(self.info), name=f"notifier-u-{self.id}") diff --git a/app/DownloadQueue.py b/app/DownloadQueue.py index 11ffbf57..91632f07 100644 --- a/app/DownloadQueue.py +++ b/app/DownloadQueue.py @@ -41,7 +41,8 @@ class DownloadQueue: LOG.info(f'Using {self.config.max_workers} workers for downloading.') asyncio.create_task( - self.__download_pool() if self.config.max_workers > 1 else self.__download() + self.__download_pool() if self.config.max_workers > 1 else self.__download(), + name='download_pool' if self.config.max_workers > 1 else 'download_worker' ) async def __add_entry( @@ -172,7 +173,9 @@ class DownloadQueue: itemDownload = self.queue.put(dlInfo) self.event.set() - asyncio.create_task(self.notifier.emit(NotifyEvent, itemDownload.info)) + asyncio.create_task( + self.notifier.emit(NotifyEvent, itemDownload.info), + name=f'notifier-{NotifyEvent}-{itemDownload.info.id}') return { 'status': 'ok' @@ -227,16 +230,15 @@ class DownloadQueue: started = time.perf_counter() LOG.debug(f'extract_info: checking {url=}') - with ThreadPoolExecutor(1) as executor: - entry = await asyncio.wait_for( - fut=asyncio.get_running_loop().run_in_executor( - executor, - ExtractInfo, - mergeConfig(self.config.ytdl_options, ytdlp_config), - url, - bool(self.config.ytdl_debug) - ), - timeout=self.config.extract_info_timeout) + entry = await asyncio.wait_for( + fut=asyncio.get_running_loop().run_in_executor( + None, + ExtractInfo, + mergeConfig(self.config.ytdl_options, ytdlp_config), + url, + bool(self.config.ytdl_debug) + ), + timeout=self.config.extract_info_timeout) if not entry: return {'status': 'error', 'msg': 'Unable to extract info check logs.'} @@ -277,17 +279,18 @@ class DownloadQueue: itemMessage = f"{id=} {item.info.id=} {item.info.title=}" - if item.started() is True: + if item.running(): LOG.debug(f'Canceling {itemMessage}') item.cancel() LOG.info(f'Cancelled {itemMessage}') + await item.close() else: - item.close() + await item.close() LOG.debug(f'Deleting from queue {itemMessage}') self.queue.delete(id) - asyncio.create_task(self.notifier.canceled(id)) + asyncio.create_task(self.notifier.canceled(id), name=f'notifier-c-{id}') self.done.put(item) - asyncio.create_task(self.notifier.completed(item)) + asyncio.create_task(self.notifier.completed(item), name=f'notifier-d-{id}') LOG.info(f'Deleted from queue {itemMessage}') status[id] = 'ok' @@ -309,7 +312,7 @@ class DownloadQueue: itemMessage = f"{id=} {item.info.id=} {item.info.title=}" LOG.debug(f'Deleting completed download {itemMessage}') self.done.delete(id) - asyncio.create_task(self.notifier.cleared(id)) + asyncio.create_task(self.notifier.cleared(id), name=f'notifier-c-{id}') LOG.info(f'Deleted completed download {itemMessage}') status[id] = 'ok' @@ -388,31 +391,32 @@ class DownloadQueue: async def __downloadFile(self, id: str, entry: Download): LOG.info(f'Queuing {id=} - {entry.info.title=} - {entry.info.url=} - {entry.info.folder=}.') - await entry.start(self.notifier) + try: + await entry.start(self.notifier) - if entry.info.status != 'finished': - if entry.tmpfilename and os.path.isfile(entry.tmpfilename): - try: - os.remove(entry.tmpfilename) - entry.tmpfilename = None - except: - pass + if entry.info.status != 'finished': + if entry.tmpfilename and os.path.isfile(entry.tmpfilename): + try: + os.remove(entry.tmpfilename) + entry.tmpfilename = None + except: + pass - entry.info.status = 'error' - - entry.close() + entry.info.status = 'error' + finally: + await entry.close() if self.queue.exists(key=id): self.queue.delete(key=id) if entry.is_canceled() is True: - asyncio.create_task(self.notifier.canceled(id)) + asyncio.create_task(self.notifier.canceled(id), name=f'notifier-c-{id}') entry.info.status = 'canceled' entry.info.error = 'Canceled by user.' self.done.put(value=entry) - asyncio.create_task(self.notifier.completed(entry.info)) + asyncio.create_task(self.notifier.completed(entry.info), name=f'notifier-d-{id}') self.event.set() diff --git a/app/main.py b/app/main.py index 65cc68f1..5e5a4edc 100644 --- a/app/main.py +++ b/app/main.py @@ -106,6 +106,7 @@ class Main: self.start() def start(self): + start: str = f'YTPTube v{self.config.version} - listening on http://{self.config.host}:{self.config.port}' web.run_app( self.app, host=self.config.host, @@ -113,8 +114,7 @@ class Main: reuse_port=True, loop=self.loop, access_log=None, - print=lambda _: print( - f'YTPTube v{self.config.version} - listening on http://{self.config.host}:{self.config.port}'), + print=lambda _: LOG.info(start) ) async def connect(self, sid, _): diff --git a/frontend/src/App.vue b/frontend/src/App.vue index cf1b2d52..9e836951 100644 --- a/frontend/src/App.vue +++ b/frontend/src/App.vue @@ -148,6 +148,19 @@ const deleteItem = (type, item) => { headers: { 'Content-Type': 'application/json' } + }).then(resp => resp.json()).then(json => { + for (const key in items) { + const itemId = items[key]; + if (itemId in json && json[itemId] === 'ok') { + if (true === (itemId in completed)) { + delete completed[itemId]; + } + if (true === (itemId in downloading)) { + toast.info('Download canceled: ' + downloading[itemId]?.title); + delete downloading[itemId]; + } + } + } }).catch((error) => { console.log(error); toast.error('Failed to delete/cancel item/s. ' + error);