synchronizers: CountDownLatch vs CyclicBarrier vs Phaser vs Semaphore
One three-round worker pipeline implemented with CountDownLatch (one latch per round), CyclicBarrier (reusable, with a barrier action), and Phaser (reusable, plus dynamic mid-run registration), alongside Semaphore solving the genuinely different problem of bounding concurrent access. Covers timeout behavior (CyclicBarrier permanently breaks on one timeout, Phaser doesn't) and virtual-thread compatibility, including the corrected finding that Object.wait() releases its monitor and never pinned, unlike Thread.sleep() inside synchronized. Co-Authored-By: Claude Sonnet 5 <[email protected]> Claude-Session: https://claude.ai/code/session_01FhzLY5p6okFva3qsnsRyvM
This commit is contained in:
Claude Sonnet 5
@@ -9,6 +9,7 @@ article; each module's own README has that article's version table, quickstart,
|
||||
| [`locks`](locks/) | synchronized vs ReentrantLock vs StampedLock: Benchmarks and a Decision Table |
|
||||
| [`atomics`](atomics/) | Java Atomics and VarHandle: CAS, LongAdder, and When Atomics Beat Locks |
|
||||
| [`vt-pinning`](vt-pinning/) | Diagnosing Virtual Thread Pinning in Production: JFR Events, jcmd, and Real Fixes |
|
||||
| [`synchronizers`](synchronizers/) | CountDownLatch vs CyclicBarrier vs Phaser vs Semaphore in Java |
|
||||
|
||||
## License
|
||||
|
||||
|
||||
@@ -17,6 +17,7 @@
|
||||
<module>locks</module>
|
||||
<module>atomics</module>
|
||||
<module>vt-pinning</module>
|
||||
<module>synchronizers</module>
|
||||
</modules>
|
||||
|
||||
<properties>
|
||||
|
||||
@@ -0,0 +1,80 @@
|
||||
# synchronizers
|
||||
|
||||
Companion code for the ankurm.com post *"CountDownLatch vs CyclicBarrier vs Phaser vs Semaphore
|
||||
in Java."* Fifth module in `java-core-examples`, the Java-core / concurrency series.
|
||||
|
||||
The same three-round worker pipeline, implemented once per class where it's a natural fit
|
||||
(`CountDownLatch`, `CyclicBarrier`, `Phaser`), plus `Semaphore` solving the genuinely different
|
||||
problem it actually solves (bounding concurrent access, not waiting for everyone to arrive) inside
|
||||
the same pipeline shape. Reuse semantics, timeout behavior, and virtual-thread compatibility are
|
||||
each demonstrated directly rather than asserted.
|
||||
|
||||
## Versions this was built and tested against
|
||||
|
||||
| Component | Version | Notes |
|
||||
|---|---|---|
|
||||
| JDK (primary) | 25.0.4.1+1 (Temurin, LTS) | |
|
||||
| JDK (comparison) | 21.0.10 (OpenJDK) | For the virtual-thread pinning contrast only. |
|
||||
| JUnit Jupiter | 5.11.0 | Correctness tests, 20 repeats per barrier test. |
|
||||
| Maven | 3.9.11 | |
|
||||
| Hardware | 2 vCPU x86-64 VM | Same sandbox as the rest of this series. |
|
||||
|
||||
## Quickstart
|
||||
|
||||
```bash
|
||||
export JAVA_HOME=/path/to/jdk-25
|
||||
mvn package
|
||||
java -cp target/classes com.ankurm.synchronizers.CyclicBarrierPipeline
|
||||
java -cp target/classes com.ankurm.synchronizers.PhaserPipeline
|
||||
java -cp target/classes com.ankurm.synchronizers.SemaphoreResourceGuard
|
||||
```
|
||||
|
||||
`scripts/run-all.sh` regenerates every file in `output/` (needs `JDK25_HOME` and `JDK21_HOME`).
|
||||
|
||||
## What's in here
|
||||
|
||||
| File | What it shows |
|
||||
|---|---|
|
||||
| `.../CountDownLatchPipeline.java` | A starting-gate latch, plus one fresh latch per round - the workaround `CyclicBarrier` exists to remove. |
|
||||
| `.../CyclicBarrierPipeline.java` | One reusable barrier across all three rounds, with a barrier action. |
|
||||
| `.../PhaserPipeline.java` | Same barrier behavior, plus a fifth worker dynamically registering after round 1. |
|
||||
| `.../SemaphoreResourceGuard.java` | The same pipeline, but `Semaphore` bounding concurrent access to a shared resource - a different concern from the other three. |
|
||||
| `.../TimeoutBehaviorDemo.java` | What each class does when the thing it's waiting for never arrives. |
|
||||
| `.../VirtualThreadCompatibilityDemo.java` | None of the four pin a virtual thread, even on JDK 21, because none are built on `synchronized`. |
|
||||
| `.../SynchronizedWaitPinsDemo.java` | The contrast case, and a correction: `synchronized` + `Object.wait()` does NOT pin (it releases the monitor); `synchronized` + `Thread.sleep()` does. |
|
||||
| `src/test/.../RoundOrderingCorrectnessTest.java` | Asserts no worker's round N+1 ever starts before every worker's round N finished (`CyclicBarrier`, `Phaser`, 20 repeats each), and `Semaphore` never exceeds its permit count. |
|
||||
| `output/01`-`04` | Each pipeline's real run. |
|
||||
| `output/05` | Timeout behavior across all four. |
|
||||
| `output/06` | Virtual-thread compatibility, JDK 21, both the four synchronizers and the `synchronized`+`wait`/`sleep` contrast. |
|
||||
| `output/07` | JUnit correctness run. |
|
||||
|
||||
## Reading the results honestly (2-vCPU sandbox)
|
||||
|
||||
**`Semaphore` isn't a fourth way to do what the other three do.** `CountDownLatch`, `CyclicBarrier`,
|
||||
and `Phaser` all answer "has everyone reached this point?" `Semaphore` answers "how many callers
|
||||
are allowed through at once?" - a genuinely different question, not a stylistic alternative.
|
||||
`output/04` shows `Semaphore` bounding a resource to 2 concurrent callers, independent of round
|
||||
boundaries or worker identity; forcing it into a "wait for the group" role would misrepresent what
|
||||
it's for.
|
||||
|
||||
**A timeout doesn't do the same thing on all four** (`output/05`). `CountDownLatch.await(timeout)`
|
||||
returns `false` and stays reusable to check again. `Semaphore.tryAcquire(timeout)` behaves the same
|
||||
way. `CyclicBarrier.await(timeout)` throws `TimeoutException` and **permanently breaks the barrier
|
||||
for every other waiter** until an explicit `reset()` - one straggler poisons the whole group.
|
||||
`Phaser.awaitAdvanceInterruptibly(phase, timeout, unit)` throws `TimeoutException` too, but the
|
||||
phaser itself is untouched and a later, fully-arrived round on the same instance succeeds normally
|
||||
with no reset needed. That difference alone is a reason to reach for `Phaser` over `CyclicBarrier`
|
||||
in anything where a slow round shouldn't take down every other round after it.
|
||||
|
||||
**`Object.wait()` doesn't pin a virtual thread - `Thread.sleep()` inside the same `synchronized`
|
||||
block does** (`output/06`), and this repo initially assumed otherwise before checking. `wait()`
|
||||
fully releases the monitor for the duration of the wait, so there's nothing held while parked and
|
||||
nothing to pin; JEP 491 (JDK 24) only needed to fix the case where a blocking call runs *while
|
||||
still holding* the monitor. All four `java.util.concurrent` classes in this module were already
|
||||
safe on virtual threads on JDK 21, years before JEP 491, because none of them are built on
|
||||
`synchronized` in the first place - a fact worth knowing before assuming every pre-JDK-24
|
||||
concurrency primitive needed the same fix.
|
||||
|
||||
## License
|
||||
|
||||
MIT - see the [repo-wide LICENSE](../LICENSE).
|
||||
@@ -0,0 +1,17 @@
|
||||
$ java CountDownLatchPipeline
|
||||
|
||||
main: all 4 workers created, still waiting at the start gate
|
||||
main: releasing the start gate
|
||||
worker-0: finished round 1
|
||||
worker-1: finished round 1
|
||||
worker-2: finished round 1
|
||||
worker-3: finished round 1
|
||||
worker-0: finished round 2
|
||||
worker-1: finished round 2
|
||||
worker-2: finished round 2
|
||||
worker-3: finished round 2
|
||||
worker-0: finished round 3
|
||||
worker-1: finished round 3
|
||||
worker-2: finished round 3
|
||||
worker-3: finished round 3
|
||||
done
|
||||
@@ -0,0 +1,18 @@
|
||||
$ java CyclicBarrierPipeline
|
||||
|
||||
worker-0: finished round 1
|
||||
worker-1: finished round 1
|
||||
worker-2: finished round 1
|
||||
worker-3: finished round 1
|
||||
barrier action: round 1 complete, all 4 workers arrived
|
||||
worker-0: finished round 2
|
||||
worker-1: finished round 2
|
||||
worker-2: finished round 2
|
||||
worker-3: finished round 2
|
||||
barrier action: round 2 complete, all 4 workers arrived
|
||||
worker-0: finished round 3
|
||||
worker-1: finished round 3
|
||||
worker-2: finished round 3
|
||||
worker-3: finished round 3
|
||||
barrier action: round 3 complete, all 4 workers arrived
|
||||
done
|
||||
@@ -0,0 +1,21 @@
|
||||
$ java PhaserPipeline
|
||||
|
||||
worker-0: finished round 1
|
||||
worker-1: finished round 1
|
||||
worker-2: finished round 1
|
||||
worker-3: finished round 1
|
||||
phaser onAdvance: phase 1 complete, 4 registered parties -> advancing
|
||||
worker-4: registers after round 1 (phaser now has 5 parties)
|
||||
worker-0: finished round 2
|
||||
worker-1: finished round 2
|
||||
worker-2: finished round 2
|
||||
worker-3: finished round 2
|
||||
worker-4: finished round 2
|
||||
phaser onAdvance: phase 2 complete, 5 registered parties -> advancing
|
||||
worker-0: finished round 3
|
||||
worker-1: finished round 3
|
||||
worker-2: finished round 3
|
||||
worker-3: finished round 3
|
||||
worker-4: finished round 3
|
||||
phaser onAdvance: phase 3 complete, 5 registered parties -> advancing
|
||||
done
|
||||
@@ -0,0 +1,16 @@
|
||||
$ java SemaphoreResourceGuard
|
||||
|
||||
worker-0 round 1: acquired permit, 1 callers concurrently in the resource (availablePermits=0)
|
||||
worker-1 round 1: acquired permit, 2 callers concurrently in the resource (availablePermits=0)
|
||||
worker-2 round 1: acquired permit, 2 callers concurrently in the resource (availablePermits=0)
|
||||
worker-3 round 1: acquired permit, 2 callers concurrently in the resource (availablePermits=0)
|
||||
worker-3 round 2: acquired permit, 1 callers concurrently in the resource (availablePermits=1)
|
||||
worker-0 round 2: acquired permit, 2 callers concurrently in the resource (availablePermits=0)
|
||||
worker-1 round 2: acquired permit, 1 callers concurrently in the resource (availablePermits=1)
|
||||
worker-2 round 2: acquired permit, 2 callers concurrently in the resource (availablePermits=0)
|
||||
worker-2 round 3: acquired permit, 1 callers concurrently in the resource (availablePermits=1)
|
||||
worker-3 round 3: acquired permit, 2 callers concurrently in the resource (availablePermits=0)
|
||||
worker-0 round 3: acquired permit, 1 callers concurrently in the resource (availablePermits=1)
|
||||
worker-1 round 3: acquired permit, 2 callers concurrently in the resource (availablePermits=0)
|
||||
permits=2 maxObservedConcurrency=2
|
||||
done
|
||||
@@ -0,0 +1,20 @@
|
||||
$ java TimeoutBehaviorDemo
|
||||
|
||||
=== CountDownLatch.await(timeout): expects 2, only 1 ever counts down ===
|
||||
await returned: false (false = timed out, latch still usable to re-check later)
|
||||
getCount() after timeout: 1
|
||||
|
||||
=== CyclicBarrier.await(timeout): expects 2 parties, only 1 arrives ===
|
||||
await threw TimeoutException, as expected
|
||||
barrier.isBroken() after the timeout: true
|
||||
a second, later await() on the SAME barrier: BrokenBarrierException immediately - one timeout poisons the barrier for everyone until reset()
|
||||
|
||||
=== Phaser.awaitAdvanceInterruptibly(phase, timeout): expects 2 parties, only 1 arrives ===
|
||||
awaitAdvanceInterruptibly threw TimeoutException, as expected
|
||||
phaser.getPhase() after the timeout: 0 (unchanged - still phase 0, not advanced or broken)
|
||||
a later round on the SAME phaser, once both parties actually arrive: succeeds normally, now at phase 1
|
||||
|
||||
=== Semaphore.tryAcquire(timeout): 0 permits available ===
|
||||
tryAcquire returned: false (false = timed out, no exception, no broken state)
|
||||
a later tryAcquire after a release: true - fully independent of the earlier timeout
|
||||
done
|
||||
@@ -0,0 +1,25 @@
|
||||
$ java -Djdk.tracePinnedThreads=full VirtualThreadCompatibilityDemo (pre-JDK-24 runtime)
|
||||
|
||||
java.version=21.0.10
|
||||
jdk.tracePinnedThreads=full
|
||||
CountDownLatch.await(): completed, no trace above this line for it
|
||||
CyclicBarrier.await(): completed, no trace above this line for it
|
||||
Phaser.arriveAndAwaitAdvance(): completed, no trace above this line for it
|
||||
Semaphore.acquire(): completed, no trace above this line for it
|
||||
done - no pinning trace printed above for any of the four
|
||||
|
||||
$ java -Djdk.tracePinnedThreads=full SynchronizedWaitPinsDemo (pre-JDK-24 runtime, contrast case)
|
||||
|
||||
java.version=21.0.10
|
||||
jdk.tracePinnedThreads=full
|
||||
synchronized + Object.wait(): completed, no pinning trace above this line for it
|
||||
VirtualThread[#21]/runnable@ForkJoinPool-1-worker-1 reason:MONITOR
|
||||
java.base/java.lang.VirtualThread$VThreadContinuation.onPinned(VirtualThread.java:199)
|
||||
java.base/jdk.internal.vm.Continuation.onPinned0(Continuation.java:393)
|
||||
java.base/java.lang.VirtualThread.parkNanos(VirtualThread.java:635)
|
||||
java.base/java.lang.VirtualThread.sleepNanos(VirtualThread.java:807)
|
||||
java.base/java.lang.Thread.sleep(Thread.java:507)
|
||||
com.ankurm.synchronizers.SynchronizedWaitPinsDemo.lambda$main$1(SynchronizedWaitPinsDemo.java:48) <== monitors:1
|
||||
java.base/java.lang.VirtualThread.run(VirtualThread.java:329)
|
||||
synchronized + Thread.sleep(): completed, pinning trace (if any) appears above this line for it
|
||||
done
|
||||
@@ -0,0 +1,4 @@
|
||||
-------------------------------------------------------------------------------
|
||||
Test set: com.ankurm.synchronizers.RoundOrderingCorrectnessTest
|
||||
-------------------------------------------------------------------------------
|
||||
Tests run: 41, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 0.514 s -- in com.ankurm.synchronizers.RoundOrderingCorrectnessTest
|
||||
@@ -0,0 +1,43 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<parent>
|
||||
<groupId>com.ankurm</groupId>
|
||||
<artifactId>java-core-examples</artifactId>
|
||||
<version>1.0</version>
|
||||
</parent>
|
||||
|
||||
<artifactId>synchronizers</artifactId>
|
||||
<name>synchronizers</name>
|
||||
<description>CountDownLatch vs CyclicBarrier vs Phaser vs Semaphore: one three-round worker pipeline implemented four ways, reuse semantics, timeout behavior, and virtual-thread compatibility.</description>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.junit.jupiter</groupId>
|
||||
<artifactId>junit-jupiter</artifactId>
|
||||
<version>5.11.0</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-compiler-plugin</artifactId>
|
||||
<version>3.13.0</version>
|
||||
<configuration>
|
||||
<release>25</release>
|
||||
</configuration>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-surefire-plugin</artifactId>
|
||||
<version>3.2.5</version>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
</project>
|
||||
Executable
+69
@@ -0,0 +1,69 @@
|
||||
#!/usr/bin/env bash
|
||||
# Regenerates every file in output/. Requires:
|
||||
# - JDK25_HOME pointing at a JDK 24+ install (this repo used Temurin 25.0.4.1+1)
|
||||
# - JDK21_HOME pointing at a pre-JDK-24 install (this repo used OpenJDK 21.0.10), for the
|
||||
# virtual-thread pinning contrast at the end
|
||||
set -euo pipefail
|
||||
cd "$(dirname "$0")/.."
|
||||
|
||||
: "${JDK25_HOME:?Set JDK25_HOME to a JDK 24+ install}"
|
||||
: "${JDK21_HOME:?Set JDK21_HOME to a pre-JDK-24 install}"
|
||||
|
||||
mkdir -p target/classes target/classes-21
|
||||
"$JDK25_HOME/bin/javac" --release 25 -d target/classes $(find src/main/java -name "*.java")
|
||||
"$JDK21_HOME/bin/javac" --release 21 -d target/classes-21 $(find src/main/java -name "*.java")
|
||||
|
||||
echo "--- 01: CountDownLatch pipeline ---"
|
||||
{
|
||||
echo '$ java CountDownLatchPipeline'
|
||||
echo ""
|
||||
"$JDK25_HOME/bin/java" -cp target/classes com.ankurm.synchronizers.CountDownLatchPipeline 2>&1 | grep -v "Picked up"
|
||||
} > output/01-countdownlatch-pipeline.txt
|
||||
|
||||
echo "--- 02: CyclicBarrier pipeline ---"
|
||||
{
|
||||
echo '$ java CyclicBarrierPipeline'
|
||||
echo ""
|
||||
"$JDK25_HOME/bin/java" -cp target/classes com.ankurm.synchronizers.CyclicBarrierPipeline 2>&1 | grep -v "Picked up"
|
||||
} > output/02-cyclicbarrier-pipeline.txt
|
||||
|
||||
echo "--- 03: Phaser pipeline (dynamic registration) ---"
|
||||
{
|
||||
echo '$ java PhaserPipeline'
|
||||
echo ""
|
||||
"$JDK25_HOME/bin/java" -cp target/classes com.ankurm.synchronizers.PhaserPipeline 2>&1 | grep -v "Picked up"
|
||||
} > output/03-phaser-pipeline.txt
|
||||
|
||||
echo "--- 04: Semaphore resource guard ---"
|
||||
{
|
||||
echo '$ java SemaphoreResourceGuard'
|
||||
echo ""
|
||||
"$JDK25_HOME/bin/java" -cp target/classes com.ankurm.synchronizers.SemaphoreResourceGuard 2>&1 | grep -v "Picked up"
|
||||
} > output/04-semaphore-resource-guard.txt
|
||||
|
||||
echo "--- 05: timeout behavior across all four ---"
|
||||
{
|
||||
echo '$ java TimeoutBehaviorDemo'
|
||||
echo ""
|
||||
"$JDK25_HOME/bin/java" -cp target/classes com.ankurm.synchronizers.TimeoutBehaviorDemo 2>&1 | grep -v "Picked up"
|
||||
} > output/05-timeout-behavior.txt
|
||||
|
||||
echo "--- 06: virtual-thread compatibility, all four, on JDK21 (pre-JEP-491) ---"
|
||||
{
|
||||
echo '$ java -Djdk.tracePinnedThreads=full VirtualThreadCompatibilityDemo (pre-JDK-24 runtime)'
|
||||
echo ""
|
||||
"$JDK21_HOME/bin/java" -Djdk.tracePinnedThreads=full -cp target/classes-21 \
|
||||
com.ankurm.synchronizers.VirtualThreadCompatibilityDemo 2>&1 | grep -v "Picked up"
|
||||
echo ""
|
||||
echo '$ java -Djdk.tracePinnedThreads=full SynchronizedWaitPinsDemo (pre-JDK-24 runtime, contrast case)'
|
||||
echo ""
|
||||
"$JDK21_HOME/bin/java" -Djdk.tracePinnedThreads=full -cp target/classes-21 \
|
||||
com.ankurm.synchronizers.SynchronizedWaitPinsDemo 2>&1 | grep -v "Picked up"
|
||||
} > output/06-virtual-thread-compatibility.txt
|
||||
|
||||
echo "--- 07: correctness tests ---"
|
||||
"$JDK25_HOME/bin/java" --version > /dev/null # sanity
|
||||
(cd "$(pwd)" && JAVA_HOME="$JDK25_HOME" mvn -q -f pom.xml test 2>&1 | grep -v "Picked up\|WARNING" || true)
|
||||
cp target/surefire-reports/com.ankurm.synchronizers.RoundOrderingCorrectnessTest.txt output/07-round-ordering-correctness.txt
|
||||
|
||||
echo "Done. See output/."
|
||||
@@ -0,0 +1,70 @@
|
||||
package com.ankurm.synchronizers;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.CopyOnWriteArrayList;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* The same three-round worker pipeline every class in this module runs, done with
|
||||
* {@link CountDownLatch}. Two uses, both idiomatic:
|
||||
*
|
||||
* <ol>
|
||||
* <li>A single "starting gate" latch releases all workers at once instead of letting them
|
||||
* race to start as soon as each is spawned.</li>
|
||||
* <li>One NEW latch per round enforces "everyone finishes round N before round N+1 starts" -
|
||||
* because a {@code CountDownLatch} counts down to zero exactly once and cannot be reset.
|
||||
* That's the limitation {@link CyclicBarrierPipeline} exists to remove.</li>
|
||||
* </ol>
|
||||
*/
|
||||
public final class CountDownLatchPipeline {
|
||||
|
||||
private static final int WORKERS = 4;
|
||||
private static final int ROUNDS = 3;
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
List<String> log = new CopyOnWriteArrayList<>();
|
||||
CountDownLatch startGate = new CountDownLatch(1);
|
||||
|
||||
Thread[] workers = new Thread[WORKERS];
|
||||
for (int w = 0; w < WORKERS; w++) {
|
||||
int id = w;
|
||||
workers[w] = new Thread(() -> runWorker(id, startGate, log));
|
||||
workers[w].start();
|
||||
}
|
||||
|
||||
log.add("main: all " + WORKERS + " workers created, still waiting at the start gate");
|
||||
Thread.sleep(50); // let them actually reach await() before we release them
|
||||
log.add("main: releasing the start gate");
|
||||
startGate.countDown();
|
||||
|
||||
for (Thread t : workers) {
|
||||
t.join();
|
||||
}
|
||||
log.forEach(System.out::println);
|
||||
System.out.println("done");
|
||||
}
|
||||
|
||||
private static void runWorker(int id, CountDownLatch startGate, List<String> log) {
|
||||
try {
|
||||
startGate.await();
|
||||
CountDownLatch roundLatch = null;
|
||||
for (int round = 1; round <= ROUNDS; round++) {
|
||||
// A fresh latch every round - CountDownLatch cannot be reused once it hits zero.
|
||||
roundLatch = ROUND_LATCHES[round - 1];
|
||||
Thread.sleep(20 + (id * 5)); // simulate uneven round work
|
||||
log.add("worker-" + id + ": finished round " + round);
|
||||
roundLatch.countDown();
|
||||
roundLatch.await(5, TimeUnit.SECONDS);
|
||||
}
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
|
||||
// One latch per round, sized to the worker count, created up front so every worker sees the
|
||||
// same instances. In real code this bookkeeping is exactly what CyclicBarrier automates.
|
||||
private static final CountDownLatch[] ROUND_LATCHES = {
|
||||
new CountDownLatch(WORKERS), new CountDownLatch(WORKERS), new CountDownLatch(WORKERS)
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
package com.ankurm.synchronizers;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.concurrent.BrokenBarrierException;
|
||||
import java.util.concurrent.CopyOnWriteArrayList;
|
||||
import java.util.concurrent.CyclicBarrier;
|
||||
|
||||
/**
|
||||
* The same three-round pipeline as {@link CountDownLatchPipeline}, but with the one thing a
|
||||
* {@code CountDownLatch} can't do: the barrier resets itself automatically after every round, so
|
||||
* one {@code CyclicBarrier} instance - not one per round - carries the whole pipeline. It also
|
||||
* runs an optional barrier action exactly once per round, on whichever thread happens to be the
|
||||
* last to arrive, useful for round-boundary bookkeeping that shouldn't run once per worker.
|
||||
*/
|
||||
public final class CyclicBarrierPipeline {
|
||||
|
||||
private static final int WORKERS = 4;
|
||||
private static final int ROUNDS = 3;
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
List<String> log = new CopyOnWriteArrayList<>();
|
||||
int[] roundCounter = {0};
|
||||
|
||||
CyclicBarrier barrier = new CyclicBarrier(WORKERS, () -> {
|
||||
roundCounter[0]++;
|
||||
log.add("barrier action: round " + roundCounter[0] + " complete, all " + WORKERS + " workers arrived");
|
||||
});
|
||||
|
||||
Thread[] workers = new Thread[WORKERS];
|
||||
for (int w = 0; w < WORKERS; w++) {
|
||||
int id = w;
|
||||
workers[w] = new Thread(() -> runWorker(id, barrier, log));
|
||||
workers[w].start();
|
||||
}
|
||||
for (Thread t : workers) {
|
||||
t.join();
|
||||
}
|
||||
|
||||
log.forEach(System.out::println);
|
||||
System.out.println("done");
|
||||
}
|
||||
|
||||
private static void runWorker(int id, CyclicBarrier barrier, List<String> log) {
|
||||
try {
|
||||
for (int round = 1; round <= ROUNDS; round++) {
|
||||
Thread.sleep(20 + (id * 5));
|
||||
log.add("worker-" + id + ": finished round " + round);
|
||||
barrier.await(); // same barrier object, every round - it resets on its own
|
||||
}
|
||||
} catch (InterruptedException | BrokenBarrierException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,68 @@
|
||||
package com.ankurm.synchronizers;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CopyOnWriteArrayList;
|
||||
import java.util.concurrent.Phaser;
|
||||
|
||||
/**
|
||||
* The same pipeline again, with {@link Phaser} - which does everything
|
||||
* {@link CyclicBarrierPipeline} does (a reusable, self-resetting barrier with a per-phase
|
||||
* action, via {@link Phaser#onAdvance}), plus the one thing neither {@code CountDownLatch} nor
|
||||
* {@code CyclicBarrier} can: a party count fixed at construction time. Here, a fifth worker
|
||||
* {@link Phaser#register()}s itself only after round 1 has already completed, and the barrier
|
||||
* for round 2 correctly waits for all five - something that requires building a brand new
|
||||
* {@code CyclicBarrier} (and re-pointing every existing worker at it) to do with that class.
|
||||
*/
|
||||
public final class PhaserPipeline {
|
||||
|
||||
private static final int INITIAL_WORKERS = 4;
|
||||
private static final int ROUNDS = 3;
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
List<String> log = new CopyOnWriteArrayList<>();
|
||||
|
||||
Phaser phaser = new Phaser(INITIAL_WORKERS) {
|
||||
@Override
|
||||
protected boolean onAdvance(int phase, int registeredParties) {
|
||||
log.add("phaser onAdvance: phase " + (phase + 1) + " complete, "
|
||||
+ registeredParties + " registered parties -> advancing");
|
||||
return phase + 1 >= ROUNDS || registeredParties == 0;
|
||||
}
|
||||
};
|
||||
|
||||
Thread[] workers = new Thread[INITIAL_WORKERS];
|
||||
for (int w = 0; w < INITIAL_WORKERS; w++) {
|
||||
int id = w;
|
||||
workers[w] = new Thread(() -> runWorker(id, phaser, log, 1));
|
||||
workers[w].start();
|
||||
}
|
||||
for (Thread t : workers) {
|
||||
t.join();
|
||||
}
|
||||
|
||||
log.forEach(System.out::println);
|
||||
System.out.println("done");
|
||||
}
|
||||
|
||||
private static void runWorker(int id, Phaser phaser, List<String> log, int startRound) {
|
||||
try {
|
||||
for (int round = startRound; round <= ROUNDS; round++) {
|
||||
Thread.sleep(20 + (id * 5));
|
||||
log.add("worker-" + id + ": finished round " + round);
|
||||
phaser.arriveAndAwaitAdvance();
|
||||
|
||||
// The late-joining worker: registers itself right after round 1 finishes, then
|
||||
// runs rounds 2 and 3 like everyone else. Nobody else needed to change.
|
||||
if (id == 0 && round == 1) {
|
||||
int lateId = 4;
|
||||
phaser.register();
|
||||
log.add("worker-" + lateId + ": registers after round 1 (phaser now has "
|
||||
+ phaser.getRegisteredParties() + " parties)");
|
||||
new Thread(() -> runWorker(lateId, phaser, log, 2)).start();
|
||||
}
|
||||
}
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
package com.ankurm.synchronizers;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CopyOnWriteArrayList;
|
||||
import java.util.concurrent.CyclicBarrier;
|
||||
import java.util.concurrent.Semaphore;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
/**
|
||||
* {@link Semaphore} solves a genuinely different problem than the other three classes in this
|
||||
* module, even though it gets listed alongside them constantly. {@code CountDownLatch},
|
||||
* {@code CyclicBarrier}, and {@code Phaser} all answer "has everyone reached this point yet?" -
|
||||
* {@code Semaphore} answers "how many callers are allowed through this point at once?" Nothing
|
||||
* here stops multiple threads from acquiring at wildly different times; it just caps how many
|
||||
* hold a permit simultaneously.
|
||||
*
|
||||
* <p>Same three-round pipeline shape as the other demos, but this time each worker also needs a
|
||||
* permit from a 2-permit {@code Semaphore} before it can call a simulated expensive shared
|
||||
* resource (a rate-limited downstream API, a connection pool) once per round - independent of
|
||||
* which round the barrier logic is in, and independent of worker identity.
|
||||
*/
|
||||
public final class SemaphoreResourceGuard {
|
||||
|
||||
private static final int WORKERS = 4;
|
||||
private static final int ROUNDS = 3;
|
||||
private static final int PERMITS = 2;
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
List<String> log = new CopyOnWriteArrayList<>();
|
||||
Semaphore resourceGuard = new Semaphore(PERMITS, true); // fair, for a predictable transcript
|
||||
AtomicInteger concurrentInResource = new AtomicInteger();
|
||||
AtomicInteger maxObservedConcurrency = new AtomicInteger();
|
||||
CyclicBarrier roundBarrier = new CyclicBarrier(WORKERS);
|
||||
|
||||
Thread[] workers = new Thread[WORKERS];
|
||||
for (int w = 0; w < WORKERS; w++) {
|
||||
int id = w;
|
||||
workers[w] = new Thread(() -> runWorker(
|
||||
id, resourceGuard, roundBarrier, concurrentInResource, maxObservedConcurrency, log));
|
||||
workers[w].start();
|
||||
}
|
||||
for (Thread t : workers) {
|
||||
t.join();
|
||||
}
|
||||
|
||||
log.forEach(System.out::println);
|
||||
System.out.println("permits=" + PERMITS + " maxObservedConcurrency=" + maxObservedConcurrency.get());
|
||||
System.out.println("done");
|
||||
}
|
||||
|
||||
private static void runWorker(int id, Semaphore resourceGuard, CyclicBarrier roundBarrier,
|
||||
AtomicInteger concurrentInResource, AtomicInteger maxObservedConcurrency,
|
||||
List<String> log) {
|
||||
try {
|
||||
for (int round = 1; round <= ROUNDS; round++) {
|
||||
resourceGuard.acquire();
|
||||
try {
|
||||
int inFlight = concurrentInResource.incrementAndGet();
|
||||
maxObservedConcurrency.updateAndGet(max -> Math.max(max, inFlight));
|
||||
log.add("worker-" + id + " round " + round + ": acquired permit, "
|
||||
+ inFlight + " callers concurrently in the resource"
|
||||
+ " (availablePermits=" + resourceGuard.availablePermits() + ")");
|
||||
Thread.sleep(15);
|
||||
} finally {
|
||||
concurrentInResource.decrementAndGet();
|
||||
resourceGuard.release();
|
||||
}
|
||||
roundBarrier.await(); // unrelated barrier concern, same as the other demos
|
||||
}
|
||||
} catch (Exception e) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,59 @@
|
||||
package com.ankurm.synchronizers;
|
||||
|
||||
/**
|
||||
* The contrast case for {@link VirtualThreadCompatibilityDemo} - and a correction to a common
|
||||
* assumption along the way. The naive expectation is that hand-rolling "wait for a signal" with
|
||||
* {@code synchronized} + {@code Object.wait()}/{@code notifyAll()} instead of a
|
||||
* {@code java.util.concurrent} class would pin, the same way {@code synchronized} + a blocking
|
||||
* call pinned before JEP 491. It doesn't: {@code Object.wait()} fully releases the monitor for
|
||||
* the duration of the wait, so there's no lock held while parked, and nothing to pin. What
|
||||
* genuinely pins on a pre-JDK-24 runtime is a blocking call made <em>while still holding</em> the
|
||||
* monitor - {@code Thread.sleep()} inside {@code synchronized}, exactly like
|
||||
* {@code MonitorPinningDemo} in the {@code vt-pinning} module. Both cases run here, back to back,
|
||||
* under the same trace flag, so the difference is visible in one transcript instead of asserted.
|
||||
*/
|
||||
public final class SynchronizedWaitPinsDemo {
|
||||
|
||||
private static final Object LOCK = new Object();
|
||||
private static boolean signalled = false;
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
System.out.println("java.version=" + System.getProperty("java.version"));
|
||||
System.out.println("jdk.tracePinnedThreads=" + System.getProperty("jdk.tracePinnedThreads"));
|
||||
|
||||
// Case 1: synchronized + Object.wait() - releases the monitor while parked, does not pin.
|
||||
Thread waitThread = Thread.ofVirtual().start(() -> {
|
||||
synchronized (LOCK) {
|
||||
try {
|
||||
while (!signalled) {
|
||||
LOCK.wait();
|
||||
}
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
});
|
||||
Thread.sleep(100);
|
||||
synchronized (LOCK) {
|
||||
signalled = true;
|
||||
LOCK.notifyAll();
|
||||
}
|
||||
waitThread.join();
|
||||
System.out.println("synchronized + Object.wait(): completed, no pinning trace above this line for it");
|
||||
|
||||
// Case 2: synchronized + Thread.sleep() - still holds the monitor while parked, DOES pin.
|
||||
Thread sleepThread = Thread.ofVirtual().start(() -> {
|
||||
synchronized (LOCK) {
|
||||
try {
|
||||
Thread.sleep(150);
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
});
|
||||
sleepThread.join();
|
||||
System.out.println("synchronized + Thread.sleep(): completed, pinning trace (if any) appears above this line for it");
|
||||
|
||||
System.out.println("done");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
package com.ankurm.synchronizers;
|
||||
|
||||
import java.util.concurrent.BrokenBarrierException;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.CyclicBarrier;
|
||||
import java.util.concurrent.Phaser;
|
||||
import java.util.concurrent.Semaphore;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.TimeoutException;
|
||||
|
||||
/**
|
||||
* What each class does when the thing it's waiting for never happens, deliberately forced by
|
||||
* under-supplying arrivals or permits. The four don't behave identically, which matters more
|
||||
* than it looks: a timed-out {@code CyclicBarrier} poisons itself for every other waiter, and
|
||||
* the other three don't.
|
||||
*/
|
||||
public final class TimeoutBehaviorDemo {
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
countDownLatchTimeout();
|
||||
cyclicBarrierTimeout();
|
||||
phaserTimeout();
|
||||
semaphoreTimeout();
|
||||
System.out.println("done");
|
||||
}
|
||||
|
||||
private static void countDownLatchTimeout() throws InterruptedException {
|
||||
System.out.println("=== CountDownLatch.await(timeout): expects 2, only 1 ever counts down ===");
|
||||
CountDownLatch latch = new CountDownLatch(2);
|
||||
latch.countDown(); // only one of two - the second never arrives
|
||||
boolean released = latch.await(200, TimeUnit.MILLISECONDS);
|
||||
System.out.println("await returned: " + released + " (false = timed out, latch still usable to re-check later)");
|
||||
System.out.println("getCount() after timeout: " + latch.getCount());
|
||||
}
|
||||
|
||||
private static void cyclicBarrierTimeout() throws InterruptedException {
|
||||
System.out.println();
|
||||
System.out.println("=== CyclicBarrier.await(timeout): expects 2 parties, only 1 arrives ===");
|
||||
CyclicBarrier barrier = new CyclicBarrier(2);
|
||||
try {
|
||||
barrier.await(200, TimeUnit.MILLISECONDS);
|
||||
} catch (TimeoutException e) {
|
||||
System.out.println("await threw TimeoutException, as expected");
|
||||
System.out.println("barrier.isBroken() after the timeout: " + barrier.isBroken());
|
||||
} catch (BrokenBarrierException e) {
|
||||
System.out.println("unexpected BrokenBarrierException");
|
||||
}
|
||||
// A second, unrelated thread trying to use the SAME barrier instance afterward:
|
||||
try {
|
||||
barrier.await(50, TimeUnit.MILLISECONDS);
|
||||
} catch (BrokenBarrierException e) {
|
||||
System.out.println("a second, later await() on the SAME barrier: BrokenBarrierException immediately"
|
||||
+ " - one timeout poisons the barrier for everyone until reset()");
|
||||
} catch (TimeoutException e) {
|
||||
System.out.println("unexpected TimeoutException on the second await");
|
||||
}
|
||||
}
|
||||
|
||||
private static void phaserTimeout() {
|
||||
System.out.println();
|
||||
System.out.println("=== Phaser.awaitAdvanceInterruptibly(phase, timeout): expects 2 parties, only 1 arrives ===");
|
||||
Phaser phaser = new Phaser(2);
|
||||
int phase = phaser.arrive(); // this party arrives; the second one never does
|
||||
try {
|
||||
phaser.awaitAdvanceInterruptibly(phase, 200, TimeUnit.MILLISECONDS);
|
||||
} catch (TimeoutException e) {
|
||||
System.out.println("awaitAdvanceInterruptibly threw TimeoutException, as expected");
|
||||
System.out.println("phaser.getPhase() after the timeout: " + phaser.getPhase()
|
||||
+ " (unchanged - still phase " + phase + ", not advanced or broken)");
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
// A later, correctly-completed round on the SAME phaser instance, no reset needed:
|
||||
phaser.arriveAndAwaitAdvance();
|
||||
System.out.println("a later round on the SAME phaser, once both parties actually arrive: "
|
||||
+ "succeeds normally, now at phase " + phaser.getPhase());
|
||||
}
|
||||
|
||||
private static void semaphoreTimeout() throws InterruptedException {
|
||||
System.out.println();
|
||||
System.out.println("=== Semaphore.tryAcquire(timeout): 0 permits available ===");
|
||||
Semaphore semaphore = new Semaphore(0);
|
||||
boolean acquired = semaphore.tryAcquire(200, TimeUnit.MILLISECONDS);
|
||||
System.out.println("tryAcquire returned: " + acquired + " (false = timed out, no exception, no broken state)");
|
||||
semaphore.release();
|
||||
boolean acquiredAfterRelease = semaphore.tryAcquire(50, TimeUnit.MILLISECONDS);
|
||||
System.out.println("a later tryAcquire after a release: " + acquiredAfterRelease + " - fully independent of the earlier timeout");
|
||||
}
|
||||
}
|
||||
+96
@@ -0,0 +1,96 @@
|
||||
package com.ankurm.synchronizers;
|
||||
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.CyclicBarrier;
|
||||
import java.util.concurrent.Phaser;
|
||||
import java.util.concurrent.Semaphore;
|
||||
|
||||
/**
|
||||
* All four classes in this module are built on {@link java.util.concurrent.locks.AbstractQueuedSynchronizer}
|
||||
* (Phaser uses its own compare-and-swap state machine, not AQS directly, but the same
|
||||
* park/unpark mechanism), not on {@code synchronized}. A virtual thread blocking inside any of
|
||||
* them parks via {@code LockSupport}, which was always safe to unmount from - none of the four
|
||||
* ever pinned a carrier, on any JDK version, including JDK 21 well before
|
||||
* <a href="https://openjdk.org/jeps/491">JEP 491</a> fixed {@code synchronized}. This demo blocks
|
||||
* a virtual thread inside each one in turn, run with {@code -Djdk.tracePinnedThreads=full}: if
|
||||
* it printed anything, that would be news. It doesn't. Compare against
|
||||
* {@link SynchronizedWaitPinsDemo}, the one construct in this module family that genuinely does
|
||||
* pin on a pre-JDK-24 runtime.
|
||||
*/
|
||||
public final class VirtualThreadCompatibilityDemo {
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
System.out.println("java.version=" + System.getProperty("java.version"));
|
||||
System.out.println("jdk.tracePinnedThreads=" + System.getProperty("jdk.tracePinnedThreads"));
|
||||
|
||||
runOnVirtualThread("CountDownLatch.await()", () -> {
|
||||
CountDownLatch latch = new CountDownLatch(1);
|
||||
Thread.ofPlatform().start(() -> {
|
||||
sleepQuietly(100);
|
||||
latch.countDown();
|
||||
});
|
||||
latch.await();
|
||||
});
|
||||
|
||||
runOnVirtualThread("CyclicBarrier.await()", () -> {
|
||||
CyclicBarrier barrier = new CyclicBarrier(2);
|
||||
Thread.ofPlatform().start(() -> {
|
||||
sleepQuietly(100);
|
||||
awaitQuietly(barrier);
|
||||
});
|
||||
barrier.await();
|
||||
});
|
||||
|
||||
runOnVirtualThread("Phaser.arriveAndAwaitAdvance()", () -> {
|
||||
Phaser phaser = new Phaser(2);
|
||||
Thread.ofPlatform().start(() -> {
|
||||
sleepQuietly(100);
|
||||
phaser.arriveAndAwaitAdvance();
|
||||
});
|
||||
phaser.arriveAndAwaitAdvance();
|
||||
});
|
||||
|
||||
runOnVirtualThread("Semaphore.acquire()", () -> {
|
||||
Semaphore semaphore = new Semaphore(0);
|
||||
Thread.ofPlatform().start(() -> {
|
||||
sleepQuietly(100);
|
||||
semaphore.release();
|
||||
});
|
||||
semaphore.acquire();
|
||||
});
|
||||
|
||||
System.out.println("done - no pinning trace printed above for any of the four");
|
||||
}
|
||||
|
||||
private interface BlockingAction {
|
||||
void run() throws Exception;
|
||||
}
|
||||
|
||||
private static void runOnVirtualThread(String label, BlockingAction action) throws InterruptedException {
|
||||
Thread t = Thread.ofVirtual().start(() -> {
|
||||
try {
|
||||
action.run();
|
||||
} catch (Exception e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
});
|
||||
t.join();
|
||||
System.out.println(label + ": completed, no trace above this line for it");
|
||||
}
|
||||
|
||||
private static void sleepQuietly(long ms) {
|
||||
try {
|
||||
Thread.sleep(ms);
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
|
||||
private static void awaitQuietly(CyclicBarrier barrier) {
|
||||
try {
|
||||
barrier.await();
|
||||
} catch (Exception e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
+137
@@ -0,0 +1,137 @@
|
||||
package com.ankurm.synchronizers;
|
||||
|
||||
import org.junit.jupiter.api.RepeatedTest;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.util.concurrent.BrokenBarrierException;
|
||||
import java.util.concurrent.CyclicBarrier;
|
||||
import java.util.concurrent.Phaser;
|
||||
import java.util.concurrent.Semaphore;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
/**
|
||||
* Sanity checks, not throughput or pinning proofs - see the module README and the article's
|
||||
* own quoted transcripts for those. These assert the one property that actually matters for a
|
||||
* round barrier: no worker's round-N+1 work is ever observed to start before every worker's
|
||||
* round-N work has finished, repeated to catch a scheduling-dependent race a single run wouldn't.
|
||||
*/
|
||||
class RoundOrderingCorrectnessTest {
|
||||
|
||||
private static final int WORKERS = 6;
|
||||
private static final int ROUNDS = 4;
|
||||
|
||||
@RepeatedTest(20)
|
||||
void cyclicBarrierNeverStartsARoundEarly() throws Exception {
|
||||
AtomicInteger currentRound = new AtomicInteger(1);
|
||||
AtomicInteger[] finishedThisRound = newCounters();
|
||||
CyclicBarrier barrier = new CyclicBarrier(WORKERS, () -> currentRound.incrementAndGet());
|
||||
|
||||
Thread[] workers = new Thread[WORKERS];
|
||||
for (int w = 0; w < WORKERS; w++) {
|
||||
workers[w] = new Thread(() -> {
|
||||
for (int round = 1; round <= ROUNDS; round++) {
|
||||
// The assertion: this worker's round MUST equal currentRound at the moment
|
||||
// it starts working - if some other worker's barrier.await() returned before
|
||||
// this one arrived, currentRound would already have advanced past `round`.
|
||||
assertEquals(round, currentRound.get(),
|
||||
"worker started round " + round + " but currentRound was already " + currentRound.get());
|
||||
finishedThisRound[round - 1].incrementAndGet();
|
||||
try {
|
||||
barrier.await();
|
||||
} catch (InterruptedException | BrokenBarrierException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
});
|
||||
workers[w].start();
|
||||
}
|
||||
for (Thread t : workers) {
|
||||
t.join();
|
||||
}
|
||||
for (AtomicInteger count : finishedThisRound) {
|
||||
assertEquals(WORKERS, count.get(), "every worker must have finished every round exactly once");
|
||||
}
|
||||
}
|
||||
|
||||
@RepeatedTest(20)
|
||||
void phaserNeverStartsARoundEarly() {
|
||||
AtomicInteger currentRound = new AtomicInteger(1);
|
||||
AtomicInteger[] finishedThisRound = newCounters();
|
||||
Phaser phaser = new Phaser(WORKERS) {
|
||||
@Override
|
||||
protected boolean onAdvance(int phase, int registeredParties) {
|
||||
currentRound.incrementAndGet();
|
||||
return phase + 1 >= ROUNDS || registeredParties == 0;
|
||||
}
|
||||
};
|
||||
|
||||
Thread[] workers = new Thread[WORKERS];
|
||||
for (int w = 0; w < WORKERS; w++) {
|
||||
workers[w] = new Thread(() -> {
|
||||
for (int round = 1; round <= ROUNDS; round++) {
|
||||
assertEquals(round, currentRound.get(),
|
||||
"worker started round " + round + " but currentRound was already " + currentRound.get());
|
||||
finishedThisRound[round - 1].incrementAndGet();
|
||||
phaser.arriveAndAwaitAdvance();
|
||||
}
|
||||
});
|
||||
workers[w].start();
|
||||
}
|
||||
for (Thread t : workers) {
|
||||
try {
|
||||
t.join();
|
||||
} catch (InterruptedException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
for (AtomicInteger count : finishedThisRound) {
|
||||
assertEquals(WORKERS, count.get(), "every worker must have finished every round exactly once");
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void semaphoreNeverExceedsItsPermitCount() throws Exception {
|
||||
int permits = 3;
|
||||
int callers = 30;
|
||||
Semaphore semaphore = new Semaphore(permits);
|
||||
AtomicInteger concurrent = new AtomicInteger();
|
||||
AtomicInteger maxObserved = new AtomicInteger();
|
||||
|
||||
Thread[] threads = new Thread[callers];
|
||||
for (int i = 0; i < callers; i++) {
|
||||
threads[i] = new Thread(() -> {
|
||||
try {
|
||||
semaphore.acquire();
|
||||
try {
|
||||
int inFlight = concurrent.incrementAndGet();
|
||||
maxObserved.updateAndGet(max -> Math.max(max, inFlight));
|
||||
Thread.sleep(5);
|
||||
} finally {
|
||||
concurrent.decrementAndGet();
|
||||
semaphore.release();
|
||||
}
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
});
|
||||
threads[i].start();
|
||||
}
|
||||
for (Thread t : threads) {
|
||||
t.join();
|
||||
}
|
||||
assertTrue(maxObserved.get() <= permits,
|
||||
"observed " + maxObserved.get() + " concurrent callers with only " + permits + " permits");
|
||||
assertEquals(permits, semaphore.availablePermits(), "all permits must be returned when every caller finishes");
|
||||
}
|
||||
|
||||
private static AtomicInteger[] newCounters() {
|
||||
AtomicInteger[] counters = new AtomicInteger[ROUNDS];
|
||||
for (int i = 0; i < ROUNDS; i++) {
|
||||
counters[i] = new AtomicInteger();
|
||||
}
|
||||
return counters;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user