Add two standalone modules: Stream Gatherers (JEP 485) and Scoped Values vs ThreadLocal (JEP 506)
gatherers/: windowFixed, windowSliding, fold, scan, mapConcurrent, a custom Gatherer (dedupeConsecutive, takeUntil), and a Collector-vs-Gatherer comparison. Verified: windowSliding emits one short window on a too-short stream rather than shrinking to empty like windowFixed; mapConcurrent waits for in-flight siblings to finish (~901ms) rather than cancelling them when one mapper throws. scoped-values/: ScopedValue API and rebinding, why plain threads (virtual or platform) do not inherit a binding while a StructuredTaskScope subtask does, and the memory cost vs InheritableThreadLocal measured two ways -- a noisy heap-delta first pass, then a precise jcmd GC.class_histogram object census (5 shared Carrier objects vs ~700,000 copied ThreadLocalMap/Entry objects for 100,000 threads). Companion code for the ankurm.com posts "Java Stream Gatherers (JEP 485)" and "Scoped Values vs ThreadLocal in Java 25: Migration Guide with Virtual Threads". Co-Authored-By: Claude Sonnet 5 <[email protected]> Claude-Session: https://claude.ai/code/session_01FtpJvZfg4nvLvtzgJTDWpB
This commit is contained in:
Claude
@@ -0,0 +1,15 @@
|
||||
/**
|
||||
* Tiny assertion helper so every demo is self-checking: a claim that stops being true turns the
|
||||
* run red instead of quietly printing something different. Passing checks are echoed, so the
|
||||
* transcript shows exactly what was verified.
|
||||
*/
|
||||
final class Check {
|
||||
private Check() {}
|
||||
|
||||
static void that(boolean condition, String claim) {
|
||||
if (!condition) {
|
||||
throw new AssertionError("CHECK FAILED: " + claim);
|
||||
}
|
||||
System.out.println("CHECK ok : " + claim);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,68 @@
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.stream.Collectors;
|
||||
import java.util.stream.Gatherers;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
/**
|
||||
* When each is the right tool. A Collector always reduces a stream to exactly one final value,
|
||||
* at the very end of the pipeline -- it cannot appear in the middle, and it cannot stop the
|
||||
* source early. A Gatherer is an intermediate operation: it produces a new Stream, can emit zero,
|
||||
* one, or many elements per input, and (as CustomGatherer.takeUntil showed) can stop pulling from
|
||||
* upstream before the source is exhausted. Reach for a Collector when the answer is genuinely one
|
||||
* value; reach for a Gatherer when you need to keep streaming afterward, or the transform itself
|
||||
* needs private state, ordering, or the ability to bail out early.
|
||||
* Chapter: docs/16-gatherers.md
|
||||
*/
|
||||
public class CollectorVsGatherer {
|
||||
|
||||
public static void main(String[] args) {
|
||||
collectorIsRightForATerminalAggregate();
|
||||
gathererIsRightForAnIntermediateStatefulTransform();
|
||||
aGathererResultCanKeepBeingAStream();
|
||||
}
|
||||
|
||||
static void collectorIsRightForATerminalAggregate() {
|
||||
// groupingBy needs to see the whole stream and produce ONE Map. There's no intermediate
|
||||
// stage here to preserve -- a Gatherer would just add ceremony around the same result.
|
||||
record Order(String customer, double amount) {}
|
||||
List<Order> orders = List.of(
|
||||
new Order("amy", 12.50), new Order("bo", 4.00),
|
||||
new Order("amy", 7.25), new Order("bo", 19.99));
|
||||
|
||||
Map<String, Double> totals = orders.stream()
|
||||
.collect(Collectors.groupingBy(Order::customer, Collectors.summingDouble(Order::amount)));
|
||||
|
||||
System.out.println("Collectors.groupingBy totals = " + totals);
|
||||
Check.that(totals.get("amy") == 19.75, "amy's total via groupingBy/summingDouble is 19.75");
|
||||
Check.that(totals.get("bo") == 23.99, "bo's total via groupingBy/summingDouble is 23.99");
|
||||
}
|
||||
|
||||
static void gathererIsRightForAnIntermediateStatefulTransform() {
|
||||
// A running total that still needs filtering and mapping done to it afterward. Doing this
|
||||
// with a Collector would mean collecting to a List<Integer> of running totals first, then
|
||||
// starting a SECOND stream over that list -- two passes, and the "it's still one
|
||||
// pipeline" property is gone. gather() keeps it as one lazy pipeline throughout.
|
||||
List<String> overThreshold = Stream.of(5, 10, 15, 20, 25)
|
||||
.gather(Gatherers.scan(() -> 0, Integer::sum)) // running totals: 5,15,30,50,75
|
||||
.filter(total -> total > 20)
|
||||
.map(total -> "total=" + total)
|
||||
.toList();
|
||||
|
||||
System.out.println("scan -> filter -> map, one pipeline = " + overThreshold);
|
||||
Check.that(overThreshold.equals(List.of("total=30", "total=50", "total=75")),
|
||||
"scan feeding straight into filter/map in the same pipeline gives [total=30, total=50, total=75]");
|
||||
}
|
||||
|
||||
static void aGathererResultCanKeepBeingAStream() {
|
||||
// The defining shape difference, made concrete: a Collector's result type is whatever the
|
||||
// Collector was parameterized with (here, a single Long). A Gatherer's result is always a
|
||||
// Stream<R> you can keep chaining -- gather() returns Stream<R>, not R.
|
||||
long distinctWindowCount = Stream.of(1, 2, 3, 4, 5, 6, 7)
|
||||
.gather(Gatherers.windowFixed(2))
|
||||
.collect(Collectors.counting());
|
||||
System.out.println("windowFixed(2) over 7 elements, then collect(counting()) = " + distinctWindowCount);
|
||||
Check.that(distinctWindowCount == 4,
|
||||
"collect(counting()) after gather() proves the gather result is still a Stream you can collect further: 4 windows");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,82 @@
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.function.Predicate;
|
||||
import java.util.stream.Gatherer;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
/**
|
||||
* Writing a Gatherer from scratch. Gatherer<T, A, R> is four composable pieces:
|
||||
* initializer (private mutable state A), integrator (per-element: read/write state A, push 0+ R
|
||||
* downstream, return whether to keep going), combiner (merge two states for parallel streams --
|
||||
* omit it and the gatherer just runs sequentially), finisher (flush anything left when the
|
||||
* upstream is exhausted). Two custom gatherers here: dedupeConsecutive (stateful, one-in/0-or-1
|
||||
* out) and takeUntil (short-circuiting -- something no Collector can do, because a Collector
|
||||
* never gets a say in whether the source stops producing).
|
||||
* Chapter: docs/16-gatherers.md
|
||||
*/
|
||||
public class CustomGatherer {
|
||||
|
||||
public static void main(String[] args) {
|
||||
dedupeConsecutiveDemo();
|
||||
takeUntilStopsTheUpstream();
|
||||
}
|
||||
|
||||
/** Collapses consecutive equal elements, like Unix `uniq` -- keeps non-adjacent duplicates. */
|
||||
static <T> Gatherer<T, ?, T> dedupeConsecutive() {
|
||||
return Gatherer.ofSequential(
|
||||
// initializer: a single-slot box holding "last element pushed downstream", or
|
||||
// null meaning "nothing pushed yet". A List<Object> of size 1 doubles as a
|
||||
// nullable box without a wrapper class.
|
||||
() -> new ArrayList<Object>(List.of()),
|
||||
// integrator: push only when this element differs from the last one we pushed.
|
||||
(state, element, downstream) -> {
|
||||
boolean isFirst = state.isEmpty();
|
||||
boolean changed = isFirst || !state.get(0).equals(element);
|
||||
if (changed) {
|
||||
if (isFirst) {
|
||||
state.add(element);
|
||||
} else {
|
||||
state.set(0, element);
|
||||
}
|
||||
return downstream.push(element);
|
||||
}
|
||||
return true; // swallowed a duplicate; keep pulling from upstream
|
||||
});
|
||||
}
|
||||
|
||||
static void dedupeConsecutiveDemo() {
|
||||
List<String> deduped = Stream.of("a", "a", "b", "b", "b", "a", "c", "c")
|
||||
.gather(CustomGatherer.<String>dedupeConsecutive())
|
||||
.toList();
|
||||
System.out.println("dedupeConsecutive of a,a,b,b,b,a,c,c = " + deduped);
|
||||
Check.that(deduped.equals(List.of("a", "b", "a", "c")),
|
||||
"consecutive runs collapse to [a, b, a, c]; the second 'a' survives because it isn't adjacent to the first");
|
||||
}
|
||||
|
||||
/**
|
||||
* takeWhile(p) stops BEFORE the first failing element. takeUntil(p) stops AFTER it --
|
||||
* includes the element that fails the predicate, then no more. Not expressible with
|
||||
* takeWhile(p.negate()) plus one extra element without re-running the predicate; building it
|
||||
* as a Gatherer means integrator itself decides when to stop pulling from upstream.
|
||||
*/
|
||||
static <T> Gatherer<T, ?, T> takeUntil(Predicate<? super T> stopAfter) {
|
||||
return Gatherer.ofSequential((state, element, downstream) -> {
|
||||
downstream.push(element);
|
||||
return !stopAfter.test(element); // false = stop; upstream is not pulled again
|
||||
});
|
||||
}
|
||||
|
||||
static void takeUntilStopsTheUpstream() {
|
||||
List<Integer> upstreamSeen = new ArrayList<>();
|
||||
List<Integer> result = Stream.of(1, 2, 3, 4, 5, 6)
|
||||
.peek(upstreamSeen::add)
|
||||
.gather(takeUntil((Integer n) -> n == 3))
|
||||
.toList();
|
||||
|
||||
System.out.println("takeUntil(n == 3) result = " + result + ", upstream elements actually pulled = " + upstreamSeen);
|
||||
Check.that(result.equals(List.of(1, 2, 3)), "takeUntil includes the element that satisfies the stop condition: [1, 2, 3]");
|
||||
Check.that(upstreamSeen.equals(List.of(1, 2, 3)),
|
||||
"the upstream peek() only saw 1, 2, 3 -- elements 4, 5, 6 were never pulled, this genuinely short-circuits");
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,104 @@
|
||||
import java.util.List;
|
||||
import java.util.stream.Gatherers;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
/**
|
||||
* The five built-in gatherers (JEP 485, final in Java 24, no preview flag on 25+):
|
||||
* windowFixed, windowSliding, fold, scan, mapConcurrent. This file covers the first four --
|
||||
* mapConcurrent gets its own file because it needs real (slow) work to say anything interesting.
|
||||
* Chapter: docs/16-gatherers.md
|
||||
*/
|
||||
public class GathererBasics {
|
||||
|
||||
public static void main(String[] args) {
|
||||
windowFixed();
|
||||
windowSliding();
|
||||
fold();
|
||||
scan();
|
||||
foldIsNotScan();
|
||||
}
|
||||
|
||||
static void windowFixed() {
|
||||
// Chunks the stream into lists of exactly `n`, except possibly the last one, which is
|
||||
// shorter if the stream doesn't divide evenly. No overlap between windows.
|
||||
List<List<Integer>> windows = Stream.of(1, 2, 3, 4, 5, 6, 7)
|
||||
.gather(Gatherers.windowFixed(3))
|
||||
.toList();
|
||||
System.out.println("windowFixed(3) of 1..7 = " + windows);
|
||||
Check.that(windows.equals(List.of(List.of(1, 2, 3), List.of(4, 5, 6), List.of(7))),
|
||||
"windowFixed(3) groups into [1,2,3][4,5,6][7], last window is the short remainder");
|
||||
|
||||
try {
|
||||
Stream.of(1).gather(Gatherers.windowFixed(0));
|
||||
Check.that(false, "windowFixed(0) should reject a non-positive window size");
|
||||
} catch (IllegalArgumentException expected) {
|
||||
Check.that(true, "windowFixed(0) throws IllegalArgumentException: " + expected.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
static void windowSliding() {
|
||||
// A window that advances by 1 each time, so consecutive windows overlap. Stops producing
|
||||
// windows once fewer than `n` elements remain -- it never emits a short window.
|
||||
List<List<Integer>> windows = Stream.of(1, 2, 3, 4, 5)
|
||||
.gather(Gatherers.windowSliding(3))
|
||||
.toList();
|
||||
System.out.println("windowSliding(3) of 1..5 = " + windows);
|
||||
Check.that(windows.equals(List.of(List.of(1, 2, 3), List.of(2, 3, 4), List.of(3, 4, 5))),
|
||||
"windowSliding(3) of 5 elements produces exactly 3 overlapping windows, each length 3");
|
||||
|
||||
// Easy to assume this behaves like windowFixed's leftover-shrinks-to-fit rule, or like
|
||||
// Collectors.windowed variants elsewhere that drop a too-short remainder -- neither is
|
||||
// right. windowSliding still emits ONE window holding whatever is there, as long as the
|
||||
// stream isn't empty.
|
||||
List<List<Integer>> tooFew = Stream.of(1, 2).gather(Gatherers.windowSliding(3)).toList();
|
||||
Check.that(tooFew.equals(List.of(List.of(1, 2))), "windowSliding(3) on a 2-element stream still emits one short window: [[1, 2]]");
|
||||
|
||||
List<List<Integer>> empty = Stream.<Integer>of().gather(Gatherers.windowSliding(3)).toList();
|
||||
Check.that(empty.isEmpty(), "windowSliding(3) on an empty stream emits nothing");
|
||||
}
|
||||
|
||||
static void fold() {
|
||||
// fold(supplier, BiFunction) is a sequential-only, non-associative reduction: the result
|
||||
// type R doesn't have to match the element type T, and there's no combiner because fold
|
||||
// never runs in parallel. That's exactly what lets it do things Stream.reduce can't --
|
||||
// like building a String left-to-right from a Stream<Integer>.
|
||||
String joined = Stream.of(1, 2, 3, 4)
|
||||
.gather(Gatherers.fold(() -> "", (acc, n) -> acc + "[" + n + "]"))
|
||||
.findFirst()
|
||||
.orElseThrow();
|
||||
System.out.println("fold building a String: " + joined);
|
||||
Check.that(joined.equals("[1][2][3][4]"), "fold(\"\", acc+\"[n]\") over 1..4 builds \"[1][2][3][4]\" left to right");
|
||||
|
||||
// fold, like Collectors.reducing, emits exactly one final value -- it's an intermediate
|
||||
// op that happens to produce a stream of size 1, not a running total.
|
||||
long count = Stream.of(1, 2, 3).gather(Gatherers.fold(() -> 0, Integer::sum)).count();
|
||||
Check.that(count == 1, "fold's output stream has exactly one element, the final accumulation");
|
||||
}
|
||||
|
||||
static void scan() {
|
||||
// scan(supplier, BiFunction) uses the identical signature to fold but emits every
|
||||
// intermediate accumulation, not just the last one -- a running total, a prefix-sum. This
|
||||
// is something no Collector can express: a Collector always reduces to one final value,
|
||||
// never a sequence of intermediate ones.
|
||||
List<Integer> runningTotals = Stream.of(1, 2, 3, 4, 5)
|
||||
.gather(Gatherers.scan(() -> 0, Integer::sum))
|
||||
.toList();
|
||||
System.out.println("scan running total of 1..5 = " + runningTotals);
|
||||
Check.that(runningTotals.equals(List.of(1, 3, 6, 10, 15)), "scan(0, +) over 1..5 emits every prefix sum: [1,3,6,10,15]");
|
||||
}
|
||||
|
||||
static void foldIsNotScan() {
|
||||
// Same supplier and BiFunction, different gatherer -- proof the two aren't the same
|
||||
// operation wearing different names.
|
||||
var supplier = (java.util.function.Supplier<Integer>) () -> 0;
|
||||
var adder = (java.util.function.BiFunction<Integer, Integer, Integer>) Integer::sum;
|
||||
|
||||
List<Integer> foldResult = Stream.of(1, 2, 3).gather(Gatherers.fold(supplier, adder)).toList();
|
||||
List<Integer> scanResult = Stream.of(1, 2, 3).gather(Gatherers.scan(supplier, adder)).toList();
|
||||
|
||||
System.out.println("fold(0,+) over 1..3 = " + foldResult);
|
||||
System.out.println("scan(0,+) over 1..3 = " + scanResult);
|
||||
Check.that(foldResult.equals(List.of(6)), "fold(0,+) over 1..3 emits just the final sum [6]");
|
||||
Check.that(scanResult.equals(List.of(1, 3, 6)), "scan(0,+) over 1..3 emits every partial sum [1,3,6]");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,120 @@
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.ConcurrentLinkedQueue;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.stream.Gatherers;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
/**
|
||||
* Gatherers.mapConcurrent(maxConcurrency, mapper): the one built-in gatherer whose whole point is
|
||||
* concurrency, not stream shape. It runs the mapper for each element on its own virtual thread (a
|
||||
* new one per element, not a fixed pool of virtual threads), bounded to at most `maxConcurrency`
|
||||
* in flight, and preserves encounter order in the output regardless of which finishes first.
|
||||
* Chapter: docs/16-gatherers.md
|
||||
*/
|
||||
public class MapConcurrentDemo {
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
orderIsPreservedEvenWhenCompletionOrderIsNot();
|
||||
concurrencyIsBounded();
|
||||
runsOnVirtualThreadsNotThePlatformPool();
|
||||
aFailureWaitsForTheOthersRatherThanCancellingThem();
|
||||
}
|
||||
|
||||
static void orderIsPreservedEvenWhenCompletionOrderIsNot() {
|
||||
// Task N sleeps for (5-N)*40ms, so task 1 finishes LAST and task 4 finishes FIRST -- yet
|
||||
// the output list comes back in input order, 1..4, because mapConcurrent buffers results
|
||||
// and only pushes downstream once every earlier slot is filled.
|
||||
List<Integer> results = Stream.of(1, 2, 3, 4)
|
||||
.gather(Gatherers.mapConcurrent(4, n -> {
|
||||
sleep(Duration.ofMillis((5 - n) * 40L));
|
||||
return n * 10;
|
||||
}))
|
||||
.toList();
|
||||
|
||||
System.out.println("mapConcurrent(4, ...) results, completion order reversed: " + results);
|
||||
Check.that(results.equals(List.of(10, 20, 30, 40)),
|
||||
"mapConcurrent output preserves encounter order [10,20,30,40] even though task 1 finishes last");
|
||||
}
|
||||
|
||||
static void concurrencyIsBounded() {
|
||||
// 8 tasks, each holding a permit for 60ms, capped at maxConcurrency=2. If the cap were
|
||||
// real only in name, all 8 would overlap and the peak-in-flight counter would read 8.
|
||||
AtomicInteger inFlight = new AtomicInteger(0);
|
||||
AtomicInteger peak = new AtomicInteger(0);
|
||||
|
||||
List<Integer> results = Stream.of(1, 2, 3, 4, 5, 6, 7, 8)
|
||||
.gather(Gatherers.mapConcurrent(2, n -> {
|
||||
int now = inFlight.incrementAndGet();
|
||||
peak.updateAndGet(p -> Math.max(p, now));
|
||||
sleep(Duration.ofMillis(60));
|
||||
inFlight.decrementAndGet();
|
||||
return n;
|
||||
}))
|
||||
.toList();
|
||||
|
||||
System.out.println("mapConcurrent(2, ...) over 8 tasks, peak concurrent in-flight = " + peak.get());
|
||||
Check.that(results.size() == 8, "all 8 tasks completed");
|
||||
Check.that(peak.get() <= 2, "peak in-flight never exceeded maxConcurrency=2, measured peak=" + peak.get());
|
||||
Check.that(peak.get() == 2, "the cap is actually reached, not just respected -- measured peak=" + peak.get());
|
||||
}
|
||||
|
||||
static void runsOnVirtualThreadsNotThePlatformPool() {
|
||||
// Each element's mapper runs on its own Thread.ofVirtual() thread, not a shared
|
||||
// ForkJoinPool.commonPool() worker the way Stream.parallel() would use.
|
||||
ConcurrentLinkedQueue<String> threadNames = new ConcurrentLinkedQueue<>();
|
||||
AtomicInteger distinctThreads = new AtomicInteger(0);
|
||||
|
||||
Stream.of(1, 2, 3, 4, 5, 6)
|
||||
.gather(Gatherers.mapConcurrent(6, n -> {
|
||||
Thread t = Thread.currentThread();
|
||||
threadNames.add(t + " virtual=" + t.isVirtual());
|
||||
return n;
|
||||
}))
|
||||
.toList();
|
||||
|
||||
threadNames.forEach(System.out::println);
|
||||
boolean allVirtual = threadNames.stream().allMatch(s -> s.endsWith("virtual=true"));
|
||||
long distinct = threadNames.stream().distinct().count();
|
||||
Check.that(allVirtual, "every mapConcurrent worker thread reports isVirtual() == true");
|
||||
Check.that(distinct == 6, "6 elements produced 6 distinct virtual threads -- one per element, not a shared pool");
|
||||
}
|
||||
|
||||
static void aFailureWaitsForTheOthersRatherThanCancellingThem() {
|
||||
// Easy to assume a fail-fast gatherer interrupts its siblings the moment one mapper
|
||||
// throws -- structured concurrency's StructuredTaskScope does exactly that. mapConcurrent
|
||||
// does NOT: it lets every already-started mapper run to completion (or timeout) before
|
||||
// the exception surfaces. Element 3 throws immediately here; elements 1, 2, 4, 5 each
|
||||
// sleep 900ms. If the failure were fail-fast we'd see it in well under 900ms -- instead
|
||||
// the whole call blocks for as long as the slowest sibling.
|
||||
Instant start = Instant.now();
|
||||
try {
|
||||
Stream.of(1, 2, 3, 4, 5)
|
||||
.gather(Gatherers.mapConcurrent(5, n -> {
|
||||
if (n == 3) {
|
||||
throw new IllegalStateException("boom at element 3");
|
||||
}
|
||||
sleep(Duration.ofMillis(900));
|
||||
return n;
|
||||
}))
|
||||
.toList();
|
||||
Check.that(false, "expected the failure to propagate");
|
||||
} catch (IllegalStateException e) {
|
||||
Duration elapsed = Duration.between(start, Instant.now());
|
||||
System.out.println("mapConcurrent propagated: " + e.getMessage() + " after " + elapsed.toMillis() + "ms");
|
||||
Check.that(elapsed.toMillis() >= 850,
|
||||
"the failure only surfaced after " + elapsed.toMillis()
|
||||
+ "ms -- mapConcurrent waited for the other 4 in-flight tasks, it did not cancel them");
|
||||
}
|
||||
}
|
||||
|
||||
static void sleep(Duration d) {
|
||||
try {
|
||||
Thread.sleep(d);
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user