finished rwlock

Author: wieceslaw

coordination/src/main/java/tech/ydb/coordination/recipes/election/LeaderElection.java

@@ -277,7 +277,7 @@ private void waitStartedState() throws InterruptedException {
             localState = state.get();
         }
 
-        if (localState == State.INITIAL || localState == State.CLOSED || localState == State.FAILED) {
+        if (localState != State.STARTED) {
             throw new IllegalStateException("Unexpected state: " + localState.name());
         }
     }

coordination/src/main/java/tech/ydb/coordination/recipes/locks/InterProcessLock.java

@@ -36,11 +36,11 @@ boolean acquire(Duration waitDuration) throws Exception, LockAlreadyAcquiredExce
      * Releases the lock if it is held by this process.
      *
      * @return false if there was nothing to release
-     * @throws InterruptedException       if the thread is interrupted
+     * @throws Exception                  if an unexpected error occurs
      * @throws LockReleaseFailedException if the lock release fails
      * @throws LockStateException         if the lock is in invalid state for release
      */
-    boolean release() throws InterruptedException, LockReleaseFailedException, LockStateException;
+    boolean release() throws Exception, LockReleaseFailedException, LockStateException;
 
 
     /**

coordination/src/main/java/tech/ydb/coordination/recipes/locks/InterProcessMutex.java

@@ -3,6 +3,7 @@
 import java.io.Closeable;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.Objects;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicReference;
@@ -24,12 +25,14 @@
 public class InterProcessMutex implements InterProcessLock, Closeable {
     private static final Logger logger = LoggerFactory.getLogger(InterProcessMutex.class);
 
-    private final AtomicReference<State> state = new AtomicReference<>(State.INITIAL);
+    private final String lockName;
     private final CoordinationSession coordinationSession;
     private final Future<?> sessionConnectionTask;
     private final LockInternals lockInternals;
     private final ListenableContainer<CoordinationSession.State> sessionListenable;
 
+    private final AtomicReference<State> state = new AtomicReference<>(State.INITIAL);
+
     /**
      * Internal state machine states
      */
@@ -82,15 +85,14 @@ public InterProcessMutex(
         }
 
         state.set(State.STARTING);
-        logger.debug("Initializing InterProcessMutex for lock '{}'", lockName);
-
+        this.lockName = lockName;
         this.coordinationSession = client.createSession(coordinationNodePath);
         this.sessionListenable = new ListenableContainer<>();
         this.lockInternals = new LockInternals(coordinationSession, lockName);
 
         coordinationSession.addStateListener(sessionState -> {
             if (sessionState == CoordinationSession.State.LOST || sessionState == CoordinationSession.State.CLOSED) {
-                logger.error("Coordination session unexpectedly changed to {} state, marking mutex as FAILED",
+                logger.error("Coordination session unexpectedly changed to '{}' state, marking mutex as 'FAILED'",
                         sessionState);
                 state.set(State.FAILED);
             }
@@ -109,9 +111,8 @@ public InterProcessMutex(
 
         if (settings.isWaitConnection()) {
             try {
-                logger.debug("Waiting for session connection to complete...");
+                logger.debug("Waiting for session connection to complete for lock '{}'", lockName);
                 sessionConnectionTask.get();
-                logger.debug("Session connection completed");
             } catch (InterruptedException e) {
                 Thread.currentThread().interrupt();
                 logger.error("Interrupted while waiting for session connection for lock '{}'", lockName, e);
@@ -128,36 +129,38 @@ public InterProcessMutex(
     @Override
     public void acquire() throws Exception {
         checkState();
-        logger.debug("Attempting to acquire lock...");
+        logger.debug("Attempting to acquire lock '{}'", lockName);
         lockInternals.tryAcquire(
                 null,
                 true,
                 null
         );
-        logger.debug("Lock acquired successfully");
+        logger.debug("Lock '{}' acquired successfully", lockName);
     }
 
     @Override
     public boolean acquire(Duration waitDuration) throws Exception {
+        Objects.requireNonNull(waitDuration, "wait duration must not be null");
+
         checkState();
-        logger.debug("Attempting to acquire lock with timeout {}...", waitDuration);
+        logger.debug("Attempting to acquire lock '{}' with timeout {}", lockName, waitDuration);
         Instant deadline = Instant.now().plus(waitDuration);
         boolean acquired = lockInternals.tryAcquire(
                 deadline,
                 true,
                 null
         ) != null;
-        logger.debug("Lock acquisition {}successful", acquired ? "" : "un");
+        logger.debug("Lock '{}' acquisition {}successful", lockName, acquired ? "" : "un");
         return acquired;
     }
 
     @Override
     public boolean release() throws InterruptedException {
         checkState();
-        logger.debug("Attempting to release lock...");
+        logger.debug("Attempting to release lock '{}'", lockName);
         boolean released = lockInternals.release();
         if (released) {
-            logger.debug("Lock released successfully");
+            logger.debug("Lock '{}' released successfully", lockName);
         } else {
             logger.debug("No lock to release");
         }
@@ -184,29 +187,27 @@ public Listenable<CoordinationSession.State> getSessionListenable() {
 
     @Override
     public void close() {
-        logger.debug("Closing InterProcessMutex...");
+        logger.debug("Closing mutex '{}'", lockName);
         state.set(State.CLOSED);
         try {
             lockInternals.close();
         } catch (Exception e) {
-            logger.warn("Error while closing lock internals", e);
+            logger.warn("Error while closing lock internals '{}'", lockName, e);
         }
-        logger.info("InterProcessMutex closed");
+        logger.info("Mutex '{}' closed", lockName);
     }
 
     private void checkState() throws LockStateException {
         State currentState = state.get();
         if (currentState == State.FAILED) {
-            throw new LockStateException("Lock '" + lockInternals.getLockName() + "' is in FAILED state",
-                    lockInternals.getLockName());
+            throw new LockStateException("Lock '" + lockName + "' is in FAILED state", lockName);
         }
         if (currentState == State.CLOSED) {
-            throw new LockStateException("Lock '" + lockInternals.getLockName() + "' is already closed",
-                    lockInternals.getLockName());
+            throw new LockStateException("Lock '" + lockName + "' is already closed", lockName);
         }
         if (currentState != State.STARTED) {
-            throw new LockStateException("Lock '" + lockInternals.getLockName() + "' is not ready (current state: "
-                    + currentState + ")", lockInternals.getLockName());
+            throw new LockStateException("Lock '" + lockName + "' is not ready (current state: " + currentState + ")",
+                    lockName);
         }
     }
 }