fix and re-enable testQueueConcurrencyUnderRecovery (#45)

Fixes #44

Author: devhawk

src/main/java/dev/dbos/transact/interceptor/BaseInvocationHandler.java

@@ -50,7 +50,7 @@ protected Object handleWorkflow(Method method, Object[] args, Workflow workflow)
 
         String workflowName = workflow.name().isEmpty() ? method.getName() : workflow.name();
 
-        String msg = String.format("Before: Starting workflow '%s' (timeout: %ds)%n",
+        String msg = String.format("Before: Starting workflow '%s' (timeout: %ds)",
                 workflowName,
                 workflow.timeout());
 

src/main/java/dev/dbos/transact/queue/QueueService.java

@@ -98,7 +98,7 @@ private void pollForWorkflows() {
 
         } finally {
             shutdownLatch.countDown();
-            logger.info("QueuesPolThread has ended. Exiting " + Thread.currentThread().getId());
+            logger.info("QueuesPollThread has ended. Exiting " + Thread.currentThread().getId());
         }
     }
 

src/test/java/dev/dbos/transact/conductor/TestWebSocketServer.java

@@ -1,8 +1,8 @@
 package dev.dbos.transact.conductor;
 
-import dev.dbos.transact.utils.ManualResetEvent;
-
 import java.net.InetSocketAddress;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
 
 import org.java_websocket.WebSocket;
 import org.java_websocket.framing.Framedata;
@@ -38,7 +38,7 @@ default void onClose(WebSocket conn, int code, String reason, boolean remote) {
 
     private static Logger logger = LoggerFactory.getLogger(TestWebSocketServer.class);
     private WebSocketTestListener listener;
-    private ManualResetEvent startEvent = new ManualResetEvent(false);
+    private Semaphore startEvent = new Semaphore(0);
 
     public TestWebSocketServer(int port) {
         super(new InetSocketAddress(port));
@@ -48,12 +48,8 @@ public void setListener(WebSocketTestListener listener) {
         this.listener = listener;
     }
 
-    public void waitStart() throws InterruptedException {
-        startEvent.waitOne();
-    }
-
     public boolean waitStart(long millis) throws InterruptedException {
-        return startEvent.waitOne(millis);
+        return startEvent.tryAcquire(millis, TimeUnit.MILLISECONDS);
     }
 
     @Override
@@ -66,7 +62,7 @@ public void onOpen(WebSocket conn, ClientHandshake handshake) {
 
     @Override
     public void onClose(WebSocket conn, int code, String reason, boolean remote) {
-        startEvent.reset();
+        startEvent.drainPermits();
         logger.info("onClose");
         if (listener != null) {
             listener.onClose(conn, code, reason, remote);
@@ -91,7 +87,7 @@ public void onWebsocketPing(WebSocket conn, Framedata f) {
 
     @Override
     public void onError(WebSocket conn, Exception ex) {
-        startEvent.reset();
+        startEvent.drainPermits();
         logger.error("onError", ex);
         if (listener != null) {
             listener.onError(conn, ex);
@@ -100,7 +96,7 @@ public void onError(WebSocket conn, Exception ex) {
 
     @Override
     public void onStart() {
-        startEvent.set();
+        startEvent.release();
         logger.info("onStart {}", getPort());
         if (listener != null) {
             listener.onStart();

src/test/java/dev/dbos/transact/queue/ConcurrencyTestService.java

@@ -3,5 +3,5 @@
 public interface ConcurrencyTestService {
     public int noopWorkflow(int i);
 
-    public int blockedWorkflow(int i);
+    public int blockedWorkflow(int i) throws InterruptedException;
 }

src/test/java/dev/dbos/transact/queue/ConcurrencyTestServiceImpl.java

@@ -1,14 +1,20 @@
 package dev.dbos.transact.queue;
 
-import dev.dbos.transact.utils.ManualResetEvent;
 import dev.dbos.transact.workflow.Workflow;
 
 import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Semaphore;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class ConcurrencyTestServiceImpl implements ConcurrencyTestService {
 
-    public ManualResetEvent event = new ManualResetEvent(false);
-    public List<ManualResetEvent> wfEvents = List.of(new ManualResetEvent(false), new ManualResetEvent(false));
+    private static final Logger logger = LoggerFactory.getLogger(ConcurrencyTestServiceImpl.class);
+
+    public CountDownLatch latch = new CountDownLatch(1);
+    public List<Semaphore> wfSemaphores = List.of(new Semaphore(0), new Semaphore(0));
     public int counter = 0;
 
     @Workflow(name = "noopWorkflow")
@@ -17,11 +23,11 @@ public int noopWorkflow(int i) {
     }
 
     @Workflow(name = "blockedWorkflow")
-    public int blockedWorkflow(int i) {
-        wfEvents.get(i).set();
+    public int blockedWorkflow(int i) throws InterruptedException {
+        logger.info("release {} semaphore", i);
+        wfSemaphores.get(i).release();
         counter++;
-        event.waitOne();
+        latch.await();
         return i;
     }
-
 }

src/test/java/dev/dbos/transact/queue/QueuesTest.java

@@ -12,7 +12,6 @@
 import dev.dbos.transact.database.SystemDatabase;
 import dev.dbos.transact.execution.DBOSExecutor;
 import dev.dbos.transact.utils.DBUtils;
-import dev.dbos.transact.utils.ManualResetEvent;
 import dev.dbos.transact.workflow.WorkflowHandle;
 import dev.dbos.transact.workflow.WorkflowState;
 import dev.dbos.transact.workflow.WorkflowStatus;
@@ -26,6 +25,7 @@
 import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.Semaphore;
 
 import javax.sql.DataSource;
 
@@ -514,9 +514,9 @@ public void testQueueConcurrencyUnderRecovery() throws Exception {
             handle3 = dbos.startWorkflow(() -> service.noopWorkflow(2));
         }
 
-        for (ManualResetEvent e : impl.wfEvents) {
-            e.waitOne();
-            e.reset();
+        for (Semaphore e : impl.wfSemaphores) {
+            e.acquire();
+            e.drainPermits();
         }
 
         assertEquals(2, impl.counter);
@@ -551,22 +551,22 @@ public void testQueueConcurrencyUnderRecovery() throws Exception {
         assertTrue(expectedWorkflowIds.contains(localHandles.get(0).getWorkflowId()));
         assertTrue(expectedWorkflowIds.contains(localHandles.get(1).getWorkflowId()));
 
-        for (ManualResetEvent e : impl.wfEvents) {
-            e.waitOne();
+        for (int i = 0; i < impl.wfSemaphores.size(); i++) {
+            logger.info("acquire {} semaphore", i);
+            impl.wfSemaphores.get(i).acquire();
         }
 
         assertEquals(4, impl.counter);
         assertEquals(WorkflowState.PENDING.toString(), handle1.getStatus().getStatus());
         assertEquals(WorkflowState.PENDING.toString(), handle2.getStatus().getStatus());
         assertEquals(WorkflowState.ENQUEUED.toString(), handle3.getStatus().getStatus());
 
-        impl.event.set();
+        impl.latch.countDown();
         assertEquals(0, handle1.getResult());
         assertEquals(1, handle2.getResult());
         assertEquals(2, handle3.getResult());
         assertEquals("local", handle3.getStatus().getExecutorId());
 
         assertTrue(DBUtils.queueEntriesAreCleanedUp(dataSource));
-
     }
 }