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

Refactor: stagger tasks creation for playlist processing

Browse source
This commit is contained in:
arabcoders 2025-12-30 17:05:59 +03:00
parent f2eebbcd3d
commit 9595a69b50
1 changed files with 29 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

36
app/library/DownloadQueue.py
View file

@ -322,8 +322,6 @@ class DownloadQueue(metaclass=Singleton):
playlistCount = entry.get("playlist_count") playlistCount = entry.get("playlist_count")
playlistCount: int = int(playlistCount) if playlistCount else len(entries) playlistCount: int = int(playlistCount) if playlistCount else len(entries)
results = []
playlist_keys: dict[str, Any] = { playlist_keys: dict[str, Any] = {
"playlist_count": playlistCount, "playlist_count": playlistCount,
"playlist": entry.get("title") or entry.get("id"), "playlist": entry.get("title") or entry.get("id"),
@ -339,18 +337,16 @@ class DownloadQueue(metaclass=Singleton):
} }
async def playlist_processor(i: int, etr: dict): async def playlist_processor(i: int, etr: dict):
acquired = False
try: try:
item_name: str = ( item_name: str = (
f"'{entry.get('title')}: {i}/{playlist_keys['n_entries']}' - '{etr.get('id')}: {etr.get('title')}'" f"'{entry.get('title')}: {i}/{playlist_keys['n_entries']}' - '{etr.get('id')}: {etr.get('title')}'"
) )
LOG.debug(f"Waiting to acquire lock for {item_name}") LOG.debug(f"Waiting to acquire lock for {item_name}")
await self.processors.acquire() await self.processors.acquire()
acquired = True
LOG.debug(f"Acquired lock for {item_name}") LOG.debug(f"Acquired lock for {item_name}")
if self.is_paused():
LOG.warning(f"Download is paused. Skipping processing of '{item_name}'.")
return {"status": "ok"}
LOG.info(f"Processing '{item_name}'.") LOG.info(f"Processing '{item_name}'.")
_status, _msg = ytdlp_reject(entry=etr, yt_params=yt_params) _status, _msg = ytdlp_reject(entry=etr, yt_params=yt_params)
@ -382,6 +378,7 @@ class DownloadQueue(metaclass=Singleton):
return await self.add(item=newItem, already=already) return await self.add(item=newItem, already=already)
finally: finally:
if acquired:
self.processors.release() self.processors.release()
max_downloads: int = -1 max_downloads: int = -1
@ -389,11 +386,11 @@ class DownloadQueue(metaclass=Singleton):
if ytdlp_opts.get("max_downloads") and isinstance(ytdlp_opts.get("max_downloads"), int): if ytdlp_opts.get("max_downloads") and isinstance(ytdlp_opts.get("max_downloads"), int):
max_downloads: int = ytdlp_opts.get("max_downloads") max_downloads: int = ytdlp_opts.get("max_downloads")
tasks: list[asyncio.Task] = [] batch_size: int = max(50, int(self.config.playlist_items_concurrency) * 10)
for i, etr in enumerate(entries, start=1):
if max_downloads > 0 and i > max_downloads:
break
async def run_batch(batch: list[tuple[int, dict]]) -> list[dict]:
tasks: list[asyncio.Task] = []
for i, etr in batch:
task = asyncio.create_task( task = asyncio.create_task(
playlist_processor(i, etr), playlist_processor(i, etr),
name=f"playlist_processor_{etr.get('id')}_{i}", name=f"playlist_processor_{etr.get('id')}_{i}",
@ -401,7 +398,24 @@ class DownloadQueue(metaclass=Singleton):
task.add_done_callback(self._handle_task_exception) task.add_done_callback(self._handle_task_exception)
tasks.append(task) tasks.append(task)
results: list[dict] = await asyncio.gather(*tasks) batch_results: list[dict] = await asyncio.gather(*tasks)
await asyncio.sleep(0)
return batch_results
results: list[dict] = []
batch: list[tuple[int, dict]] = []
for i, etr in enumerate(entries, start=1):
if max_downloads > 0 and i > max_downloads:
break
batch.append((i, etr))
if len(batch) >= batch_size:
results.extend(await run_batch(batch))
batch.clear()
if batch:
results.extend(await run_batch(batch))
log_msg: str = f"Playlist '{playlist_name}' processing completed with '{len(results)}' entries." log_msg: str = f"Playlist '{playlist_name}' processing completed with '{len(results)}' entries."
if max_downloads > 0 and len(entries) > max_downloads: if max_downloads > 0 and len(entries) > max_downloads: