Thread Pool C++ 11

병렬 처리를 위한 Thread Pool C++

 

요즘 코드를 보니, Thread Pool를 이용한 코드가 많이 보였다. 그래서 검색. 일단 예제 코드 keep, 컴파일하여 테스트해보려 했으나, VS2010에서는 동작 안한다고 한다. [참조] - 이유인 즉슨, Visual Studio 2010에서는 C++ 11이 구현되어 있지 않가고 한다. Visual Studio 2013에서 테스트한 결과 정상 동작 이를 잘 활용하려면, 람다 표현식에 대해 약간의 공부가 필요할 듯 함 ( 람다 표현식에 관한 정리 참조 ) 병렬처리 시 중요한 부분은 캡쳐블록임. 공유 변수에 대한 처리와 관련이 깊기 때문이다. 그리고, 기존 프로그램에 쉽게 적용하기 위해, 람다 표현식에서 멤버 함수를 지정할 수 있는 여러가지 방법 예제가 필요함.

ThreadPool.h

#ifndef THREAD_POOL_H
#define THREAD_POOL_H

#include <vector>
#include <queue>
#include <memory>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <future>
#include <functional>
#include <stdexcept>

class ThreadPool
{
public:
	ThreadPool(size_t capacity) : m_IsStop(false)
	{
		// the constructor just launches some amount of workers
		for(size_t i = 0; i < capacity; ++i)
		{
			m_Workers.push_back(std::thread([&]()
			{
				while(true)
				{
					std::unique_lock<std::mutex> lock(m_QueMutex);

					while(!m_IsStop && m_Tasks.empty())
					{
						m_Condition.wait(lock);
					}

					if (m_IsStop && m_Tasks.empty())  return;

					std::function<void()> task(m_Tasks.front());
					m_Tasks.pop();
					lock.unlock();

					task();
				}
			}));
		}
	}

	~ThreadPool()
	{
		{
			std::unique_lock<std::mutex> lock(m_QueMutex);
			m_IsStop = true;
		}

		m_Condition.notify_all();

		for(size_t i = 0; i < m_Workers.size(); ++i)
		{
			m_Workers[i].join();
		}
	}

	template <class _F, class ... _Args>
	auto Enqueue(_F&& f, _Args&& ... args)
		->std::future<decltype(std::forward<_F>(f)(
		std::forward<_Args>(args)...))>
	{
		typedef decltype(std::forward<_F>(f)(
			std::forward<_Args>(args)...)) return_type;

		// don't allow enqueueing after stopping the pool
		if (m_IsStop)
		{
			throw std::runtime_error("enqueue on stopped ThreadPool");
		}

		auto task = std::make_shared<std::packaged_task<return_type()>>
			(
			std::bind(std::forward<_F>(f), std::forward<_Args>(args)...)
			);

		std::future<return_type> result = task->get_future();

		{
			std::unique_lock<std::mutex> lock(m_QueMutex);
			m_Tasks.push([task](){ (*task)(); });
		}

		m_Condition.notify_one();

		return result;
	}

private:
	// need to keep track of threads so we can join them
	std::vector<std::thread> m_Workers;

	// the task queue
	std::queue<std::function<void()>> m_Tasks;

	// synchronization
	std::mutex m_QueMutex;
	std::condition_variable m_Condition;
	bool m_IsStop;
};


#endif

main.cpp

#include <iostream>
#include <chrono>
#include "ThreadPool.h"


int main(int argc, const char * argv[])
{
	ThreadPool pool(4);
	std::vector<std::future<int>> results;

	for(int i = 0; i < 4; ++i)
	{
		results.push_back(pool.Enqueue([i](){
			std::cout << "hello " << i << std::endl;
			std::this_thread::sleep_for(std::chrono::seconds(2));
			std::cout << "world " << i << std::endl;
			return i * i;
		}));
	}

	for(size_t i = 0; i < results.size(); ++i)
	{
		int value = results[i].get();
		std::cout << value << ' ';
	}

	std::cout << std::endl;

	return 0;
}

클래스 내에서 멤버함수를 사용하는 방법 #1

int sumfunc_static(int i)
{
	int sum = 0;
	for (int k = i * 100; k < i * 100 + 100; k++)
		sum += k;
	return sum;
}

int CMFCApplication1Dlg::sumfunc_memfunc(int i)
{
	int sum = 0;
	for (int k = i * 100; k < i * 100 + 100; k++)
		sum += k;
	return sum;
}

void CMFCApplication1Dlg::Test()
{
	ThreadPool pool(4);
	std::vector<std::future<int>> results;

	for (int i = 0; i < 20; ++i)
	{
		/*auto lamda = &CMFCApplication1Dlg::sumfunc_memfunc;
		(this->*lamda)(i);*/
		/*auto lamda = &sumfunc_static;
		lamda(i);*/
		
		results.push_back(pool.Enqueue([&,i](){ return sumfunc_memfunc(i); }));
	}
  
	for (size_t i = 0; i < results.size(); ++i)
	{
		int value = results[i].get();
		std::cout << value << std::endl;
	}

	std::cout << std::endl;

	return ;
}