QFuture Memoryleak

Question

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

Тестовая программа вычисляет что-то простое и работает до сих пор. Только использование памяти постоянно увеличивается.

Файл проекта QT:

QT -= gui
QT += concurrent widgets
CONFIG += c++11 console
CONFIG -= app_bundle
DEFINES += QT_DEPRECATED_WARNINGS
SOURCES += main.cpp

Файл программы QT:

#include <QCoreApplication>
#include <qdebug.h>
#include <qtconcurrentrun.h>

double parallel_function(int instance){
    return (double)(instance)*10.0;
}

int main(int argc, char *argv[])
{
    QCoreApplication a(argc, argv);
    int nr_of_threads = 8;
    double result_sum,temp_var;
    for(qint32 i = 0; i<100000000; i++){
      QFuture<double> * future = new QFuture<double>[nr_of_threads];

      for(int thread = 0; thread < nr_of_threads; thread++){
        future[thread] = QtConcurrent::run(parallel_function,thread);
      }
      for(int thread = 0; thread < nr_of_threads; thread++){
        future[thread].waitForFinished();
        temp_var = future[thread].result();
        qDebug()<<"result: " << temp_var;
        result_sum += temp_var;
      }
  }

  qDebug()<<"total: "<<result_sum;
  return a.exec();
}

Как я уже заметил, QtConcurrent::run(parallel_function,thread) выделяет память, но не освобождает память после future[thread].waitForFinished(),

Что здесь не так?

Answer 1

У вас утечка памяти, потому что future массив не удаляется. добавлять delete[] future в конце внешнего цикла.

for(qint32 i = 0; i<100000000; i++)
{
   QFuture<double> * future = new QFuture<double>[nr_of_threads];

   for(int thread = 0; thread < nr_of_threads; thread++){
     future[thread] = QtConcurrent::run(parallel_function,thread);
   }
   for(int thread = 0; thread < nr_of_threads; thread++){
      future[thread].waitForFinished();
      temp_var = future[thread].result();
      qDebug()<<"result: " << temp_var;
      result_sum += temp_var;
   }

   delete[] future; // <--
}

Answer 2

Вот как это может выглядеть - обратите внимание, насколько все проще! Вы полностью заняты ручным управлением памятью: почему? Прежде всего, QFuture это значение. Вы можете очень эффективно хранить его в любом векторном контейнере, который будет управлять памятью за вас. Вы можете выполнить итерацию такого контейнера, используя range-for. И т.п.

QT = concurrent   # dependencies are automatic, you don't use widgets
CONFIG += c++14 console
CONFIG -= app_bundle
SOURCES = main.cpp

Хотя пример синтетический и map_function Это очень просто, стоит подумать, как сделать все максимально эффективно и выразительно. Ваш алгоритм является типичной операцией уменьшения карты, и blockingMappedReduce имеет половину накладных расходов на ручное выполнение всей работы.

Прежде всего, давайте исправим исходную проблему в C++ вместо некоторых C-with-plus Франкенштейна.

// https://github.com/KubaO/stackrun/tree/master/questions/future-ranges-49107082
/* QtConcurrent will include QtCore as well */
#include <QtConcurrent>
#include <algorithm>
#include <iterator>

using result_type = double;

static result_type map_function(int instance){
   return instance * result_type(10);
}

static void sum_modifier(result_type &result, result_type value) {
   result += value;
}

static result_type sum_function(result_type result, result_type value) {
   return result + value;
}

result_type sum_approach1(int const N) {
   QVector<QFuture<result_type>> futures(N);
   int id = 0;
   for (auto &future : futures)
      future = QtConcurrent::run(map_function, id++);
   return std::accumulate(futures.cbegin(), futures.cend(), result_type{}, sum_function);
}

Нет ручного управления памятью и явного разделения на "потоки" - это было бессмысленно, поскольку платформа параллельного выполнения знает о количестве потоков. Так что это уже лучше!

Но это кажется довольно расточительным: каждое будущее внутренне выделяется хотя бы один раз (!).

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

#include <iterator>

template <typename tag> class num_iterator : public std::iterator<tag, int, int, const int*, int> {
   int num = 0;
   using self = num_iterator;
   using base = std::iterator<tag, int, int, const int*, int>;
public:
   explicit num_iterator(int num = 0) : num(num) {}
   self &operator++() { num ++; return *this; }
   self &operator--() { num --; return *this; }
   self &operator+=(typename base::difference_type d) { num += d; return *this; }
   friend typename base::difference_type operator-(self lhs, self rhs) { return lhs.num - rhs.num; }
   bool operator==(self o) const { return num == o.num; }
   bool operator!=(self o) const { return !(*this == o); }
   typename base::reference operator*() const { return num; }
};

using num_f_iterator = num_iterator<std::forward_iterator_tag>;

result_type sum_approach2(int const N) {
   auto results = QtConcurrent::blockingMapped<QVector<result_type>>(num_f_iterator{0}, num_f_iterator{N}, map_function);
   return std::accumulate(results.cbegin(), results.cend(), result_type{}, sum_function);
}

using num_b_iterator = num_iterator<std::bidirectional_iterator_tag>;

result_type sum_approach3(int const N) {
   auto results = QtConcurrent::blockingMapped<QVector<result_type>>(num_b_iterator{0}, num_b_iterator{N}, map_function);
   return std::accumulate(results.cbegin(), results.cend(), result_type{}, sum_function);
}

Можем ли мы бросить std::accumulate и использовать blockingMappedReduced вместо? Конечно:

result_type sum_approach4(int const N) {
   return QtConcurrent::blockingMappedReduced(num_b_iterator{0}, num_b_iterator{N},
                                              map_function, sum_modifier);
}

Мы также можем попробовать итератор с произвольным доступом:

using num_r_iterator = num_iterator<std::random_access_iterator_tag>;

result_type sum_approach5(int const N) {
   return QtConcurrent::blockingMappedReduced(num_r_iterator{0}, num_r_iterator{N},
                                              map_function, sum_modifier);
}

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

#include <numeric>

result_type sum_approach6(int const N) {
   QVector<int> sequence(N);
   std::iota(sequence.begin(), sequence.end(), 0);
   return QtConcurrent::blockingMappedReduced(sequence, map_function, sum_modifier);
}

Конечно, наша цель - сравнить все это:

template <typename F> void benchmark(F fun, double const N) {
   QElapsedTimer timer;
   timer.start();
   auto result = fun(N);
   qDebug() << "sum:" << fixed << result << "took" << timer.elapsed()/N << "ms/item";
}

int main() {
   const int N = 1000000;
   benchmark(sum_approach1, N);
   benchmark(sum_approach2, N);
   benchmark(sum_approach3, N);
   benchmark(sum_approach4, N);
   benchmark(sum_approach5, N);
   benchmark(sum_approach6, N);
}

На моей системе в сборке релиза вывод:

sum: 4999995000000.000000 took 0.015778 ms/item
sum: 4999995000000.000000 took 0.003631 ms/item
sum: 4999995000000.000000 took 0.003610 ms/item
sum: 4999995000000.000000 took 0.005414 ms/item
sum: 4999995000000.000000 took 0.000011 ms/item
sum: 4999995000000.000000 took 0.000008 ms/item

Обратите внимание, что использование map-Reduction для случайной итерируемой последовательности имеет более чем на 3 порядка меньшую нагрузку, чем использование QtConcurrent::run и на 2 порядка быстрее, чем неслучайно итерируемые решения.