Moving periodic download logic into manager so the manager can call it when it is enabled.

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)

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

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
 

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.