java - que - html 5.0 caracteristicas

Cambios internos por límite y flujo desordenado.

Question

Básicamente esto surgió al tratar de responder otra pregunta. Supongamos que este código:

AtomicInteger i = new AtomicInteger(0);
AtomicInteger count = new AtomicInteger(0);
IntStream.generate(() -> i.incrementAndGet())
.parallel()
.peek(x -> count.incrementAndGet())
.limit(5)
.forEach(System.out::println);
System.out.println("count = " + count);

Comprendo el hecho de que IntStream#generate es un flujo infinito desordenado y para que termine debe haber una operación de cortocircuito ( limit en este caso). También entiendo que el Supplier es libre de ser llamado tantas veces como se sienta la implementación de Stream antes de que alcance ese límite.

Ejecutar esto bajo java-8, imprimiría el count siempre 512 (puede que no sea siempre, pero es así en mi máquina).

En contraste, ejecutar esto bajo java-10 rara vez excede de 5 . Entonces, mi pregunta es qué cambió internamente que el cortocircuito sea mucho mejor (estoy tratando de responder a esto por mi cuenta teniendo las fuentes y tratando de hacer algunas diferencias ...)

Answer 1

El cambio ocurrió en algún lugar entre Java 9, beta 103 y Java 9, beta 120 ( JDK‑8154387 ).

La clase responsable es StreamSpliterators.UnorderedSliceSpliterator.OfInt , resp. su super clase StreamSpliterators.UnorderedSliceSpliterator .

La versión antigua de la clase parecía

abstract static class UnorderedSliceSpliterator<T, T_SPLITR extends Spliterator<T>> {
static final int CHUNK_SIZE = 1 << 7;
// The spliterator to slice
protected final T_SPLITR s;
protected final boolean unlimited;
private final long skipThreshold;
private final AtomicLong permits;
UnorderedSliceSpliterator(T_SPLITR s, long skip, long limit) {
this.s = s;
this.unlimited = limit < 0;
this.skipThreshold = limit >= 0 ? limit : 0;
this.permits = new AtomicLong(limit >= 0 ? skip + limit : skip);
}
UnorderedSliceSpliterator(T_SPLITR s,
UnorderedSliceSpliterator<T, T_SPLITR> parent) {
this.s = s;
this.unlimited = parent.unlimited;
this.permits = parent.permits;
this.skipThreshold = parent.skipThreshold;
}

...

 @Override
public void forEachRemaining(Consumer<? super T> action) {
Objects.requireNonNull(action);
ArrayBuffer.OfRef<T> sb = null;
PermitStatus permitStatus;
while ((permitStatus = permitStatus()) != PermitStatus.NO_MORE) {
if (permitStatus == PermitStatus.MAYBE_MORE) {
// Optimistically traverse elements up to a threshold of CHUNK_SIZE
if (sb == null)
sb = new ArrayBuffer.OfRef<>(CHUNK_SIZE);
else
sb.reset();
long permitsRequested = 0;
do { } while (s.tryAdvance(sb) && ++permitsRequested < CHUNK_SIZE);
if (permitsRequested == 0)
return;
sb.forEach(action, acquirePermits(permitsRequested));
}
else {
// Must be UNLIMITED; let ''er rip
s.forEachRemaining(action);
return;
}
}
}

Como podemos ver, trata de almacenar en búfer hasta CHUNK_SIZE = 1 << 7 elementos en cada separador, que puede terminar en “número de núcleos de CPU” x 128 elementos.

Por el contrario, la nueva versión parece

abstract static class UnorderedSliceSpliterator<T, T_SPLITR extends Spliterator<T>> {
static final int CHUNK_SIZE = 1 << 7;
// The spliterator to slice
protected final T_SPLITR s;
protected final boolean unlimited;
protected final int chunkSize;
private final long skipThreshold;
private final AtomicLong permits;
UnorderedSliceSpliterator(T_SPLITR s, long skip, long limit) {
this.s = s;
this.unlimited = limit < 0;
this.skipThreshold = limit >= 0 ? limit : 0;
this.chunkSize = limit >= 0 ? (int)Math.min(CHUNK_SIZE,
((skip + limit) / AbstractTask.LEAF_TARGET) + 1) : CHUNK_SIZE;
this.permits = new AtomicLong(limit >= 0 ? skip + limit : skip);
}
UnorderedSliceSpliterator(T_SPLITR s,
UnorderedSliceSpliterator<T, T_SPLITR> parent) {
this.s = s;
this.unlimited = parent.unlimited;
this.permits = parent.permits;
this.skipThreshold = parent.skipThreshold;
this.chunkSize = parent.chunkSize;
}

...

 @Override
public void forEachRemaining(Consumer<? super T> action) {
Objects.requireNonNull(action);
ArrayBuffer.OfRef<T> sb = null;
PermitStatus permitStatus;
while ((permitStatus = permitStatus()) != PermitStatus.NO_MORE) {
if (permitStatus == PermitStatus.MAYBE_MORE) {
// Optimistically traverse elements up to a threshold of chunkSize
if (sb == null)
sb = new ArrayBuffer.OfRef<>(chunkSize);
else
sb.reset();
long permitsRequested = 0;
do { } while (s.tryAdvance(sb) && ++permitsRequested < chunkSize);
if (permitsRequested == 0)
return;
sb.forEach(action, acquirePermits(permitsRequested));
}
else {
// Must be UNLIMITED; let ''er rip
s.forEachRemaining(action);
return;
}
}
}

Así que ahora hay un campo de instancia chunkSize . Cuando hay un límite definido y la expresión ((skip + limit) / AbstractTask.LEAF_TARGET) + 1 evalúa a un valor más pequeño que CHUNK_SIZE , se CHUNK_SIZE ese valor más pequeño. Así que cuando chunkSize límites pequeños, el chunkSize será mucho más pequeño. En su caso con un límite de 5 , el tamaño del fragmento siempre será 1 .