[card-cache] optimizations to avoid deadlocks

- deadlocks happened during cleanups (it was inefficient and load tests simulating cleanup + heavy load situations showed that the deadlock is real)
- Locking at the `context` level. Each time we wanna clean up, we create new context and all directories/processes are within that context.
- Each time we want to cleanup the cache, we switch the `context` where processes are gettting written/spawned. Context also sets the read/write directory of the caches.
- So that during cleanup it only locks to change the `context` and then it can take it's sweet time to cleanup the dictionary in memory and the directory.
- Also making default running of service for each card to 60 seconds (helps make things snappier)
- optimizations to avoid **any** concurrent access of shared resource
- Added `timings` dict in card cache to optimize loading cycles (Ensured that it is set based on a per-card basis)
- Changed defaults for minimum amount to time to wait for cards in the cache process.

Author: valayDave

services/ui_backend_service/data/cache/card_cache_manager.py

@@ -6,7 +6,7 @@
 import uuid
 from asyncio.subprocess import Process
 import asyncio
-from .card_cache_service import CardCache, cleanup_non_running_caches
+from .card_cache_service import CardCache, safe_wipe_dir
 import contextlib
 
 
@@ -26,11 +26,11 @@
     "CARD_CACHE_PROCESS_MAX_UPTIME", 3 * 60  # 3 minutes
 )
 CARD_CACHE_PROCESS_NO_CARD_WAIT_TIME = os.environ.get(
-    "CARD_CACHE_PROCESS_NO_CARD_WAIT_TIME", 4  # 4 seconds
+    "CARD_CACHE_PROCESS_NO_CARD_WAIT_TIME", 20  # 20 seconds
 )
-DEFAULT_CACHE_STORAGE_PATH = "/tmp"
-CACHE_STORAGE_PATH = os.environ.get(
-    "CARD_CACHE_STORAGE_PATH", DEFAULT_CACHE_STORAGE_PATH
+DEFAULT_CACHE_STORAGE_PATH_ROOT = "/tmp"
+CACHE_STORAGE_PATH_ROOT = os.environ.get(
+    "CARD_CACHE_STORAGE_PATH_ROOT", DEFAULT_CACHE_STORAGE_PATH_ROOT
 )
 CACHE_SERVICE_LOG_STORAGE_ROOT = os.environ.get("CACHE_SERVICE_LOG_STORAGE_ROOT", None)
 
@@ -57,35 +57,106 @@ async def process_is_running(process: Process):
     return status == "running"
 
 
-class AsyncProcessManager:
+class AsyncCardCacheProcessManager:
 
     processes = {
-        # "procid": {
-        #     "proc": asyncio.subprocess.Process,
-        #     "started": time.time()
+        # "<current-context>":{
+        #  "processes": {
+        #       "<procid>": {
+        #           "proc": asyncio.subprocess.Process,
+        #           "started": time.time()
+        #       }
+        #   },
+        #  "write_directory": "<write_directory>"
         # }
     }
 
+    def get_context_dict(self, context):
+        _x = self.processes.get(context, None)
+        if _x is None:
+            return _x
+        return _x.copy()
+
+    @property
+    def current_process_dict(self):
+        return self.processes[self._current_context]["processes"]
+
+    @property
+    def current_write_directory(self):
+        return self.processes[self._current_context]["write_directory"]
+
     def __init__(self, logger) -> None:
-        self.lock = asyncio.Lock()
+        self.context_lock = asyncio.Lock()
         self.logger = logger
+        self._current_context = None
+        self.update_context()
+
+    async def set_new_context(self):
+        async with self.context_lock:
+            old_context = self._current_context
+            self.update_context()
+            return old_context
+
+    async def remove_context(self, context):
+        async with self.context_lock:
+            if context in self.processes:
+                del self.processes[context]
+
+    def update_context(self):
+        _ctx_dict, _ctx = self.create_context_dict()
+        self.processes.update(_ctx_dict)
+        self._current_context = _ctx
+
+    def create_context_dict(self):
+        _ctx = uuid.uuid4().hex[:8]
+        return {
+            _ctx: {
+                "processes": {},
+                "write_directory": os.path.join(CACHE_STORAGE_PATH_ROOT, _ctx),
+            }
+        }, _ctx
 
-    def _register_process(self, procid, proc):
-        self.processes[procid] = {
+    def _register_process(self, procid, proc, stdout, stderr):
+        self.current_process_dict[procid] = {
             "proc": proc,
             "started": time.time(),
+            "stdout": stdout,
+            "stderr": stderr
         }
 
-    async def add(self, procid, cmd, logs_file_path=CACHE_SERVICE_LOG_STORAGE_ROOT):
+    def _make_command(self, pathspec):
+        return [
+            str(i)
+            for i in [
+                sys.executable,
+                PATH_TO_CACHE_SERVICE,
+                "task-updates",
+                pathspec,
+                "--uptime-seconds",
+                CARD_CACHE_PROCESS_MAX_UPTIME,
+                "--list-frequency",
+                CARD_LIST_POLLING_FREQUENCY,
+                "--data-update-frequency",
+                DATA_UPDATE_POLLING_FREQUENCY,
+                "--html-update-frequency",
+                CARD_UPDATE_POLLING_FREQUENCY,
+                "--cache-path",
+                self.current_write_directory,
+                "--max-no-card-wait-time",
+                CARD_CACHE_PROCESS_NO_CARD_WAIT_TIME,
+            ]
+        ]
+
+    async def add(self, procid, pathspec, logs_file_path=CACHE_SERVICE_LOG_STORAGE_ROOT):
         running_proc = await self.is_running(procid)
         if running_proc:
             return procid, "running"
         # The lock helps to ensure that the processes only get added one at a time
         # This is important because the processes are added to a shared dictionary
-        async with self.lock:
+        async with self.context_lock:
             proc, started_on = self.get(procid)
             if proc is not None:
-                await self.remove(procid, delete_item=True)
+                await self.remove_current(procid, delete_item=True)
 
             logs_file = None
             if logs_file_path is not None:
@@ -96,12 +167,12 @@ async def add(self, procid, cmd, logs_file_path=CACHE_SERVICE_LOG_STORAGE_ROOT):
                     ),
                     "w",
                 )
-
+            cmd = self._make_command(pathspec)
             await self.spawn(procid, cmd, logs_file, logs_file)
             return procid, "started"
 
     def get(self, procid):
-        proc_dict = self.processes.get(procid, None)
+        proc_dict = self.current_process_dict.get(procid, None)
         if proc_dict is not None:
             return proc_dict["proc"], proc_dict["started"]
         return None, None
@@ -113,81 +184,95 @@ async def spawn(self, procid, cmd, stdout_file, std_err_file=None):
             stderr=std_err_file,
             shell=False,
         )
-        self._register_process(procid, proc)
+        self._register_process(procid, proc, stdout_file, std_err_file)
+
+    async def clean_terminate_process(self, process: Process):
+        try:
+            self.logger.error("Terminating process %s " % process.pid)
+            if process.returncode is None:
+                process.terminate()
+            await process.wait()
+        except Exception as e:
+            self.logger.error(f"Error terminating process: {str(type(e))}")
+
+    async def clean_close_process(self, process):
+        try:
+            await process.wait()
+        except Exception as e:
+            self.logger.error(f"Error closing process: {str(e)}")
+
+    def _close_stdout_and_err(self, process_context_dict):
+        if process_context_dict["stdout"] is not None:
+            process_context_dict["stdout"].close()
+        if process_context_dict["stderr"] is not None:
+            process_context_dict["stderr"].close()
 
-    async def remove(self, procid, delete_item=True):
-        if procid not in self.processes:
+    async def remove_current(self, procid, delete_item=True):
+        if procid not in self.current_process_dict:
             return
         self.logger.info("Removing process: %s" % procid)
-        await self.processes[procid]["proc"].wait()
-        self.logger.info("Process removed: %s" % procid)
-        if self.processes[procid]["proc"].stdout is not None:
-            self.processes[procid]["proc"].stdout.close()
+        await self.clean_close_process(self.current_process_dict[procid]["proc"])
+        self._close_stdout_and_err(self.current_process_dict[procid])
         if delete_item:
-            del self.processes[procid]
+            del self.current_process_dict[procid]
 
     async def cleanup(self):
-        # The lock ensures that when the dictionary is being modified,
-        # no other process can modify it at the same time.
-        async with self.lock:
-            removal_keys = []
-            for procid in self.processes:
-                running_proc = await self.is_running(procid)
-                if running_proc:
-                    continue
-                removal_keys.append(procid)
-                await self.remove(procid, delete_item=False)
-            for procid in removal_keys:
-                del self.processes[procid]
-            return removal_keys
+        old_context = await self.set_new_context()
+        _ctx_dict = self.get_context_dict(old_context)
+        if _ctx_dict is None:
+            return []
+        # Two things to remove (old Keys and old directories)
+        wait_keys = []
+        for pid in _ctx_dict["processes"]:
+            status = await process_status(_ctx_dict["processes"][pid]["proc"])
+            if status in ["completed", "failed"]:
+                await self.clean_close_process(_ctx_dict["processes"][pid]["proc"])
+                self._close_stdout_and_err(_ctx_dict["processes"][pid]["proc"])
+            else:
+                wait_keys.append(pid)
+        if len(wait_keys) > 0:
+            # sleeping for MAX_PROCESS_TIME so that all the processes writing should have finished
+            # AND any upstream request accessing the cache-dir should have been attenteded
+            await asyncio.sleep(CARD_CACHE_PROCESS_MAX_UPTIME)
+            for pid in wait_keys:
+                await self.clean_terminate_process(_ctx_dict["processes"][pid]["proc"])
+                self._close_stdout_and_err(_ctx_dict["processes"][pid]["proc"])
+
+        _write_dir = _ctx_dict["write_directory"]
+        self.logger.info("Removing directory: %s" % _write_dir)
+        safe_wipe_dir(_write_dir)
+        await self.remove_context(old_context)
 
     async def is_running(self, procid):
-        if procid not in self.processes:
+        if procid not in self.current_process_dict:
             return False
-        return await process_is_running(self.processes[procid]["proc"])
+        return await process_is_running(self.current_process_dict[procid]["proc"])
 
     async def get_status(self, procid):
-        if procid not in self.processes:
+        if procid not in self.current_process_dict:
             return None
-        return await process_status(self.processes[procid]["proc"])
+        return await process_status(self.current_process_dict[procid]["proc"])
 
     async def running_processes(self):
-        return [procid for procid in self.processes if await self.is_running(procid)]
+        return [
+            procid
+            for procid in self.current_process_dict
+            if await self.is_running(procid)
+        ]
 
 
 class CardCacheManager:
     def __init__(self) -> None:
         self.logger = logging.getLogger("CardCacheManager")
-        self._process_manager = AsyncProcessManager(self.logger)
+        self._process_manager = AsyncCardCacheProcessManager(self.logger)
         self._manager_id = uuid.uuid4().hex
         self.logger.info("CardCacheManager initialized")
 
-    def _make_task_command(self, pathspec):
-        return [
-            str(i)
-            for i in [
-                sys.executable,
-                PATH_TO_CACHE_SERVICE,
-                "task-updates",
-                pathspec,
-                "--uptime-seconds",
-                CARD_CACHE_PROCESS_MAX_UPTIME,
-                "--list-frequency",
-                CARD_LIST_POLLING_FREQUENCY,
-                "--data-update-frequency",
-                DATA_UPDATE_POLLING_FREQUENCY,
-                "--html-update-frequency",
-                CARD_UPDATE_POLLING_FREQUENCY,
-                "--cache-path",
-                CACHE_STORAGE_PATH,
-                "--max-no-card-wait-time",
-                CARD_CACHE_PROCESS_NO_CARD_WAIT_TIME
-            ]
-        ]
-
     def get_local_cache(self, pathspec, card_hash):
         cache = CardCache.load_from_disk(
-            pathspec, card_hash, CACHE_STORAGE_PATH,
+            pathspec,
+            card_hash,
+            self._process_manager.current_write_directory,
         )
         return cache
 
@@ -196,40 +281,22 @@ async def register(self, pathspec):
         is_running = await self._process_manager.is_running(proc_id)
         if is_running:
             return proc_id, "running"
-
-        cmd = self._make_task_command(pathspec)
-        self.logger.info(
-            "Registering task [%s]" % (pathspec)
+        self.logger.info("Registering task [%s]" % (pathspec))
+        _id, status = await self._process_manager.add(
+            proc_id, pathspec, logs_file_path=CACHE_SERVICE_LOG_STORAGE_ROOT
         )
-        _id, status = await self._process_manager.add(proc_id, cmd, logs_file_path=CACHE_SERVICE_LOG_STORAGE_ROOT)
         return _id, status
 
     async def get_status(self, pathspec):
         return await self._process_manager.get_status(pathspec)
 
-    async def start_process_cleanup_routine(self, interval=60):
-        try:
-            while True:
-                cleanup_keys = await self._process_manager.cleanup()  # Perform the cleanup
-                if len(cleanup_keys) > 0:
-                    self.logger.info(
-                        "Cleaned up processes: %s" % ", ".join(cleanup_keys)
-                    )
-                await asyncio.sleep(interval)  # Wait for a specified interval before the next cleanup
-        except asyncio.CancelledError:
-            self.logger.info("Process cleanup routine cancelled")
-
-    async def cleanup_disk_routine(self, interval=60 * 60 * 4):
+    async def regular_cleanup_routine(self, interval=60 * 60 * 24 * 5):
         try:
             while True:
                 await asyncio.sleep(interval)
-                # The lock ensure that new processes are not getting created or
-                # processes are not getting removed when the disk cleanup is happening.
-                async with self._process_manager.lock:
-                    running_proc_ids = await self._process_manager.running_processes()
-                    cleanup_non_running_caches(CACHE_STORAGE_PATH, CardCache.CACHE_DIR, running_proc_ids)
+                await self._process_manager.cleanup()
         except asyncio.CancelledError:
-            self.logger.info("Disk cleanup routine cancelled")
+            self.logger.info("Process/Directory cleanup routine cancelled")
 
 
 async def verify_process_has_crashed(cache_manager: CardCacheManager, pathspec):
@@ -254,7 +321,10 @@ def _get_html_or_refresh(local_cache: CardCache):
 
 
 async def wait_until_card_is_ready(
-    cache_manager: CardCacheManager, local_cache: CardCache, max_wait_time=3, frequency=0.1
+    cache_manager: CardCacheManager,
+    local_cache: CardCache,
+    max_wait_time=3,
+    frequency=0.1,
 ):
     html = None
     start_time = time.time()
@@ -264,7 +334,9 @@ async def wait_until_card_is_ready(
         html = _get_html_or_refresh(local_cache)
         if html is not None:
             break
-        process_failed = await verify_process_has_crashed(cache_manager, local_cache.pathspec)
+        process_failed = await verify_process_has_crashed(
+            cache_manager, local_cache.pathspec
+        )
         if process_failed:
             cache_manager.logger.error(
                 f"Card {local_cache.card_hash} has crashed for {local_cache.pathspec}"