Merge pull request #12 from danieladams456/concurrency-fix

Merge Status Concurrency Fix

Author: danieladams456

src/main/java/com/github/danieladams456/statusvoter/StatusVoter.java

@@ -14,10 +14,14 @@ public StatusVoter() {
     /**
      * Merges the incoming classification with the existing status classification.
      * The status classification with the higher score is retained.
+     * <p>
+     * This method is marked as synchronized to avoid concurrency bugs.
+     * It is not a huge penalty to mark the whole method as such since it
+     * is fast in comparison to the work this status is tracking.
      *
      * @param newClassification the new, incoming classification to merge with the existing classification
      */
-    public void merge(StatusClassification newClassification) {
+    public synchronized void merge(StatusClassification newClassification) {
         // null value is reset to a legitimate non-null value
         if (null == newClassification) {
             newClassification = StatusClassification.INTERNAL_STATUS_MERGE_ERROR;

src/test/java/com/github/danieladams456/statusvoter/ConcurrencyTest.java

@@ -0,0 +1,58 @@
+package com.github.danieladams456.statusvoter;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static com.github.danieladams456.statusvoter.TestData.STATUSES_IN_ORDER;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class ConcurrencyTest {
+    /**
+     * This test checks for atomic compare-and-set merges.
+     * Set each status on a separate thread, and always expect the final result to be the most severe.
+     */
+    @Test
+    void multipleConcurrentUpdates() throws InterruptedException {
+        final int testIterations = 10_000;
+        int failureCount = 0;  // should not need AtomicInteger
+
+        for (int i = 0; i < testIterations; i++) {
+            try (ExecutorService executor = Executors.newFixedThreadPool(STATUSES_IN_ORDER.size())) {
+                CountDownLatch startLatch = new CountDownLatch(1);
+                CountDownLatch doneLatch = new CountDownLatch(STATUSES_IN_ORDER.size());
+                StatusVoter voter = new StatusVoter();
+
+                for (StatusClassification statusClassification : STATUSES_IN_ORDER) {
+                    executor.submit(() -> {
+                        try {
+                            startLatch.await();
+                            Thread.sleep(1);
+                            voter.merge(statusClassification);
+                        } catch (InterruptedException e) {
+                            Thread.currentThread().interrupt();
+                        } finally {
+                            doneLatch.countDown();
+                        }
+                    });
+                }
+
+                // single countDown() will unblock
+                startLatch.countDown();
+                // wait until all threads have completed their operations, then shut down thread pool.
+                doneLatch.await();
+                executor.shutdown();
+                assertThat(executor.awaitTermination(100, TimeUnit.MILLISECONDS)).isTrue();
+
+                // result should always be the most critical status
+                if (voter.getClassification() != StatusClassification.INTERNAL_ERROR) {
+                    failureCount++;
+                }
+            }
+        }
+        assertThat(failureCount).isEqualTo(0);
+    }
+}

src/test/java/com/github/danieladams456/statusvoter/StateTransitionTest.java

@@ -5,20 +5,10 @@
 import org.junit.jupiter.params.provider.NullAndEmptySource;
 import org.junit.jupiter.params.provider.ValueSource;
 
-import java.util.List;
-
+import static com.github.danieladams456.statusvoter.TestData.STATUSES_IN_ORDER;
 import static org.assertj.core.api.Assertions.assertThat;
 
 public class StateTransitionTest {
-    private static final List<StatusClassification> statusesInOrder = List.of(
-            StatusClassification.INITIAL,
-            StatusClassification.SUCCESS,
-            StatusClassification.UNCLEAR_ATTRIBUTION_ERROR,
-            StatusClassification.CUSTOMER_DATA_ERROR,
-            StatusClassification.CUSTOMER_DATA_VALIDATION_ERROR,
-            StatusClassification.INTERNAL_STATUS_MERGE_ERROR,
-            StatusClassification.INTERNAL_ERROR
-    );
 
     @Test
     void testStatusInit() {
@@ -35,16 +25,16 @@ void testStatusInit() {
      */
     @Test
     void testStatusOnlyMovesUp() {
-        for (int i = 0; i < statusesInOrder.size(); i++)
-            for (int j = 0; j < statusesInOrder.size(); j++)
-                for (int k = 0; k < statusesInOrder.size(); k++) {
+        for (int i = 0; i < STATUSES_IN_ORDER.size(); i++)
+            for (int j = 0; j < STATUSES_IN_ORDER.size(); j++)
+                for (int k = 0; k < STATUSES_IN_ORDER.size(); k++) {
                     StatusVoter voter = new StatusVoter();
-                    voter.merge(statusesInOrder.get(i));
-                    voter.merge(statusesInOrder.get(j));
-                    voter.merge(statusesInOrder.get(k));
+                    voter.merge(STATUSES_IN_ORDER.get(i));
+                    voter.merge(STATUSES_IN_ORDER.get(j));
+                    voter.merge(STATUSES_IN_ORDER.get(k));
 
                     var maxIndex = Math.max(Math.max(i, j), k);
-                    var expectedClassification = statusesInOrder.get(maxIndex);
+                    var expectedClassification = STATUSES_IN_ORDER.get(maxIndex);
                     assertThat(voter.getClassification()).isEqualTo(expectedClassification);
                     assertThat(voter).hasToString(expectedClassification.name());
                 }

src/test/java/com/github/danieladams456/statusvoter/TestData.java

@@ -0,0 +1,15 @@
+package com.github.danieladams456.statusvoter;
+
+import java.util.List;
+
+public class TestData {
+    static final List<StatusClassification> STATUSES_IN_ORDER = List.of(
+            StatusClassification.INITIAL,
+            StatusClassification.SUCCESS,
+            StatusClassification.UNCLEAR_ATTRIBUTION_ERROR,
+            StatusClassification.CUSTOMER_DATA_ERROR,
+            StatusClassification.CUSTOMER_DATA_VALIDATION_ERROR,
+            StatusClassification.INTERNAL_STATUS_MERGE_ERROR,
+            StatusClassification.INTERNAL_ERROR
+    );
+}