Алгоритм поиска префиксов с использованием многоядерных

Question

У меня есть задача отфильтровать список (вектор) из слов по префиксу.Алгоритм должен использовать современные многоядерные процессоры.

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

//      PrintWriter writer = new PrintWriter("C:\\DemoList.txt", "UTF-8");
//      
//      for(char i = 'A'; i<= 'Z'; i++) {
//          for(char j = 'A'; j<= 'Z'; j++) {
//              for(char n = 'A'; n<= 'Z'; n++) {
//                  for(char m = 'A'; m<= 'Z'; m++) {
//                      writer.println("" + i + j + n + m );
//                  }
//                      
//              }
//          }
//      }
    List<String> allLines = Files.readAllLines(Paths.get("C:\\", "DemoList.txt"));
    Collections.shuffle(allLines);
    String pattern = "AA";

    List<String> result = new ArrayList<>();
    int cores = Runtime.getRuntime().availableProcessors();
    int threadsNum = allLines.size() / cores;

    long start_time = System.nanoTime();

    for (String word : allLines) {
        if (word.startsWith(pattern))
            result.add(word);

    }

    long end_time = System.nanoTime();
    double difference = (end_time - start_time) / 1e6;
    System.out.println("Time difference in Milliseconds with Brute-Force: " + difference);

//With Parallisim:
    long new_start_time = System.nanoTime();

    List<String> filteredList = allLines.parallelStream().filter(s -> s.startsWith(pattern))
            .collect(Collectors.toList());

    long new_end_time = System.nanoTime();

    double new_difference = (new_end_time - new_start_time) / 1e6;
    System.out.println("Time difference in Milliseconds with Stream from Java 8: " + new_difference);   

Результат: Разница во времени в миллисекундах с помощью Brute-Force: 33.033602 Разница во времени в миллисекундах с Stream от Java 8: 65.017069

Каждый поток должен фильтровать диапазон из списка.

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

Answer 1

Из вашего примера кода ваш метод измерения времени граничит с Micro Benchmarking, для которого простое измерение времени для одного выполнения вводит в заблуждение.

Вы можете подробно обсудить это в следующем посте StackOverflow: Как написать правильный микробанчмарк в Java?

Я написал более точный тест для получения более точного измерения вашего примера кода. Код работает на QuadCore i7 с многопоточностью (но это ноутбук, из-за управления питанием и теплом, результаты могут быть слегка смещены по отношению к многопоточному коду, который производит больше тепла).

@Benchmark
public void testSequentialFor(Blackhole bh, Words words) {
    List<String> filtered = new ArrayList<>();
    for (String word : words.toSort) {
        if (word.startsWith(words.prefix)) {
            filtered.add(word);
        }
    }
    bh.consume(filtered);
}

@Benchmark
public void testParallelStream(Blackhole bh, Words words) {
    bh.consume(words.toSort.parallelStream()
            .filter(w -> w.startsWith(words.prefix))
            .collect(Collectors.toList())
    );
}

@Benchmark
public void testManualThreading(Blackhole bh, Words words) {
    // This is quick and dirty, bit gives a decent baseline as to
    // what a manually threaded partitionning can achieve.
    List<Future<List<String>>> async = new ArrayList<>();
    // this has to be optimized to avoid creating "almost empty" work units
    int batchSize = words.size / ForkJoinPool.commonPool().getParallelism();
    int numberOfDispatchedWords = 0;
    while (numberOfDispatchedWords < words.toSort.size()) {
        int start = numberOfDispatchedWords;
        int end = numberOfDispatchedWords + batchSize;
        async.add(words.threadPool.submit(() -> {
            List<String> batch = words.toSort.subList(start, Math.min(end, words.toSort.size()));
            List<String> result = new ArrayList<>();
            for (String word : batch) {
                if (word.startsWith(words.prefix)) {
                    result.add(word);
                }
            }
            return result;
        }));
        numberOfDispatchedWords += batchSize;
    }
    List<String> result = new ArrayList<>();
    for (Future<List<String>> asyncResult : async) {
        try {
            result.addAll(asyncResult.get());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    bh.consume(result);
}

@State(Scope.Benchmark)
public static class Words {

    ExecutorService threadPool = ForkJoinPool.commonPool();

    List<String> toSort;

    @Param({"100", "1000", "10000", "100000"})
    private int size;

    private String prefix = "AA";

    @Setup
    public void prepare() {
        //a 4 to 13 letters long, random word
        //for more precision, it should not be that random (use a fixed seed), but given the simple nature of the fitlering, I guess it's ok this way
        Supplier<String> wordCreator = () -> RandomStringUtils.random(4 + ThreadLocalRandom.current().nextInt(10));
        toSort = Stream.generate(wordCreator).limit(size).collect(Collectors.toList());
    }
}

Вот результаты

Benchmark                     (size)   Mode  Cnt        Score       Error  Units
PerfTest.testManualThreading     100  thrpt    6    95971,811 ±  1100,589  ops/s
PerfTest.testManualThreading    1000  thrpt    6    76293,983 ±  1632,959  ops/s
PerfTest.testManualThreading   10000  thrpt    6    34630,814 ±  2660,058  ops/s
PerfTest.testManualThreading  100000  thrpt    6     5956,552 ±   529,368  ops/s
PerfTest.testParallelStream      100  thrpt    6    69965,462 ±   451,418  ops/s
PerfTest.testParallelStream     1000  thrpt    6    59968,271 ±   774,859  ops/s
PerfTest.testParallelStream    10000  thrpt    6    29079,957 ±   513,244  ops/s
PerfTest.testParallelStream   100000  thrpt    6     4217,146 ±   172,781  ops/s
PerfTest.testSequentialFor       100  thrpt    6  3553930,640 ± 21142,150  ops/s
PerfTest.testSequentialFor      1000  thrpt    6   356217,937 ±  7446,137  ops/s
PerfTest.testSequentialFor     10000  thrpt    6    28894,748 ±   674,929  ops/s
PerfTest.testSequentialFor    100000  thrpt    6     1725,735 ±    13,273  ops/s

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

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

Я бы не стал сначала пытаться сортировать, затем выполнять бинарный поиск, поскольку фильтрация - это операция O (N), а сортировка O (Nlog (N)) (поверх которой вы должны добавить бинарный поиск). Поэтому, если у вас нет очень точной схемы данных, я не вижу, чтобы она работала в ваших интересах.

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

Answer 2

Начиная с Java 8, вы можете использовать потоки для решения проблемы в несколько строк:

List<String> yourList = new ArrayList<>(); // A list whose size requires parallelism
String yourPrefix = ""; // Your custom prefix
final List<String> filteredList = yourList.parallelStream()
               .filter(s -> s.startsWith(yourPrefix))
               .collect(Collectors.toList());

Я предлагаю вам это чтение и посмотреть на этот вопрос чтобы понять, поможет ли вам параллелизм или нет.