Double-buffer SideEffectQueue to avoid blocking enqueue during flush

Author: Jasper Geurtz

src/main/java/bwapi/SideEffectQueue.java

@@ -8,6 +8,7 @@
 class SideEffectQueue {
 
     private ArrayList<SideEffect> queue = new ArrayList<>();
+    private ArrayList<SideEffect> flushQueue = new ArrayList<>();
 
     /**
      * Includes a side effect to be sent back to BWAPI in the future.
@@ -25,8 +26,19 @@ synchronized void enqueue(SideEffect sideEffect) {
      * @param liveGameData
      * The live game frame's data, using the BWAPI shared memory.
      */
-    synchronized void flushTo(ClientData.GameData liveGameData) {
-        queue.forEach(x -> x.apply(liveGameData));
-        queue.clear();
+    void flushTo(ClientData.GameData liveGameData) {
+        final ArrayList<SideEffect> toFlush;
+        synchronized (this) {
+            if (queue.isEmpty()) {
+                return;
+            }
+            toFlush = queue;
+            queue = flushQueue;
+            flushQueue = toFlush;
+        }
+        for (int i = 0, size = toFlush.size(); i < size; i++) {
+            toFlush.get(i).apply(liveGameData);
+        }
+        toFlush.clear();
     }
 }

src/test/java/bwapi/SynchronizationTest.java

@@ -388,4 +388,46 @@ public void async_DisablesLatencyCompensation() {
         environmentAsync.onFrame(2, () -> { assertThrows(IllegalStateException.class, () -> environmentAsync.bwClient.getGame().setLatCom(true)); });
         environmentAsync.runGame(3);
     }
+
+    @Test
+    public void sideEffectQueue_DoesNotBlockEnqueueDuringFlush() throws Exception {
+        SideEffectQueue queue = new SideEffectQueue();
+        CountDownLatch inFlush = new CountDownLatch(1);
+        CountDownLatch enqueueDone = new CountDownLatch(1);
+        AtomicBoolean secondApplied = new AtomicBoolean(false);
+
+        SideEffect first = org.mockito.Mockito.mock(SideEffect.class);
+        org.mockito.Mockito.doAnswer(inv -> {
+            inFlush.countDown();
+            assertTrue("enqueue should not be blocked while flushTo is applying side effects",
+                    enqueueDone.await(5, TimeUnit.SECONDS));
+            return null;
+        }).when(first).apply(null);
+
+        SideEffect second = org.mockito.Mockito.mock(SideEffect.class);
+        org.mockito.Mockito.doAnswer(inv -> {
+            secondApplied.set(true);
+            return null;
+        }).when(second).apply(null);
+
+        queue.enqueue(first);
+
+        Thread enqueuer = new Thread(() -> {
+            try {
+                assertTrue(inFlush.await(5, TimeUnit.SECONDS));
+                queue.enqueue(second);
+                enqueueDone.countDown();
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        });
+        enqueuer.start();
+
+        queue.flushTo(null);
+        enqueuer.join(5000);
+        assertFalse("Second side effect should remain queued for the next flush", secondApplied.get());
+
+        queue.flushTo(null);
+        assertTrue("Second side effect should be applied on the next flush", secondApplied.get());
+    }
 }