Local executor was not cleaning its files. Also, add a mechanism to clean

old submit scripts to avoid gazillions of small files in ~/.psij/work.

Author: hategan

src/psij/executors/batch/batch_scheduler_executor.py

@@ -16,7 +16,7 @@
 from psij import JobExecutor, JobExecutorConfig, Launcher, Job, SubmitException, \
     JobStatus, JobState
 from psij.executors.batch.template_function_library import ALL as FUNCTION_LIBRARY
-from psij.utils import _StatusUpdater
+from psij.utils import _StatusUpdater, _FileCleaner
 
 UNKNOWN_ERROR = 'PSIJ: Unknown error'
 
@@ -206,6 +206,7 @@ def __init__(self, url: Optional[str] = None,
         assert config
         self.work_directory = config.work_directory / self.name
         self.work_directory.mkdir(parents=True, exist_ok=True)
+        cast(_FileCleaner, _FileCleaner.get_instance()).clean(self.work_directory)
 
     def submit(self, job: Job) -> None:
         """See :func:`~psij.JobExecutor.submit`."""

src/psij/executors/local.py

@@ -19,7 +19,7 @@
 from psij import JobExecutor
 from psij.executors.batch.batch_scheduler_executor import _env_to_mustache
 from psij.executors.batch.script_generator import TemplatedScriptGenerator
-from psij.utils import SingletonThread, _StatusUpdater
+from psij.utils import SingletonThread, _StatusUpdater, _FileCleaner
 
 from psij.executors.batch.template_function_library import ALL as FUNCTION_LIBRARY
 
@@ -232,6 +232,7 @@ def __init__(self, url: Optional[str] = None,
         self._reaper = _ProcessReaper.get_instance()
         self._work_dir = Path.home() / '.psij' / 'work' / 'local'
         self._work_dir.mkdir(parents=True, exist_ok=True)
+        cast(_FileCleaner, _FileCleaner.get_instance()).clean(self._work_dir)
         self._status_updater = cast(_StatusUpdater, _StatusUpdater.get_instance())
         self.generator = TemplatedScriptGenerator(config, Path(__file__).parent / 'local'
                                                   / 'local.mustache')
@@ -339,13 +340,22 @@ def _process_done(self, p: _ProcessEntry) -> None:
                 message = p.out
             state = JobState.FAILED
 
+        if state.final:
+            self._clean_submit_file(p.job)
         # We need to ensure that the status updater has processed all updates that
         # have been sent up to this point
         self._status_updater.flush()
         self._status_updater.unregister_job(p.job)
         self._set_job_status(p.job, JobStatus(state, time=p.done_time, exit_code=p.exit_code,
                                               message=message))
 
+    def _clean_submit_file(self, job: Job) -> None:
+        submit_file_path = self._work_dir / (job.id + '.job')
+        try:
+            submit_file_path.unlink()
+        except FileNotFoundError:
+            pass
+
     def list(self) -> List[str]:
         """
         Return a list of ids representing jobs that are running on the underlying implementation.

src/psij/utils.py

@@ -2,11 +2,13 @@
 import io
 import logging
 import os
+import queue
 import random
 import socket
 import tempfile
 import threading
 import time
+from datetime import datetime, timedelta
 from pathlib import Path
 from typing import Type, Dict, Optional, Tuple, Set, List
 
@@ -17,6 +19,9 @@
 logger = logging.getLogger(__name__)
 
 
+_MAX_FILE_AGE_DAYS = 30
+
+
 class SingletonThread(threading.Thread):
     """
     A convenience class to return a thread that is guaranteed to be unique to this process.
@@ -207,3 +212,37 @@ def _process_update_data(self, data: bytes) -> None:
                     pass
             if job:
                 executor._set_job_status(job, JobStatus(state))
+
+
+class _FileCleaner(SingletonThread):
+    def __init__(self) -> None:
+        super().__init__()
+        self.name = 'File Cleaner'
+        self.daemon = True
+        self.queue: queue.SimpleQueue[Path] = queue.SimpleQueue()
+
+    def clean(self, path: Path) -> None:
+        self.queue.put(path)
+
+    def run(self) -> None:
+        while True:
+            try:
+                path = self.queue.get(block=True, timeout=1)
+                try:
+                    self._do_clean(path)
+                except Exception as ex:
+                    print(f'Warning: cannot clean {path}: {ex}')
+            except queue.Empty:
+                pass
+
+    def _do_clean(self, path: Path) -> None:
+        now = datetime.now()
+        max_age = timedelta(days=_MAX_FILE_AGE_DAYS)
+        for child in path.iterdir():
+            m_time = datetime.fromtimestamp(child.lstat().st_mtime)
+            if now - m_time > max_age:
+                try:
+                    child.unlink()
+                except FileNotFoundError:
+                    # we try our best
+                    pass