danvics/ytptube
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Switch to single Manager instead of one manager per download

Browse source
This commit is contained in:
ArabCoders 2025-01-19 20:01:46 +03:00
parent 2147a21aa4
commit 2b8e79c1c4
2 changed files with 34 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

45
app/library/Download.py
View file

@ -6,8 +6,6 @@ import multiprocessing
import os
import shutil
import time
from multiprocessing.managers import SyncManager
from email.utils import formatdate
import yt_dlp
@ -15,9 +13,9 @@ import yt_dlp
from .AsyncPool import Terminator
from .config import Config
from .Emitter import Emitter
from .ffprobe import ffprobe
from .ItemDTO import ItemDTO
from .Utils import get_opts, jsonCookie, mergeConfig
from .ffprobe import ffprobe
LOG = logging.getLogger("download")
@ -38,13 +36,14 @@ class Download:
debug: bool = False
tempPath: str = None
emitter: Emitter = None
manager: SyncManager = None
canceled: bool = False
is_live: bool = False
info_dict: dict = None
"yt-dlp metadata dict."
update_task = None
cancel_in_progress: bool = False
bad_live_options: list = [
"concurrent_fragment_downloads",
"fragment_retries",
@ -188,9 +187,8 @@ class Download:
LOG.info(f'Task id="{self.info.id}" PID="{os.getpid()}" title="{self.info.title}" completed.')
async def start(self, emitter: Emitter):
self.manager = multiprocessing.Manager()
self.emitter = emitter
self.status_queue = self.manager.Queue()
self.status_queue = Config.get_manager().Queue()
# Create temp dir for each download.
self.tempPath = os.path.join(self.temp_dir, hashlib.shake_256(f"D-{self.info.id}".encode("utf-8")).hexdigest(5))
@ -223,12 +221,15 @@ class Download:
Close download process.
This method MUST be called to clear resources.
"""
if not self.started():
if not self.started() or self.cancel_in_progress:
return False
self.cancel_in_progress = True
procId = self.proc.ident
LOG.info(f"Closing PID='{procId}' download process.")
try:
LOG.info(f"Closing download process: '{procId}'.")
try:
if "update_task" in self.__dict__ and self.update_task.done() is False:
self.update_task.cancel()
@ -239,19 +240,16 @@ class Download:
self.kill()
loop = asyncio.get_running_loop()
tasks = []
if self.proc.is_alive():
tasks.append(loop.run_in_executor(None, self.proc.join))
LOG.debug(f"Waiting for PID='{procId}' to close.")
await loop.run_in_executor(None, self.proc.join)
LOG.debug(f"PID='{procId}' closed.")
if self.status_queue:
LOG.debug(f"Closing status queue for PID='{procId}'.")
self.status_queue.put(Terminator())
if self.manager:
tasks.append(loop.run_in_executor(None, self.manager.shutdown))
if len(tasks) > 0:
await asyncio.gather(*tasks)
LOG.debug(f"Closed status queue for PID='{procId}'.")
if self.proc:
self.proc.close()
@ -259,7 +257,7 @@ class Download:
self.delete_temp()
LOG.debug(f"Closed download process: '{procId}'.")
LOG.debug(f"Closed PID='{procId}' download process.")
return True
except Exception as e:
@ -293,7 +291,7 @@ class Download:
if self.tempKeep is True or not self.tempPath:
return
if self.info.status != "finished" and self.is_live:
if "finished" != self.info.status and self.is_live:
LOG.warning(
f"Keeping live temp folder '{self.tempPath}', as the reported status is not finished '{self.info.status}'."
)
@ -320,11 +318,9 @@ class Download:
self.update_task = asyncio.get_running_loop().run_in_executor(None, self.status_queue.get)
status = await self.update_task
except (asyncio.CancelledError, OSError, FileNotFoundError):
LOG.debug(f"Closing progress update for: {self.info._id=}.")
return
if status is None or status.__class__ is Terminator:
LOG.debug(f"Closing progress update for: {self.info._id=}.")
if status is None or isinstance(status, Terminator):
return
if status.get("id") != self.id or len(status) < 2:
@ -352,7 +348,7 @@ class Download:
self.info.status = status.get("status", self.info.status)
self.info.msg = status.get("msg")
if self.info.status == "error" and "error" in status:
if "error" == self.info.status and "error" in status:
self.info.error = status.get("error")
asyncio.create_task(
self.emitter.error(message=self.info.error, data=self.info), name=f"emitter-e-{self.id}"
@ -361,7 +357,10 @@ class Download:
if "downloaded_bytes" in status:
total = status.get("total_bytes") or status.get("total_bytes_estimate")
if total:
self.info.percent = status["downloaded_bytes"] / total * 100
try:
self.info.percent = status["downloaded_bytes"] / total * 100
except ZeroDivisionError:
self.info.percent = 0
self.info.total_bytes = total
self.info.speed = status.get("speed")

13
app/library/config.py
View file

@ -1,16 +1,18 @@
import json
import logging
import multiprocessing
import os
import re
import sys
import time
from multiprocessing.managers import SyncManager
from pathlib import Path
import coloredlogs
from dotenv import load_dotenv
from yt_dlp.version import __version__ as YTDLP_VERSION
from .Utils import load_file, mergeDict, IGNORED_KEYS
from .Utils import IGNORED_KEYS, load_file, mergeDict
from .version import APP_VERSION
@ -131,11 +133,20 @@ class Config:
"default_preset",
)
_manager: SyncManager | None = None
@staticmethod
def get_instance():
"""Static access method."""
return Config() if not Config.__instance else Config.__instance
@staticmethod
def get_manager() -> SyncManager:
if not Config._manager:
Config._manager = multiprocessing.Manager()
return Config._manager
def __init__(self):
"""Virtually private constructor."""
if Config.__instance is not None: