Параллельная обработка больших коллекций данных через QtConcurrent::map и reduce

Введение

Когда нужно применить одну и ту же операцию к каждому элементу большой коллекции (например, изменить размер тысяч изображений или провести вычисления над массивом данных), часто такая обработка естественно параллелится — элементы независимы друг от друга. QtConcurrent предоставляет готовые функции map(), mappedReduced() и их вариации, которые автоматически распределяют обработку коллекции по доступным ядрам процессора, избавляя от необходимости вручную делить данные между потоками.

Концепция

QtConcurrent::map(container, function) применяет function к каждому элементу container параллельно, изменяя элементы «на месте» (in-place), и возвращает QFuture<void>. QtConcurrent::mapped(container, function) похож, но создаёт новую коллекцию с результатами, не изменяя исходную. QtConcurrent::mappedReduced(container, mapFunction, reduceFunction) идёт дальше: после параллельного применения mapFunction к каждому элементу результаты последовательно «сворачиваются» в одно значение через reduceFunction — классическая модель map-reduce в миниатюре.

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

Пример кода

#include <QCoreApplication>
#include <QtConcurrent/QtConcurrent>
#include <QVector>
#include <QDebug>
#include <QElapsedTimer>
#include <cmath>

bool isPrime(int n)
{
    if (n < 2) return false;
    for (int i = 2; i * i <= n; ++i) {
        if (n % i == 0) return false;
    }
    return true;
}

int countDivisors(int n)
{
    int count = 0;
    for (int i = 1; i <= n; ++i) {
        if (n % i == 0) count++;
    }
    return count;
}

void sumReduce(int &result, int value)
{
    result += value;
}

int main(int argc, char *argv[])
{
    QCoreApplication app(argc, argv);

    QVector<int> numbers;
    for (int i = 1; i <= 50000; ++i) {
        numbers.append(i);
    }

    QElapsedTimer timer;
    timer.start();

    // mappedReduced: для каждого числа считаем число делителей,
    // затем параллельно суммируем все результаты в одно значение
    int totalDivisors = QtConcurrent::blockingMappedReduced<int>(
        numbers, countDivisors, sumReduce);

    qDebug() << "Сумма количества делителей:" << totalDivisors;
    qDebug() << "Затрачено мс:" << timer.elapsed();

    // filtered: параллельный отбор простых чисел
    QVector<int> primes = QtConcurrent::blockingFiltered(numbers, isPrime);
    qDebug() << "Найдено простых чисел:" << primes.size();

    return 0;
}

Пояснения к коду

QtConcurrent::blockingMappedReduced<int>() — блокирующая версия mappedReduced, которая сразу возвращает итоговый результат без необходимости работать с QFuture явно. Для каждого числа в numbers параллельно вызывается countDivisors, а результаты сворачиваются функцией sumReduce в общую сумму. Использование блокирующей версии оправдано здесь, поскольку весь смысл вызова — дождаться итогового агрегированного значения перед тем, как продолжить выполнение.

QtConcurrent::blockingFiltered демонстрирует ещё одну удобную функцию — параллельную фильтрацию коллекции по предикату isPrime, возвращающую только элементы, удовлетворяющие условию.

Подводные камни

• reduceFunction в mappedReduced должна быть потокобезопасной относительно общего результата, если не указан флаг QtConcurrent::OrderedReduce/UnorderedReduce явно регулирующий, как Qt сериализует вызовы свёртки — по умолчанию Qt гарантирует, что сама свёртка выполняется последовательно (не параллельно), но порядок применения может быть не таким, как порядок элементов в исходной коллекции, если не указан OrderedReduce.
• Слишком мелкая гранулярность задач. Если функция, применяемая к каждому элементу, очень лёгкая (например, простое сложение двух чисел), накладные расходы на распределение по потокам могут превысить выигрыш от параллелизма — QtConcurrent хорош для задач с заметной вычислительной стоимостью на элемент.
• Блокирующие версии (blockingMap, blockingFiltered и другие) недопустимы в GUI-потоке для длительных операций — они заморозят интерфейс точно так же, как обычный синхронный вызов; для интерфейсных приложений нужно использовать неблокирующие версии с QFutureWatcher.
• Изменение общих данных вне контейнера внутри функции map (например, запись в глобальную переменную из функции, переданной в QtConcurrent::map) нарушает предположение о независимости элементов и создаёт гонку данных, которую сам QtConcurrent не отслеживает и не предотвращает.