JDK Stream реверс-инженеринг (реконструкция)16.08.2021 23:16

Привет all!

Вступление

Приглянулась мне однажды идея реверс-инженеринга (реконструкции) StreamAPI из JDK8. Что и как из этого вышло опишу далее.

Ссылки

Актуальный репозиторий

Ветка актуальная для статьи

Вкратце

StreamAPI — это технология появившаяся в Java 8 позволяющая манипулировать данными в наборах (напр. коллекциями) в функциональном стиле (лямбда-выражениями). Более подробно про стримы можно почитать тут.

Зачем и для чего

Для обучения, для более глубокого понимания исследуемой темы. В процессе написания данной статьи было получено немало опыта, как по стримам, так и в целом по Java и программированию. Поэтому эта риверсия является неплохой обучающей практикой. Я бы порекомендовал и новичкам и тем кто уровнем повыше выполнить самостоятельную реализацию стрима. Поверьте, независимо от вашего опыта, вас ждут открытия :)

Название

Название было выбрано следующим образом: Stream Reversed → StreamRe → StreamEr → Streamer

Возможные возможности/невозможности

Поскольку это обучающий реверс-инженеринг, целью которого является максимально понятно реализовать уже реализованное, то не стоит ожидать от стримера эффективного расхода памяти и быстродействия уровня Enterprise. Более того, в конце статьи я приведу пример сравнения производительности разных стримов, в котором стример уступает оригиналу из JDK по быстродействию.

Так же из реализации была исключена возможность распараллеливания стримов (Parallel stream), т.к. приемлемая реализация этого подхода потребует иных принципов построения и выходит за рамки этого материала.

Описанная тут реализация сохраняет следующие преимущества гибкости StreamAPI:

• чтения данных не происходит до вызова одного из терминальных методов. Т.е., пока мы собираем стример, мы никак не влияем на источник данных, не читаем его, а только формируем набор правил, по которым будет работать конвеер стрима, когда он будет запущен вызовом одного из терминальных методов.

• после вызова терминального метода, данные последовательно читаются из родительского набора и так же последовательно проходят по цепочке операций, не накапливаясь при этом в коллекциях, массивах и т.д. Конечно, за исключением метода сортировки, который по очевидным причинам требует предварительного накопления данных.

Spliterator vs. Iterator

Для того чтобы Stream мог функционировать, ему необходим источник данных. Стандартная реализация JDK (далее «оригинал»), под капотом, для чтения источника, использует сплитератор — Spliterator.

Основная задача сплитератора — разделять данные на блоки, т.е. порцировать их. Порцирование используется «оригиналом» для воможности распараллеливания стримов, когда разные порции обрабатываются разными потоками. Более подробно, о сплитераторах можно почитать тут.

Поскольку мы не будем реализовывать parallel для стримера, то и в разделении данных на блоки у нас тоже нет необходимости. Для простоты примера хватит итераторa — Iterator, поэтому «под капотом» именно через него и будем получать данные из родительского источника.

Жизненный цикл (внутренние состояния)

Жизненный цикл стримера я разделил на три состояния:

• Ожидание (WAITING) — начальное состояние стримера. В этом состоянии экземпляры создаются. Пока стример находится в этом состоянии, мы можем конструировать его из операций и вызвать один из терминальных методов когда потребуется «включить конвеер».

• В работе (OPERATED) — в это состояние стример переходит после вызова любого терминального метода. Это состояние означает, что либо стример находится в работе — «конвеер запущен», либо готов к запуску, т.е. уже сконструирован, а значит вызовы как конвеерных так и терминальных методов более невозможны.

• Завершен (CLOSED) — Это состояние означает, что стример завершил выполнение работы, ссылка на внешний итератор-источник обнулена (RefCount для GC).В это состояние стример переходит после того как:

  • завершилась работа любого из терминальных методов. Даже если в источнике остались данные. Например findFirst() вернул «первый» элемент. Данные возможно еще остались, но стример отработал свою задачу и может освободить не используемые ссылки.

  • В источнике закончились данные — hasNext() изменил свое состояние с true на false.

  • Извне, был вызван метод явного закрытия — close () и при этом стример находился в WAITING состоянии. Данное условие (WAITING) является обязательным, поскольку мы не можем по запросу завершать работающий стрим. Таковы правила, далее я это рассмотрю.

Подготовка

Разделим методы, стрима которые будем реализовывать на три группы:

Порождающие (factory): empty, of, generate, iterate, concat

Промежуточные (intermediate)/конвеерные: peek, onClose, distinct, filter, skip, limit, sorted, map, mapToInt, mapToLong, mapToDouble, flatMap, flatMapToInt, flatMapToLong, flatMapToDouble

Завершающие (terminal): spliterator, parallel, unordered, forEachOrdered, collect, min, max, reduce, count, forEach, allMatch, anyMatch, noneMatch, findFirst, findAny, iterator, toArray

Прочие: close, isParallel, sequential

Итак, создадим проект с начальной структурой и классом Streamer, реализующий интерфейс java.util.stream.Stream. Позволим IDE сгенерировать пустую реализацию всех методов (перечислены выше). Сгенерированные методы заглушим при помощи UnsupportedOperationException.

В итоге должно получиться примерно так.

Так же, сразу напишем реализацию простых методов — «однострочников» чтобы более к ним не возвращаться.

@Override
public Optional findFirst() {
    return findAny(); //поскольку у нас упорядоченный стрим, то первый элемент 				(First) и есть "произвольный" (Any)
}

@Override
public boolean isParallel() {
    return false; //мы не поддерживаем параллелизм, поэтому всегда false
}

@Override
public Stream sequential() {
    return this; //мы "последовательны", поэтому вернем себя же
}

@Override
public void forEachOrdered(Consumer action) {
    forEach(action); //опять же, мы упорядочены источником, поэтому в нашем 							случае forEach и forEachOrdered эквивалентны
}

@Override
public Spliterator spliterator() {
    return Spliterators.spliteratorUnknownSize(this.iterator(), Spliterator.ORDERED); //создадим сплитератор на основе «внутреннего» 												итератора
}

@Override
public Stream unordered() {
    return this; //так же, можно вернуть себя
}

Создание экземпляров

Под капотом, экземпляры будут создаваться единственым закрытым (private) конструктором, который в качестве аргумента принимает внешний итератор-источник. Этот итератор и будем использовать в качестве источника данных. Клиенты же, как и в оригинале, будут получать экземпляры стримера из статических фабрик. Стоит добавить, что к статической фабрике of () я дополнительно добавил перегруженные методы получения экземпляров Streamer из коллекций, перечисляемых (Iterable) типов, и непосредственно из самих Iterator`ов.

Примеры порождения стримера:

package pw.komarov.streamer;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;

public class StreamerInstancesCreationExamples {
    public static void main(String[] args) {
        Streamer.empty(); //пустой

        Streamer.of(new Object()); //единичный объект
        Streamer.of(new Integer[]{1, 4, 8, 17}); //массив
        Streamer.of(Arrays.asList(7.34, 9, 18.7, 3)); //Iterable (List)
        Streamer.of("Foo", "Bar", "Juice", "hello", "streamer"); //из констант

        //Infinite
        Streamer.generate(() -> ThreadLocalRandom.current().nextInt()); //бесконечный (рэндом-число)
        Streamer.generate(() ->
            {
                List strings = Arrays.asList("randomly", "returned", "string", "value");
                return strings.get(ThreadLocalRandom.current().nextInt(strings.size()));
            }); //рэндом значение

        Streamer.iterate(100, (i) -> i * 2); //последовательность {100,200,400.........n}
    }
}

Методы generate() и iterate() порождают бесконечный стрим, который на каждом шаге получает значение из бесконечного итератора, у которого hasNext() всегда == true и «заглушен» метод forEachRemaining():

private static abstract class AbstractInfiniteIterator implements Iterator {
    @Override
    public boolean hasNext() {
        return true;
    }

    @Override
    public void forEachRemaining(Consumer consumer) {
        throw new UnsupportedOperationException();
    }
}

Итератор для generate():

private static class InfiniteGenerator extends AbstractInfiniteIterator {
    private final Supplier supplier;
    
    InfiniteGenerator(Supplier supplier) {
        this.supplier = supplier;
    }

    @Override
    public E next() {
        return supplier.get();
    }
}

Далее, сначала создаем экземпляр этого генерирующего итератора, и затем из него стример:

public static  Streamer generate(Supplier supplier) {
    return of(new InfiniteGenerator<>(supplier));
}

Похожим образом реализован и iterate():

public static class InfiniteIterator extends AbstractInfiniteIterator {
    private E value; //значение предыдущего шага, при первом вызове — initial

    private final UnaryOperator unaryOperator; //клиенсткая функция 							генерации значения

    InfiniteIterator(E initial, UnaryOperator unaryOperator) {
        this.value = initial;
        this.unaryOperator = unaryOperator;
    }

    @Override
    public E next() {
        E prev = this.value;
        this.value = unaryOperator.apply(prev);
        return prev;
    }
}

public static  Streamer iterate(E initial, UnaryOperator unaryOperator {
    return of(new InfiniteIterator<>(initial, unaryOperator));
}

В итоге должно получиться примерно так.

Закрытие/завершение

Опишем два метода закрытия/завершения стримера. Первый — internalClose() для внутреннего использования. Вызывать его будем когда работа стримера логически завершена. Например закончились данные в источнике или завершена работа одного из терминальных методов. В общем, в тех случаях, когда использование стримера более невозможно. Этот метод будет так же обнулять ссылки на внешние ресурсы (чтобы уменьшить RefCount для GC) и переводить стример в CLOSED состояние.

Второй метод — внешнего закрытия, реализует close() интерфейса AutoCloseable. Фактически же, завершает стример только из состояния WAITING. Это сделано для того, чтобы внешний вызов не мог повлиять на работу выполняющегося стрима. Так работает оригинал. На мой взгляд это поведение не логично. И вот почему… Предположим, что стрим выполняет тяжеловесную операцию одним из терминальных методов. В какой то момент (к примеру, пользователь запросил отмену действия), мы понимаем что больше не нуждаемся в этой тяжеловесной работе и хотим ее принудительно прекратить. Стрим исполняется в другом потоке, но у нас есть указатель на этот стрим. Вызываем close () в надежде прекратить выполнение операции, но он продолжает работать как ни в чем небывало… А жаль… Ведь так хотелось… :).

Второй важной частью работы этого метода является вызов пользовательских onClose последовательностей. Но и тут скрывается подвох. В оригинальном стриме эти onClose выполняются только в случае явного вызова метода close (). Т.е. если стрим завершил работу, допустим найдено искомое (min, max и т.д.), то onClose будут просто проигнорированы, а ведь возможно там были важные финализаторы… При описанном поведении инструмент предоставляемый методом onClose () вообще не представляет практической ценности, поскольку те же самые операции можно вызвать «вручную» из клиентского кода, после вызова close () например. Можно будет даже более гибко обработать возможные исключения.

Ну что же, имеем то, что имеем… поэтому для поддержания совместимости реализуем эти особенности в том же виде:

private enum State {WAITING, OPERATED, CLOSED}

private State state = State.WAITING;

private final List onCloseSequences = new LinkedList<>();

@Override
public void close() {
    if (state == State.WAITING)
        internalClose();

    //обработаем (выполним) клиентские onClose последовательности...
    RuntimeException rte = null;
    for (Iterator iterator = onCloseSequences.iterator(); 					iterator.hasNext(); ) {
        Runnable runnable = iterator.next();
        try {
            runnable.run();
        } catch (RuntimeException e) {
            if (rte == null) //если это первое исключение в цепочке...
                rte = e; //...сохраним его
            else //если не первое...
                rte.addSuppressed(e); //...сохраним его в suppressed первого
        } finally {
            iterator.remove();
        }
    }

    if (rte != null)
        throw rte;
}

private void internalClose() {
    externalIterator = null;

    state = State.CLOSED;
}

private void throwIfNotWaiting() {
    if (state != State.WAITING)
        throw new IllegalStateException("stream has already been operated upon 		or closed");
}

@Override
public Stream onClose(Runnable closeHandler) {
    throwIfNotWaiting();

    onCloseSequences.add(closeHandler);

    return this;
}

Контракт onClose () для стрима гласит, что первое исключение погашается и сохраняется, прочие исключение (если они есть), добавляются в suppressed первого. И если было первое, то оно и бросается после выполнения всех onClose`ов. Этот контракт так же сохранен в реализации приведенной выше.

Расстановка по шаблону

Ранее мы реализовали метод проверки текущего состояния стримера, который бросает IllegalStateException если стример не в WAITING состоянии. Теперь пришло время его расставить в места где это нужно. А нужно это сделать во всех терминальных и конвеерных методах, кроме «однострочников» описанных ранее, т.к. они все равно ссылаются на эти методы.

Поскольку конвеерные методы будут работать по принципу Builder`a — иметь возможность телескопического построения (прим.: object.method1().method2().method3().methodN()…), то каждый из этих методов должен возвращать экземпляр себя. В итоге шаблон конвеерного метода приобрел такой вид:

{
    throwIfNotWaiting();
        
          //todo: тут будет создание и добавление операций

    return this;
}

Каждый терминальный метод должен переводить стример из WAITING в OPERATED состояние, а по завершению работы — корректно закрывать его. Резюмируя вышесказанное, «шаблон» терминального метода приобретает такой вид:

{
    throwIfNotWaiting(); //IllegalStateException если пытаемся использовать 				запущенный или завершенный стример

    state = State.OPERATED; //переведем в OPERATED

	try {
	    ;//todo: терминальные операции…
	} finally {
	    internalClose(); //выполним завершение
	}
    throw new UnsupportedOperationException("will be soon");  //чтобы не забыть 					про return :)
}

В итоге получилось так.

Промежуточные операции (intermediate/conveyor)

Ну вот мы и подошли к логике работы стримера. Как известно, стрим состоит из набора операций, которые последовательно применяются к данным которые представлены этому стриму. Поставим вопрос, как будем хранить и как будем «строить» наборы этих операций? Тут все очень просто.

Для обозначения самой операции, объявим интерфейс:

private interface IntermediateOperation {}

Набор операций — список элементов этого интерфейса:

private List intermediateOperations = new LinkedList<>();

А добавлять в этот список конкретные операции будем из конвеерных методов.

Из всех конвеерных операций стрима выделим отдельную группу — фильтрующие операции. Это операции, которые на основании некоторого условия (предиката), зависящего от типа операции, определяют — пройдет ли элемент данных далее по конвееру или будет отброшен на текущем шаге. Вот список всех конвеерные методов, относящихся к фильтрующим операциям: skip(), limit(), distinct(), filter().

Для обозначения этих операций, объявим еще один интерфейс:

private interface FilteringOperation extends IntermediateOperation, 	Predicate {}

Predicate является функциональным интерфейсом (FunctionalInterface, подробнее https://habr.com/ru/post/512730/), и его функциональный метод — boolean test(). Реализацией этого метода в конкретной операции мы и будем определять, пройдет ли элемент по конвееру дальше, или будет «отброшен».

Вот так будет выглядеть класс конкретной операции (в приведенном случае skip):

private static class SkipOperation implements FilteringOperation {
    private final long totalCount; //количество элементов которые требуется 								"пропустить"
    private long processedCount; //количество уже "пропущеных" элементов 								текущей операцией
    
    SkipOperation(long totalCount) {
        this.totalCount = totalCount;
    }

    @Override
    public boolean test(Object o) {
        if (processedCount < totalCount) {
            processedCount++;
            
            return true; //пропустим элемент далее
        }

        return false; //отбросим/отфильтруем элемент
    }
}

@Override
public Stream skip(long n) {
    throwIfNotWaiting(); //проверим текущее состояние

    intermediateOperations.add(new SkipOperation(n)); //создадим Skip-операцию, 								и добавим ее в список операций.

    return this; //вернем экземпляр «себя» для возможности телескопического 											построения
}

По такому же принципу реализуем добавление остальных фильтрующих операций:

//limit()
private long filteredByLimit; //количество "отсеяных" limit'ом элементов

private class LimitOperation implements FilteringOperation {
    private final long maxSize; //собственно и есть лимит

    LimitOperation(long maxSize) {
        this.maxSize = maxSize;
    }

    @Override
    public boolean test(Object o) {
        return maxSize < ++filteredByLimit;
    }
}

@Override
public Stream limit(long maxSize) {
    throwIfNotWaiting();

    intermediateOperations.add(new LimitOperation(maxSize));

    return this;
}

//distinct()

private static class DistinctOperation implements FilteringOperation {
    private Set