Use a lock file for precaching

Ensure that no two workers on the same machine simultaneously
try to precache files. All workers except one skip precaching.

Author: fagu

cms/db/filecacher.py

@@ -30,6 +30,7 @@
 import logging
 import os
 import tempfile
+import fcntl
 from abc import ABCMeta, abstractmethod
 
 import gevent
@@ -543,6 +544,36 @@ def _create_directory_or_die(directory):
             logger.error(msg)
             raise RuntimeError(msg)
 
+    def precache_lock(self):
+        """Lock the (shared) cache for precaching if it is currently unlocked.
+
+        Locking is optional: Any process can perform normal cache operations
+        at any time whether the cache is locked or not.
+
+        The locking mechanism's only purpose is to avoid wasting resources by
+        ensuring that on each machine, only one worker precaches at any time.
+
+        return (fileobj|None): The lock file if the cache was previously
+            unlocked. Closing the file object will release the lock.
+            None if the cache was already locked.
+
+        """
+        lock_file = os.path.join(self.file_dir, "cache_lock")
+        fobj = open(lock_file, 'w')
+        returned = False
+        try:
+            fcntl.flock(fobj, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError:
+            # This exception is raised only if the errno is EWOULDBLOCK,
+            # which means that the file is already locked.
+            return None
+        else:
+            returned = True
+            return fobj
+        finally:
+            if not returned:
+                fobj.close()
+
     def _load(self, digest, cache_only):
         """Load a file into the cache and open it for reading.
 

cms/service/Worker.py

@@ -6,6 +6,7 @@
 # Copyright © 2010-2012 Matteo Boscariol <[email protected]>
 # Copyright © 2013-2015 Luca Wehrstedt <[email protected]>
 # Copyright © 2016 Luca Versari <[email protected]>
+# Copyright © 2021 Fabian Gundlach <[email protected]>
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as
@@ -70,24 +71,35 @@ def precache_files(self, contest_id):
         contest_id (int): the id of the contest
 
         """
-        # In order to avoid a long-living connection, first fetch the
-        # complete list of files and then download the files; since
-        # this is just pre-caching, possible race conditions are not
-        # dangerous
-        logger.info("Precaching files for contest %d.", contest_id)
-        with SessionGen() as session:
-            contest = Contest.get_from_id(contest_id, session)
-            files = enumerate_files(session, contest, skip_submissions=True,
-                                    skip_user_tests=True, skip_print_jobs=True)
-        for digest in files:
-            try:
-                self.file_cacher.cache_file(digest)
-            except KeyError:
-                # No problem (at this stage) if we cannot find the
-                # file
-                pass
-
-        logger.info("Precaching finished.")
+        lock = self.file_cacher.precache_lock()
+        if lock is None:
+            # Another worker is already precaching. Hence, this worker doesn't
+            # need to do anything.
+            logger.info("Another worker is already precaching files for "
+                        "contest %d.", contest_id)
+            return
+        with lock:
+            # In order to avoid a long-living connection, first fetch the
+            # complete list of files and then download the files; since
+            # this is just pre-caching, possible race conditions are not
+            # dangerous
+            logger.info("Precaching files for contest %d.", contest_id)
+            with SessionGen() as session:
+                contest = Contest.get_from_id(contest_id, session)
+                files = enumerate_files(session,
+                                        contest,
+                                        skip_submissions=True,
+                                        skip_user_tests=True,
+                                        skip_print_jobs=True)
+            for digest in files:
+                try:
+                    self.file_cacher.cache_file(digest)
+                except KeyError:
+                    # No problem (at this stage) if we cannot find the
+                    # file
+                    pass
+
+            logger.info("Precaching finished.")
 
     @rpc_method
     def execute_job_group(self, job_group_dict):