threadpool.cpp

#pragma 
#include <iostream>
#include <thread>
#include <mutex>
#include <future>
#include <condition_variable>
#include <deque>

namespace Jen 
{
	class TaskPool
	{
	private:
		using func = std::function<void()>;
		std::deque<func> tasks;
		std::condition_variable cv;
		
		bool is_empty()
		{
			bool ret;
			{
				std::unique_lock<std::mutex> lock(task_mutex);
				ret = tasks.empty();
			}
			return ret;
		}
	public:
		std::mutex task_mutex;

		template <typename F>
		void add_task(F f)
		{
			try{
				std::unique_lock<std::mutex> lg(task_mutex);
				tasks.push_back(func(f));
			}
			catch (std::exception e)
			{
				std::cerr << "add task : " << e.what() << std::endl;
			}
			cv.notify_one();
		}

		func get_task()
		{
			func ret;
			try{
				std::unique_lock<std::mutex> lg(task_mutex);
				ret = tasks.front();
				tasks.pop_front();

			}
			catch (std::exception e)
			{
				std::cerr << "get task : " << e.what() << std::endl;
			}
			return ret;
		}

		void wait_task()
		{
			if (is_empty())
			{
				std::unique_lock<std::mutex> lock(task_mutex, std::defer_lock);
				cv.wait(lock, [] {return true; });
			}
		}
	};

	class Worker
	{
	public:
		Worker(TaskPool &s, std::shared_future<void> sf) : pool(s), end(sf) { }
		void operator()();
	private:
		TaskPool &pool;
		std::shared_future<void> end;
	};

	void Worker::operator()()
	{
		try {
			std::function<void()> task;
			while (true)
			{
				{
					std::unique_lock<std::mutex> lg(pool.task_mutex);
					while (end.valid())
					{
						pool.wait_task();
						task = pool.get_task();
					}
				}

				if (!end.valid())
				{
					return;
				}

				task();
			}
		}
		catch (std::exception e)
		{
 			std::cerr << "worker ()operator : " << e.what() << std::endl;
		}
	}

	class ThreadPool
	{
	private:
		std::vector<std::thread> Threads;
		std::promise<void> pr;
	public:
		ThreadPool(TaskPool& tp, const int pool_size)
		{
			std::shared_future<void> sf = std::shared_future<void>(pr.get_future());
			for (int i = 0; i < pool_size; i++)
			{
				Threads.push_back(std::thread(Worker(tp, sf)));
			}
		}

		ThreadPool::~ThreadPool()
		{
			pr.set_value();

			// join them
			for (size_t i = 0; i<Threads.size(); ++i)
				Threads[i].join();
		}
	};
}


void hahah()
{
	int i = 0;
	while (i < 1000000000)
	{
		++i;
	}
	std::cout << "task fin" << std::endl;
}

int main()
{
	Jen::TaskPool taskp;
	Jen::ThreadPool tp(taskp, 2);
	for (int i = 0; i < 4; i++)
	{
		taskp.add_task(hahah);
	}
}

worker ()operator : device or resource busy: device or resource busy
worker ()operator : device or resource busy: device or resource busy
계속하려면 아무 키나 누르십시오 . . .