Kezunlin's Blog

Live and Learn

0%

how to implement thread pool with boost in c++

Posted on Edited on In

Thread Pool with Boost

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
#include <iostream>    //std::cout std::endl
#include <thread>      //std::thread
#include <future>      //std::future std::promise
#include <utility>     //std::ref
#include <chrono>      //std::chrono::seconds

#include <boost/thread.hpp>
#include <boost/bind.hpp>
#include <boost/asio.hpp>

class ThreadPool {
public:
	explicit ThreadPool(size_t size) : work_(io_service_) {
		for (size_t i = 0; i < size; ++i) {
			workers_.create_thread(
				boost::bind(&boost::asio::io_service::run, &io_service_));
		}
	}

	~ThreadPool() {
		std::cout << "~ThreadPool" << std::endl;
		io_service_.stop(); // stop before join_all
		workers_.join_all();
	}

	// Add new work item to the pool.
	template<class F>
	void Enqueue(F f) {
		io_service_.post(f);
		//sync, return immediately
	}

private:
	boost::thread_group workers_;
	boost::asio::io_service io_service_;
	boost::asio::io_service::work work_;
};

boost::mutex io_mutex;

void count(int id)
{
	for (int i = 0; i < 10; i++)
	{
		boost::mutex::scoped_lock lock(io_mutex);
		std::cout << id << ":" << i << std::endl;
	}
}

void test_thread()
{
	boost::thread thrd1(boost::bind(&count, 1));
	boost::thread thrd2(boost::bind(&count, 2));
	thrd1.join();
	thrd2.join();
}

void print(int i)
{
	boost::mutex::scoped_lock lock(io_mutex);
	std::cout << "print() #" << boost::this_thread::get_id() << std::endl;
	std::cout << "hello " << i << std::endl;
	boost::this_thread::sleep(boost::posix_time::seconds(1));
	std::cout << "world " << i << std::endl;
}

void test_thread_pool()
{
	// Create a thread pool of 4 worker threads.
	ThreadPool pool(4);

	// Queue a bunch of work items.
	for (int i = 0; i < 8; ++i) {
		pool.Enqueue(boost::bind(&print, i));
	}
}

void do_task(std::promise<int> &promiseObj) {
	boost::mutex::scoped_lock lock(io_mutex);
	std::cout << "Inside thread: " << std::this_thread::get_id() << std::endl;
	std::cout << "Inside thread: sleep 2 seconds... " << std::endl;
	std::this_thread::sleep_for(std::chrono::seconds(2));
	promiseObj.set_value(35);
}

void test_future()
{
	std::promise<int> promiseObj;
	std::future<int> futureObj = promiseObj.get_future();
	std::cout << "create thread..." << std::endl;
	std::thread th(do_task, std::ref(promiseObj));

	std::cout << "futureObj.get() block main thread." << std::endl;
	std::cout << futureObj.get() << std::endl;

	th.join();
	std::cout << "after join" << std::endl;
}

/*
std::bind
bind预先绑定的参数需要传具体的变量或值进去，对于预先绑定的参数，是pass-by-value的；[使用std::ref()可以pass by reference]
对于不事先绑定的参数，需要传std::placeholders进去，从_1开始，依次递增。placeholder是pass-by-reference的；
*/

int main(int argc, char* argv[])
{
	//test_thread();
	//test_thread_pool();
	test_future();
	return 0;
}

Reference

History

  • 20180523: created.