Производительность параллельных вычислений. Java 8 [5 min reading]

Benchmark

В предыдущих двух постах мы разобрались с тем, как устроен data parallelism в Stream API. Теперь можем двигаться к самому важному - что все это нам дает. Вроде бы все очевидно - чем больше ядер - тем больше задач мы можем выполнить параллельно - тем быстрее мы получаем результат. Но давайте по порядку.


Вернемся к нашему примеру из прошлого поста и сравним время выполнения. OpenJDK предлагает для этого удобный инструмент. Как это работает?

Добавляем зависимость:

        <dependency>
            <groupId>org.openjdk.jmh</groupId>
            <artifactId>jmh-core</artifactId>
            <version>0.1</version>
        </dependency>

Задаем количество прогревающих и тестовых запусков, замерять будем среднее время выполнения операции:

@State(Scope.Thread)
@BenchmarkMode(Mode.AverageTime)
public class SerialAndParallelBenchmark {
    public static void main(String[] args) throws RunnerException {
        Options opt = new OptionsBuilder()
                .include(".*")
                .warmupIterations(10)
                .measurementIterations(10)
                .forks(1)
                .outputFormat(OutputFormatType.TextReport)
                .build();

        new Runner(opt).run();
    }
    @Setup
    public void init() {
        [...]
    }
[...]
}

Аннотируем методы, время выполнения которых необходимо замерять:

    @GenerateMicroBenchmark
    public void sumReduce() {
        ReduceExample.sumReduce(integers);
    }

    @GenerateMicroBenchmark
    public void sumParallelReduce() {
        ReduceExample.sumParallelReduce(integers);
    }

Performance

В этих методах мы с помощью reduce суммируем значения 1000 элементов, получаем такой результат:

Benchmark             Mode Thr     Count  Sec         Mean   Mean error    Units
sumParallelReduce     avgt   1        10    1        0.016        0.003    ms/op
sumReduce             avgt   1        10    1        0.007        0.001    ms/op

Нас интересует колонка Mean. Что мы видим, для тысячи элементов последовательное выполнение вычислений быстрее. Для 10 000 тысяч:

Benchmark             Mode Thr     Count  Sec         Mean   Mean error    Units
sumParallelReduce     avgt   1        10    1        0.079        0.013    ms/op
sumReduce             avgt   1        10    1        0.053        0.002    ms/op

Время исполнения уже почти сравнялось. Теперь посмотрим при 100 000 элементов:

Benchmark             Mode Thr     Count  Sec         Mean   Mean error    Units
sumParallelReduce     avgt   1        10    1        0.446        0.076    ms/op
sumReduce             avgt   1        10    1        0.553        0.025    ms/op

Теперь параллельные вычисления выполнились быстрее. Особого прироста в производительности мы не наблюдаем, а при небольшом размере данных даже тратим больше времени выполняя задачу параллельно. Давайте попробуем разобраться в чем же дело.

Так как параллельные вычисления используют Fork/Join, то параллельное вычисление состоит из следующих этапов: декомпозиция задачи, вычисления в подзадачах, слияние результатов:

Таким образом мы имеем overhead в виде декомпозиции и слияния, которые тоже требует время на выполнение. Потому важно больше времени проводить на уровне вычисления (ForkJoinTask#compute) в подзадачах.

Рассмотрим другой пример. Есть структура данных:

{
   "teams":[
      {
         "name":"Real Madrid",
         "players":[
            {
               "name":"Ronaldo",
               "score":35
            },
            {
               "name":"Bale",
               "score":24
            }
         ]
      },
      {
         "name":"Barcelona",
         "players":[
            {
               "name":"Suarez",
               "score":40
            },
            {
               "name":"Messi",
               "score":26
            }
         ]
      }
   ]
}

И метод, который вычисляет суммы голов забитых всеми игроками всех команд

    @GenerateMicroBenchmark
    public void sum() {
        teams.stream()
                .flatMap(team -> team.getPlayers().stream())
                .mapToInt(Player::getScore)
                .sum();
    }

    @GenerateMicroBenchmark
    public void sumParallel() {
        teams.parallelStream()
                .flatMap(team -> team.getPlayers().parallelStream())
                .mapToInt(Player::getScore)
                .sum();
    }

Результаты:

кол-во команд	последовательно	параллельно
1 000	0.059	0.046
10 000	0.919	0.603
100 000	8.169	5.844
1 000 000	157.504	51.575

Здесь мы видим, что при больших размерах коллекций, параллельные вычисления происходят быстрее. Два вида операций: flatMap и mapToInt выполняются параллельно на уровне подзадач, sum на уровне слияния.

Вывод

Учитывая временные затраты на декомпозицию и слияние, параллельные вычисления позволяют сэкономить время только при больших размерах данных. Также большое значение структура нашего stream pipeline: промежуточные операции (map, filter, flatMap) выполняются параллельно, терминальная(reduce, collect, sum) совершает слияние. Поэтому, если объем работы, которую можно распараллелить достаточно большой, можно сэкономить на времени выполнения, если нет, тогда операция слияния может сделать параллельное выполнение еще более затратным.

Универсального рецепта нет. Для каждого отдельного случая нужно проводить измерения: measure, don't guess

Материалы:
Java 8 Lambdas.Functional Programming for the Masses by Richard Warburton