Внутренние изменения для лимитного и неупорядоченного потока

В основном это появилось при попытке ответить на другой вопрос. Предположим, этот код:

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);

Я понимаю тот факт, что IntStream#generate — это неупорядоченный бесконечный поток, и для его завершения должна быть операция короткого замыкания (в данном случае limit). Я также понимаю, что Supplier может вызываться столько раз, сколько будет ощущаться реализацией Stream, прежде чем она достигнет этого предела.

Запуск этого под java-8 будет печатать count всегда 512 (может быть, не всегда, но на моей машине это так).

В отличие от этого под java-10 редко превышает 5. Итак, мой вопрос заключается в том, что изменилось внутри, что короткое замыкание происходит намного лучше (я пытаюсь ответить на это самостоятельно, имея исходники и пытаясь сделать некоторые различия...)


person Eugene    schedule 27.04.2018    source источник


Ответы (1)


Изменение произошло где-то между Java 9, бета 103 и Java 9, бета 120 (JDK‑8154387< /а>).

Ответственный класс — StreamSpliterators.UnorderedSliceSpliterator.OfInt, соответственно. его суперкласс StreamSpliterators.UnorderedSliceSpliterator.

Старая версия класса выглядела так

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;
                }
            }
        }

Как мы видим, он пытается буферизовать до CHUNK_SIZE = 1 << 7 элементов в каждом разделителе, что может закончиться «количеством ядер ЦП» × 128 элементов.

Напротив, новая версия выглядит

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;
                }
            }
        }

Итак, теперь есть поле экземпляра chunkSize. Когда существует определенный предел и выражение ((skip + limit) / AbstractTask.LEAF_TARGET) + 1 оценивается как меньшее значение, чем CHUNK_SIZE, будет использоваться это меньшее значение. Таким образом, при небольших лимитах chunkSize будет намного меньше. В вашем случае с ограничением 5 размер чанка всегда будет 1.

person Holger    schedule 27.04.2018