Нахождение числа простых чисел менее 10 миллионов с помощью многопоточной программы

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

Я создал последовательную программу, которая вычисляет количество простых чисел от 1 до 10 миллионов, и рассчитал ее 50 раз с помощью тестовой программы. Вот код для последовательной программы:

import java.util.Locale;
import java.text.NumberFormat;
/**
 * A serial program that calculates the number of primes less than a
 * certain number (which is hard coded). Used as a basis for
 * benchmarking the multi-threaded programs that do the same thing.
 *
 * @author Tirthankar Mazumder
 * @version 1.2
 * @date 2nd May, 2021
 */
public class PrimalityTestSerial {
    public static final long max = 10_000_000;
    public static void main(String[] args) {
        final long startTime = System.currentTimeMillis();
        
        long num_primes = primeCalculator();
        
        final long endTime = System.currentTimeMillis();
        
        NumberFormat nf = NumberFormat.getInstance(Locale.US);
         
        System.out.println("Number of primes less than " + nf.format(max) + ": " + num_primes);
        System.out.println("Took " + (endTime - startTime) + " ms.");
        System.out.println();
    }
    
    private static boolean isPrime(long l) {
        long upper_bound = (long) Math.floor(Math.sqrt(l));
        
        for (long i = 2; i <= upper_bound; i++) {
            if (l % i == 0)
                return false;
        }
        
        return true;
    }
    
    public static long primeCalculator() {
        long num_primes = 0;
        
        for (long i = 2; i <= max; i++) {
            if (isPrime(i))
                num_primes++;
        }
        return num_primes;
    }
}

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

/**
 * A worker class for calculating the number of primes from start to end,
 * which are private member variables. Instances of this class are used
 * in the multithreaded version of PrimalityTestSerial.
 *
 * @author Tirthankar Mazumder
 * @version 1.2
 * @date 3rd May, 2021
 */
public class PrimalityTestWorker1 implements Runnable {
    //Member variables
    public static long totalPrimeCount = 0;
    private final long start;
    private final long end;
    
    public PrimalityTestWorker1(long start, long end) {
        this.start = start;
        this.end = end;
    }
    
    private synchronized void increment(long num) {
        totalPrimeCount += num;
    }
    
    private static boolean isPrime(long l) {
        long upper_bound = (long) Math.floor(Math.sqrt(l));
        
        for (long i = 2; i <= upper_bound; i++) {
            if (l % i == 0)
                return false;
        }
        
        return true;
    }
    
    private void numPrimes() {
        long primeCount = 0;
        for (long i = start; i <= end; i++) {
            if (isPrime(i))
                primeCount++;
        }
        increment(primeCount);
    }
    
    public void run() {
        numPrimes();
        Thread.yield();
    }
}

Вот основная программа, которая использует экземпляры PrimalityTest1Worker в качестве потоков:

import java.util.Locale;
import java.text.NumberFormat;
/**
 * The master program for the multithreaded primality test that creates
 * objects of the PrimalityTestWorker1 to make threads, and then collates
 * the results and prints them to stdout.
 *
 * @author Tirthankar Mazumder
 * @version 1.0=2
 * @date 3rd May, 2021
 */
public class PrimalityTestParallel1Runner {
    public static final int cores = Runtime.getRuntime().availableProcessors();
    //We will spawn as many threads as there are cores on the system, and not
    //more than that because we are not I/O bound here.

    public static final long max = PrimalityTestSerial.max;
    //For consistency.

    public static void main(String[] args) {
        long startTime = System.currentTimeMillis();

        primeCalculator();

        long endTime = System.currentTimeMillis();
        
        NumberFormat nf = NumberFormat.getInstance(Locale.US);
        
        System.out.println("Number of primes less than " + nf.format(max) + ": " +
            PrimalityTestWorker1.totalPrimeCount);
            
        System.out.println("Took " + (endTime - startTime) + " ms.");
        System.out.println();
    }
    
    public static void primeCalculator() {
        Thread arrThreads[] = new Thread[cores];

        long chunk = max / cores;
        long threadStart = 2;
        long threadEnd = threadStart + chunk;

        for (int i = 0; i < cores; i++) {
            Thread t = new Thread(new PrimalityTestWorker1(threadStart, threadEnd));
            t.start();
            arrThreads[i] = t;

            threadStart = threadEnd + 1;
            threadEnd = (threadEnd + chunk > max) ? max : threadEnd + chunk;
        }

        for (int i = 0; i < cores; i++) {
            try {
                arrThreads[i].join(); 
            } catch (InterruptedException e) {
                System.out.println("Was interrupted.");
                return;
            }
        }
    }
}

Наконец, вот код программы тестирования, которая запускает каждую программу 50 раз, а затем вычисляет среднее время выполнения:

import java.util.Arrays;
/**
 * A wrapper class that handles benchmarking the performances of
 * PrimalityTestSerial and PrimalityTestParallel1Runner and then
 * prints information about the results to stdout.
 *
 * @author Tirthankar Mazumder
 * @version 1.0
 * @date 8th May, 2021
 */
public class PrimalityTestSuite {
    public static final int n = 50;
    //Number of test runs to perform
    
    public static void main(String[] args) {
        long totalSerialTime = 0;
        long totalParallelTime = 0;
        
        long serialTimes[] = new long[n];
        
        double avgSerialTime = 0;
        double avgParallelTime = 0;
        
        System.out.println("Starting Serial runs...");
        long startTime = System.currentTimeMillis();
        for (int i = 0; i < n; i++) {
            PrimalityTestSerial.primeCalculator();
            serialTimes[i] = System.currentTimeMillis();
        }
        
        for (int i = 0; i < n; i++) {
            serialTimes[i] -= startTime;
            for (int j = 0; j < i; j++) {
                serialTimes[i] -= serialTimes[j];
                //to get rid of the time taken by the previous runs
            }
            avgSerialTime += serialTimes[i];
        }
        
        avgSerialTime /= n;
        
        long parallelTimes[] = new long[n];
        
        System.out.println("Starting parallel runs...");
        startTime = System.currentTimeMillis();
        for (int i = 0; i < n; i++) {
            PrimalityTestParallel1Runner.primeCalculator();
            parallelTimes[i] = System.currentTimeMillis();
        }
        
        for (int i = 0; i < n; i++) {
            parallelTimes[i] -= startTime;
            for (int j = 0; j < i; j++) {
                parallelTimes[i] -= parallelTimes[j];
                //to get rid of the time taken by the previous runs
            }
            avgParallelTime += parallelTimes[i];
        }
        
        avgParallelTime /= n;
        
        Arrays.sort(serialTimes);
        Arrays.sort(parallelTimes);
        
        double bestThreeSerialAvg = (serialTimes[0] + serialTimes[1]
                                     + serialTimes[2]) / 3;
        double bestThreeParallelAvg = (parallelTimes[0] + parallelTimes[1]
                                     + parallelTimes[2]) / 3;
        
        System.out.println();
        System.out.println("Results:");
        
        System.out.println("Average of " + n + " Serial Runs: " + avgSerialTime + " ms.");
        System.out.println("Average of " + n + " Parallel Runs: " + avgParallelTime + " ms.");
        
        System.out.println();
        System.out.println("Average speed-up: " + avgSerialTime / avgParallelTime + "x");
        System.out.println();
        
        System.out.println("Average of best 3 Serial Runs: " + bestThreeSerialAvg + " ms.");
        System.out.println("Average of best 3 Parallel Runs: " + bestThreeParallelAvg + " ms.");
        
        System.out.println();
        System.out.println("Average speed-up (w.r.t. best run times): "
                            + bestThreeSerialAvg / bestThreeParallelAvg + "x");
        System.out.println();
    }
}

Вот результаты тестовой программы:

Starting Serial runs...
Starting parallel runs...

Results:
Average of 50 Serial Runs: 4378.92 ms.
Average of 50 Parallel Runs: 1529.2 ms.

Average speed-up: 2.8635364896678x

Average of best 3 Serial Runs: 4328.0 ms.
Average of best 3 Parallel Runs: 1297.0 ms.

Average speed-up (w.r.t. best run times): 3.3369313801079414x

Отсюда очевидно, что среднее ускорение просто примерно в 3 раза. Однако это удивительно, потому что я ожидаю, что многопоточная программа будет работать в 7-8 раз быстрее (поскольку последовательная программа использует только 1 ядро, тогда как многопоточная программа должна использовать все 8 ядер в моей системе).

Итак, мой вопрос: почему многопоточная программа работает не так быстро, как я ожидал?

threadEnd = (threadEnd + chunk > max) ? max : threadEnd + chunk;

Ваш primeCalculatorс одинакового размера куски вбегают неравное время. Призывы к primeCalculator для, скажем, 1-20 будут значительно быстрее, чем вызовы primeCalculator для, скажем, 10000-10020. Когда выполнено 1-20, 10000-10020 все еще будет работать, поэтому производительность не снижается до 1/8.

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

В крайнем случае, блоки имеют размер 1. Если вы знаете, сколько потоков будет (мы видим, что у вас есть cores потоков), то это довольно просто реализовать: сообщить каждому потоку t общее количество потоков n и начальное значение x чтобы проверить простоту, затем добавьте n после каждой проверки. То есть вместо предоставления threadStart а также threadEnd ценность, предоставьте threadStart Значение между 1 а также n, а threadIncrement значение (это просто n; в качестве альтернативы вы можете сделать это n глобальная переменная, которую могут видеть все потоки, поэтому вам не нужно передавать ее в рабочий конструктор). Затем в каждом потоке вы делаете что-то вроде этого:

x = threadStart;

while (x < max) {
    if (isPrime(x)) {
        // ...
    }
    x += n;
}

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

Также:

Если ваш процессор имеет SMT (например, Intel Hyper Threading), один поток на физическое ядро, вероятно, в основном максимально увеличивает пропускную способность целочисленного деления. (Составные числа с малыми простыми множителями будут означать, что вы относительно часто выходите из цикла, позволяя SMT выполнять свою работу, позволяя другому потоку делать больше, пока этот поток восстанавливается после неверного предсказания ветвления.)

Но все же с SMT ускорение в лучшем случае может масштабироваться только с количеством физических ядер, а не с количеством, которое вы определяете с помощью availableProcessors. Например, 4x на процессоре 4c8t (4 ядра 8 потоков), таком как i7-6700k, который ОС обнаружит как имеющий 8 логических ядер, но имеющий 4 физических.

(Деление — очень медленная операция и не полностью конвейерная даже в современных высокопроизводительных процессорах x86. Т.е. он не может начинать новое деление каждый такт, как это возможно для умножения, не говоря уже о том, чтобы делать несколько за тактовый цикл, как с add / sub. Например, Intel Skylake имеет пропускную способность в один idiv r32 за 6 циклов (https://uops.info/ / https://agner.org/optimize/). Он микрокодирован как 10 мопов, но за 6 циклов интерфейс с 4-ю шириной может выдать 24 мупа, так что есть много места для накладных расходов цикла test / jcc и inc / cmp / jcc в этом простом цикле, который просто запускает idiv несколько раз. делать знаковое целочисленное деление.)

Возможно, именно поэтому ваше неравномерное распределение работы, определенное @Peilonrayz, не повредит так сильно, как могло бы: после завершения «простого» потока любой поток, имеющий физическое ядро, может добиться более быстрого прогресса.

Эта зависимость от планирования потоков также может быть причиной того, что ваше среднее значение из трех лучших (3,33x) значительно лучше, чем ваше среднее значение ускорения в 2,86 раза. Наилучший случай возможен только тогда, когда последний (самый трудоемкий) поток разделяет ядро ​​с первым потоком (завершается первым). Я предполагаю тебя делать есть процессор с SMT. (Но я совершенно не уверен в этом предсказании.)

Кстати, неравномерная стоимость фрагментов — вот что OpenMP schedule(dynamic) для при использовании C с OpenMP для автоматического распараллеливания цикла. (https://stackoverflow.com/questions/10850155/whats-the-difference-between-static-and-dynamic-schedule-in-openmp). Разбейте проблему на более мелкие куски и пусть рабочий поток запускает новый каждый раз, когда вы его заканчиваете.

Предложение @Jivan Pal чередовать числа, которые вы проверяете, по потокам — это альтернативный способ сделать работу единообразной, если вы пропускаете четные числа! Каждое четное число больше 2 является составным. (OTOH, если вы настаиваете на грубой проверке каждого из них, тогда в идеале ОС будет планировать потоки, проверяющие четные числа, на те же ядра, что и реальная работа, поэтому они не будут так сильно конкурировать. Или, по крайней мере, эти 4 потока будут выходить из большого количества раньше, чем остальные, оставляя 4 потока проверки нечетного числа.)

Кстати, это нормально в качестве эксперимента / эталона для многопоточности, но если вы действительно хотите быстро подсчитывать простые числа, вы, безусловно, захотите улучшить алгоритм, который выполняется каждым потоком. Здесь есть тривиальный множитель 2, проверяя только нечетные числа как делители: for(int i=3 ; i < rootn ; i+=2). И рассматривая только нечетные числа как простые кандидаты во внешнем цикле. С другой стороны, вы поднимаетесь только до sqrt (n), что означает, что вы делаете гораздо меньше работы, чем еще более наивный i < n петли грубой силы.

Или гораздо лучший вариант, используйте решето из эратосфенов, и да, CodeReview.SE имеет для этого целый тег. (Но это труднее распараллелить.) Конечно, не имеет значения, насколько неэффективен ваш алгоритм, когда вы просто сравниваете относительное ускорение.

В общем, добавить больше потоков / ядер для решения проблемы иногда легко (для досадно параллельных задач, таких как проверка методом грубой силы), но повышение эффективности каждого потока также очень ценно. (И стоит меньше общих циклов процессора / энергии.) В частности, для этой проблемы просеивание на одном ядре должно быть быстрее, чем перебор на всех ядрах, даже с большим Xeon или, возможно, даже с кластером.

Или, как комментарии @qwr, даже более конкретная математика для простых чисел (https://en.wikipedia.org/wiki/Meissel%E2%80%93Lehmer_algorithm) позволяет подсчет их без находка их. Простые числа — это своего рода особый случай по сравнению с другими проблемами реального мира, учитывая, сколько умных вычислений было сделано для них — алгоритмические оптимизации даже больше, чем это часто возможно.

Для простых чисел объем работы зависит от $ O ( sqrt {n}) $, количество пробных отделов. Простые числа OTOH встречаются реже, чем больше числа: у большинства чисел есть несколько меньших факторов, поэтому вы не часто полностью $ sqrt {n} $ итераций.