diff --git a/docker/archive/main.py b/docker/archive/main.py index e452742b..2bbb37ba 100644 --- a/docker/archive/main.py +++ b/docker/archive/main.py @@ -57,7 +57,7 @@ async def call_single_file(url) -> bytes: See https://github.com/gildas-lormeau/SingleFile """ logger.info(f'archiving {url}') - cmd = ['/usr/src/app/node_modules/single-file/cli/single-file', url, + cmd = ['/usr/src/app/node_modules/single-file-cli/single-file', url, '--browser-executable-path', '/usr/bin/chromium-browser', '--browser-args', '["--no-sandbox"]', '--dump-content'] logger.debug(f'archive cmd: {cmd}') @@ -139,4 +139,4 @@ async def post_archive(request: Request): if __name__ == '__main__': - app.run('0.0.0.0', 8080, workers=4) + app.run('0.0.0.0', 8080, workers=4, auto_reload=True) diff --git a/main.py b/main.py index f16fec0c..5f67766e 100755 --- a/main.py +++ b/main.py @@ -9,7 +9,7 @@ import pytz from sanic import Sanic from sanic.signals import Event -from wrolpi import root_api, BEFORE_STARTUP_FUNCTIONS, after_startup, admin +from wrolpi import root_api, BEFORE_STARTUP_FUNCTIONS, admin, after_startup from wrolpi.common import logger, get_config, import_modules, check_media_directory, limit_concurrent, wrol_mode_enabled from wrolpi.dates import set_timezone from wrolpi.downloader import download_manager @@ -176,14 +176,14 @@ async def set_log_level(args): @limit_concurrent(1) async def periodic_downloads(app: Sanic, loop): """ - A simple function that perpetually calls downloader_manager.do_downloads() after sleeping. + Starts the perpetual downloader on download manager. Limited to only one process. """ # Set all downloads to new. download_manager.reset_downloads() - if wrol_mode_enabled: + if wrol_mode_enabled(): logger.warning(f'Not starting download manager because WROL Mode is enabled.') download_manager.disable() return @@ -195,13 +195,7 @@ async def periodic_downloads(app: Sanic, loop): return download_manager.enable() - - async def _periodic_download(): - await download_manager.do_downloads() - await asyncio.sleep(30) - await app.add_task(_periodic_download()) # noqa - - await app.add_task(_periodic_download()) + await app.add_task(download_manager.perpetual_download()) @after_startup diff --git a/modules/archive/__init__.py b/modules/archive/__init__.py index 91fd8403..f41fa331 100644 --- a/modules/archive/__init__.py +++ b/modules/archive/__init__.py @@ -46,7 +46,7 @@ class ArchiveDownloader(Downloader, ABC): return DownloadResult(success=True, location=f'/archive/{archive.id}') @optional_session - def already_downloaded(self, urls: List[str], session: Session = None) -> List: + def already_downloaded(self, *urls: List[str], session: Session = None) -> List: archives = list(session.query(Archive).filter(Archive.url.in_(urls))) return archives diff --git a/wrolpi/downloader.py b/wrolpi/downloader.py index bc08a36d..8f544951 100644 --- a/wrolpi/downloader.py +++ b/wrolpi/downloader.py @@ -4,7 +4,6 @@ import os import pathlib import traceback from abc import ABC -from asyncio import Task from dataclasses import dataclass, field from datetime import timedelta, datetime from enum import Enum @@ -180,7 +179,7 @@ class Downloader: raise NotImplementedError() @optional_session - def already_downloaded(self, urls: List[str], session: Session = None): + def already_downloaded(self, *urls: List[str], session: Session = None): raise NotImplementedError() @property @@ -272,18 +271,22 @@ class DownloadManager: URLs to be downloaded, then only one worker will be busy. """ priority_sorter = partial(sorted, key=attrgetter('priority')) + manager = multiprocessing.Event() def __init__(self): self.instances: Tuple[Downloader] = tuple() self._instances = dict() self.disabled = multiprocessing.Event() self.stopped = multiprocessing.Event() + self.download_queue: multiprocessing.Queue = multiprocessing.Queue() - self.workers: List[Task] = [] + self.workers: List[Dict] = [] self.worker_count: int = 1 + self.data = multiprocessing.Manager().dict() # We haven't started downloads yet, so no domains are downloading. self.data['processing_domains'] = [] + self.data['workers'] = dict() def register_downloader(self, instance: Downloader): if not isinstance(instance, Downloader): @@ -305,14 +308,15 @@ class DownloadManager: from wrolpi.db import get_db_session pid = os.getpid() - worker_logger = logger.getChild(f'download_worker.{pid}.{num}') + name = f'{pid}.{num}' + worker_logger = logger.getChild(f'download_worker.{name}') status = 'disabled' if self.disabled.is_set() else 'enabled' worker_logger.info(f'Starting up. DownloadManager is {status}.') while True: if self.stopped.is_set(): - # Service is restarting, close the worker. + # Service may be restarting, close the worker. return disabled = self.disabled.is_set() @@ -391,7 +395,6 @@ class DownloadManager: except Empty: # No work yet. await asyncio.sleep(0.1) - continue except Exception as e: worker_logger.warning(f'Unexpected error', exc_info=e) @@ -412,6 +415,7 @@ class DownloadManager: def start_workers(self, loop=None): """Start all download worker tasks. Does nothing if they are already running.""" if not self.workers: + logger.warning('Starting workers') if not loop: try: loop = asyncio.get_running_loop() @@ -437,6 +441,28 @@ class DownloadManager: for task in self.workers: task.cancel() + async def perpetual_download(self): + """ + A method that calls itself forever. It will queue new downloads when they are ready. + + Only one of these can be running at a time. + """ + if self.manager.is_set(): + # Only one manager needs to be running. + return + + self.manager.set() + + async def _perpetual_download(): + if self.stopped.is_set(): + return + + await download_manager.do_downloads() + await asyncio.sleep(5) + asyncio.create_task(_perpetual_download()) + + asyncio.create_task(_perpetual_download()) + def get_downloader(self, url: str) -> Tuple[Downloader, Dict]: for i in self.instances: valid, info = i.valid_url(url) @@ -743,6 +769,7 @@ class DownloadManager: self.start_workers(loop) try: + asyncio.create_task(self.perpetual_download()) asyncio.create_task(self.do_downloads()) except RuntimeError: # This may not work while testing.