Merge pull request #31 from ENCODE-DCC/dev

v0.4.1

autouri/__init__.py

@@ -5,4 +5,4 @@
 from .s3uri import S3URI
 
 __all__ = ["AbsPath", "AutoURI", "URIBase", "GCSURI", "HTTPURL", "S3URI"]
-__version__ = "0.4.0"
+__version__ = "0.4.1"

autouri/gcsuri.py

@@ -26,6 +26,7 @@
 
 from .autouri import AutoURI, URIBase
 from .metadata import URIMetadata, get_seconds_from_epoch, parse_md5_str
+from .ntp_now import now_utc
 
 logger = logging.getLogger(__name__)
 
@@ -64,6 +65,14 @@ def add_google_app_creds_to_env(service_account_key_file):
 
 
 class GCSURILock(BaseFileLock):
+    """Class constants:
+    - LOCK_FILE_EXPIRATION_SEC:
+        Expiration of a lock file based on its modtime in seconds
+        If expired then such lock file is ignored.
+    """
+
+    LOCK_FILE_EXPIRATION_SEC = 1800
+
     def __init__(self, lock_file, timeout=900, poll_interval=10.0, no_lock=False):
         super().__init__(lock_file, timeout=timeout)
         self._poll_interval = poll_interval
@@ -80,14 +89,19 @@ def _acquire(self):
         u = GCSURI(self._lock_file)
         str_id = str(id(self))
         try:
-            if not u.exists:
+            if (
+                not u.exists
+                or now_utc().timestamp() > u.mtime + GCSURILock.LOCK_FILE_EXPIRATION_SEC
+            ):
                 u.write(str_id, no_lock=True)
-                time.sleep(self._lock_read_delay)
-            if u.read() == str_id:
                 self._lock_file_fd = id(self)
+
+            elif u.read() == str_id:
+                self._lock_file_fd = id(self)
+
         except (Forbidden, NotFound):
             raise
-        except ClientError:
+        except (ClientError, ValueError):
             pass
         return None
 

autouri/s3uri.py

@@ -2,7 +2,6 @@
     S3 Object versioning must be turned off
 """
 import logging
-import time
 from tempfile import NamedTemporaryFile
 from typing import Optional, Tuple
 
@@ -13,6 +12,7 @@
 
 from .autouri import AutoURI, URIBase
 from .metadata import URIMetadata, get_seconds_from_epoch, parse_md5_str
+from .ntp_now import now_utc
 
 logger = logging.getLogger(__name__)
 
@@ -28,8 +28,15 @@ class S3URILock(BaseFileLock):
     This module first checks if .lock does not exist, then tries to write .lock with id(self).
     It waits for a short time (self._lock_read_delay) and checks if written .lock has the same id(self).
     self._lock_read_delay is set as poll_interval/10.
+
+    Class constants:
+    - LOCK_FILE_EXPIRATION_SEC:
+        Expiration of a lock file based on its modtime in seconds
+        If expired then such lock file is ignored.
     """
 
+    LOCK_FILE_EXPIRATION_SEC = 1800
+
     def __init__(self, lock_file, timeout=900, poll_interval=10.0, no_lock=False):
         super().__init__(lock_file, timeout=timeout)
         self._poll_interval = poll_interval
@@ -46,11 +53,16 @@ def _acquire(self):
         u = S3URI(self._lock_file)
         str_id = str(id(self))
         try:
-            if not u.exists:
+            if (
+                not u.exists
+                or now_utc().timestamp() > u.mtime + S3URILock.LOCK_FILE_EXPIRATION_SEC
+            ):
                 u.write(str_id, no_lock=True)
-                time.sleep(self._lock_read_delay)
-            if u.read() == str_id:
                 self._lock_file_fd = id(self)
+
+            elif u.read() == str_id:
+                self._lock_file_fd = id(self)
+
         except ClientError as e:
             status = e.response["ResponseMetadata"]["HTTPStatusCode"]
             if status in (403, 404):