Merge pull request #84 from arabcoders/dev

Cleanup after download finish.

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.manager = multiprocessing.Manager()
-        self.loop = asyncio.get_running_loop()
         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.notifier.updated(self.info), name=f"notifier-{self.id}")
-        asyncio.create_task(self.progress_update())
+        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}'.")
+            LOG.info(f"Closing download process: '{procId}'.")
-            self.proc.close()
+            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)
+            self.update_task = asyncio.get_running_loop().run_in_executor(None, self.status_queue.get)
-            if status.get('id') != self.id or len(status) < 2:
+
+            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))
+                asyncio.create_task(self.notifier.updated(self.info), name=f"notifier-u-{self.id}")
-                return
+                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}")

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(
-                entry = await asyncio.wait_for(
+                fut=asyncio.get_running_loop().run_in_executor(
-                    fut=asyncio.get_running_loop().run_in_executor(
+                    None,
-                        executor,
+                    ExtractInfo,
-                        ExtractInfo,
+                    mergeConfig(self.config.ytdl_options, ytdlp_config),
-                        mergeConfig(self.config.ytdl_options, ytdlp_config),
+                    url,
-                        url,
+                    bool(self.config.ytdl_debug)
-                        bool(self.config.ytdl_debug)
+                ),
-                    ),
+                timeout=self.config.extract_info_timeout)
-                    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.info.status != 'finished':
-            if entry.tmpfilename and os.path.isfile(entry.tmpfilename):
+                if entry.tmpfilename and os.path.isfile(entry.tmpfilename):
-                try:
+                    try:
-                    os.remove(entry.tmpfilename)
+                        os.remove(entry.tmpfilename)
-                    entry.tmpfilename = None
+                        entry.tmpfilename = None
-                except:
+                    except:
-                    pass
+                        pass
 
-            entry.info.status = 'error'
+                entry.info.status = 'error'
-
+        finally:
-        entry.close()
+            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()
 

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(
+            print=lambda _: LOG.info(start)
-                f'YTPTube v{self.config.version} - listening on http://{self.config.host}:{self.config.port}'),
         )
 
     async def connect(self, sid, _):

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);