Improved torture tests

DavyLandman · DavyLandman · commit d6b1f2818942 · 2024-10-14T15:19:51.000+02:00
diff --git a/src/main/java/engineering/swat/watch/impl/BundledSubscription.java b/src/main/java/engineering/swat/watch/impl/BundledSubscription.java
@@ -57,10 +57,11 @@ public Closeable subscribe(Key target, Consumer<Event> eventListener) throws IOE
             Subscription<Event> active = this.subscriptions.computeIfAbsent(target, t -> new Subscription<>());
             // after this, there will only be 1 instance of active subscription in the map.
             // but we might have a race with remove, which can close the subscript between our get and our addition
+            // since this code is very hard to get right without locks, and shouldn't be run too often
+            // we take a big lock around the subscription management
             synchronized(active) {
                 if (active.closed) {
-                    // we lost the race with closing the subscription
-                    // so we retry
+                    // we lost the race with closing the subscription, so we retry
                     continue;
                 }
                 active.add(eventListener);
diff --git a/src/main/java/engineering/swat/watch/impl/SubscriptionKey.java b/src/main/java/engineering/swat/watch/impl/SubscriptionKey.java
@@ -36,4 +36,9 @@ public boolean equals(@Nullable Object obj) {
     public int hashCode() {
         return Objects.hash(path, recursive);
     }
+
+    @Override
+    public String toString() {
+        return path.toString() + (recursive ? "[recursive]" : "");
+    }
 }
diff --git a/src/test/java/engineering/swat/watch/TortureTests.java b/src/test/java/engineering/swat/watch/TortureTests.java
@@ -1,6 +1,5 @@
 package engineering.swat.watch;
 
-import static org.awaitility.Awaitility.doNotCatchUncaughtExceptionsByDefault;
 import static org.awaitility.Awaitility.await;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
@@ -11,8 +10,8 @@
 import java.time.Duration;
 import java.time.LocalTime;
 import java.util.Random;
+import java.util.Collections;
 import java.util.Set;
-import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedDeque;
 import java.util.concurrent.Executor;
@@ -27,10 +26,8 @@
 import org.apache.logging.log4j.Logger;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.RepeatedTest;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable;
-import org.junit.jupiter.api.condition.EnabledIf;
 import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
 
 import engineering.swat.watch.WatchEvent.Kind;
@@ -173,24 +170,30 @@ void pressureOnFSShouldNotMissNewFilesAnything() throws InterruptedException, IO
         }
     }
 
-    @Test
+    private final int TORTURE_REGISTRATION_THREADS = THREADS * 500;
+
+    @RepeatedTest(failureThreshold=1, value = 20)
     void manyRegistrationsForSamePath() throws InterruptedException, IOException {
         var startRegistering = new Semaphore(0);
+        var startedWatching = new Semaphore(0);
         var startDeregistring = new Semaphore(0);
         var done = new Semaphore(0);
         var seen = ConcurrentHashMap.<Path>newKeySet();
         var exceptions = new LinkedBlockingDeque<Exception>();
-        for (int t = 0; t < THREADS * 100; t++) {
+
+        for (int t = 0; t < TORTURE_REGISTRATION_THREADS; t++) {
             var r = new Thread(() -> {
                 try {
                     var watcher = Watcher
                         .watch(testDir.getTestDirectory(), WatchScope.PATH_AND_CHILDREN)
-                        .onEvent(e -> { if (e.getKind() == Kind.CREATED) seen.add(e.calculateFullPath()); });
+                        .onEvent(e -> seen.add(e.calculateFullPath()));
                     startRegistering.acquire();
                     try (var c = watcher.start()) {
+                        startedWatching.release();
                         startDeregistring.acquire();
                     }
                     catch(Exception e) {
+                        startedWatching.release();
                         exceptions.push(e);
                     }
                 } catch (InterruptedException e1) {
@@ -203,29 +206,39 @@ void manyRegistrationsForSamePath() throws InterruptedException, IOException {
             r.start();
         }
 
-        startRegistering.release(THREADS * 100);
-        startDeregistring.release((THREADS * 100) - 1);
-        done.acquire((THREADS * 100) - 1);
-        assertTrue(seen.isEmpty(), "No events should have been sent");
-        Files.writeString(testDir.getTestDirectory().resolve("test124.txt"), "Hello World");
-        await("We should see only one event")
-            .failFast(() -> !exceptions.isEmpty())
-            .timeout(TestHelper.LONG_WAIT)
-            .pollInterval(Duration.ofMillis(10))
-            .until(seen::size, s -> s == 1);
-        if (!exceptions.isEmpty()) {
-            fail(exceptions.pop());
+        try {
+            startRegistering.release(TORTURE_REGISTRATION_THREADS);
+            startDeregistring.release(TORTURE_REGISTRATION_THREADS - 1);
+            startedWatching.acquire(TORTURE_REGISTRATION_THREADS); // make sure they area ll started
+            done.acquire(TORTURE_REGISTRATION_THREADS - 1);
+            assertTrue(seen.isEmpty(), "No events should have been sent");
+            var target = testDir.getTestDirectory().resolve("test124.txt");
+            //logger.info("Writing: {}", target);
+            Files.writeString(target, "Hello World");
+            var expected = Collections.singleton(target);
+            await("We should see only one event")
+                .failFast(() -> !exceptions.isEmpty())
+                .timeout(TestHelper.LONG_WAIT)
+                .pollInterval(Duration.ofMillis(10))
+                .until(() -> seen, expected::equals);
+            if (!exceptions.isEmpty()) {
+                fail(exceptions.pop());
+            }
+        }
+        finally {
+            startDeregistring.release(TORTURE_REGISTRATION_THREADS);
         }
-        startDeregistring.release();
     }
 
-    @Test
+    @RepeatedTest(failureThreshold=1, value = 20)
     void manyRegisterAndUnregisterSameTime() throws InterruptedException, IOException {
         var startRegistering = new Semaphore(0);
+        var startedWatching = new Semaphore(0);
         var stopAll = new Semaphore(0);
         var done = new Semaphore(0);
-        var seen = new ConcurrentLinkedDeque<Path>();
+        var seen = ConcurrentHashMap.<Long>newKeySet();
         var exceptions = new LinkedBlockingDeque<Exception>();
+        var target = testDir.getTestDirectory().resolve("test124.txt");
         int amountOfWatchersActive = 0;
         try {
             for (int t = 0; t < THREADS; t++) {
@@ -239,10 +252,14 @@ void manyRegisterAndUnregisterSameTime() throws InterruptedException, IOExceptio
                         for (int k = 0; k < 1000; k++) {
                             var watcher = Watcher
                                 .watch(testDir.getTestDirectory(), WatchScope.PATH_AND_CHILDREN)
-                                .onEvent(e -> { if (e.getKind() == Kind.CREATED) seen.add(e.calculateFullPath()); });
+                                .onEvent(e -> {
+                                    if (e.calculateFullPath().equals(target)) {
+                                        seen.add(Thread.currentThread().getId());
+                                    }
+                                });
                             try (var c = watcher.start()) {
                                 if (finishWatching && k + 1 == 1000) {
-                                    logger.info("Waiting on stop signal");
+                                    startedWatching.release();
                                     stopAll.acquire();
                                 }
                             }
@@ -253,6 +270,7 @@ void manyRegisterAndUnregisterSameTime() throws InterruptedException, IOExceptio
                     } catch (InterruptedException e1) {
                     }
                     finally {
+                        startedWatching.release();
                         done.release();
                     }
                 });
@@ -262,9 +280,10 @@ void manyRegisterAndUnregisterSameTime() throws InterruptedException, IOExceptio
 
             startRegistering.release(THREADS);
             done.acquire(THREADS - amountOfWatchersActive);
+            startedWatching.acquire(THREADS);
             assertTrue(seen.isEmpty(), "No events should have been sent");
-            Files.writeString(testDir.getTestDirectory().resolve("test124.txt"), "Hello World");
-            await("We should see only exactly the events we expect")
+            Files.writeString(target, "Hello World");
+            await("We should see only exactly the " + amountOfWatchersActive + " events we expect")
                 .failFast(() -> !exceptions.isEmpty())
                 .pollDelay(TestHelper.NORMAL_WAIT.minusMillis(100))
                 .until(seen::size, Predicate.isEqual(amountOfWatchersActive))
diff --git a/src/test/java/engineering/swat/watch/impl/BundlingTests.java b/src/test/java/engineering/swat/watch/impl/BundlingTests.java
@@ -11,14 +11,17 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.awaitility.Awaitility;
+import org.hamcrest.core.IsEqual;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.RepeatedTest;
 import org.junit.jupiter.api.Test;
 
 import engineering.swat.watch.TestHelper;
@@ -36,7 +39,7 @@ private static class FakeSubscribable implements ISubscribable<Long, Boolean>  {
         public Closeable subscribe(Long target, Consumer<Boolean> eventListener) throws IOException {
             subs.put(target, eventListener);
             return () -> {
-                subs.remove(target);
+                subs.remove(target, eventListener);
             };
         }
 
@@ -60,7 +63,7 @@ static void setupEverything() {
         Awaitility.setDefaultTimeout(TestHelper.LONG_WAIT.getSeconds(), TimeUnit.SECONDS);
     }
 
-    private static final long SUBs = 100;
+    private static final int SUBs = 100;
     private static final long MSGs = 100_000;
 
     @Test
@@ -101,4 +104,44 @@ void manySubscriptions() throws IOException {
 
 
     }
+
+    @RepeatedTest(failureThreshold = 1, value=50)
+    void parallelSubscriptions() throws IOException, InterruptedException {
+        var hits = new AtomicInteger();
+        var endPointReached = new Semaphore(0);
+        var waitingForClose = new Semaphore(0);
+        var done = new Semaphore(0);
+
+        int active = 0;
+        for (int j = 0; j < SUBs; j++) {
+            boolean keepAround = j % 2 == 0;
+            if (keepAround) {
+                active++;
+            }
+            var t = new Thread(() -> {
+                for (int k =0; k < 1000; k++) {
+                    try (var c = target.subscribe(Long.valueOf(0), b -> hits.incrementAndGet())) {
+                        if (keepAround && k + 1 == 1000) {
+                            endPointReached.release();
+                            waitingForClose.acquire();
+                        }
+                    } catch (Exception ignored) {
+                        logger.catching(ignored);
+                    }
+                }
+                done.release();
+            });
+            t.setDaemon(true);
+            t.start();
+        }
+
+        endPointReached.acquire(active);
+        done.acquire(SUBs - active);
+        fakeSubs.publish(Long.valueOf(0));
+
+        await("Subscriptions should have hit")
+            .untilAtomic(hits, IsEqual.equalTo(active));
+        waitingForClose.release(active);
+    }
+
 }

Original file line number	Diff line number	Diff line change
`@@ -36,4 +36,9 @@ public boolean equals(@Nullable Object obj) {`
`36`	`36`	`public int hashCode() {`
`37`	`37`	`return Objects.hash(path, recursive);`
`38`	`38`	`}`
	`39`	`+`
	`40`	`+ @Override`
	`41`	`+ public String toString() {`
	`42`	`+ return path.toString() + (recursive ? "[recursive]" : "");`
	`43`	`+ }`
`39`	`44`	`}`