Merge pull request #11815 from vespa-engine/jvenstad/wrap-curator-mutex-with-reentarnt-lock

Hold a JVM-wide reentrant lock to grab mutex — helps ZK stale reads?

1 files changed, 22 insertions, 6 deletions

diff --git a/zkfacade/src/main/java/com/yahoo/vespa/curator/Lock.java b/zkfacade/src/main/java/com/yahoo/vespa/curator/Lock.java
index 30af89d0ea8..2e554d39e44 100644
--- a/zkfacade/src/main/java/com/yahoo/vespa/curator/Lock.java
+++ b/zkfacade/src/main/java/com/yahoo/vespa/curator/Lock.java
@@ -7,6 +7,7 @@ import org.apache.curator.framework.recipes.locks.InterProcessLock;
 
 import java.time.Duration;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
 
 /**
  * A cluster-wide re-entrant mutex which is released on (the last symmetric) close
@@ -15,32 +16,47 @@ import java.util.concurrent.TimeUnit;
  */
 public class Lock implements Mutex {
 
+    private final ReentrantLock lock;
     private final InterProcessLock mutex;
     private final String lockPath;
 
     public Lock(String lockPath, Curator curator) {
         this.lockPath = lockPath;
+        this.lock = new ReentrantLock(true);
         mutex = curator.createMutex(lockPath);
     }
 
     /** Take the lock with the given timeout. This may be called multiple times from the same thread - each matched by a close */
     public void acquire(Duration timeout) throws UncheckedTimeoutException {
-        boolean acquired;
         try {
-            acquired = mutex.acquire(timeout.toMillis(), TimeUnit.MILLISECONDS);
+            if ( ! mutex.acquire(timeout.toMillis(), TimeUnit.MILLISECONDS))
+                throw new UncheckedTimeoutException("Timed out after waiting " + timeout +
+                                                    " to acquire lock '" + lockPath + "'");
+            if ( ! lock.tryLock()) { // Should be available to only this thread, while holding the above mutex.
+                release();
+                throw new IllegalStateException("InterProcessMutex acquired, but guarded lock held by someone else, for lock '" + lockPath + "'");
+            }
+        }
+        catch (UncheckedTimeoutException | IllegalStateException e) {
+            throw e;
         }
         catch (Exception e) {
             throw new RuntimeException("Exception acquiring lock '" + lockPath + "'", e);
         }
-
-        if (! acquired)
-            throw new UncheckedTimeoutException("Timed out after waiting " + timeout +
-                                                " to acquire lock '" + lockPath + "'");
     }
 
     @Override
     public void close() {
         try {
+            lock.unlock();
+        }
+        finally {
+            release();
+        }
+    }
+
+    private void release() {
+        try {
             mutex.release();
         }
         catch (Exception e) {