diff --git a/async-test-lib/src/main/java/se/deversity/asynctest/diagnostics/CompletableFutureObtrudeDetector.java b/async-test-lib/src/main/java/se/deversity/asynctest/diagnostics/CompletableFutureObtrudeDetector.java index 3e34e23bf..0f2f9e8f9 100644 --- a/async-test-lib/src/main/java/se/deversity/asynctest/diagnostics/CompletableFutureObtrudeDetector.java +++ b/async-test-lib/src/main/java/se/deversity/asynctest/diagnostics/CompletableFutureObtrudeDetector.java @@ -50,8 +50,10 @@ State newState(Object instance, String label) { public void recordObtrude(CompletableFuture future, String label, Thread thread) { if (future == null || thread == null) return; State s = stateFor(future, label, "CompletableFuture"); - s.obtrudeCount.incrementAndGet(); + // The thread first, then the count: stateFor publishes the state before either is set, and + // analyze reads the count first, so a count it sees above zero has a thread name behind it. s.lastObtrudedByThread = thread.getName(); + s.obtrudeCount.incrementAndGet(); } /** * Analyses what has been recorded about the observation and builds the report for it. @@ -61,9 +63,16 @@ public void recordObtrude(CompletableFuture future, String label, Thread thre public Report analyze() { Report r = new Report(); for (State s : states()) { + // One read of each field, count first (see recordObtrude). A state registered by a + // recordObtrude still under way has no count yet and nothing to report. + int count = s.obtrudeCount.get(); + String lastThread = s.lastObtrudedByThread; + if (count == 0 || lastThread == null) { + continue; + } String msg = String.format( "CompletableFuture '%s' obtruded %d times (last by thread '%s') — obtruding values or exceptions forces downstream pipelines to execute with outdated/inconsistent states, introducing publication races.", - s.label, s.obtrudeCount.get(), s.lastObtrudedByThread + s.label, count, lastThread ); r.violations.add(msg); r.structuredViolations.add(new Violation( @@ -73,8 +82,8 @@ public Report analyze() { List.of(), Map.of( "label", s.label, - "obtrudeCount", s.obtrudeCount, - "lastObtrudedByThread", s.lastObtrudedByThread + "obtrudeCount", count, + "lastObtrudedByThread", lastThread ), Instant.now() )); diff --git a/async-test-lib/src/main/java/se/deversity/asynctest/diagnostics/ThreadPoolDeadlockDetector.java b/async-test-lib/src/main/java/se/deversity/asynctest/diagnostics/ThreadPoolDeadlockDetector.java index 23cac90af..f1340c22c 100644 --- a/async-test-lib/src/main/java/se/deversity/asynctest/diagnostics/ThreadPoolDeadlockDetector.java +++ b/async-test-lib/src/main/java/se/deversity/asynctest/diagnostics/ThreadPoolDeadlockDetector.java @@ -70,6 +70,8 @@ static final class PoolState { final int poolSize; final AtomicInteger activeTaskCount = new AtomicInteger(0); final AtomicInteger nestedSubmissionCount = new AtomicInteger(0); + /** This pool's share of the detector-wide deadlockRiskCount, so analyze can total what it lists. */ + final AtomicInteger deadlockRisks = new AtomicInteger(0); final List nestedSubmissions = Collections.synchronizedList(new ArrayList<>()); PoolState(String name, int poolSize) { @@ -133,6 +135,7 @@ public void recordNestedSubmission(ExecutorService pool, String name) { // Check for deadlock risk: active tasks >= pool size if (activeTasks >= state.poolSize) { + state.deadlockRisks.incrementAndGet(); deadlockRiskCount.incrementAndGet(); } } @@ -166,20 +169,28 @@ public ThreadPoolDeadlockReport analyze() { return DetectorFailurePolicy.checkedReport(this, report801); } + // The total comes from the pools this pass lists, not from deadlockRiskCount read after the + // loop: a submission recorded between the two made the report claim an issue it listed no + // pool for. Risks are read before the nested count, which recordNestedSubmission raises + // first, so a pool with a risk always makes the list. List risks = new ArrayList<>(); + int totalRisks = 0; for (PoolState state : states()) { - if (state.nestedSubmissionCount.get() > 0) { + int poolRisks = state.deadlockRisks.get(); + int nested = state.nestedSubmissionCount.get(); + if (nested > 0) { + totalRisks += poolRisks; risks.add(new PoolDeadlockRisk( state.name, state.poolSize, - state.nestedSubmissionCount.get(), + nested, state.activeTaskCount.get(), new ArrayList<>(state.nestedSubmissions) )); } } - ThreadPoolDeadlockReport report801 = new ThreadPoolDeadlockReport(risks, deadlockRiskCount.get()); + ThreadPoolDeadlockReport report801 = new ThreadPoolDeadlockReport(risks, totalRisks); report801.fillStructuredViolations(); return DetectorFailurePolicy.checkedReport(this, report801); } diff --git a/async-test-lib/src/test/java/se/deversity/asynctest/diagnostics/CompletableFutureObtrudeDetectorTest.java b/async-test-lib/src/test/java/se/deversity/asynctest/diagnostics/CompletableFutureObtrudeDetectorTest.java index 69ea83b61..141aee291 100644 --- a/async-test-lib/src/test/java/se/deversity/asynctest/diagnostics/CompletableFutureObtrudeDetectorTest.java +++ b/async-test-lib/src/test/java/se/deversity/asynctest/diagnostics/CompletableFutureObtrudeDetectorTest.java @@ -78,4 +78,21 @@ void doesNotKeepARecordedSubjectAlive() throws InterruptedException { CompletableFutureObtrudeDetector d = new CompletableFutureObtrudeDetector(); AbstractInstanceDetectorTest.assertNotRetained(CompletableFuture::new, f -> d.recordObtrude(f, null, Thread.currentThread())); } + + /** + * A recordObtrude still under way has registered the future but not yet set its thread or + * count. analyze() put that null thread into Map.of and threw, so a worker still recording + * dropped the run's whole report (DetectorAnalysisUnderConcurrentWritesTest, #1212). This + * builds the half-recorded state directly instead of waiting for the scheduler to. + */ + @Test + void aFutureRegisteredButNotYetRecordedIsNotReported() { + CompletableFutureObtrudeDetector d = new CompletableFutureObtrudeDetector(); + d.stateFor(new CompletableFuture(), "mid-record", "CompletableFuture"); + + CompletableFutureObtrudeDetector.Report report = assertDoesNotThrow(d::analyze); + + assertFalse(report.hasIssues()); + assertTrue(report.structuredViolations.isEmpty()); + } } diff --git a/async-test-lib/src/test/java/se/deversity/asynctest/diagnostics/ThreadPoolDeadlockDetectorTest.java b/async-test-lib/src/test/java/se/deversity/asynctest/diagnostics/ThreadPoolDeadlockDetectorTest.java index 688a3c95b..e0f2b0c59 100644 --- a/async-test-lib/src/test/java/se/deversity/asynctest/diagnostics/ThreadPoolDeadlockDetectorTest.java +++ b/async-test-lib/src/test/java/se/deversity/asynctest/diagnostics/ThreadPoolDeadlockDetectorTest.java @@ -261,4 +261,31 @@ void doesNotKeepARecordedSubjectAlive() throws InterruptedException { ThreadPoolDeadlockDetector d = new ThreadPoolDeadlockDetector(); AbstractInstanceDetectorTest.assertNotRetained(() -> new java.util.concurrent.ThreadPoolExecutor(1, 1, 0, java.util.concurrent.TimeUnit.SECONDS, new java.util.concurrent.LinkedBlockingQueue()), e -> d.registerPool(e, null)); } + + /** + * The report's total and its pool list must come from the same pass. analyze() listed the + * pools, then read the detector-wide risk count; a submission recorded in between (or a pool + * whose weak key was cleared) left a total above zero with no pool listed, and the report + * claimed an issue it had no structured violation for (#1212). Dropping the pool's state + * while the detector-wide count stays reproduces that gap without depending on timing. + */ + @Test + void aRiskWithNoListedPoolDoesNotMakeAnIssue() { + ThreadPoolDeadlockDetector detector = new ThreadPoolDeadlockDetector(); + ExecutorService pool = Executors.newSingleThreadExecutor(); + try { + detector.registerPool(pool, "single"); + detector.recordNestedSubmission(pool, "single"); + assertEquals(1, detector.getDeadlockRiskCount()); + detector.clearStates(); + + ThreadPoolDeadlockDetector.ThreadPoolDeadlockReport report = + assertDoesNotThrow(detector::analyze); + + assertFalse(report.hasIssues()); + assertEquals(report.hasIssues(), !report.structuredViolations.isEmpty()); + } finally { + pool.shutdownNow(); + } + } } diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 1cce2a526..9f3f1812f 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -13,6 +13,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- `CompletableFutureObtrudeDetector` and `ThreadPoolDeadlockDetector` no longer fail analysis while + a worker is still recording, which dropped the run's whole report (#1212). The first put a future's + thread name into its violation before `recordObtrude` had set it, a `NullPointerException`; a future + registered but not yet counted is now left out. The second listed the pools, then read its risk total + afterwards, so a submission recorded in between reported an issue with no structured violation; the + total now comes from the pools the same pass lists. - `AtomicNonAtomicUpdate` no longer reports a plain store whose value was never built from the atomic read it followed. `gauge.get()` used for a log line, then `gauge.set(queue.size())`, lost no update, and was reported as one (#1187). The agent now analyses each class's bytes, decides per store