From 49fb444060432509fa1fd4d3da6b558b446e1a13 Mon Sep 17 00:00:00 2001 From: fabrizzio-dotCMS Date: Wed, 19 Aug 2026 17:20:31 -0600 Subject: [PATCH 1/4] #34154: test(java25): runnable demos for carrier scheduling and object headers Four main-class demos under com.dotcms.jdk, built for the Java 21-25 tech talk. They are the evidence half of the talk: each turns a claim about the JVM into something an audience can watch happen, using this codebase's own classes and its own incidents. - VirtualThreadCarrierTimelineDemo draws one row per carrier and one column per 100 ms tick, showing which virtual thread was mounted. mode=file comes out identical to a CPU hog while mode=socket frees both carriers at once, which is the shape of #37038. - VirtualThreadCarrierStarvationDemo contrasts a virtual-thread executor with a fixed platform pool: both start 2 of 10 tasks, but only the platform run lets an unrelated virtual thread proceed. - VirtualThreadYieldVsParkDemo separates yielding from parking on the same two carriers. - CompactObjectHeadersDemo relaunches itself in two child JVMs, one per setting of -XX:+/-UseCompactObjectHeaders, and measures the real ContentSearchHit and SiteSearchHit records with the thread allocation counter plus a heap-delta cross-check. No production code is touched and nothing new runs in CI beyond compilation. Co-Authored-By: Claude Opus 5 (1M context) --- .../dotcms/jdk/CompactObjectHeadersDemo.java | 455 ++++++++++++++ .../VirtualThreadCarrierStarvationDemo.java | 98 +++ .../jdk/VirtualThreadCarrierTimelineDemo.java | 593 ++++++++++++++++++ .../jdk/VirtualThreadYieldVsParkDemo.java | 84 +++ 4 files changed, 1230 insertions(+) create mode 100644 dotCMS/src/test/java/com/dotcms/jdk/CompactObjectHeadersDemo.java create mode 100644 dotCMS/src/test/java/com/dotcms/jdk/VirtualThreadCarrierStarvationDemo.java create mode 100644 dotCMS/src/test/java/com/dotcms/jdk/VirtualThreadCarrierTimelineDemo.java create mode 100644 dotCMS/src/test/java/com/dotcms/jdk/VirtualThreadYieldVsParkDemo.java diff --git a/dotCMS/src/test/java/com/dotcms/jdk/CompactObjectHeadersDemo.java b/dotCMS/src/test/java/com/dotcms/jdk/CompactObjectHeadersDemo.java new file mode 100644 index 000000000000..d78811a12139 --- /dev/null +++ b/dotCMS/src/test/java/com/dotcms/jdk/CompactObjectHeadersDemo.java @@ -0,0 +1,455 @@ +package com.dotcms.jdk; + +import com.sun.management.HotSpotDiagnosticMXBean; +import com.sun.management.ThreadMXBean; +import java.lang.management.ManagementFactory; +import java.lang.reflect.Constructor; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.function.Supplier; + +/** + * Live demo for the Java 25 talk: what {@code -XX:+UseCompactObjectHeaders} (JEP 519) actually + * costs and buys, measured on dotCMS's own objects. + * + *

Every Java object carries a fixed prefix the JVM owns and your code never sees: the + * object header. Classically 12 bytes — an 8-byte mark word (identity hash, lock state, GC + * age, forwarding pointer; multiplexed, so a plain object leaves most of it unused) plus a 4-byte + * compressed klass pointer (which class this is). Compact headers narrow the klass pointer to 22 + * bits and fold it into the spare room of the mark word, giving a single 8-byte header. + * Same information, one word less, on every object in the heap. + * + *

The flag is not on by default in JDK 25 ({@code UseCompactObjectHeaders = false + * {product}}). dotCMS opted in deliberately: {@code container/tomcat9/bin/setenv.sh} for the + * production container and {@code parent/pom.xml} for the surefire/failsafe JVMs. + * + *

Why this demo relaunches itself

+ * + * The flag is read once at JVM startup, so no single process can show both sides and no JUnit test + * can compare them — which is exactly why {@code CacheSizingUtilTest} was rewritten to assert + * layout-independent invariants instead of byte counts. This class therefore spawns two child + * JVMs, identical but for {@code -XX:+/-UseCompactObjectHeaders}, and prints their results side by + * side. + * + *

How it measures

+ * + * Two independent instruments, neither of which is a sizing API reporting on itself: + * + *
    + *
  1. Allocation counter (the headline number) — {@code ThreadMXBean + * .getCurrentThreadAllocatedBytes()} around the fill loop, divided by N. The JVM counts the + * bytes it actually handed out, so this is exact and does not depend on the collector running. + *
  2. Heap delta (the cross-check) — used heap before and after the fill, with the backing + * array allocated before the baseline so its own N reference slots are excluded. Noisy + * by nature, but it answers the fair question the counter cannot: the bytes are not merely + * un-allocated, they are genuinely not resident. + *
+ * + * Payloads are shared singletons, so what the first three rows measure is the object shell itself. + * Both numbers are printed; the headline is snapped to the object alignment, which is exact rather + * than cosmetic because every footprint — and every sum of footprints — is a multiple of it. + * + *

When the dotCMS classpath is present the real records + * {@code com.dotcms.content.index.domain.ContentSearchHit} / {@code SiteSearchHit} are loaded and + * measured. Standalone, the demo falls back to local twins with an identical field layout and says + * so in the output. + * + *

What to look for on the slide

+ * + * 
+ *   shape                12-byte hdr   8-byte hdr   saved
+ *   java.lang.Object            16 B          8 B     8 B   the header and nothing else
+ *   ContentSearchHit            40 B         32 B     8 B   5 refs + 1 float = 24 B of real data
+ *   SiteSearchHit               40 B         40 B     0 B   one MORE field, and it costs nothing
+ *   realistic hit             3176 B       2992 B    184 B   the same hit carrying its _source
+ *
+ * + *

Every row is exact and reproducible run to run — that is what the allocation counter buys. The + * last row is a graph of some sixty objects, so its absolute size depends on how this demo builds the + * map; what carries over to production is the ratio, not the constant. + * + * The second and third rows are the point. {@code SiteSearchHit} carries one extra reference than + * {@code ContentSearchHit}, yet with 12-byte headers both weigh the same: objects are aligned + * to 8 bytes, so 12+24=36 rounds up to 40 and the extra field lands in padding that was already paid + * for. With 8-byte headers the short shape lands exactly on 32 and the long one is the one rounding + * up — so the very same field now costs 8 bytes. + * + *

The lesson is not "objects got 4 bytes smaller". It is that compact headers change which + * refactors are worth doing: splitting these two shapes apart (#36899) returned nothing before + * the flag and returns 8 bytes per hit after it. + * + *

The final row keeps that honest, and is the row to end on. Against a hit carrying its own + * 20-field {@code _source}, the split is worth 0.25% while the flag itself is worth 5.8% — + * because the flag shrinks every object in the graph, the map nodes and strings and char + * arrays, not just the record you were looking at. That is also why the payoff shows up in the + * caches, where dotCMS holds millions of small objects, rather than in any one data class. + * + *

Run it — one command prints both columns: + * + * 

+ *   # standalone, no build required (uses local twins)
+ *   java dotCMS/src/test/java/com/dotcms/jdk/CompactObjectHeadersDemo.java
+ *
+ *   # against the real dotCMS records (after ./mvnw install -pl :dotcms-core --am -DskipTests)
+ *   java -cp dotCMS/target/classes:dotCMS/target/test-classes com.dotcms.jdk.CompactObjectHeadersDemo
+ *
+ * + * Options: {@code -Dn=1000000} instances per shape, {@code -Dchild=on|off} to run a single side in + * the current JVM instead of spawning children. + * + *

Not a JUnit test on purpose: the whole point is a comparison across two JVMs, and a test runs in + * the one surefire started — which is precisely why {@code CacheSizingUtilTest} had to give up on byte + * counts. It is a {@code main} demo, compiled by the real build so it cannot silently rot, and named + * {@code *Demo} so surefire skips it. {@code System.out} is deliberate: the console output is + * the artifact being shown to an audience; the Logger-only rule targets production code. + * + * @author Fabrizio Araya + * @see JEP 519 — Compact Object Headers + * @see VirtualThreadCarrierTimelineDemo + */ +public final class CompactObjectHeadersDemo { + + /** Instances allocated per shape. Large enough that per-object rounding is invisible. */ + private static final int N = Integer.getInteger("n", 1_000_000); + + /** Instances allocated and discarded before measuring, to move JIT/reflection warmup out of the window. */ + private static final int WARMUP = 20_000; + + /** Payloads shared by every instance, so only the object shell shows up in the heap delta. */ + private static final String SHARED_ID = "shared-id"; + private static final String SHARED_INDEX = "shared-index"; + private static final Map SHARED_MAP = Map.of(); + private static final List SHARED_LIST = List.of(); + + private CompactObjectHeadersDemo() { + } + + public static void main(final String[] args) throws Exception { + final String child = System.getProperty("child"); + if (child != null) { + runOneSide(); + return; + } + runBothSides(); + } + + // ───────────────────────────────────────────────────────────────────────── + // Shapes under measurement + // ───────────────────────────────────────────────────────────────────────── + + /** Local stand-in for {@code ContentSearchHit}: 5 references + 1 float. */ + record ContentHitTwin(String getId, String getIndex, Map getSourceAsMap, + float getScore, Map getFields, List getSortValues) { + } + + /** Local stand-in for {@code SiteSearchHit}: the same plus one highlights reference. */ + record SiteHitTwin(String getId, String getIndex, Map getSourceAsMap, + float getScore, Map getFields, List getSortValues, + Map> getHighlights) { + } + + /** + * One shape to measure. + * + * @param divisor how much to scale N down for this shape — a shape that retains kilobytes cannot + * be allocated a million times, and does not need to be: the measurement noise is + * a fraction of a byte per instance either way. + */ + private record Shape(String label, String note, int divisor, Supplier factory) { + } + + private static List shapes() { + final List shapes = new ArrayList<>(); + + shapes.add(new Shape("java.lang.Object", "header only, no fields", 1, Object::new)); + + final Supplier realContent = realHitFactory( + "com.dotcms.content.index.domain.ContentSearchHit", false); + final Supplier realSite = realHitFactory( + "com.dotcms.content.index.domain.SiteSearchHit", true); + + shapes.add(new Shape("ContentSearchHit", + realContent != null ? "real record, 5 refs + 1 float" : "TWIN, 5 refs + 1 float", 1, + realContent != null ? realContent + : () -> new ContentHitTwin(SHARED_ID, SHARED_INDEX, SHARED_MAP, 1.0f, + SHARED_MAP, SHARED_LIST))); + + shapes.add(new Shape("SiteSearchHit", + realSite != null ? "real record, 6 refs + 1 float" : "TWIN, 6 refs + 1 float", 1, + realSite != null ? realSite + : () -> new SiteHitTwin(SHARED_ID, SHARED_INDEX, SHARED_MAP, 1.0f, + SHARED_MAP, SHARED_LIST, Map.of()))); + + shapes.add(new Shape("realistic hit", "+ its own 20-field _source map", 20, + CompactObjectHeadersDemo::realisticHit)); + + return shapes; + } + + /** + * Builds a factory for a real dotCMS hit record when it is on the classpath, or {@code null} + * when running standalone. The canonical constructor is + * {@code (String, String, Map, float, Map, List[, Map])}. + */ + private static Supplier realHitFactory(final String className, final boolean withHighlights) { + try { + final Class type = Class.forName(className); + final Constructor ctor = withHighlights + ? type.getDeclaredConstructor(String.class, String.class, Map.class, float.class, + Map.class, List.class, Map.class) + : type.getDeclaredConstructor(String.class, String.class, Map.class, float.class, + Map.class, List.class); + final Object[] argv = withHighlights + ? new Object[] {SHARED_ID, SHARED_INDEX, SHARED_MAP, 1.0f, SHARED_MAP, + SHARED_LIST, SHARED_MAP} + : new Object[] {SHARED_ID, SHARED_INDEX, SHARED_MAP, 1.0f, SHARED_MAP, + SHARED_LIST}; + ctor.newInstance(argv); // fail fast here rather than inside the measurement loop + return () -> { + try { + return ctor.newInstance(argv); + } catch (ReflectiveOperationException e) { + throw new IllegalStateException(e); + } + }; + } catch (ReflectiveOperationException | LinkageError notOnClasspath) { + return null; + } + } + + /** + * A hit as it actually arrives from the engine: the record shell plus a private 20-entry + * {@code _source} map. Nothing is shared, so the heap delta is the whole retained cost — which is + * the honesty check against the shell-only rows above. + */ + private static Object realisticHit() { + final Map source = new LinkedHashMap<>(); + for (int i = 0; i < 20; i++) { + source.put("field_" + i, "value_" + i); + } + return new SiteHitTwin("id-" + source.hashCode(), "working-index", source, 1.0f, + new HashMap<>(), List.of(), Map.of()); + } + + // ───────────────────────────────────────────────────────────────────────── + // Single side: measure in THIS JVM and print machine-readable rows + // ───────────────────────────────────────────────────────────────────────── + + private static final ThreadMXBean THREADS = + (ThreadMXBean) ManagementFactory.getThreadMXBean(); + + private static void runOneSide() { + for (final Shape shape : shapes()) { + final int instances = N / shape.divisor(); + final double allocated = allocatedBytesPerInstance(shape.factory(), instances); + final double heap = heapDeltaPerInstance(shape.factory(), instances); + System.out.println("ROW\t" + shape.label() + "\t" + snapToAlignment(allocated) + "\t" + + String.format("%.2f", allocated) + "\t" + String.format("%.0f", heap) + "\t" + + shape.note()); + } + } + + /** + * Bytes the JVM handed this thread per instance — the exact figure. Warmup runs first so the JIT + * and the reflection caches do their one-time allocation outside the measured window, and the + * backing array is allocated before it so its own slots are not counted. + */ + private static double allocatedBytesPerInstance(final Supplier factory, final int instances) { + warmUp(factory); + final Object[] holder = new Object[instances]; + + final long before = THREADS.getCurrentThreadAllocatedBytes(); + for (int i = 0; i < instances; i++) { + holder[i] = factory.get(); + } + final long after = THREADS.getCurrentThreadAllocatedBytes(); + + if (holder[instances - 1] == null) { + throw new IllegalStateException("unreachable"); + } + return (double) (after - before) / instances; + } + + /** + * Rounds a measurement to the nearest multiple of {@code ObjectAlignmentInBytes} (8 here). + * + *

This is not cosmetic. The JVM pads every object up to the alignment, so a true footprint is + * always a multiple of it — and so is any sum of them. The heap delta below carries a fraction of + * a byte of noise per instance (JIT and reflection allocate a little of their own), which snapping + * removes without inventing anything. The raw average is printed too so the correction is visible. + */ + private static long snapToAlignment(final double raw) { + final long alignment = Long.parseLong(vmOption("ObjectAlignmentInBytes")); + return Math.round(raw / alignment) * alignment; + } + + /** Lets the JIT compile the factory and reflection build its caches before anything is measured. */ + private static void warmUp(final Supplier factory) { + final Object[] warmup = new Object[WARMUP]; + for (int i = 0; i < WARMUP; i++) { + warmup[i] = factory.get(); + } + if (warmup[WARMUP - 1] == null) { + throw new IllegalStateException("unreachable"); + } + } + + /** + * Heap delta per instance — the independent cross-check. The backing array is allocated and + * settled before the fill, so its own reference slots are excluded. This one is genuinely + * noisy: it depends on the collector having settled, and it under-reports for shapes that retain + * large graphs. Read it as confirmation of the allocation counter, never instead of it. + */ + private static double heapDeltaPerInstance(final Supplier factory, final int instances) { + warmUp(factory); + final Object[] holder = new Object[instances]; + final long baseline = settledHeap(); + + for (int i = 0; i < instances; i++) { + holder[i] = factory.get(); + } + + final long filled = settledHeap(); + final double perInstance = (double) (filled - baseline) / instances; + + // Keep the array strongly reachable across the second measurement. + if (holder[instances - 1] == null) { + throw new IllegalStateException("unreachable"); + } + return perInstance; + } + + /** Used heap after coaxing the collector, so the delta reflects live objects only. */ + private static long settledHeap() { + final Runtime runtime = Runtime.getRuntime(); + long used = Long.MAX_VALUE; + for (int attempt = 0; attempt < 6; attempt++) { + System.gc(); + final long now = runtime.totalMemory() - runtime.freeMemory(); + if (now >= used) { + break; // stopped shrinking + } + used = now; + } + return used; + } + + // ───────────────────────────────────────────────────────────────────────── + // Parent side: spawn both children and print the comparison + // ───────────────────────────────────────────────────────────────────────── + + private static void runBothSides() throws Exception { + System.out.println(); + System.out.println("Compact object headers (JEP 519) — measured on " + N + " instances per shape"); + System.out.println("JVM: " + Runtime.version() + " compressed oops: " + vmOption("UseCompressedOops") + + " object alignment: " + vmOption("ObjectAlignmentInBytes") + " bytes"); + System.out.println("Default for UseCompactObjectHeaders on this JDK: " + vmOption("UseCompactObjectHeaders") + + " (dotCMS turns it on in setenv.sh and parent/pom.xml)"); + + final Map classicRaw = new LinkedHashMap<>(); + final Map compactRaw = new LinkedHashMap<>(); + final Map classic = measureIn("-XX:-UseCompactObjectHeaders", classicRaw); + final Map compact = measureIn("-XX:+UseCompactObjectHeaders", compactRaw); + final Map notes = new LinkedHashMap<>(); + shapes().forEach(s -> notes.put(s.label(), s.note())); + + System.out.println(); + System.out.printf("%-20s %14s %14s %10s %s%n", + "shape", "12-byte hdr", "8-byte hdr", "saved", ""); + System.out.println("-".repeat(86)); + for (final String label : notes.keySet()) { + final long before = classic.getOrDefault(label, -1L); + final long after = compact.getOrDefault(label, -1L); + System.out.printf("%-20s %11d B %11d B %8d B %s%n", + label, before, after, before - after, notes.get(label)); + } + System.out.println(); + System.out.println("Read the two hit rows together: one extra reference field costs " + + (classic.getOrDefault("SiteSearchHit", 0L) - classic.getOrDefault("ContentSearchHit", 0L)) + + " B with the classic header and " + + (compact.getOrDefault("SiteSearchHit", 0L) - compact.getOrDefault("ContentSearchHit", 0L)) + + " B with the compact one."); + System.out.println("Same field, same code. 8-byte alignment decides what you actually pay."); + + final long hitBefore = classic.getOrDefault("realistic hit", 0L); + final long hitAfter = compact.getOrDefault("realistic hit", 0L); + final long splitSaving = classic.getOrDefault("SiteSearchHit", 0L) + - compact.getOrDefault("ContentSearchHit", 0L); + System.out.println(); + System.out.printf("But keep the magnitudes straight. On a hit that carries its own _source, " + + "the flag returns %d B of %d (%.1f%%), while splitting the two record shapes " + + "apart returns %d B of the same %d (%.2f%%).%n", + hitBefore - hitAfter, hitBefore, 100.0 * (hitBefore - hitAfter) / hitBefore, + splitSaving, hitBefore, 100.0 * splitSaving / hitBefore); + System.out.println("The flag pays off because it shrinks EVERY object in the graph — the map " + + "nodes, the strings, the char arrays — not because it shrank your record."); + System.out.println(); + System.out.println("Unrounded measurements — allocation counter, then the heap-delta cross-check:"); + System.out.printf(" %-20s %26s %26s%n", "", "12-byte header", "8-byte header"); + System.out.printf(" %-20s %13s %12s %13s %12s%n", + "", "allocated", "heap", "allocated", "heap"); + for (final String label : notes.keySet()) { + System.out.printf(" %-20s %s %s%n", + label, classicRaw.get(label), compactRaw.get(label)); + } + System.out.println(); + } + + /** Runs this class in a child JVM with the given flag and parses its ROW lines. */ + private static Map measureIn(final String headerFlag, final Map raw) + throws Exception { + final String java = ProcessHandle.current().info().command() + .orElse(System.getProperty("java.home") + "/bin/java"); + + final List command = new ArrayList<>(List.of(java, + headerFlag, + "-Xmx2g", + "-Dchild=" + headerFlag, + "-Dn=" + N)); + + // Launched as `java Demo.java`, this class lives in a memory class loader and no child could + // find it on a class path — so hand the child the source file and let it compile it again. + final String sourceFile = System.getProperty("jdk.launcher.sourcefile"); + if (sourceFile != null) { + command.add(sourceFile); + } else { + // --enable-preview: dotCMS compiles with it, and preview-marked classes refuse to load + // without it. Harmless when the class path holds no preview classes. + command.addAll(List.of("--enable-preview", + "-cp", System.getProperty("java.class.path"), + CompactObjectHeadersDemo.class.getName())); + } + + final Process process = new ProcessBuilder(command).redirectErrorStream(true).start(); + final Map rows = new LinkedHashMap<>(); + final List transcript = new ArrayList<>(); + try (var reader = process.inputReader()) { + String line; + while ((line = reader.readLine()) != null) { + transcript.add(line); + if (line.startsWith("ROW\t")) { + final String[] parts = line.split("\t"); + rows.put(parts[1], Long.parseLong(parts[2])); + raw.put(parts[1], String.format("%10s B %10s B", parts[3], parts[4])); + } + } + } + if (process.waitFor() != 0 || rows.isEmpty()) { + transcript.forEach(System.out::println); + throw new IllegalStateException("child JVM failed for " + headerFlag); + } + return rows; + } + + private static String vmOption(final String name) { + try { + return ManagementFactory.getPlatformMXBean(HotSpotDiagnosticMXBean.class) + .getVMOption(name).getValue(); + } catch (RuntimeException unsupported) { + return "n/a"; + } + } +} diff --git a/dotCMS/src/test/java/com/dotcms/jdk/VirtualThreadCarrierStarvationDemo.java b/dotCMS/src/test/java/com/dotcms/jdk/VirtualThreadCarrierStarvationDemo.java new file mode 100644 index 000000000000..eab2b8328ee2 --- /dev/null +++ b/dotCMS/src/test/java/com/dotcms/jdk/VirtualThreadCarrierStarvationDemo.java @@ -0,0 +1,98 @@ +package com.dotcms.jdk; + +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * Live demo for the Java 25 talk: a virtual-thread executor never runs out of workers — it runs + * out of JVM. + * + *

Ten tasks, each an uninterruptible {@code while(true)} CPU loop, are submitted to either a + * virtual-thread-per-task executor or a fixed pool of two platform threads. In both cases only two + * tasks ever execute, because there are only two carriers / two pool threads. The difference is the + * blast radius: + * + * 

+ *                                  VT executor      fixed platform pool (2)
+ *   tasks that started               2 of 10               2 of 10
+ *   unrelated code elsewhere ran?      NO                    YES
+ *
+ * + *

With platform threads the starvation is contained: eight tasks wait in your queue, + * under your bound, and the OS preempts the two runners so the rest of the process keeps + * going. With virtual threads the scarce resource is the carrier pool, which is + * {@code availableProcessors()} wide, global to the JVM and shared with every other virtual + * thread in the process — so an unrelated task that never touched this executor is starved too. + * The VT scheduler is cooperative and will not take a carrier back by force. + * + *

This is the mechanism behind issue #37038 (fixed in PR #37041): a blocking file read pins its + * carrier for the whole call exactly like this CPU loop does, and with 2-4 carriers in a container a + * handful of slow reads on a network mount stalled content indexing process-wide. + * + *

Run it (no flags needed — it pins the carrier count itself so the demo is reproducible + * on a 24-core laptop): + * + * 

+ *   java -Dmode=vt       dotCMS/src/test/java/com/dotcms/jdk/VirtualThreadCarrierStarvationDemo.java
+ *   java -Dmode=platform dotCMS/src/test/java/com/dotcms/jdk/VirtualThreadCarrierStarvationDemo.java
+ *
+ * + *

Not a JUnit test on purpose: the hog tasks ignore interrupts and can never be reclaimed, so + * running them inside a shared surefire JVM would starve the rest of the suite. It is a {@code main} + * demo, compiled by the real build so it cannot silently rot, and named {@code *Demo} so surefire + * does not pick it up. {@code System.out} is deliberate here — the console output is the + * artifact being shown to an audience; the Logger-only rule targets production code. + * + * @see VirtualThreadYieldVsParkDemo + * @see VirtualThreadCarrierTimelineDemo + */ +public class VirtualThreadCarrierStarvationDemo { + + private static final int CARRIERS = 2; + private static final int TASKS = 10; + + public static void main(final String[] args) throws InterruptedException { + // Read lazily on first virtual thread creation, so setting it here is enough: the demo does + // not depend on the audience's core count, and needs no command-line flag. + if (System.getProperty("jdk.virtualThreadScheduler.parallelism") == null) { + System.setProperty("jdk.virtualThreadScheduler.parallelism", String.valueOf(CARRIERS)); + } + + final String mode = System.getProperty("mode", "vt"); + final boolean virtual = "vt".equals(mode); + final AtomicInteger started = new AtomicInteger(); + + System.out.println("mode = " + mode + + " | carriers = " + System.getProperty("jdk.virtualThreadScheduler.parallelism") + + " | availableProcessors = " + Runtime.getRuntime().availableProcessors()); + + final ExecutorService pool = virtual + ? Executors.newVirtualThreadPerTaskExecutor() + : Executors.newFixedThreadPool(CARRIERS); + + for (int i = 0; i < TASKS; i++) { + pool.submit(() -> { + started.incrementAndGet(); + long spin = 0; + while (true) { // never blocks, never parks, never yields the carrier + spin++; + } + }); + } + + Thread.sleep(500); + System.out.println("tasks that STARTED: " + started.get() + " of " + TASKS + + (virtual ? " (all " + TASKS + " virtual threads exist — 8 never ran a line)" : "")); + + // A completely unrelated part of the system, which knows nothing about the pool above. + Thread.ofVirtual().name("innocent").start( + () -> System.out.println(">>> the INNOCENT virtual thread RAN")); + + Thread.sleep(2000); + System.out.println("--- done: no '>>>' above means unrelated code was starved too ---"); + + // The hogs ignore interrupts by design, so there is nothing to shut down gracefully. + System.exit(0); + } +} diff --git a/dotCMS/src/test/java/com/dotcms/jdk/VirtualThreadCarrierTimelineDemo.java b/dotCMS/src/test/java/com/dotcms/jdk/VirtualThreadCarrierTimelineDemo.java new file mode 100644 index 000000000000..4aeef8fd6ea3 --- /dev/null +++ b/dotCMS/src/test/java/com/dotcms/jdk/VirtualThreadCarrierTimelineDemo.java @@ -0,0 +1,593 @@ +package com.dotcms.jdk; + +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; + +/** + * Live demo for the Java 25 talk: you can watch the carriers. Six virtual threads are handed + * to two carriers, and the demo prints, tick by tick, which virtual thread each carrier was actually + * running. + * + *

How to read the picture. One row is one carrier: a real OS thread, the only + * thing in the JVM that can actually execute code. One cell is one 100 ms tick, oldest on + * the left, newest on the right. Inside the cell: + * + * 

+ *   1 .. 6   which of the six virtual threads was mounted on that carrier during that tick
+ *   .        nobody was mounted: the carrier was idle and free for any other task in the JVM
+ *   C        the unrelated 'canary' virtual thread, submitted mid-run, that only wants a carrier
+ *            for a microsecond
+ *
+ * + *

So a row that keeps repeating the same digit means one virtual thread took that carrier and + * never gave it back. A row whose digits keep changing, with dots in between, means the tasks are + * releasing the carrier and taking turns on it: + * + * 

+ *   mode=cpu    (never releases)             mode=sleep    (releases on every sleep)
+ *   carrier-1 | 111111111111111111           carrier-1 | 1.4.2.6.3.1.5.2.
+ *   carrier-2 | 222222222222222222           carrier-2 | 2.5.3.1.4.6.2.3.
+ *   2 of 6 tasks ever got a carrier          6 of 6 tasks got a carrier
+ *   canary: NEVER RAN                        canary: ran in 1 ms
+ *
+ * + *

Same six tasks, same two carriers, same executor in both cases. The only difference is whether + * the blocking call releases the carrier - which is the whole point: a virtual thread is + * cheap, a carrier is not, and the carrier pool is {@code availableProcessors()} wide, global to the + * JVM and shared with every other virtual thread in the process. Nothing takes a carrier back by + * force; the scheduler is cooperative. + * + *

How the carrier is observed. {@code VirtualThread.toString()} appends the carrier while + * the thread is mounted and omits it while it is not: + * + * 

+ *   VirtualThread[#26,task-1]/runnable@ForkJoinPool-1-worker-1   <- mounted, prints as a digit
+ *   VirtualThread[#28,task-3]/timed_waiting                      <- unmounted, prints as a dot
+ *
+ * + * The JDK names the carriers {@code ForkJoinPool-1-worker-N}; the demo relabels them + * {@code carrier-N}, because "worker" is exactly the word that makes an audience think it is looking + * at the virtual threads instead of at the scarce OS threads underneath them. + * + * A platform thread polls that every tick. It has to be a platform thread: run the monitor on + * a virtual thread ({@code -Dmonitor=virtual}) and the tool itself freezes mid-drawing, which is the + * same failure the rest of the JVM is suffering. + * + *

Modes ({@code -Dmode=}): + * + * 

+ *   compare  (default)  all four phases: sleep, socket, file, cpu - in that order
+ *   sleep               Thread.sleep in a loop                 - unmounts on every sleep
+ *   socket              read() on a socket nobody writes to    - unmounts once, forever
+ *   file                read() on a FIFO nobody writes to      - blocking FILE i/o: measure it
+ *   cpu                 uninterruptible while(true) loop       - never unmounts
+ *
+ * + *

The order of {@code compare} is itself part of the lesson. {@code sleep} and {@code socket} give + * the carrier back, so the JVM survives them; {@code file} parks its tasks in a read that keeps the + * carrier, but writing to the FIFO releases them, so the phase can clean up after itself; {@code cpu} + * cannot be cleaned up at all - its tasks ignore interrupts - so it has to run last. A phase that + * starts with no carrier left to obtain says {@code SKIPPED} instead of drawing an empty grid. + * + *

{@code file} is the shape of issue #37038 (fixed in PR #37041): a blocking file read on a + * network mount, on a container with 2-4 carriers, stalled content indexing process-wide. Run + * {@code file} and {@code socket} back to back - the rows tell you which kind of blocking Loom knows + * how to unmount and which one just eats a carrier. + * + * 

+ *   java dotCMS/src/test/java/com/dotcms/jdk/VirtualThreadCarrierTimelineDemo.java
+ *   java -Dmode=file   dotCMS/src/test/java/com/dotcms/jdk/VirtualThreadCarrierTimelineDemo.java
+ *   java -Dmode=socket -Dcolor=false ...VirtualThreadCarrierTimelineDemo.java
+ *
+ * + *

Not a JUnit test on purpose: the cpu tasks ignore interrupts and can never be reclaimed, so + * running them inside a shared surefire JVM would starve the rest of the suite. It is a {@code main} + * demo, compiled by the real build so it cannot silently rot, and named {@code *Demo} so surefire + * skips it. {@code System.out} is deliberate: the console output is the artifact being shown + * to an audience; the Logger-only rule targets production code. + * + * @see VirtualThreadCarrierStarvationDemo + * @see VirtualThreadYieldVsParkDemo + */ +public class VirtualThreadCarrierTimelineDemo { + + private static final int CARRIERS = 2; + private static final int TASKS = 6; + private static final int TICK_MS = 100; + private static final int TICKS = 40; + private static final int CANARY_TICK = 5; + private static final char FREE = '.'; + private static final char CANARY = 'C'; + private static final boolean COLOR = !"false".equals(System.getProperty("color")); + private static final String CSI = ((char) 27) + "["; + + /** What each of the six tasks does, and how its rows should be read. */ + private record Workload(String name, String detail, String expectation, Runnable body) { + + } + + public static void main(final String[] args) throws Exception { + // Read lazily on first virtual thread creation, so setting it here is enough: the demo does + // not depend on the audience's core count and needs no command-line flag. + if (System.getProperty("jdk.virtualThreadScheduler.parallelism") == null) { + System.setProperty("jdk.virtualThreadScheduler.parallelism", String.valueOf(CARRIERS)); + } + final String mode = System.getProperty("mode", "compare"); + if ("compare".equals(mode)) { + // Order matters, and it is part of the lesson. sleep and socket release the carrier, so + // the JVM survives them. file parks its tasks in a blocking read that keeps the carrier, + // but they can be unblocked by writing to the FIFO, so the phase cleans up after itself. + // cpu cannot be cleaned up - the tasks ignore interrupts - so it has to be the last one. + runPhase("sleep"); + //runPhase("socket"); + //runPhase("file"); + runPhase("cpu"); // terminal: nothing in this JVM gets a carrier after this + } else { + runPhase(mode); + } + System.out.flush(); + System.exit(0); // the cpu tasks ignore interrupts; there is nothing to shut down + } + + private static void runPhase(final String mode) throws Exception { + final List cleanup = new ArrayList<>(); + final AtomicLong releases = new AtomicLong(); + // Every phase must leave the JVM as it found it, or the next phase measures the leftovers + // instead of its own workload. Whatever can retire, retires when this flips. + final AtomicBoolean running = new AtomicBoolean(true); + final Workload workload = workload(mode, releases, running, cleanup); + + System.out.println(); + System.out.println("mode = " + workload.name() + " | " + workload.detail()); + System.out.println("carriers = " + System.getProperty("jdk.virtualThreadScheduler.parallelism") + + " of " + Runtime.getRuntime().availableProcessors() + " cpus" + + " | tasks = " + TASKS + " virtual threads" + + " | tick = " + TICK_MS + " ms"); + System.out.println(); + printLegend(workload); + System.out.println(); + + if (!carrierAvailable()) { + System.out.println(" SKIPPED: no carrier can be obtained any more. An earlier phase took" + + " both carriers and"); + System.out.println(" never gave them back, so there is nothing left to measure" + + " here. Run this"); + System.out.println(" mode in its own JVM: -Dmode=" + mode); + System.out.println(); + running.set(false); + closeAll(cleanup); + return; + } + + final Map labels = new LinkedHashMap<>(); + final List watched = new ArrayList<>(); + final Timeline timeline = new Timeline(labels); + final Runnable body = workload.body(); + for (int i = 0; i < TASKS; i++) { + // The label is decided here, not after start(), because the task reports its own carrier + // the instant it starts running - which can happen before start() has even returned. + final char label = (char) ('1' + i); + final Thread task = Thread.ofVirtual().name("task-" + (i + 1)).start(() -> { + timeline.selfReport(label); + body.run(); + }); + labels.put(task.threadId(), label); + watched.add(task); + } + + final AtomicLong canaryRanAt = new AtomicLong(-1); + final AtomicLong canarySubmittedAt = new AtomicLong(); + final Runnable monitor = () -> { + for (int tick = 0; tick < TICKS; tick++) { + if (tick == CANARY_TICK) { + // A completely unrelated part of the system, which knows nothing about the six + // tasks above and only wants a carrier for a microsecond. + canarySubmittedAt.set(System.nanoTime()); + final Thread canary = Thread.ofVirtual().name("canary").start(() -> { + timeline.selfReport(CANARY); + canaryRanAt.set(System.nanoTime() - canarySubmittedAt.get()); + }); + labels.put(canary.threadId(), CANARY); + watched.add(canary); + } + timeline.sample(watched); + timeline.draw(); + sleep(TICK_MS); + } + }; + + if ("virtual".equals(System.getProperty("monitor"))) { + Thread.ofVirtual().name("monitor").start(monitor).join(); + } else { + monitor.run(); + } + + System.out.println(); + System.out.printf(" tasks that ever held a carrier : %d of %d%n", + timeline.everMounted().stream().filter(label -> label != CANARY).count(), TASKS); + System.out.printf(" times a carrier changed hands : %d (digit changes above)%n", + timeline.handOffs()); + System.out.printf(" blocking calls that RELEASED a carrier : %d (counted by the tasks)%n", + releases.get()); + System.out.printf(" unrelated 'canary' virtual thread : %s%n", + canaryRanAt.get() < 0 + ? "NEVER RAN in " + (TICKS * TICK_MS) + " ms <-- the whole JVM is starved" + : String.format("ran %.2f ms after it was submitted", + canaryRanAt.get() / 1_000_000d)); + System.out.println(); + + running.set(false); // tell the tasks to retire... + closeAll(cleanup); // ...and unblock the ones parked in a read that ignores the flag + } + + /** + * The audience has to be told what the drawing means before it starts moving, so the legend is + * printed once per phase, right above the rows it explains. + */ + private static void printLegend(final Workload workload) { + System.out.println(" HOW TO READ THE ROWS BELOW"); + legend("one row", "one CARRIER: a real OS thread, the only thing that can run code"); + legend("one cell", "one " + TICK_MS + " ms tick - oldest on the left, newest on the right"); + legend("1 .. " + TASKS, "which virtual thread that carrier was running during that tick"); + legend(String.valueOf(FREE), "carrier IDLE: nothing was mounted, so it was free for anyone else"); + legend(String.valueOf(CANARY), "an unrelated virtual thread, submitted at tick " + CANARY_TICK + + ", that only needs a carrier for a microsecond"); + legend("same digit", "one virtual thread is holding that carrier and never gives it back"); + legend("digits change", "the tasks release the carrier, so they take turns on it"); + legend("expect", workload.expectation()); + } + + /** Fixed-width key column, wrapped text: a legend nobody can read is not a legend. */ + private static void legend(final String key, final String text) { + final String pad = " %-14s %s%n"; + final String indent = " ".repeat(17); + final StringBuilder line = new StringBuilder(); + boolean first = true; + for (final String word : text.split(" ")) { + if (line.length() + word.length() + 1 > 74) { + System.out.printf(first ? pad : "%s%s%n", first ? key : indent, line); + line.setLength(0); + first = false; + } + line.append(line.isEmpty() ? "" : " ").append(word); + } + System.out.printf(first ? pad : "%s%s%n", first ? key : indent, line); + } + + /** + * Submits a throw-away virtual thread and sees whether it ever gets to run. A phase that follows + * a carrier-eating workload cannot measure anything, and an empty drawing is a worse answer than + * saying so out loud. + */ + private static boolean carrierAvailable() throws InterruptedException { + final AtomicBoolean ran = new AtomicBoolean(); + Thread.ofVirtual().name("probe").start(() -> ran.set(true)).join(1_000); + return ran.get(); + } + + private static void closeAll(final List cleanup) { + for (final AutoCloseable closeable : cleanup) { + try { + closeable.close(); + } catch (final Exception ignore) { + // demo teardown, best effort + } + } + } + + private static Workload workload(final String mode, final AtomicLong releases, + final AtomicBoolean running, final List cleanup) throws Exception { + switch (mode) { + case "cpu": + return new Workload("cpu", "while(true) spin++ (no block, no park, no yield)", + "every row stuck on one digit: two tasks own both carriers forever, " + + "the other four never get to run", + () -> { + long spin = 0; + while (true) { // deliberately ignores 'running': that IS the lesson, + spin++; // nothing can take a carrier back by force + } + }); + case "sleep": + return new Workload("sleep", "15 ms of work, then Thread.sleep(35), in a loop", + "digits keep changing, with dots in between: every sleep releases " + + "the carrier, so all six tasks take turns on the two carriers", + () -> { + while (running.get()) { + burn(15); + sleep(35); + releases.incrementAndGet(); + } + }); + case "socket": { + final ServerSocket server = new ServerSocket(0, 50, InetAddress.getLoopbackAddress()); + cleanup.add(server); + // Accept every connection and never write a byte, so all readers block in read(). + Thread.ofPlatform().daemon().name("silent-server").start(() -> { + while (!server.isClosed()) { + try { + cleanup.add(server.accept()); + } catch (final Exception closed) { + return; + } + } + }); + return new Workload("socket", "read() on a socket nobody ever writes to", + "a digit for a moment, then dots everywhere: a blocked socket read " + + "releases the carrier, so both carriers end up idle", + () -> { + if (!running.get()) { + return; // the phase is over; do not take a carrier for nothing + } + try (Socket socket = new Socket(InetAddress.getLoopbackAddress(), + server.getLocalPort())) { + releases.incrementAndGet(); + socket.getInputStream().read(); // blocks forever + } catch (final Exception e) { + throw new IllegalStateException(e); + } + }); + } + case "file": { + final Path fifo = fifo(); + cleanup.add(() -> unblockFifoReaders(fifo)); + return new Workload("file", "read() on a FIFO nobody ever writes to (" + fifo + ")", + "this is the measurement: if the rows stay stuck on a digit, blocking " + + "FILE i/o kept the carrier instead of releasing it", + () -> { + if (!running.get()) { + return; // opening the FIFO now would park forever on a carrier + } + try (InputStream in = Files.newInputStream(fifo)) { + in.read(); // blocks forever + } catch (final Exception e) { + throw new IllegalStateException(e); + } + }); + } + default: + throw new IllegalArgumentException( + "unknown mode '" + mode + "' - use compare|cpu|sleep|socket|file"); + } + } + + /** + * The point of {@code file} is that a blocking file read keeps its carrier, so when the phase ends + * both carriers are still held by tasks parked inside {@code read()}. Opening the FIFO for writing + * hands each of them a byte; they return, finish, and the carriers come back. Without this the + * phase would poison every phase after it - which is exactly what {@code cpu} does, and why + * {@code cpu} runs last. + */ + private static void unblockFifoReaders(final Path fifo) throws Exception { + final Thread writer = Thread.ofPlatform().daemon().name("fifo-writer").start(() -> { + try (OutputStream out = Files.newOutputStream(fifo)) { // returns once a reader is parked + for (int i = 0; i < TASKS * 4; i++) { + out.write('x'); + out.flush(); + sleep(25); // let the readers that were still queued get their turn too + } + } catch (final Exception ignore) { + // nobody left to unblock; the JVM is about to end the phase anyway + } + }); + writer.join(2_000); // never let teardown wedge the demo + } + + private static Path fifo() throws Exception { + final Path fifo = Path.of(System.getProperty("java.io.tmpdir"), "vt-carrier-demo.fifo"); + Files.deleteIfExists(fifo); + final int exit = new ProcessBuilder("mkfifo", fifo.toString()) + .redirectErrorStream(true).start().waitFor(); + if (exit != 0 || !Files.exists(fifo)) { + throw new IllegalStateException("mkfifo failed (exit " + exit + ") - mode=file needs a FIFO"); + } + return fifo; + } + + /** A virtual thread telling the timeline which carrier it woke up on. */ + private record Mount(String carrier, Character label) { + + } + + /** One row per carrier, one column per tick, plus the bookkeeping the summary needs. */ + private static final class Timeline { + + /** {@code worker-10} has to sort after {@code worker-2}, so compare the number, not the text. */ + private static final Comparator BY_CARRIER_NUMBER = + Comparator.comparingInt(Timeline::carrierNumber).thenComparing(Comparator.naturalOrder()); + + private final Map labels; + /** + * Sorted, not insertion-ordered: which carrier is discovered first is a race, and a row order + * that changes between runs is one more thing the audience has to explain away. + */ + private final Map lanes = new TreeMap<>(BY_CARRIER_NUMBER); + /** Mounts reported by the tasks themselves; drained by the monitor on the next tick. */ + private final ConcurrentLinkedQueue selfReported = new ConcurrentLinkedQueue<>(); + private final Map current = new LinkedHashMap<>(); + private final Set everMounted = new LinkedHashSet<>(); + private int handOffs; + private int linesDrawn; + + private Timeline(final Map labels) { + this.labels = labels; + } + + /** + * Called by each task on its own carrier, before it blocks. A {@code socket} task is mounted + * for a few microseconds - connect, then {@code read()} - so a 100 ms sampler only catches it + * by luck, and when it misses every task the drawing has no rows at all. The task reporting + * its own carrier is the same observation, just from a finer-grained observer; without it the + * picture depends on how loaded the machine is. + */ + private void selfReport(final char label) { + final String carrier = carrierOf(Thread.currentThread()); + if (carrier != null) { + selfReported.add(new Mount(carrier, label)); // the monitor owns all the other state + } + } + + private void sample(final List watched) { + final Map mounted = new LinkedHashMap<>(); + for (final Thread virtual : watched) { + final String carrier = carrierOf(virtual); + if (carrier != null) { + final Character label = labels.get(virtual.threadId()); + mounted.put(carrier, label); + everMounted.add(label); + lanes.computeIfAbsent(carrier, unused -> backFilled()); + } + } + // Mounts too short for this tick to see. Every report counts towards "ever held a + // carrier", but a single cell can only show one label: a live reading wins over a report, + // and among several reports on the same carrier the first one wins. + for (Mount report = selfReported.poll(); report != null; report = selfReported.poll()) { + everMounted.add(report.label()); + lanes.computeIfAbsent(report.carrier(), unused -> backFilled()); + mounted.putIfAbsent(report.carrier(), report.label()); + } + for (final Map.Entry lane : lanes.entrySet()) { + final Character now = mounted.get(lane.getKey()); + final Character before = current.get(lane.getKey()); + if (now != null && !now.equals(before)) { + handOffs++; + } + current.put(lane.getKey(), now); + lane.getValue().append(now == null ? FREE : now); + } + } + + /** + * A carrier is only visible while the virtual thread is mounted: + * {@code VirtualThread[#26,task-1]/runnable@ForkJoinPool-1-worker-1}. + */ + private static String carrierOf(final Thread virtual) { + final String description = virtual.toString(); + final int at = description.lastIndexOf('@'); + return at < 0 ? null : description.substring(at + 1); + } + + /** A carrier discovered late still needs a full-width row, so back-fill it as free. */ + private StringBuilder backFilled() { + final int width = lanes.isEmpty() ? 0 : lanes.values().iterator().next().length(); + return new StringBuilder(String.valueOf(FREE).repeat(width)); + } + + private void draw() { + if (linesDrawn > 0) { + System.out.print(CSI + linesDrawn + "A"); // redraw the block in place + } + // The block gets taller as carriers are discovered, so erase downwards first: without + // this the taller frame leaves the previous frame's last row stranded on screen. + System.out.print(CSI + "0J"); + if (lanes.isEmpty()) { + System.out.println(" (waiting: no task has been given a carrier yet)"); + linesDrawn = 1; + return; + } + int lines = 0; + int width = 0; + for (final Map.Entry lane : lanes.entrySet()) { + System.out.printf(" %-10s | %s%n", carrierName(lane.getKey()), colorize(lane.getValue())); + width = Math.max(width, lane.getValue().length()); + lines++; + } + System.out.printf(" %-10s | %s%n", "time", axis(width)); + lines++; + linesDrawn = lines; + } + + /** A second marker under the cells, so the row reads as elapsed time, not as an abstract band. */ + private static String axis(final int width) { + final int ticksPerSecond = Math.max(1, 1000 / TICK_MS); + final StringBuilder out = new StringBuilder(width + 4); + for (int tick = 0; tick < width; tick++) { + if (tick % ticksPerSecond == 0) { + out.append(tick / ticksPerSecond).append('s'); + } else if (out.length() <= tick) { + out.append(' '); + } + } + return out.toString(); + } + + /** + * The JDK calls them {@code ForkJoinPool-1-worker-N}, but "worker" reads as "one of my + * tasks"; what the row actually is, is a carrier. + */ + private static int carrierNumber(final String carrier) { + final int dash = carrier.lastIndexOf('-'); + try { + return dash < 0 ? Integer.MAX_VALUE : Integer.parseInt(carrier.substring(dash + 1)); + } catch (final NumberFormatException notNumbered) { + return Integer.MAX_VALUE; + } + } + + private static String carrierName(final String carrier) { + final int dash = carrier.lastIndexOf("-worker-"); + return dash < 0 ? carrier : "carrier-" + carrier.substring(dash + "-worker-".length()); + } + + private static String colorize(final CharSequence row) { + if (!COLOR) { + return row.toString(); + } + final StringBuilder out = new StringBuilder(row.length() * 12); + for (int i = 0; i < row.length(); i++) { + final char cell = row.charAt(i); + if (cell == FREE) { + out.append(CSI).append("90m").append(cell).append(CSI).append("0m"); + } else if (cell == CANARY) { + out.append(CSI).append("1;97m").append(cell).append(CSI).append("0m"); + } else { + out.append(CSI).append("1;3").append((cell - '1') % 6 + 1).append('m') + .append(cell).append(CSI).append("0m"); + } + } + return out.toString(); + } + + private Set everMounted() { + return everMounted; + } + + private int handOffs() { + return handOffs; + } + } + + /** Keeps the carrier busy without blocking, so the sampler can see who is mounted. */ + private static void burn(final long millis) { + final long until = System.nanoTime() + millis * 1_000_000L; + long spin = 0; + while (System.nanoTime() < until) { + spin++; + } + } + + private static void sleep(final long millis) { + try { + Thread.sleep(millis); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + } +} diff --git a/dotCMS/src/test/java/com/dotcms/jdk/VirtualThreadYieldVsParkDemo.java b/dotCMS/src/test/java/com/dotcms/jdk/VirtualThreadYieldVsParkDemo.java new file mode 100644 index 000000000000..ff7f72644046 --- /dev/null +++ b/dotCMS/src/test/java/com/dotcms/jdk/VirtualThreadYieldVsParkDemo.java @@ -0,0 +1,84 @@ +package com.dotcms.jdk; + +/** + * Live demo for the Java 25 talk: {@code Thread.yield()} gives up the turn, not the carrier. Only + * a park actually frees it. + * + *

Two virtual threads loop forever on two carriers while a third, unrelated virtual thread waits + * to be scheduled. What the loop body does decides whether the third one ever runs: + * + * 

+ *   mode=busy    while (true) { spin++; }        innocent NEVER runs
+ *   mode=yield   Thread.yield()                  innocent NEVER runs   <-- the surprise
+ *   mode=sleep   Thread.sleep(1)                 innocent runs immediately
+ *
+ * + *

Why {@code yield} is not enough: a carrier is a {@code ForkJoinPool} worker and it drains + * its own local queue first, LIFO. A virtual thread that yields is re-submitted to that same + * local queue, so the worker immediately picks it back up. The third thread was submitted externally + * and sits in the shared submission queue, which a worker only visits once its local queue is empty. + * {@code Thread.sleep} parks on a timer, the local queue genuinely empties, and only then does the + * worker go looking and find the waiting task. + * + *

The talk conclusion: there is no escape hatch for CPU-bound work on a virtual thread. If + * a task does not block on something that parks, it does not belong on a virtual thread. File I/O + * fails the same test for the same reason — it never parks either (see #37038 / PR #37041). + * + *

Run it (no flags needed; it pins the carrier count itself): + * + * 

+ *   java -Dmode=busy  dotCMS/src/test/java/com/dotcms/jdk/VirtualThreadYieldVsParkDemo.java
+ *   java -Dmode=yield dotCMS/src/test/java/com/dotcms/jdk/VirtualThreadYieldVsParkDemo.java
+ *   java -Dmode=sleep dotCMS/src/test/java/com/dotcms/jdk/VirtualThreadYieldVsParkDemo.java
+ *
+ * + *

Not a JUnit test on purpose, and {@code System.out} is deliberate — see the note on + * {@link VirtualThreadCarrierStarvationDemo}. + * + * @see VirtualThreadCarrierStarvationDemo + * @see VirtualThreadCarrierTimelineDemo + */ +public class VirtualThreadYieldVsParkDemo { + + private static final int CARRIERS = 2; + + public static void main(final String[] args) throws InterruptedException { + // Read lazily on first virtual thread creation, so setting it here is enough. + if (System.getProperty("jdk.virtualThreadScheduler.parallelism") == null) { + System.setProperty("jdk.virtualThreadScheduler.parallelism", String.valueOf(CARRIERS)); + } + + final String mode = System.getProperty("mode", "busy"); // busy | yield | sleep + System.out.println("mode = " + mode + + " | carriers = " + System.getProperty("jdk.virtualThreadScheduler.parallelism")); + + for (int i = 0; i < CARRIERS; i++) { + Thread.ofVirtual().name("hog-" + i).start(() -> { + long spin = 0; + while (true) { + spin++; + try { + switch (mode) { + case "yield" -> Thread.yield(); // re-queued on the SAME worker, LIFO + case "sleep" -> Thread.sleep(1); // parks: the local queue empties + default -> { /* busy: never asks to get off the carrier */ } + } + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + } + }); + } + + Thread.sleep(200); // let the hogs take both carriers + + // A completely unrelated part of the system. + Thread.ofVirtual().name("innocent").start( + () -> System.out.println(">>> the INNOCENT virtual thread RAN")); + + Thread.sleep(3000); + System.out.println("--- done: no '>>>' above means the innocent never ran ---"); + System.exit(0); + } +} From 915a7c3df53c95600f83d986f647ea5a9f881373 Mon Sep 17 00:00:00 2001 From: fabrizzio-dotCMS Date: Wed, 19 Aug 2026 22:13:56 -0600 Subject: [PATCH 2/4] #34154: test(java25): demo for when a custom Collector is worth writing Adds CustomCollectorDemo to the talk demos. Four sections, all runnable with `java .java`: - the anatomy (supplier/accumulator/combiner/finisher) on toImmutableList, the collector everyone writes first and the JDK already ships twice; - IDENTITY_FINISH declared alongside a finisher does not fail, it makes the stream skip the finisher, so a method promising an immutable list returns a plain ArrayList in silence; - List.copyOf copies and rejects null while Collections.unmodifiableList wraps and accepts it, and Stream.toList differs from Collectors.toUnmodifiableList on exactly that point; - the one worth writing: merging into a map while reporting which keys collided, which Collectors.toMap cannot express - it can only throw or discard the loser silently. All four pieces do real work there, including the combiner, which surfaces collisions neither parallel half saw alone. The javadoc also records the counterexample: the three-way IndexPolicy split in ContentletIndexAPIImpl wants Collectors.groupingBy, not a custom collector. Co-Authored-By: Claude Opus 5 (1M context) --- .../com/dotcms/jdk/CustomCollectorDemo.java | 297 ++++++++++++++++++ 1 file changed, 297 insertions(+) create mode 100644 dotCMS/src/test/java/com/dotcms/jdk/CustomCollectorDemo.java diff --git a/dotCMS/src/test/java/com/dotcms/jdk/CustomCollectorDemo.java b/dotCMS/src/test/java/com/dotcms/jdk/CustomCollectorDemo.java new file mode 100644 index 000000000000..0e134f174362 --- /dev/null +++ b/dotCMS/src/test/java/com/dotcms/jdk/CustomCollectorDemo.java @@ -0,0 +1,297 @@ +package com.dotcms.jdk; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.Function; +import java.util.stream.Collector; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +/** + * Live demo for the Java 25 talk: when a custom {@link Collector} is worth writing, and when the + * JDK already wrote it for you. + * + *

A collector is four pieces and three type parameters {@code }: what goes in, the + * mutable accumulator it builds up internally, and what comes out. That A and R may + * differ is the whole trick — accumulate into something mutable because that is what is efficient, + * and let the finisher convert it once, at the end. + * + * 

+ *   supplier      the bucket                ArrayList::new
+ *   accumulator   put one element in it     List::add
+ *   combiner      merge two buckets         (a, b) -> { a.addAll(b); return a; }
+ *   finisher      turn A into R             List::copyOf
+ *
+ * + *

Part 1 — the one NOT worth writing

+ * + * {@link #toImmutableList()} is the collector everybody writes first. It works, and it is redundant: + * {@code Stream.toList()} and {@code Collectors.toUnmodifiableList()} have shipped since Java 16 and + * 10. It earns its place here only as the anatomy lesson, plus two traps that are worth seeing once: + * + *
    + *
  • {@code IDENTITY_FINISH} silently skips the finisher. Declaring it does not fail — the + * stream simply never calls the finisher, so a method promising an immutable list hands back a + * plain mutable {@code ArrayList}. The characteristic means "the accumulator already is + * the result"; declaring it while having a finisher is lying to the stream. + *
  • Copy and view are not interchangeable. {@code List.copyOf} copies and rejects + * {@code null}; {@code Collections.unmodifiableList} wraps and accepts it. In a codebase where + * an unset content field arrives as {@code null}, that difference shows up in production, not + * in tests. + *
+ * + *

Part 2 — the one that IS worth writing

+ * + * {@link #mergingReportingConflicts} merges a stream into a map and reports which keys collided. + * {@code Collectors.toMap} offers only two bad answers to a duplicate key: throw + * {@code IllegalStateException} (losing every other collision), or take a merge function and discard + * the loser in silence. Neither can tell the caller that something clashed at all. + * + *

This is the criterion for writing your own: all four pieces have to do real work. + * + *

    + *
  • the supplier builds an accumulator holding two coordinated structures at once — no JDK + * collector does that, and {@code Collectors.teeing} cannot, because it feeds two independent + * collectors while here the detection is the merge; + *
  • the combiner is not decoration: joining two halves in parallel can surface a collision + * neither half saw alone; + *
  • the finisher returns an immutable, typed {@code record}, so the caller cannot forget to + * look at the conflicts — they are in the return type. + *
+ * + *

The dotCMS-shaped use is merging field maps from several contentlets, or consolidating settings + * from several sources, where {@code toMap} forces a choice between blowing up and silently + * overwriting. + * + *

The counterexample worth keeping in mind

+ * + * {@code ContentletIndexAPIImpl.addContentToIndex} splits contentlets three ways by + * {@code IndexPolicy} with {@code CollectionsUtils.partition} plus positional + * {@code get(0)/get(1)/get(2)} — two parallel lists the compiler never cross-checks. That one does + * not want a custom collector: {@code Collectors.groupingBy(Contentlet::getIndexPolicy)} has + * been the right answer since Java 8. Writing a collector there would be using the new toy for its + * own sake. + * + *

Run it (no build required): + * + * 

+ *   java dotCMS/src/test/java/com/dotcms/jdk/CustomCollectorDemo.java
+ *
+ * + *

Not a JUnit test on purpose: the console output is the artifact being shown to an + * audience. It is a {@code main} demo, compiled by the real build so it cannot silently rot, and named + * {@code *Demo} so surefire skips it. {@code System.out} is deliberate for the same reason; the + * Logger-only rule targets production code. + * + * @author Fabrizio Araya + * @see CompactObjectHeadersDemo + */ +public final class CustomCollectorDemo { + + private CustomCollectorDemo() { + } + + public static void main(final String[] args) { + anatomy(); + identityFinishTrap(); + copyVersusView(); + worthWriting(); + } + + // ───────────────────────────────────────────────────────────────────────── + // Part 1 — anatomy, on a collector you do not actually need + // ───────────────────────────────────────────────────────────────────────── + + /** The collector everyone writes first. Correct, and already in the JDK twice over. */ + static Collector, List> toImmutableList() { + return Collector.of( + ArrayList::new, // supplier + List::add, // accumulator + (left, right) -> { left.addAll(right); return left; }, // combiner + List::copyOf); // finisher + } + + /** Same, but freezing with a read-only view instead of a copy. Not the same thing. */ + static Collector, List> toUnmodifiableView() { + return Collector.of( + ArrayList::new, + List::add, + (left, right) -> { left.addAll(right); return left; }, + Collections::unmodifiableList); + } + + private static void anatomy() { + header("1. Anatomy — supplier, accumulator, combiner, finisher"); + + final List names = List.of("ana", "beto", "caro", "dani"); + final List result = names.stream().map(String::toUpperCase) + .collect(toImmutableList()); + + System.out.println(" result " + result); + System.out.println(" class " + result.getClass().getSimpleName() + + " <- no longer an ArrayList: the finisher ran"); + try { + result.add("EVA"); + System.out.println(" add() NO ERROR — the list is not immutable!"); + } catch (UnsupportedOperationException expected) { + System.out.println(" add() UnsupportedOperationException"); + } + + // The combiner is only ever called on a parallel stream. A broken one passes every + // sequential test and fails the day somebody writes parallelStream(). + final List parallel = names.parallelStream().map(String::toUpperCase) + .collect(toImmutableList()); + System.out.println(" in parallel " + parallel + " (same: " + + result.equals(parallel) + ") <- the combiner only runs here"); + + System.out.println(" Stream.toList() " + names.stream().toList() + + " <- which is why you rarely need to write this one"); + } + + private static void identityFinishTrap() { + header("2. Trap — IDENTITY_FINISH silently skips the finisher"); + + final Collector, List> lying = Collector.of( + ArrayList::new, + List::add, + (left, right) -> { left.addAll(right); return left; }, + List::copyOf, + Collector.Characteristics.IDENTITY_FINISH); // <- the lie + + final List shouldBeImmutable = Stream.of("a").collect(lying); + System.out.println(" declared immutable (the finisher says List::copyOf)"); + System.out.println(" actual class " + shouldBeImmutable.getClass().getSimpleName()); + shouldBeImmutable.add("b"); + System.out.println(" mutated it " + shouldBeImmutable + + " <- no exception, no warning, no failure anywhere"); + } + + private static void copyVersusView() { + header("3. Copy vs view — and what each does with null"); + + final List withNull = Arrays.asList("a", null); + System.out.println(" unmodifiableList " + withNull.stream().collect(toUnmodifiableView()) + + " <- wraps; nulls survive"); + try { + withNull.stream().collect(toImmutableList()); + System.out.println(" List.copyOf accepted null"); + } catch (NullPointerException expected) { + System.out.println(" List.copyOf NullPointerException <- copies; rejects null"); + } + System.out.println(" Stream.toList() " + withNull.stream().toList() + + " <- immutable AND null-tolerant"); + try { + withNull.stream().collect(Collectors.toUnmodifiableList()); + } catch (NullPointerException expected) { + System.out.println(" toUnmodifiableList() NullPointerException <- immutable, null-hostile"); + } + } + + // ───────────────────────────────────────────────────────────────────────── + // Part 2 — the collector worth writing + // ───────────────────────────────────────────────────────────────────────── + + /** + * What the collector returns: the merged map and the keys that collided. Both immutable, so + * the caller cannot mutate the result and cannot overlook the conflicts. + */ + record MergeResult(Map merged, Set conflicts) { + + boolean clean() { + return conflicts.isEmpty(); + } + } + + /** The mutable accumulator. It never escapes the collector — the finisher freezes it. */ + private static final class Acc { + + private final Map map = new LinkedHashMap<>(); + private final Set conflicts = new LinkedHashSet<>(); + + void put(final K key, final V value) { + // containsKey BEFORE the put: afterwards it is always true. And containsKey rather than + // `map.put(...) != null`, so a key whose previous value was null still counts as a clash. + if (map.containsKey(key)) { + conflicts.add(key); + } + map.put(key, value); + } + + Acc merge(final Acc other) { + other.map.forEach(this::put); // may discover a clash neither half saw alone + this.conflicts.addAll(other.conflicts); + return this; + } + } + + /** + * Merges into a map and reports which keys collided — the answer {@code Collectors.toMap} cannot + * give, since its only options are to throw or to discard the loser in silence. + */ + static Collector> mergingReportingConflicts( + final Function keyFn, final Function valueFn) { + return Collector.of( + Acc::new, + (acc, element) -> acc.put(keyFn.apply(element), valueFn.apply(element)), + Acc::merge, + acc -> new MergeResult<>( + Collections.unmodifiableMap(new LinkedHashMap<>(acc.map)), + Set.copyOf(acc.conflicts))); + } + + /** A field contributed by some contentlet — the shape this collector exists for. */ + record Field(String name, String value) { + } + + private static void worthWriting() { + header("4. Worth writing — merge a map AND report the collisions"); + + final List fields = List.of( + new Field("title", "Home"), + new Field("body", "..."), + new Field("title", "Home v2"), // <- the collision + new Field("author", "ana")); + + final MergeResult merged = + fields.stream().collect(mergingReportingConflicts(Field::name, Field::value)); + + System.out.println(" merged " + merged.merged()); + System.out.println(" conflicts " + merged.conflicts()); + System.out.println(" clean() " + merged.clean()); + try { + merged.merged().put("x", "y"); + System.out.println(" put() NO ERROR — not immutable!"); + } catch (UnsupportedOperationException expected) { + System.out.println(" put() UnsupportedOperationException"); + } + + final MergeResult parallel = + fields.parallelStream().collect(mergingReportingConflicts(Field::name, Field::value)); + System.out.println(" in parallel " + parallel.merged() + + " conflicts=" + parallel.conflicts() + " <- the combiner detects them too"); + + System.out.println(); + System.out.println(" What Collectors.toMap offers instead:"); + try { + fields.stream().collect(Collectors.toMap(Field::name, Field::value)); + } catch (IllegalStateException expected) { + System.out.println(" toMap(k, v) IllegalStateException" + + " <- and it cannot say how many others clashed"); + } + System.out.println(" toMap(k, v, merge) " + + fields.stream().collect(Collectors.toMap(Field::name, Field::value, (a, b) -> b)) + + " <- did anything clash? no way to know"); + } + + private static void header(final String title) { + System.out.println(); + System.out.println(title); + System.out.println("-".repeat(78)); + } +} From 3410ea41ca31508c714031a177f020dcdafff975 Mon Sep 17 00:00:00 2001 From: fabrizzio-dotCMS Date: Wed, 19 Aug 2026 22:36:07 -0600 Subject: [PATCH 3/4] #34154: test(java25): demo for stream gatherers Adds StreamGatherersDemo to the talk demos. Four sections, runnable with `java .java`: - batching, where windowFixed emits the short final window on its own - the flush that hand-written accumulators forget; - scan and fold, the built-ins that carry state between elements; - writing your own, including that returning false from the integrator short-circuits the stream, which no Collector can do; - mapConcurrent, which runs each element on a virtual thread under an explicit bound and preserves encounter order. Measured 1246 ms sequential vs 347 ms at concurrency 4. The javadoc records the two call sites this lands on: OSIndexAPIImpl collects a list purely so Lists.partition can cut it up, and PopulateContentletAsJSONUtil splits the batch flush across the accumulator and its caller, duplicated for inserts and updates. Co-Authored-By: Claude Opus 5 (1M context) --- .../com/dotcms/jdk/StreamGatherersDemo.java | 253 ++++++++++++++++++ 1 file changed, 253 insertions(+) create mode 100644 dotCMS/src/test/java/com/dotcms/jdk/StreamGatherersDemo.java diff --git a/dotCMS/src/test/java/com/dotcms/jdk/StreamGatherersDemo.java b/dotCMS/src/test/java/com/dotcms/jdk/StreamGatherersDemo.java new file mode 100644 index 000000000000..d62cf5b07e70 --- /dev/null +++ b/dotCMS/src/test/java/com/dotcms/jdk/StreamGatherersDemo.java @@ -0,0 +1,253 @@ +package com.dotcms.jdk; + +import java.util.List; +import java.util.Objects; +import java.util.stream.Gatherer; +import java.util.stream.Gatherers; +import java.util.stream.IntStream; + +/** + * Live demo for the Java 25 talk: stream gatherers — the extension point the middle of a stream + * never had. + * + *

A {@code Stream}'s intermediate vocabulary closed in Java 8 and never grew: {@code map}, + * {@code filter}, {@code limit}, {@code skip}, {@code sorted}, {@code distinct}, {@code peek}, + * {@code flatMap}. Since Java 8 you have been able to write your own terminal operation — that + * is what a {@code Collector} is. There was never a way to write your own intermediate one. + * + *

The reason it matters is that the existing operations are amnesic. {@code map} sees one element + * and produces one; {@code filter} sees one and decides keep-or-drop; even {@code flatMap}, which can + * change the element count, decides looking at a single element. None can remember anything about what + * it already saw — which rules out a whole family of ordinary operations: + * + * 

+ *   batches of 50                     must accumulate 50 before emitting anything
+ *   sliding window                    must remember the previous elements
+ *   running total                     must remember the sum
+ *   collapse consecutive duplicates   must remember the previous element
+ *   stop once a condition holds       must remember that it held
+ *
+ * + * All of them need state between elements, and there was nowhere to put it. So the way out was + * always the same: {@code collect(toList())}, leave the stream, and finish by hand with a {@code for} + * and a variable outside it. A gatherer is where that state goes — {@code .gather(...)} takes a + * stream and returns a stream, so the pipeline survives. + * + *

The two costs of doing it by hand, both of which are in this codebase

+ * + *
    + *
  1. Materialising a list you only wanted to walk in pieces. + * {@code OSIndexAPIImpl.getIndexAlias} (around lines 846-855) does + * {@code stream().map(...).collect(toList())} and immediately + * {@code Lists.partition(physicalNames, ALIAS_LOOKUP_BATCH_SIZE)}. The whole list exists solely + * so that it can be cut up. That is {@code .gather(windowFixed(N))} written the long way. + * {@code Lists.partition} appears in a dozen more places. + *
  2. Forgetting the tail. {@code PopulateContentletAsJSONUtil.processInsertRecord} + * (around 466-471) flushes when the batch reaches {@code MAX_BATCH_SIZE}, and its caller + * (around 332-334) has to remember {@code if (!paramsInsert.isEmpty()) doInsertBatch(...)} for + * the partly-filled remainder. Two places that must agree, duplicated again for updates. + * {@code windowFixed} emits the short final window on its own — watch for the lone + * {@code [id7]} in the output below. + *
+ * + *

What the JDK ships

+ * + * {@code windowFixed}, {@code windowSliding}, {@code fold}, {@code scan} and {@code mapConcurrent}. + * Gatherers are final since Java 24 (JEP 485), so none of this needs {@code --enable-preview}. + * + *

{@code mapConcurrent} is the one that connects this topic to Loom: it runs each element on a + * virtual thread, under a concurrency limit you choose, and preserves encounter order. + * It is the JDK's answer to "parallelise the I/O in this stream" — the thing {@code parallelStream()} + * never did well, because it uses the common ForkJoinPool, sized for CPU work and shared with the + * whole process. + * + *

Run it (no build required; takes a couple of seconds for the concurrency section): + * + * 

+ *   java dotCMS/src/test/java/com/dotcms/jdk/StreamGatherersDemo.java
+ *
+ * + *

Not a JUnit test on purpose: the console output is the artifact being shown to an + * audience. It is a {@code main} demo, compiled by the real build so it cannot silently rot, and named + * {@code *Demo} so surefire skips it. {@code System.out} is deliberate for the same reason; the + * Logger-only rule targets production code. + * + * @author Fabrizio Araya + * @see JEP 485 — Stream Gatherers + * @see CustomCollectorDemo + */ +public final class StreamGatherersDemo { + + /** Stands in for the index names {@code OSIndexAPIImpl} batches before asking OpenSearch. */ + private static final List INDEX_NAMES = + List.of("idx1", "idx2", "idx3", "idx4", "idx5", "idx6", "idx7"); + + private StreamGatherersDemo() { + } + + public static void main(final String[] args) { + batching(); + statefulBuiltIns(); + writingYourOwn(); + concurrentMapping(); + } + + // ───────────────────────────────────────────────────────────────────────── + // 1. The batching case, which is the one already written by hand here + // ───────────────────────────────────────────────────────────────────────── + + private static void batching() { + header("1. Batching — and the partial final window nobody remembers to flush"); + + System.out.println(" source " + INDEX_NAMES + " (7 items, batch size 3)"); + + final String label = " windowFixed(3) "; + final String windows = INDEX_NAMES.stream().gather(Gatherers.windowFixed(3)).toList().toString(); + System.out.println(label + windows); + // Point at the short final window wherever it lands, rather than at a hand-counted column. + System.out.println(" ".repeat(label.length() + windows.lastIndexOf('[')) + + "^^^^^^^ the short tail, emitted for free"); + System.out.println(" windowSliding(3) " + + INDEX_NAMES.stream().gather(Gatherers.windowSliding(3)).toList()); + System.out.println(); + System.out.println(" The list is never materialised: windowFixed is a stream step, so the"); + System.out.println(" batches are produced lazily as the source is consumed. Compare with"); + System.out.println(" collect(toList()) followed by Lists.partition(...), which must build the"); + System.out.println(" whole list first purely in order to cut it up."); + } + + // ───────────────────────────────────────────────────────────────────────── + // 2. The other built-ins that carry state + // ───────────────────────────────────────────────────────────────────────── + + private static void statefulBuiltIns() { + header("2. Carrying state — scan and fold"); + + final List amounts = List.of(10, 20, 30, 40); + System.out.println(" source " + amounts); + System.out.println(" scan (running total) " + + amounts.stream().gather(Gatherers.scan(() -> 0, Integer::sum)).toList() + + " <- one output per input"); + System.out.println(" fold (single value) " + + amounts.stream().gather(Gatherers.fold(() -> 0, Integer::sum)).toList() + + " <- one output in total"); + } + + // ───────────────────────────────────────────────────────────────────────── + // 3. Writing your own: state, plus a boolean that can end the stream + // ───────────────────────────────────────────────────────────────────────── + + /** Remembers the previous element, so runs of equal values collapse into one. */ + private static final class Previous { + + private T value; + } + + /** + * Collapses consecutive duplicates. Note this is not {@code distinct()}: a value may + * reappear later, it just may not repeat back to back. + */ + static Gatherer collapsingRuns() { + return Gatherer.ofSequential( + Previous::new, + (state, element, downstream) -> { + if (Objects.equals(state.value, element)) { + return true; // swallow it, keep going + } + state.value = element; + return downstream.push(element); + }); + } + + /** Remembers that the cut-off condition already fired. */ + private static final class Latch { + + private boolean tripped; + } + + /** + * Emits elements until the first {@code ERROR} and then ends the stream. Returning {@code false} + * from the integrator short-circuits — the rest of the source is never consumed, which no + * {@code Collector} can do. + */ + static Gatherer untilError() { + return Gatherer.ofSequential( + Latch::new, + (state, element, downstream) -> { + if (state.tripped || "ERROR".equals(element)) { + state.tripped = true; + return false; // false ends the stream here + } + return downstream.push(element); + }); + } + + private static void writingYourOwn() { + header("3. Writing your own — initial state + what to do with each element"); + + final List statuses = List.of("OK", "OK", "OK", "ERROR", "ERROR", "OK"); + System.out.println(" source " + statuses); + System.out.println(" collapsingRuns() " + statuses.stream().gather(collapsingRuns()).toList() + + " <- runs collapse, later repeats survive"); + System.out.println(" untilError() " + statuses.stream().gather(untilError()).toList() + + " <- returning false ends the stream"); + System.out.println(); + System.out.println(" A gatherer can therefore short-circuit, which a Collector cannot: a"); + System.out.println(" collector always drains the whole stream before it can produce anything."); + } + + // ───────────────────────────────────────────────────────────────────────── + // 4. mapConcurrent — where this topic meets Loom + // ───────────────────────────────────────────────────────────────────────── + + private static void concurrentMapping() { + header("4. mapConcurrent — bounded concurrency on virtual threads, order preserved"); + + final List ids = IntStream.rangeClosed(1, 12).boxed().toList(); + + final long startSequential = System.nanoTime(); + final List sequential = ids.stream().map(StreamGatherersDemo::slowIo).toList(); + final long sequentialMs = millisSince(startSequential); + + final long startConcurrent = System.nanoTime(); + final List concurrent = + ids.stream().gather(Gatherers.mapConcurrent(4, StreamGatherersDemo::slowIo)).toList(); + final long concurrentMs = millisSince(startConcurrent); + + System.out.printf(" sequential %5d ms%n", sequentialMs); + System.out.printf(" mapConcurrent(4) %5d ms%n", concurrentMs); + System.out.println(" same order as input " + sequential.equals(concurrent)); + System.out.println(" ran on virtual threads " + ranVirtual(ids)); + System.out.println(); + System.out.println(" The bound is explicit and local. parallelStream() would instead borrow"); + System.out.println(" the common ForkJoinPool — sized for CPU work and shared process-wide —"); + System.out.println(" which is the wrong resource for blocking I/O."); + } + + /** Stands in for a blocking call: 100 ms of waiting, no CPU. */ + private static String slowIo(final int id) { + try { + Thread.sleep(100); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + } + return "r" + id; + } + + /** Reports whether mapConcurrent actually ran the mapper on virtual threads. */ + private static boolean ranVirtual(final List ids) { + return ids.stream() + .gather(Gatherers.mapConcurrent(4, id -> Thread.currentThread().isVirtual())) + .allMatch(Boolean::booleanValue); + } + + private static long millisSince(final long startNanos) { + return (System.nanoTime() - startNanos) / 1_000_000; + } + + private static void header(final String title) { + System.out.println(); + System.out.println(title); + System.out.println("-".repeat(78)); + } +} From e8c9ad237620874e45fa3ac2b840745afedfed30 Mon Sep 17 00:00:00 2001 From: fabrizzio-dotCMS Date: Thu, 20 Aug 2026 14:44:28 -0600 Subject: [PATCH 4/4] #34154: test(java25): didactic walkthrough of writing a Gatherer SimplestGathererDidactic takes the API apart one step at a time, as the teaching companion to StreamGatherersDemo (which shows what gatherers are for, not how to write one). passThrough/onlyLong/upperCase/twice are the same stateless gatherer one line apart, making the point that the only decision is how many times you call push(). numbered/whenChanged/occurrence then introduce state and show what map and filter cannot express. Runs standalone: java dotCMS/src/test/java/com/dotcms/jdk/SimplestGathererDidactic.java Co-Authored-By: Claude Opus 5 (1M context) --- .../dotcms/jdk/SimplestGathererDidactic.java | 261 ++++++++++++++++++ 1 file changed, 261 insertions(+) create mode 100644 dotCMS/src/test/java/com/dotcms/jdk/SimplestGathererDidactic.java diff --git a/dotCMS/src/test/java/com/dotcms/jdk/SimplestGathererDidactic.java b/dotCMS/src/test/java/com/dotcms/jdk/SimplestGathererDidactic.java new file mode 100644 index 000000000000..5fa8a28e34c8 --- /dev/null +++ b/dotCMS/src/test/java/com/dotcms/jdk/SimplestGathererDidactic.java @@ -0,0 +1,261 @@ +package com.dotcms.jdk; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Gatherer; + +/** + * The smallest possible gatherer, taken apart — the teaching companion to {@link StreamGatherersDemo}, + * which shows what gatherers are for rather than how to write one. + * + *

Start with {@link #passThrough()}. It does nothing at all, and that is the point: it already + * contains the entire API, so there is nothing else to learn afterwards. + * + * 

+ *   static Gatherer<String, Void, String> passThrough() {
+ *       return Gatherer.of((state, element, downstream) -> {
+ *           downstream.push(element);
+ *           return true;
+ *       });
+ *   }
+ *
+ * + *

The three type parameters

+ * + * {@code Gatherer} reads: what goes in, the state, what comes + * out. {@code Void} means "I do not need to remember anything" — the simplest case, and the + * reason {@link Gatherer#of(Gatherer.Integrator)} takes a single lambda with no state supplier. + * + *

The three things the lambda receives

+ * + * 
+ *   state        your memory between elements — null here, because Void
+ *   element      the one going past right now
+ *   downstream   the rest of the pipeline; you hand it results with push()
+ *
+ * + * And the returned {@code boolean} means "keep going". Returning {@code false} ends the stream + * on the spot, without consuming the rest of the source — something no {@code Collector} can do. + * + *

Why this one shape covers the old operations

+ * + * The only real decision is how many times you call {@code push}. Each method below differs + * from {@code passThrough} by a single line: + * + * 
+ *   source        [ana, beto, caro]
+ *   passThrough   [ana, beto, caro]                        push once, unchanged
+ *   onlyLong      [beto, caro]                             push zero or one time  -> behaves like filter
+ *   upperCase     [ANA, BETO, CARO]                        push something else    -> behaves like map
+ *   twice         [ana, ana, beto, beto, caro, caro]       push more than once    -> behaves like flatMap
+ *
+ * + * So a gatherer can do everything the pre-existing operations do. What makes it new is the + * first parameter, {@code state}. + * + *

Then: state, the parameter that was being ignored

+ * + * {@link #numbered()} and {@link #whenChanged()} stop ignoring it. Three things change, and only + * three: + * + * 
+ *   Gatherer<String, Counter, String>    the middle type is no longer Void
+ *   Gatherer.ofSequential(Counter::new,  a supplier for the initial state comes first
+ *       (counter, element, downstream)   the same lambda — but the first parameter now means something
+ *
+ * + *

The state must be an object whose field you mutate, never a plain {@code int}: the + * gatherer hands the same instance to every invocation, so the memory has to live inside something + * that survives between them. {@code ofSequential} is the honest factory here — order-dependent state + * cannot be split across threads, and it says so instead of asking for a combiner that could not be + * written correctly. + * + * 

+ *   numbered      [1. ana, 2. beto, 3. caro]     impossible with map:    output depends on position
+ *   whenChanged   [ok, error, ok]                impossible with filter: predicate sees one element
+ *
+ * + * That is the whole idea. Batching, sliding windows and running totals are this same shape with a + * richer piece of state — see {@link StreamGatherersDemo}, which also shows what the JDK already + * ships so you do not write them yourself. + * + *

Run it (no build required): + * + * 

+ *   java dotCMS/src/test/java/com/dotcms/jdk/SimplestGathererDidactic.java
+ *
+ * + *

{@code System.out} is deliberate: the console output is the artifact being shown. Not a + * JUnit test, and named so surefire skips it, matching the other demos in this package. + * + * @author Fabrizio Araya + * @see StreamGatherersDemo + */ +public final class SimplestGathererDidactic { + + private static final List SOURCE = List.of("ana", "beto", "caro"); + + /** Has a run of repeats in the middle, and one value that comes back later. */ + private static final List REPEATED = List.of("ok", "ok", "ok", "error", "error", "ok"); + + /** Words that come back, so a per-word tally has something to count. */ + private static final List WITH_REPEATS = + List.of("ana", "beto", "ana", "caro", "ana", "beto"); + + private SimplestGathererDidactic() { + } + + // ───────────────────────────────────────────────────────────────────────── + // Step 0 — the whole API, doing nothing + // ───────────────────────────────────────────────────────────────────────── + + static Gatherer passThrough() { + return Gatherer.of((state, element, downstream) -> { + downstream.push(element); // hand it to the rest of the pipeline + return true; // true = carry on with the next element + }); + } + + // ───────────────────────────────────────────────────────────────────────── + // Steps 1-3 — one line different each time + // ───────────────────────────────────────────────────────────────────────── + + /** Push zero or one time, and you have written {@code filter}. */ + static Gatherer onlyLong() { + return Gatherer.of((state, element, downstream) -> { + if (element.length() > 3) { + downstream.push(element); + } + return true; + }); + } + + /** Push something other than what came in, and you have written {@code map}. */ + static Gatherer upperCase() { + return Gatherer.of((state, element, downstream) -> { + downstream.push(element.toUpperCase()); + return true; + }); + } + + /** Push more than once, and you have written the expanding half of {@code flatMap}. */ + static Gatherer twice() { + return Gatherer.of((state, element, downstream) -> { + downstream.push(element); + downstream.push(element); + return true; + }); + } + + // ───────────────────────────────────────────────────────────────────────── + // Step 4 — using the parameter that was being ignored: state + // ───────────────────────────────────────────────────────────────────────── + + /** + * The state has to be an object whose field you mutate, never a plain {@code int}. The + * gatherer hands the same instance to every invocation, and the lambda cannot reassign its own + * parameter in a way the next invocation would see — so the memory has to live inside + * something. One tiny class is the clearest form of that. + */ + private static final class Counter { + + private int seen; + } + + /** + * Numbers the elements. Impossible with {@code map}: numbering depends on how many went past + * before, and {@code map} sees one element with no idea of its position. + */ + static Gatherer numbered() { + return Gatherer.ofSequential( + Counter::new, // 1. the initial state + (counter, element, downstream) -> { // 2. same lambda as before + counter.seen++; // ...but now it remembers + return downstream.push(counter.seen + ". " + element); + }); + } + + /** State does not have to be a number: here it is the previous element. */ + private static final class Previous { + + private String value; + } + + /** + * Emits an element only when it differs from the one just before it. Note this is not + * {@code distinct()}: a value that comes back later is emitted again, it just may not repeat back + * to back. And it is not {@code filter} either — a predicate sees one element and cannot know what + * preceded it. + */ + static Gatherer whenChanged() { + return Gatherer.ofSequential( + Previous::new, + (previous, element, downstream) -> { + if (element.equals(previous.value)) { + return true; // swallow it, but keep going + } + previous.value = element; + return downstream.push(element); + }); + } + + /** State does not have to be one value: here it is a tally, one counter per distinct word. */ + private static final class Tally { + + private final Map timesSeen = new LinkedHashMap<>(); + } + + /** + * Says how many times each word has appeared so far. Same shape as {@link #numbered()} — + * the only change is that the state went from one counter to one counter per word. + * + *

Note what it can and cannot answer. Walking the stream it knows "this is the 2nd {@code ana}", + * because that is settled by the time the element goes past. It cannot say "{@code ana} appears + * twice in total": the total is only known once the source is exhausted, and by then every element + * has already been pushed. Emitting the totals is the job of a finisher, the next step. + */ + static Gatherer occurrence() { + return Gatherer.ofSequential( + Tally::new, + (tally, element, downstream) -> { + // merge returns the NEW value, so this counts and reads in one call + final int times = tally.timesSeen.merge(element, 1, Integer::sum); + return downstream.push(element + " (" + times + ")"); + }); + } + + public static void main(final String[] args) { + System.out.println(); + System.out.println("The same gatherer, one line apart"); + System.out.println("-".repeat(78)); + print("source", SOURCE); + print("passThrough", SOURCE.stream().gather(passThrough()).toList()); + print("onlyLong", SOURCE.stream().gather(onlyLong()).toList()); + print("upperCase", SOURCE.stream().gather(upperCase()).toList()); + print("twice", SOURCE.stream().gather(twice()).toList()); + System.out.println(); + System.out.println(" The only decision is how many times you call push()."); + + System.out.println(); + System.out.println("Now using state — the parameter ignored above"); + System.out.println("-".repeat(78)); + print("source", SOURCE); + print("numbered", SOURCE.stream().gather(numbered()).toList()); + System.out.println(); + print("source", REPEATED); + print("whenChanged", REPEATED.stream().gather(whenChanged()).toList()); + System.out.println(); + print("source", WITH_REPEATS); + print("occurrence", WITH_REPEATS.stream().gather(occurrence()).toList()); + System.out.println(); + System.out.println(" numbered() cannot be written with map: the output depends on how many"); + System.out.println(" elements went past before this one. whenChanged() cannot be written with"); + System.out.println(" filter: a predicate sees one element and nothing about its neighbours."); + System.out.println(); + } + + private static void print(final String label, final List value) { + System.out.printf(" %-14s %s%n", label, value); + } +}