На выходных я решил по изучать возможности стандартной библиотеки С++11. Завёл MSVC2012 и начал писать класс пула потоков с очередью заданий. В результате, в пока далеко не законченном варианте, получился примерно такой код:
ThreadPool.h:
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.
#pragma once
#include <vector>
#include <map>
#include <thread>
#include <atomic>
#include <memory>
#include <random>
#include <deque>
#include <future>
#include <mutex>
#include <condition_variable>
#include <iostream>
typedef std::tuple<std::packaged_task<int(void*,void*)>,void*,void*> ThreadPoolTask;
class ThreadPool
{
private:
std::vector< std::pair< std::thread,std::unique_ptr< std::atomic<bool> > > > threads;
static void threadProcess(ThreadPool *self, const std::atomic<bool> *const active);
std::mutex cvLockMutex;
std::condition_variable cvThreadAwait;
std::deque<ThreadPoolTask> tasks;
const bool getHeadTask(ThreadPoolTask& task);
public:
ThreadPool(const uint32_t& threadsNumber = 0);
virtual ~ThreadPool(void);
void startNew(const uint32_t& threadsNumber = 1);
void terminateAll();
const uint32_t getCount() const;
void addTask(ThreadPoolTask& task, const bool& queueBegin = false);
};
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.
#include "ThreadPool.h"
//-------------------------------------------------------------------------------------------------
ThreadPool::ThreadPool(const uint32_t& threadsNumber)
: threads(0)
{
if(threadsNumber)
this->startNew(threadsNumber);
}
//-------------------------------------------------------------------------------------------------
ThreadPool::~ThreadPool(void)
{
this->terminateAll();
}
//-------------------------------------------------------------------------------------------------
void ThreadPool::threadProcess(ThreadPool *self, const std::atomic<bool> *const active)
{
std::cout << "START - Thread ID # " << std::this_thread::get_id() << std::endl;
std::random_device randomDevice;
std::unique_lock<std::mutex> cvLock(self->cvLockMutex);
ThreadPoolTask task;
while(active->load())
{
if(self->getHeadTask(task))
{
std::cout << "Thread ID # " << std::this_thread::get_id() << std::endl;
std::get<0>(task).operator()(std::get<1>(task),std::get<2>(task));
std::this_thread::sleep_for(std::chrono::milliseconds(randomDevice()%1000));
}
else
{
// Перейти в ожидание
self->cvThreadAwait.wait(cvLock);
}
}
std::cout << "END - Thread ID # " << std::this_thread::get_id() << std::endl;
}
//-------------------------------------------------------------------------------------------------
void ThreadPool::startNew(const uint32_t& threadsNumber)
{
for(uint32_t i = 0; i < threadsNumber; ++i)
{
std::atomic<bool> *activitiFlag = new std::atomic<bool>(true);
this->threads.push_back( std::pair<std::thread,std::unique_ptr<std::atomic<bool>>>( std::thread(ThreadPool::threadProcess, this, activitiFlag) ,std::unique_ptr<std::atomic<bool>>(activitiFlag) ) );
}
}
//-------------------------------------------------------------------------------------------------
void ThreadPool::terminateAll()
{
for(auto& thread : this->threads)
thread.second->store(false);
this->cvThreadAwait.notify_all();
for(auto& thread : this->threads)
thread.first.join();
this->threads.clear();
}
//-------------------------------------------------------------------------------------------------
const uint32_t ThreadPool::getCount() const
{
return this->threads.size();
}
//-------------------------------------------------------------------------------------------------
//-------------------------------------------------------------------------------------------------
//-------------------------------------------------------------------------------------------------
void ThreadPool::addTask(ThreadPoolTask& task, const bool& queueBegin)
{
(queueBegin ? this->tasks.push_front(std::move(task)) : this->tasks.push_back(std::move(task)));
this->cvThreadAwait.notify_one();
}
//-------------------------------------------------------------------------------------------------
const bool ThreadPool::getHeadTask(ThreadPoolTask& task)
{
if(this->tasks.empty())
return false;
task = std::move(this->tasks.front());
this->tasks.pop_front();
return true;
}
//-------------------------------------------------------------------------------------------------
Для тестов использовалось консольное приложение:
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.
int funk(void*,void*)
{
wcout << L"\t\t\tРабота в потоке!" << endl;
return 0;
}
int main(/**/)
{
setlocale(LC_ALL,"RUS");
wcout << L"Старт программы" << endl;
ThreadPool tp;
tp.addTask(ThreadPoolTask(std::packaged_task<int(void*,void*)>(funk),nullptr,nullptr));
tp.startNew(5);
_getch();
tp.addTask(ThreadPoolTask(std::packaged_task<int(void*,void*)>(funk),nullptr,nullptr));
tp.addTask(ThreadPoolTask(std::packaged_task<int(void*,void*)>(funk),nullptr,nullptr));
_getch();
tp.addTask(ThreadPoolTask(std::packaged_task<int(void*,void*)>(funk),nullptr,nullptr));
tp.addTask(ThreadPoolTask(std::packaged_task<int(void*,void*)>(funk),nullptr,nullptr));
tp.addTask(ThreadPoolTask(std::packaged_task<int(void*,void*)>(funk),nullptr,nullptr));
_getch();
tp.addTask(ThreadPoolTask(std::packaged_task<int(void*,void*)>(funk),nullptr,nullptr));
tp.addTask(ThreadPoolTask(std::packaged_task<int(void*,void*)>(funk),nullptr,nullptr));
_getch();
tp.terminateAll();
_getch();
wcout << L"Завершение программы" << endl;
}
Всё работает отлично (не считая отсутствия синхронизации, в том числе при выводе в консоль, но это не законченный вариант).
Но вот только при вызове каждой группы
tp.addTask из консольной программы, вызов
this->cvThreadAwait.notify_one(); в методе
void ThreadPool::addTask(ThreadPoolTask& task, const bool& queueBegin) будит один и тот же поток! (привёл в файле картинку вывода для наглядности) Если вместо
notify_one() использовать
notify_all() , то, как и задумано, вызываются случайные потоки на каждое задание.
Что я делаю не верно, что вызывается один и тот же поток? При этом при загрузке следующей группы заданий возможно будет использоваться другой поток, но точно так же только он один. Или может я не правильно понимаю нормальное поведение
notify_one() ?
