Как мне запустить что-то параллельное в Java?

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

Но есть и другая возможность; вообще не используйте массивы. Вы можете хранить комбинации в значениях int, поскольку каждое значение int представляет собой комбинацию битов. Это не только экономит много памяти, но вы также можете легко перебирать все возможные комбинации, просто увеличивая значение, так как перебор всех int чисел также перебирает все возможные комбинации битов¹. Единственное, что нам нужно реализовать, — это сгенерировать правильную строку для конкретного значения int, интерпретируя биты как числа в соответствии с их положением.

Для первой попытки мы можем пойти по простому пути и использовать уже существующие классы:

public static void main(String[] args) {
    long t0 = System.nanoTime();
    combinations(10, 25);
    long t1 = System.nanoTime();
    System.out.println((t1 - t0)/1_000_000+" ms");
    System.out.flush();
}
static void combinations(int start, int end) {
    for(int i = 1, stop = (1 << (end - start)) - 1; i <= stop; i++) {
        System.out.println(
            BitSet.valueOf(new long[]{i}).stream()
                  .mapToObj(b -> String.valueOf(b + start))
                  .collect(Collectors.joining(", ", "[", "]"))
        );
    }
}

Метод использует эксклюзивный конец, поэтому для вашего примера вам нужно вызвать его как combinations(0, 3), и он напечатает

[0]
[1]
[0, 1]
[2]
[0, 2]
[1, 2]
[0, 1, 2]
3 ms

^{конечно, сроки могут отличаться}

Для приведенного выше примера combinations(10, 25) он печатает все комбинации, за которыми следует 3477 ms на моей машине. Это звучит как возможность для оптимизации, но мы должны сначала подумать о том, какие операции требуют каких затрат.

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

Давайте определим долю стоимости, создав новый PrintStream, отключив автоочистку при разрывах строк и используя безумно большой буфер, способный хранить весь вывод:

public static void main(String[] args) {
    System.setOut(new PrintStream(
        new BufferedOutputStream(new FileOutputStream(FileDescriptor.out),1<<20),false));
    long t0 = System.nanoTime();
    combinations(10, 25);
    long t1 = System.nanoTime();
    System.out.flush();
    long t2 = System.nanoTime();
    System.out.println((t1 - t0)/1_000_000+" ms");
    System.out.println((t2 - t0)/1_000_000+" ms");
    System.out.flush();
}
static void combinations(int start, int end) {
    for(int i = 1, stop = (1 << (end - start)) - 1; i <= stop; i++) {
        System.out.println(
            BitSet.valueOf(new long[]{i}).stream()
                  .mapToObj(b -> String.valueOf(b + start))
                  .collect(Collectors.joining(", ", "[", "]"))
        );
    }
}

На моей машине он печатает что-то в порядке

93 ms
3340 ms

Показано, что код потратил более трех секунд на непараллелизуемую печать и всего около 100 миллисекунд на расчет. Для полноты следующий код переходит на уровень ниже для поколения String:

static void combinations(int start, int end) {
    for(int i = 1, stop = (1 << (end - start)) - 1; i <= stop; i++) {
        System.out.println(bits(i, start));
    }
}
static String bits(int bits, int offset) {
    StringBuilder sb = new StringBuilder().append('[');
    for(;;) {
        int bit = Integer.lowestOneBit(bits), num = Integer.numberOfTrailingZeros(bit);
        sb.append(num + offset);
        bits -= bit;
        if(bits == 0) break;
        sb.append(", ");
    }
    return sb.append(']').toString();
}

что вдвое сокращает время вычислений на моей машине, не оказывая при этом заметного влияния на общее время, что сейчас не должно вызывать удивления.

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

Последовательный код уже привел задачу к форме, которая сводится к итерации от начального значения к конечному значению. Теперь мы переписываем этот код в ForkJoinTask (или подходящий подкласс), который представляет собой итерацию с начальным и конечным значением. Затем мы добавляем возможность разделить эту операцию на две, разделив диапазон посередине, чтобы получить две задачи, повторяющиеся в каждой половине диапазона. Это можно повторять до тех пор, пока мы не решим, что у нас достаточно потенциально параллельных заданий, и выполняем текущую итерацию локально. После локальной обработки мы должны дождаться завершения любой задачи, которую мы выделили, чтобы гарантировать, что завершение корневой задачи подразумевает завершение всех подзадач.

public class Combinations extends RecursiveAction {
    public static void main(String[] args) {
        System.setOut(new PrintStream(new BufferedOutputStream(
            new FileOutputStream(FileDescriptor.out),1<<20),false));
        ForkJoinPool pool = (ForkJoinPool) Executors.newWorkStealingPool();
        long t0 = System.nanoTime();
        Combinations job = Combinations.get(10, 25);
        pool.execute(job);
        job.join();
        long t1 = System.nanoTime();
        System.out.flush();
        long t2 = System.nanoTime();
        System.out.println((t1 - t0)/1_000_000+" ms");
        System.out.println((t2 - t0)/1_000_000+" ms");
        System.out.flush();
    }

    public static Combinations get(int min, int max) {
        return new Combinations(min, 1, (1 << (max - min)) - 1);
    }

    final int offset, from;
    int to;

    private Combinations(int offset, int from, int to) {
        this.offset = offset;
        this.from = from;
        this.to = to;
    }

    @Override
    protected void compute() {
        ArrayDeque<Combinations> spawned = new ArrayDeque<>();
        while(getSurplusQueuedTaskCount() < 2) {
            int middle = (from + to) >>> 1;
            if(middle == from) break;
            Combinations forked = new Combinations(offset, middle, to);
            forked.fork();
            spawned.addLast(forked);
            to = middle - 1;
        }
        performLocal();
        for(;;) {
            Combinations forked = spawned.pollLast();
            if(forked == null) break;
            if(forked.tryUnfork()) forked.performLocal(); else forked.join();
        }
    }

    private void performLocal() {
        for(int i = from, stop = to; i <= stop; i++) {
            System.out.println(bits(i, offset));
        }
    }

    static String bits(int bits, int offset) {
        StringBuilder sb = new StringBuilder().append('[');
        for(;;) {
            int bit=Integer.lowestOneBit(bits), num=Integer.numberOfTrailingZeros(bit);
            sb.append(num + offset);
            bits -= bit;
            if(bits == 0) break;
            sb.append(", ");
        }
        return sb.append(']').toString();
    }
}

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

Есть два способа разделения. Примеры часто создают две или более разветвленных подзадач с последующим их объединением. Это может привести к большому количеству задач, которые просто ждут других. Альтернативой является разветвление подзадачи и изменение текущей задачи для представления другой. Здесь разветвленная задача представляет диапазон [middle, to], тогда как текущая задача изменена для представления диапазона [from, middle].

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

Это работает плавно, но, к сожалению, как и ожидалось, не ускоряет работу, так как самая дорогая часть — это печать.

¹ Использование int для представления всех комбинаций уменьшает поддерживаемую длину диапазона до 31, но имейте в виду, что такая длина диапазона подразумевает 2³¹ - 1 комбинаций, что довольно много для повторения. Если это все еще кажется ограничением, вы можете изменить код, чтобы использовать вместо него long. Поддерживаемой тогда длины диапазона 63, другими словами 2⁶³ - 1 комбинаций, достаточно, чтобы компьютер был занят до скончания века.