Remove synchronized in Waiter

Author: kagkarlsson

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/Waiter.java

@@ -15,16 +15,21 @@
 
 import java.time.Duration;
 import java.time.Instant;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Supplier;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class Waiter {
+
   private static final Logger LOG = LoggerFactory.getLogger(Waiter.class);
 
-  private final Object lock;
+  private final ReentrantLock lock;
+  private final Condition condition;
   private boolean woken = false;
   private final Duration duration;
-  private Clock clock;
+  private final Clock clock;
   private boolean isWaiting = false;
   private boolean skipNextWait = false;
 
@@ -33,66 +38,77 @@ public Waiter(Duration duration) {
   }
 
   public Waiter(Duration duration, Clock clock) {
-    this(duration, clock, new Object());
+    this.duration = duration;
+    this.clock = clock;
+    this.lock = new ReentrantLock();
+    this.condition = this.lock.newCondition();
   }
 
-  Waiter(Duration duration, Clock clock, Object lock) {
+  Waiter(Duration duration, Clock clock, ReentrantLock lock, Condition condition) {
     this.duration = duration;
     this.clock = clock;
     this.lock = lock;
+    this.condition = condition;
   }
 
   public void doWait() throws InterruptedException {
     long millis = duration.toMillis();
 
-    if (millis > 0) {
-      Instant waitUntil = clock.now().plusMillis(millis);
-
-      while (clock.now().isBefore(waitUntil)) {
-        synchronized (lock) {
-          if (skipNextWait) {
-            LOG.debug("Waiter has been notified to skip next wait-period. Skipping wait.");
-            skipNextWait = false;
-            return;
-          }
-
-          woken = false;
-          LOG.debug("Waiter start wait.");
-          this.isWaiting = true;
-          lock.wait(millis);
-          this.isWaiting = false;
-          if (woken) {
-            LOG.debug(
-                "Waiter woken, it had {}ms left to wait.",
-                (waitUntil.toEpochMilli() - clock.now().toEpochMilli()));
-            return;
-          }
+    if (millis <= 0) {
+      return;
+    }
+
+    Instant waitUntil = clock.now().plusMillis(millis);
+
+    while (clock.now().isBefore(waitUntil)) {
+      lock.lock();
+      try {
+        if (skipNextWait) {
+          LOG.debug("Waiter has been notified to skip next wait-period. Skipping wait.");
+          skipNextWait = false;
+          return;
+        }
+
+        woken = false;
+        LOG.debug("Waiter start wait.");
+        this.isWaiting = true;
+        condition.awaitNanos(millis * 1_000_000);
+        this.isWaiting = false;
+        if (woken) {
+          LOG.debug(
+            "Waiter woken, it had {}ms left to wait.",
+            (waitUntil.toEpochMilli() - clock.now().toEpochMilli()));
+          return;
         }
+
+      } finally {
+        lock.unlock();
       }
     }
   }
 
-  public boolean wake() {
-    synchronized (lock) {
+  boolean wake() {
+    return withLock(lock, () -> {
       if (!isWaiting) {
         return false;
       } else {
         woken = true;
-        lock.notify();
+        condition.signal();
         return true;
       }
-    }
+    });
   }
 
   public void wakeOrSkipNextWait() {
     // Take early lock to avoid race-conditions. Lock is also taken in wake() (lock is re-entrant)
-    synchronized (lock) {
+    withLock(lock, () -> {
       final boolean awoken = wake();
       if (!awoken) {
         LOG.debug("Waiter not waiting, instructing to skip next wait.");
         this.skipNextWait = true;
       }
-    }
+      return null;
+    });
   }
 
   public Duration getWaitDuration() {
@@ -102,4 +118,20 @@ public Duration getWaitDuration() {
   public boolean isWaiting() {
     return isWaiting;
   }
+
+  public static <T> T withLock(ReentrantLock lock, Supplier<T> action) {
+    lock.lock();
+    try {
+      return action.get();
+    } finally {
+      lock.unlock();
+    }
+  }
+
+  public static <T> T withLock(ReentrantLock lock, Runnable action) {
+    return withLock(lock, () -> {
+      action.run();
+      return null;
+    });
+  }
 }

db-scheduler/src/test/java/com/github/kagkarlsson/scheduler/WaiterTest.java

@@ -4,11 +4,14 @@
 
 import java.time.Duration;
 import java.util.concurrent.*;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 public class WaiterTest {
+
   private ExecutorService executor;
 
   @BeforeEach
@@ -45,23 +48,21 @@ public void should_wait_until_woken() throws ExecutionException, InterruptedExce
 
   @Test
   public void should_wait_for_duration_even_if_prematurely_notified()
-      throws ExecutionException, InterruptedException {
-    Object lock = new Object();
+    throws ExecutionException, InterruptedException {
+    ReentrantLock lock = new ReentrantLock();
+    Condition condition = lock.newCondition();
 
-    Waiter waiter = new Waiter(Duration.ofMillis(200), new SystemClock(), lock);
+    Waiter waiter = new Waiter(Duration.ofMillis(200), new SystemClock(), lock, condition);
     Future<Long> waitTime = executor.submit(new WaitForWaiter(waiter));
     sleep(20); // give executor time to get to wait(..)
 
-    synchronized (lock) {
-      lock.notify();
-    }
-
+    Waiter.withLock(lock, condition::signal);
     assertTrue(waitTime.get() >= 200L, "Waited: " + waitTime.get());
   }
 
   @Test
   public void should_not_wait_if_instructed_to_skip_next()
-      throws ExecutionException, InterruptedException {
+    throws ExecutionException, InterruptedException {
     Waiter waiter = new Waiter(Duration.ofMillis(1000));
     waiter.wakeOrSkipNextWait(); // set skip
     Future<Long> waitTime = executor.submit(new WaitForWaiter(waiter));
@@ -77,6 +78,7 @@ private void sleep(int millis) {
   }
 
   private static class WaitForWaiter implements Callable<Long> {
+
     private Waiter waiter;
 
     WaitForWaiter(Waiter waiter) {