Release the thread lock when IPC acquisition fails

CHANGES.rst

@@ -89,6 +89,10 @@ Compute
 Storage
 ~~~~~~~
 
+- [Local Storage] Release the thread lock if IPC lock acquisition fails,
+  allowing the same lock object to be reused after a timeout or error.
+  (#2203)
+
 - [Local Storage] Prevent source objects from being deleted when
   ``download_object`` is called with a destination that resolves
   to the source file.

libcloud/storage/drivers/local.py

@@ -102,13 +102,17 @@ def __enter__(self):
                 "in %s seconds" % (self.path, lock_acquire_timeout)
             )
 
-        success = self.ipc_lock.acquire(blocking=True, timeout=lock_acquire_timeout)
+        try:
+            success = self.ipc_lock.acquire(blocking=True, timeout=lock_acquire_timeout)
 
-        if not success:
-            raise LibcloudError(
-                "Failed to acquire IPC lock (%s) for path %s "
-                "in %s seconds" % (self.ipc_lock_path, self.path, lock_acquire_timeout)
-            )
+            if not success:
+                raise LibcloudError(
+                    "Failed to acquire IPC lock (%s) for path %s "
+                    "in %s seconds" % (self.ipc_lock_path, self.path, lock_acquire_timeout)
+                )
+        except BaseException:
+            self.thread_lock.release()
+            raise
 
     def __exit__(self, type, value, traceback):
         if self.thread_lock.locked():

libcloud/test/storage/test_local.py

@@ -22,6 +22,7 @@
 import tempfile
 import unittest
 import multiprocessing
+from unittest.mock import patch
 
 from libcloud.utils.files import exhaust_iterator
 from libcloud.common.types import LibcloudError
@@ -139,6 +140,22 @@ def test_lock_local_storage(self):
         self.assertEqual(bool(success_1.value), True, "Check didn't pass")
         self.assertEqual(bool(success_2.value), True, "Second check didn't pass")
 
+    def test_lock_local_storage_releases_thread_lock_on_ipc_failure(self):
+        for error in (None, OSError("lock unavailable"), KeyboardInterrupt()):
+            with self.subTest(error=error):
+                lock = LockLocalStorage(os.path.join(self.key, "lock-retry"))
+                first_result = False if error is None else error
+                with patch.object(lock.ipc_lock, "acquire", side_effect=[first_result, True]):
+                    expected_error = LibcloudError if error is None else type(error)
+                    with self.assertRaises(expected_error) as exc:
+                        lock.__enter__()
+                    if error is not None:
+                        self.assertIs(exc.exception, error)
+                    self.assertFalse(lock.thread_lock.locked())
+                    with lock:
+                        self.assertTrue(lock.thread_lock.locked())
+                    self.assertFalse(lock.thread_lock.locked())
+
     def test_list_containers_empty(self):
         containers = self.driver.list_containers()
         self.assertEqual(len(containers), 0)