странная утечка в многопоточном коде

@HamsterGamer · Регистрация: 21.06.2019

Студворк — интернет-сервис помощи студентам

Всем доброго времени суток, реализовал многопоточный вариант алгоритма quicksort с использованием пула потоков (чтобы он не создавал каждый проход рекурсией новый поток). Однако почему-то сам пул не умирает, причем даже в конце программы (кт кстати у меня завершается с кодом 0). То есть строка "DIE" в деструкторе пула не выводится, хотя пулом владеет умный shared_ptr, что еще больше вызывает у меня недоумение. Прошу помочь выяснить в чем же утечка! Заранее благодарю за помощь!

C++
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
#include <iostream>
#include <thread>
#include <future>
#include <atomic>
#include <condition_variable>
#include <functional>
#include <vector>
#include <cassert>
#include <queue>
 
struct policy_parallel_tag {} par;
struct policy_seq_tag {} seq;
 
struct thread_pool{
private:
    std::vector<std::thread> pool;
    std::queue<std::function<void()>> tasks;
    std::mutex m;
    std::condition_variable cv;
    std::atomic<bool> stop = false;
public:
    //inline static bool is_alive = false;
 
    explicit thread_pool(size_t pool_count = 16) {
        //is_alive = true;
        size_t const thread_count = std::min(
                (size_t)((std::thread::hardware_concurrency() - 1) ? std::thread::hardware_concurrency() - 1 : 2),
                pool_count);
        pool.reserve(thread_count);
        for (size_t i = 0; i < thread_count; i++){
            pool.emplace_back([this] {
                while (true) {
                    std::function<void()> current_task;
                    std::unique_lock unlk(m);
                    cv.wait(unlk, [&] {
                        return this->stop || !this->tasks.empty();
                    });
 
                    if (!this->stop && this->tasks.empty())
                        return;
 
                    if (!this->tasks.empty()) {
                        current_task = std::move(this->tasks.front());
                        this->tasks.pop();
                        unlk.unlock();
                        current_task();
                    }
                }
            });
        }
    }
 
    template <typename F, typename... Args>
    auto enterqueue(F&& func, Args&&... args) -> std::future<typename std::result_of<F(Args...)>::type> {
        using return_type = typename std::result_of<F(Args...)>::type;
        auto task = std::make_shared<std::packaged_task<return_type()>>
                (std::bind(std::forward<F>(func), std::forward<Args>(args)...));
        auto fut = task->get_future();
        {
            std::unique_lock lk(m);
            if (stop) throw std::logic_error("already stopped!");
            tasks.emplace([task](){(*task)();});
        }
        cv.notify_one();
        return fut;
    }
 
    ~thread_pool(){
        stop = true;
        cv.notify_all();
        for (auto & t : pool)
            t.join();
 
        //is_alive = false;
        std::cout << "DIE" << std::endl; // НЕ ВЫВОДИТСЯ
    }
};
 
template <class RandomIt>
void quicksort_impl_seq(RandomIt first, RandomIt last){
    if (first == last) {
        return;
    }
 
    auto pivot = first;
    auto lower_last_iter = std::partition(first, last, [&pivot](auto it){
        return it < *pivot;
    });
 
    quicksort_impl_seq(first, lower_last_iter);
    quicksort_impl_seq(++pivot, last);
}
 
template <class RandomIt>
void quicksort_impl_par(RandomIt first, RandomIt last, std::shared_ptr<thread_pool> tp){
    if (first == last) {
        return;
    }
 
    auto pivot = first;
    auto lower_last_iter = std::partition(first, last, [&pivot](auto it){
        return it < *pivot;
    });
 
    tp->enterqueue(quicksort_impl_par<RandomIt>, first, lower_last_iter, tp);
    tp->enterqueue(quicksort_impl_par<RandomIt>, ++pivot, last, tp);
}
 
template <class RandomIt, class Compare>
void quicksort_impl_seq(RandomIt first, RandomIt last, Compare comp){
    if (first == last) {
        return;
    }
 
    auto pivot = first;
    auto lower_last_iter = std::partition(first, last, [&pivot, &comp](auto it){
        return comp(it, *pivot);
    });
 
    quicksort_impl_seq(first, lower_last_iter, comp);
    quicksort_impl_seq(++pivot, last, comp);
}
 
template <class RandomIt, class Compare>
void quicksort_impl_par(RandomIt first, RandomIt last, Compare comp, std::shared_ptr<thread_pool> tp){
    if (first == last) {
        return;
    }
 
    auto pivot = first;
    auto lower_last_iter = std::partition(first, last, [&pivot, &comp](auto it){
        return comp(it, *pivot);
    });
 
    tp->enterqueue(quicksort_impl_par<RandomIt, Compare>, first, lower_last_iter, comp, tp);
    tp->enterqueue(quicksort_impl_par<RandomIt, Compare>, ++pivot, last, comp, tp);
}
 
template <class RandomIt, class SchemePolicy>
void pquicksort(RandomIt first, RandomIt last, SchemePolicy&& partition){
    if constexpr (std::is_same_v<std::decay_t<SchemePolicy>, policy_parallel_tag>) {
        std::shared_ptr<thread_pool> pool = std::make_shared<thread_pool>();                    // инициализация пула
        quicksort_impl_par(first, last, pool);
    } else if constexpr (std::is_same_v<std::decay_t<SchemePolicy>, policy_seq_tag>) {
        quicksort_impl_seq(first, last);
    }
}
 
template <class RandomIt, class Compare, class SchemePolicy>
void pquicksort(RandomIt first, RandomIt last, Compare comp, SchemePolicy&& partition){
    if constexpr (std::is_same_v<std::decay_t<SchemePolicy>, policy_parallel_tag>) {
        std::shared_ptr<thread_pool> pool = std::make_shared<thread_pool>();                    // инициализация пула
        quicksort_impl_par(first, last, comp, pool);
    } else if constexpr (std::is_same_v<std::decay_t<SchemePolicy>, policy_seq_tag>) {
        quicksort_impl_seq(first, last, comp);
    }
}
 
#include <random>
template <typename T>
void filler(std::vector<T> & l, size_t count){
    std::random_device rd;
    std::mt19937 mers(rd());
    std::uniform_int_distribution distrib(1, 1000);
    while (count--){
        l.push_back(distrib(mers));
    }
}
 
#include <chrono>
using namespace std::chrono_literals;
 
int main(){
    size_t const count = 100;
    std::vector<int> v;
    v.reserve(count);
    filler(v, count);
 
    pquicksort(v.begin(), v.end(), par);
    std::this_thread::sleep_for(1s);
//    while (thread_pool::is_alive) continue;
//    while (thread_pool::is_alive);
    for (auto & i : v){
        std::cout << i << "\n";
    }
    auto prev = v.begin();
    for (auto next = std::next(prev); next != v.end(); next++){
        assert(*prev <= *next);
        prev = next;
    }
}

Добавлено через 10 минут
UPDATE:
Проблема не в утечке и не в shared_ptr, почему-то такая перестановка работает:

C++
1
2
3
4
5
6
7
8
9
~thread_pool(){
        stop = true;
        cv.notify_all();
        //is_alive = false;
        std::cout << "DIE" << std::endl; // НЕ ВЫВОДИТСЯ
 
        for (auto & t : pool)
            t.join();
    }

Как-будто после всех join он не идет дальше, еще больше вопросов >_< ...

Добавлено через 7 минут
Хотя похоже сама идея с пулом потоков дурацкая, потому что с большими данными работает в разы хуже однопоточной. Нужно вероятно какой-то определенный минимальный размер подмассива для распараллеливания выставить, и в ином случае делать однопоточно

@AndryS1 · 31.01.2021, 04:55

Сообщение от HamsterGamer

кт кстати у меня завершается с кодом 0

у меня в дебаге abort вызывается в одном из потоков

Добавлено через 1 минуту
и вообще, зачем вам многопоточный вариант quicksort? уже есть std::sort с возможным распараллеливанием

Добавлено через 1 минуту

Сообщение от AndryS1

у меня в дебаге abort вызывается в одном из потоков

не всегда, но один раз из 5 точно

Code
1
2
Exception thrown at 0x00007FFAD8B13B29 in 000.exe: Microsoft C++ exception: std::system_error at memory location 0x0000005F9E7FCB38.
Debug Error!

@zayats80888 · 31.01.2021, 11:12

HamsterGamer, ваша основная функция потока безопасна по исключениям? Нет, но вы их не отлавливаете. Более того, вы их не отлавливаете в функциях задач, при том, что сами бросаете исключение.
Так же объясните, в чем смысл этого фрагмента:

Сообщение от HamsterGamer

C++
1
2
                    if (!this->stop && this->tasks.empty())
                        return;

Далее, в функции сортировки вы создаёте пул потоков и предаёте владение им самому себе(через задачу). Главный поток об этом ничего не знает, как он поймёт, что задачи выполнены?
Ну и самое главное:

Сообщение от HamsterGamer

C++
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
template <class RandomIt>
void quicksort_impl_par(RandomIt first, RandomIt last, std::shared_ptr<thread_pool> tp){
    if (first == last) {
        return;
    }
 
    auto pivot = first;
    auto lower_last_iter = std::partition(first, last, [&pivot](auto it){
        return it < *pivot;
    });
 
    tp->enterqueue(quicksort_impl_par<RandomIt>, first, lower_last_iter, tp);
    tp->enterqueue(quicksort_impl_par<RandomIt>, ++pivot, last, tp);
//                                                  ^ Что это? 
// И где собственно сортировка? std::partition не сортирует, а просто разбивает диапазон на две группы.
// Что будет, если first окажется максимальным элементом в диапазоне?
}

@HamsterGamer · 31.01.2021, 12:25 **[ТС]**

zayats80888,

Сообщение от zayats80888

Так же объясните, в чем смысл этого фрагмента:

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

Про pivot согласен, переделаю алгоритм.

Добавлено через 29 секунд
AndryS1, у меня тоже иногда код ошибки. Там где-то ub и я не понимаю, оно про алгоритм или про пул :c

@zayats80888 · 31.01.2021, 13:01

Сообщение от HamsterGamer

перейдут в эту ветку

В это ветку кода попасть не возможно, т.к.

Сообщение от HamsterGamer

stop станет равным true

А если он false, то tasks.empty() не может быть true, т.к. в этом случае поток будет сидеть в cv.wait(...).
Т.е. функция потока никогда не завершится.

Сообщение от HamsterGamer

Когда в деструкторе

Какой поток отвечает за уничтожение пула? Может ли поток присоединить(join) сам себя?

И ещё:

Сообщение от HamsterGamer

C++
1
2
3
4
auto pivot = first;
    auto lower_last_iter = std::partition(first, last, [&pivot, &comp](auto it){
        return comp(it, *pivot);
    });

*pivot по ходу алгоритма будет меняться.

@oleg-m1973 · 31.01.2021, 13:16

Сообщение от HamsterGamer

if (!this->stop && this->tasks.empty())
return;

Вместо этого надо

C++
1
2
if (this->stop)
      return;

Добавлено через 3 минуты

Сообщение от HamsterGamer

~thread_pool(){
stop = true;
cv.notify_all();

stop = true здесь должно выполняться под блокировкой, мьютекс m

Добавлено через 3 минуты

Сообщение от HamsterGamer

tp->enterqueue(quicksort_impl_par<RandomIt> , first, lower_last_iter, tp);
tp->enterqueue(quicksort_impl_par<RandomIt> , ++pivot, last, tp);

У тебя enterqueue возвращает std::future, которая в деструкторе будет ждать, пока поток не завершиться. Соответственно, здесь никакой многопоточности не будет. Проверь.

Добавлено через 1 минуту
Вот здесь у меня есть реализация примерно такого же пула потоков https://www.cyberforum.ru/blog... g5939.html

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

@zayats80888 · 31.01.2021, 13:24

Сообщение от oleg-m1973

У тебя enterqueue возвращает std::future, которая в деструкторе будет ждать, пока поток не завершиться.

Это справедливо только для future, возвращённое std::async. В данном случае блокировки не будет.

@HamsterGamer · 31.01.2021, 14:58 **[ТС]**

zayats80888, oleg-m1973, вот что исправил по вашим советам:

C++
1
2
3
4
5
6
7
8
9
10
11
12
13
 ~thread_pool(){
        {
            std::lock_guard lk(m);
            stop = true;
            cv.notify_all();
        }
 
        for (auto & t : pool)
            t.join();
 
        is_alive = false;
        std::cout << "DIE\n";
    }

и

C++
1
2
                    if (this->stop)
                        return;

Но проблемы не решило, есть подозрение что какой-то поток сам себя join'ит или еще что-то, на линуксе кидает исключение: Resource deadlock avoided, а также 5 data race'ов и все они завязаны на строке создания пула, как-будто его создает несколько потоков, хотя там только один из main'а. Я ничего не понимаю, буду тестить :|

@oleg-m1973 · 31.01.2021, 15:14

Сообщение от HamsterGamer

zayats80888, oleg-m1973, вот что исправил по вашим советам:

C++
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
pool.emplace_back([this] ()
{
    auto wait = []() -> std::function<void()>
    {
        std::unique_lock lock(m);
        while (!this->stop)
        {
            if (!tasks.empty())
            {
                auto fn = std::move(this->tasks.front());
                this->tasks.pop();
                return fn;
            }
            cv.wait(lock);
        }
 
        return {};
    };
 
    for (;;)
    {
        auto task = wait();
        if (!task)
            break;
 
        task();
    }
});

Добавлено через 4 минуты
Ну и - вполне вероятно, что пул потоков работает нормально, а твоя сортировка тупо зацикливается

@zayats80888 · 31.01.2021, 16:18

Сообщение от HamsterGamer

есть подозрение что какой-то поток сам себя join'ит или еще что-то

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

Сообщение от zayats80888

в функции сортировки вы создаёте пул потоков и предаёте владение им самому себе(через задачу)

Т.е. уничтожать пул будет сам себя. Когда останется одна последняя задача(которая владеет этим пулом через последний shared_ptr), то в функции потока, при выходе из скоупа объект current_task запустит деструктор пула. Далее в деструкторе он попытается присоединить сам себя(это же один из потоков пула). Но даже если вы это учтёте, то после уничтожения пула, функция этого потока не завершиться, а пойдет на новую итерацию, где попытается заблокировать несуществующий mutex и т.д. Что бы этого избежать, в самом простом случае, функции потока нужно владеть неким разделяемым состоянием(флагом, который не будет уничтожен при уничтожении пула, например), что бы узнать, что пул уже уничтожен.

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

@oleg-m1973 · 31.01.2021, 16:22

Сообщение от zayats80888

Т.е. уничтожать пул будет сам себя. Когда останется одна последняя задача(которая владеет этим пулом через последний shared_ptr), то в функции потока, при выходе из скоупа объект current_task запустит деструктор пула.

Кстати, да. HamsterGamer, что это вообще за хрень с shared_ptr? И зачем ты вообще создаёшь пул потоков в функции сортировки? Он должен быть глобальными, по отношению к ней.

@HamsterGamer · 31.01.2021, 16:31 **[ТС]**

oleg-m1973, zayats80888, ок понял. Я не мог понять как это он сам себя пытается уничтожить, а сейчас мне стало ясно что деструктор последнего shared_ptr будет вызываться непосредственно в таске одного из потоков. Не знаю насколько это очевидная ошибка, но я как-то ее один вообще не заметил. Большое спасибо!

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

Хотя наверное сделать что-то типо этого (но в таком случае это уже не простой quick_sort, а суп какой-то, где помимо контейнера нужно еще и пул делать):

C++
1
2
3
4
5
6
    std::chrono::time_point time = std::chrono::steady_clock::now();
    {
        thread_pool pool;
        pquicksort(v.begin(), v.end(), par, pool);
    }
    auto duration1 = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - time).count();

@oleg-m1973 · 31.01.2021, 16:45

Сообщение от HamsterGamer

Хотя наверное сделать что-то типо этого (но в таком случае это уже не простой quick_sort, а суп какой-то, где помимо контейнера нужно еще и пул делать):

Сделай хотя бы вот так

C++
1
2
3
4
5
6
7
8
9
10
11
template <class RandomIt, class Compare, class SchemePolicy>
void pquicksort(RandomIt first, RandomIt last, Compare comp, SchemePolicy&& partition)
{
    if constexpr (std::is_same_v<std::decay_t<SchemePolicy>, policy_parallel_tag>) 
    {
        static thread_pool pool{}; // инициализация пула
        quicksort_impl_par(first, last, comp, pool);
    } 
    else if constexpr (std::is_same_v<std::decay_t<SchemePolicy>, policy_seq_tag>) 
        quicksort_impl_seq(first, last, comp);
}

Добавлено через 3 минуты
Но только это тоже будет работать дико медленно. Чтобы работало хотя бы также, как параллельный std::sort, нужно чтобы в твоей функции сортировки не запускалось никаких потоков и не выделялось никакой динамической памяти.

Новые блоги и статьи Все статьи Все блоги /
Инструменты COM: Сохранение данный из VARIANT в файл и загрузка из файла в VARIANT bedvit 28.01.2026 Сохранение базовых типов COM и массивов (одномерных или двухмерных) любой вложенности (деревья) в файл, с возможностью выбора алгоритмов сжатия и шифрования. Часть библиотеки BedvitCOM Использованы. . .	Загрузка PNG с альфа-каналом на SDL3 для Android: с помощью SDL_LoadPNG (без SDL3_image) 8Observer8 28.01.2026 Содержание блога SDL3 имеет собственные средства для загрузки и отображения PNG-файлов с альфа-каналом и базовой работы с ними. В этой инструкции используется функция SDL_LoadPNG(), которая. . .	Загрузка PNG с альфа-каналом на SDL3 для Android: с помощью SDL3_image 8Observer8 27.01.2026 Содержание блога SDL3_image - это библиотека для загрузки и работы с изображениями. Эта пошаговая инструкция покажет, как загрузить и вывести на экран смартфона картинку с альфа-каналом, то есть с. . .	влияние грибов на сукцессию anaschu 26.01.2026 Бифуркационные изменения массы гриба происходят тогда, когда мы уменьшаем массу компоста в 10 раз, а скорость прироста биомассы уменьшаем в три раза. Скорость прироста биомассы может уменьшаться за. . .
Воспроизведение звукового файла с помощью SDL3_mixer при касании экрана Android 8Observer8 26.01.2026 Содержание блога SDL3_mixer - это библиотека я для воспроизведения аудио. В отличие от инструкции по добавлению текста код по проигрыванию звука уже содержится в шаблоне примера. Нужно только. . .	Установка Android SDK, NDK, JDK, CMake и т.д. 8Observer8 25.01.2026 Содержание блога Перейдите по ссылке: https:/ / developer. android. com/ studio и в самом низу страницы кликните по архиву "commandlinetools-win-xxxxxx_latest. zip" Извлеките архив и вы увидите. . .	Вывод текста со шрифтом TTF на Android с помощью библиотеки SDL3_ttf 8Observer8 25.01.2026 Содержание блога Если у вас не установлены Android SDK, NDK, JDK, и т. д. то сделайте это по следующей инструкции: Установка Android SDK, NDK, JDK, CMake и т. д. Сборка примера Скачайте. . .	Использование SDL3-callbacks вместо функции main() на Android, Desktop и WebAssembly 8Observer8 24.01.2026 Содержание блога Если вы откроете примеры для начинающих на официальном репозитории SDL3 в папке: examples, то вы увидите, что все примеры используют следующие четыре обязательные функции, а. . .

странная утечка в многопоточном коде

Решение

Решение