fix: improve file cache of regions in concurrent scenarios

Author: lihsai0

qiniu/compat.py

@@ -14,6 +14,13 @@
     # because of u'...' Unicode literals.
     import json  # noqa
 
+# -------
+# Platform
+# -------
+
+is_windows = sys.platform == 'win32'
+is_linux = sys.platform == 'linux'
+is_macos = sys.platform == 'darwin'
 
 # -------
 # Pythons

qiniu/http/regions_provider.py

@@ -6,8 +6,9 @@
 import tempfile
 import os
 import shutil
+import threading
 
-from qiniu.compat import json, b as to_bytes
+from qiniu.compat import json, b as to_bytes, is_windows, is_linux, is_macos
 from qiniu.utils import io_md5, dt2ts
 
 from .endpoint import Endpoint
@@ -24,7 +25,7 @@ def __iter__(self):
         """
         Returns
         -------
-        list[Region]
+        Generator[Region, None, None]
         """
 
 
@@ -137,27 +138,112 @@ def __init__(self, message):
         super(FileAlreadyLocked, self).__init__(message)
 
 
-class _FileLocker:
-    def __init__(self, origin_file):
-        self._origin_file = origin_file
+_file_threading_lockers_lock = threading.Lock()
+_file_threading_lockers = {}
+
+
+class _FileThreadingLocker:
+    def __init__(self, fd):
+        self._fd = fd
 
     def __enter__(self):
-        if os.access(self.lock_file_path, os.R_OK | os.W_OK):
-            raise FileAlreadyLocked('File {0} already locked'.format(self._origin_file))
-        with open(self.lock_file_path, 'w'):
-            pass
+        with _file_threading_lockers_lock:
+            global _file_threading_lockers
+            threading_lock = _file_threading_lockers.get(self._file_path, threading.Lock())
+            # Could use keyword style `acquire(blocking=False)` when min version of python update to >= 3
+            if not threading_lock.acquire(False):
+                raise FileAlreadyLocked('File {0} already locked'.format(self._file_path))
+            _file_threading_lockers[self._file_path] = threading_lock
 
     def __exit__(self, exc_type, exc_val, exc_tb):
-        os.remove(self.lock_file_path)
+        with _file_threading_lockers_lock:
+            global _file_threading_lockers
+            threading_lock = _file_threading_lockers.get(self._file_path)
+            if threading_lock and threading_lock.locked():
+                threading_lock.release()
+                del _file_threading_lockers[self._file_path]
 
     @property
-    def lock_file_path(self):
-        """
-        Returns
-        -------
-        str
-        """
-        return self._origin_file + '.lock'
+    def _file_path(self):
+        return self._fd.name
+
+
+if is_linux or is_macos:
+    import fcntl
+
+    # Use subclass of _FileThreadingLocker when min version of python update to >= 3
+    class _FileLocker:
+        def __init__(self, fd):
+            self._fd = fd
+
+        def __enter__(self):
+            try:
+                fcntl.lockf(self._fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            except IOError:
+                # Use `raise ... from ...` when min version of python update to >= 3
+                raise FileAlreadyLocked('File {0} already locked'.format(self._file_path))
+
+        def __exit__(self, exc_type, exc_val, exc_tb):
+            fcntl.lockf(self._fd, fcntl.LOCK_UN)
+
+        @property
+        def _file_path(self):
+            return self._fd.name
+
+elif is_windows:
+    import msvcrt
+
+
+    class _FileLocker:
+        def __init__(self, fd):
+            self._fd = fd
+
+        def __enter__(self):
+            try:
+                # TODO(lihs): set `_nbyte` bigger?
+                msvcrt.locking(self._fd, msvcrt.LK_LOCK | msvcrt.LK_NBLCK, 1)
+            except OSError:
+                raise FileAlreadyLocked('File {0} already locked'.format(self._file_path))
+
+        def __exit__(self, exc_type, exc_val, exc_tb):
+            msvcrt.locking(self._fd, msvcrt.LK_UNLCK, 1)
+
+        @property
+        def _file_path(self):
+            return self._fd.name
+
+else:
+    class _FileLocker:
+        def __init__(self, fd):
+            self._fd = fd
+
+        def __enter__(self):
+            try:
+                # Atomic file creation
+                open_flags = os.O_EXCL | os.O_RDWR | os.O_CREAT
+                fd = os.open(self.lock_file_path, open_flags)
+                os.close(fd)
+            except FileExistsError:
+                raise FileAlreadyLocked('File {0} already locked'.format(self._file_path))
+
+        def __exit__(self, exc_type, exc_val, exc_tb):
+            try:
+                os.remove(self.lock_file_path)
+            except FileNotFoundError:
+                pass
+
+        @property
+        def _file_path(self):
+            return self._fd.name
+
+        @property
+        def lock_file_path(self):
+            """
+            Returns
+            -------
+            str
+            """
+            return self._file_path + '.lock'
 
 
 # use dataclass instead namedtuple if min version of python update to 3.7
@@ -168,7 +254,8 @@ def lock_file_path(self):
         'persist_path',
         'last_shrink_at',
         'shrink_interval',
-        'should_shrink_expired_regions'
+        'should_shrink_expired_regions',
+        'memo_cache_lock'
     ]
 )
 
@@ -177,11 +264,12 @@ def lock_file_path(self):
     memo_cache={},
     persist_path=os.path.join(
         tempfile.gettempdir(),
-        'qn-regions-cache.jsonl'
+        'qn-py-sdk-regions-cache.jsonl'
     ),
     last_shrink_at=datetime.datetime.fromtimestamp(0),
-    shrink_interval=datetime.timedelta(-1),  # useless for now
-    should_shrink_expired_regions=False
+    shrink_interval=datetime.timedelta(days=1),
+    should_shrink_expired_regions=False,
+    memo_cache_lock=threading.Lock()
 )
 
 
@@ -323,7 +411,7 @@ def _parse_persisted_regions(persisted_data):
     return parsed_data.get('cacheKey'), regions
 
 
-def _walk_persist_cache_file(persist_path, ignore_parse_error=False):
+def _walk_persist_cache_file(persist_path, ignore_parse_error=True):
     """
     Parameters
     ----------
@@ -394,23 +482,24 @@ def __init__(
         self.base_regions_provider = base_regions_provider
 
         persist_path = kwargs.get('persist_path', None)
+        last_shrink_at = datetime.datetime.fromtimestamp(0)
         if persist_path is None:
             persist_path = _global_cache_scope.persist_path
+            last_shrink_at = _global_cache_scope.last_shrink_at
 
         shrink_interval = kwargs.get('shrink_interval', None)
         if shrink_interval is None:
-            shrink_interval = datetime.timedelta(days=1)
+            shrink_interval = _global_cache_scope.shrink_interval
 
         should_shrink_expired_regions = kwargs.get('should_shrink_expired_regions', None)
         if should_shrink_expired_regions is None:
-            should_shrink_expired_regions = False
+            should_shrink_expired_regions = _global_cache_scope.should_shrink_expired_regions
 
-        self._cache_scope = CacheScope(
-            memo_cache=_global_cache_scope.memo_cache,
+        self._cache_scope = _global_cache_scope._replace(
             persist_path=persist_path,
-            last_shrink_at=datetime.datetime.fromtimestamp(0),
+            last_shrink_at=last_shrink_at,
             shrink_interval=shrink_interval,
-            should_shrink_expired_regions=should_shrink_expired_regions,
+            should_shrink_expired_regions=should_shrink_expired_regions
         )
 
     def __iter__(self):
@@ -423,7 +512,7 @@ def __iter__(self):
             self.__get_regions_from_base_provider
         ]
 
-        regions = None
+        regions = []
         for get_regions in get_regions_fns:
             regions = get_regions(fallback=regions)
             if regions and all(r.is_live for r in regions):
@@ -439,7 +528,8 @@ def set_regions(self, regions):
         ----------
         regions: list[Region]
         """
-        self._cache_scope.memo_cache[self.cache_key] = regions
+        with self._cache_scope.memo_cache_lock:
+            self._cache_scope.memo_cache[self.cache_key] = regions
 
         if not self._cache_scope.persist_path:
             return
@@ -469,8 +559,11 @@ def persist_path(self, value):
         ----------
         value: str
         """
+        if value == self._cache_scope.persist_path:
+            return
         self._cache_scope = self._cache_scope._replace(
-            persist_path=value
+            persist_path=value,
+            last_shrink_at=datetime.datetime.fromtimestamp(0)
         )
 
     @property
@@ -586,7 +679,6 @@ def __get_regions_from_base_provider(self, fallback=None):
     def __flush_file_cache_to_memo(self):
         for cache_key, regions in _walk_persist_cache_file(
             persist_path=self._cache_scope.persist_path
-            # ignore_parse_error=True
         ):
             if cache_key not in self._cache_scope.memo_cache:
                 self._cache_scope.memo_cache[cache_key] = regions
@@ -609,12 +701,18 @@ def __should_shrink(self):
     def __shrink_cache(self):
         # shrink memory cache
         if self._cache_scope.should_shrink_expired_regions:
-            kept_memo_cache = {}
-            for k, regions in self._cache_scope.memo_cache.items():
-                live_regions = [r for r in regions if r.is_live]
-                if live_regions:
-                    kept_memo_cache[k] = live_regions
-            self._cache_scope = self._cache_scope._replace(memo_cache=kept_memo_cache)
+            memo_cache_old = self._cache_scope.memo_cache.copy()
+            # Could use keyword style `acquire(blocking=False)` when min version of python update to >= 3
+            if self._cache_scope.memo_cache_lock.acquire(False):
+                try:
+                    for k, regions in memo_cache_old.items():
+                        live_regions = [r for r in regions if r.is_live]
+                        if live_regions:
+                            self._cache_scope.memo_cache[k] = live_regions
+                        else:
+                            del self._cache_scope.memo_cache[k]
+                finally:
+                    self._cache_scope.memo_cache_lock.release()
 
         # shrink file cache
         if not self._cache_scope.persist_path:
@@ -625,7 +723,7 @@ def __shrink_cache(self):
 
         shrink_file_path = self._cache_scope.persist_path + '.shrink'
         try:
-            with _FileLocker(shrink_file_path):
+            with open(shrink_file_path, 'a') as f, _FileThreadingLocker(f), _FileLocker(f):
                 # filter data
                 shrunk_cache = {}
                 for cache_key, regions in _walk_persist_cache_file(
@@ -646,25 +744,36 @@ def __shrink_cache(self):
                         )
 
                 # write data
-                with open(shrink_file_path, 'a') as f:
-                    for cache_key, regions in shrunk_cache.items():
-                        f.write(
-                            json.dumps(
-                                {
-                                    'cacheKey': cache_key,
-                                    'regions': [_persist_region(r) for r in regions]
-                                }
-                            ) + os.linesep
-                        )
+                for cache_key, regions in shrunk_cache.items():
+                    f.write(
+                        json.dumps(
+                            {
+                                'cacheKey': cache_key,
+                                'regions': [_persist_region(r) for r in regions]
+                            }
+                        ) + os.linesep
+                    )
+
+                # make the cache file available for all users
+                if is_linux or is_macos:
+                    os.chmod(shrink_file_path, 0o666)
 
                 # rename file
                 shutil.move(shrink_file_path, self._cache_scope.persist_path)
+
+                # update last shrink time
+                self._cache_scope = self._cache_scope._replace(
+                    last_shrink_at=datetime.datetime.now()
+                )
+                global _global_cache_scope
+                if _global_cache_scope.persist_path == self._cache_scope.persist_path:
+                    _global_cache_scope = _global_cache_scope._replace(
+                        last_shrink_at=self._cache_scope.last_shrink_at
+                    )
+
         except FileAlreadyLocked:
+            # skip file shrink by another running
             pass
-        finally:
-            self._cache_scope = self._cache_scope._replace(
-                last_shrink_at=datetime.datetime.now()
-            )
 
 
 def get_default_regions_provider(

qiniu/http/response.py

@@ -57,6 +57,7 @@ def need_retry(self):
         ]):
             return False
         # https://developer.qiniu.com/fusion/kb/1352/the-http-request-return-a-status-code
+        # https://developer.qiniu.com/kodo/3928/error-responses
         if self.status_code in [
             501, 509, 573, 579, 608, 612, 614, 616, 618, 630, 631, 632, 640, 701
         ]: