boost strand 사용해보기

boost strand 는 boost 에서 비동기 처리를 할 때, 멀티 쓰레드 환경에서 동기화를 맞출 필요 없도록 직렬화 시켜주는 기술이다. 쉽게 정리된 내용이 없어 정리하게 되었다.

 

시작하기에 앞서 boost 의 io_context 를 먼저 사용할 줄 알아야 하기 때문에, 싱글 쓰레드 기반에서 io_context 예시를 먼저 들어본다.

 

1. SingleThread 에서 io_context 를 사용

#include <iostream>
#include <boost/asio.hpp>

int main()
{
	std::cout << "============== TestSingleThread =============="<< "\n";
	std::cout << "Main Thread Id :" << std::this_thread::get_id() << "\n";

	boost::asio::io_context io_context;
	boost::asio::post( io_context, []()
		{
			std::cout << "Post 1 Thread Id : " << std::this_thread::get_id() << "\n";
		} );

	boost::asio::post( io_context, []()
		{
			std::cout << "Post 2 Thread Id : " << std::this_thread::get_id() << "\n";
		} );

	boost::asio::post( io_context, []()
		{
			std::cout << "Post 3 Thread Id : " << std::this_thread::get_id() << "\n";
		} );

	io_context.run();
	std::cout << "==============================================" << "\n";
    
    return 0;
}

위 코드는 io_context 를 만들고, 해야할 일을 세개 등록한다. 그리고 run 을 했을 때 thread id 를 출력하는데 결과는 다음과 같이 모두 같은 쓰레드에서 도는 것을 확인 할 수 있다.

모두 같은 쓰레드에서 돌아간다

 

동작 원리를 간단하게 그림으로 표현해봤다

여기서 Main Thread 와 같은 곳에서 돌아가는 이유는 Main Thread 에서 io_contex.run() 을 실행했기 때문이다.

 

이제 Multi Thread 로 넘어가보자.

 

2. Multi Thread에서 io_context 를 사용

 

쓰레드는 4개를 사용하고 7명의 유저가 일을 세개씩 요청하는 상황의 코드이다

#include <iostream>
#include <boost/asio.hpp>
#include <boost/chrono.hpp>
#include <boost/thread/thread.hpp> 

int main()
{
	std::cout << "============== TestMultiThreads ==============" << "\n";
	std::cout << "Main Thread Id :" << std::this_thread::get_id() << "\n";

	boost::asio::io_context io_context;

	auto work = make_work_guard( io_context );

	int thread_cnt = 4;
	std::vector< std::thread > threads;
	for ( int i = 0; i < thread_cnt; ++i )
	{
		threads.emplace_back( [&]()
			{
				io_context.run();
			} );
	}

	std::mutex mutex;
	int user_cnt = 7;
	for ( int i = 0; i < user_cnt; ++i )
	{
		boost::asio::post( io_context, [i, &mutex]()
			{
				std::cout << "User : " << i << " Post 1 Thread Id : " << std::this_thread::get_id() << " start\n";
				boost::this_thread::sleep_for( boost::chrono::milliseconds( 100 ) );
				std::cout << "User : " << i << " Post 1 Thread Id : " << std::this_thread::get_id() << " end\n";
			} );

		boost::asio::post( io_context, [i, &mutex]()
			{
				std::cout << "User : " << i << " Post 2 Thread Id : " << std::this_thread::get_id() << " start\n";
				boost::this_thread::sleep_for( boost::chrono::milliseconds( 100 ) );
				std::cout << "User : " << i << " Post 2 Thread Id : " << std::this_thread::get_id() << " end\n";
			} );

		boost::asio::post( io_context, [i, &mutex]()
			{
				std::cout << "User : " << i << " Post 3 Thread Id : " << std::this_thread::get_id() << " start\n";
				boost::this_thread::sleep_for( boost::chrono::milliseconds( 100 ) );
				std::cout << "User : " << i << " Post 3 Thread Id : " << std::this_thread::get_id() << " end\n";
			} );
	}

	work.reset();
	for ( auto& thread : threads )
	{
		thread.join();
	}

	std::cout << "==============================================" << "\n";
    
    return 0;
}

아까와 다른점이라면, io_contex.run() 을 실행하는 쓰레드가 메인 쓰레드가 아니라는 점에 있다. 위 코드의 실행 결과는 다음과 같다.

 

위 결과의 1,2,3 번째 줄을 확인 해보면, 같은 유저가 요청한 일에 대해서 순서보장이 되지 않고 다른 쓰레드에서 돌아갈 수 있다는 것을 알 수 있다. 이것 또한 그림으로 표현해보면 아래와 같다.

여기서 순서 보장이 되지 않는 이유는 등록한 일이 쓰레드에 랜덤으로 할당 되기 때문이다.

 

예를 들면 첫번째로 등록한일이 쓰레드 1에 할당, 두번째로 등록한 테스크가이 쓰레드 2에 할당 되었다고 하면 쓰레드 2에서 더 먼저 처리 될 수 있기 때문이다.

 

만약 여러 쓰레드에서 접근하면 안되는 변수에 접근하는 테스크라면, 별도의 동기화 작업을 해야한다 ( ex 락 )

 

이를 방지하기 위해 사용하는 것이 Strand 이다.

 

3. Multi Thread에서 io_context 와 Strand를 사용

 

위에서 User 라고 표기한 단위마다 동기화가 필요하다면 각 User 마다 strand 를 추가한다.

 

코드는 다음과 같다.

 

#include <iostream>
#include <boost/asio.hpp>
#include <boost/chrono.hpp>
#include <boost/thread/thread.hpp> 

int main()
{
	std::cout << "================= TestStrand =================" << "\n";
	std::cout << "Main Thread Id :" << std::this_thread::get_id() << "\n";

	boost::asio::io_context io_context;
	auto work = make_work_guard( io_context );

	int thread_cnt = 4;
	std::vector< std::thread > threads;
	for ( int i = 0; i < thread_cnt; ++i )
	{
		threads.emplace_back( [ & ]()
			{
				io_context.run();
			} );
	}

	std::mutex mutex;
	int user_cnt = 7;
	std::vector< boost::asio::io_context::strand > strands;
	for ( int i = 0; i < user_cnt; ++i )
	{
		boost::this_thread::sleep_for( boost::chrono::milliseconds( 1 ) );
		boost::asio::io_context::strand strand( io_context );
		strands.push_back( std::move( strand ) );
	}

	for ( int i = 0; i < user_cnt; ++i )
	{
		io_context.post( strands[ i ].wrap( [ i, &mutex ]()
			{
				std::cout << "User : " << i << " Post 1 Thread Id : " << std::this_thread::get_id() << " start\n";
				boost::this_thread::sleep_for( boost::chrono::milliseconds( 100 ) );
				std::cout << "User : " << i << " Post 1 Thread Id : " << std::this_thread::get_id() << " end\n";
			} ) );

		io_context.post( strands[ i ].wrap( [ i, &mutex ]()
			{
				std::cout << "User : " << i << " Post 2 Thread Id : " << std::this_thread::get_id() << " start\n";
				boost::this_thread::sleep_for( boost::chrono::milliseconds( 100 ) );
				std::cout << "User : " << i << " Post 2 Thread Id : " << std::this_thread::get_id() << " end\n";
			} ) );

		io_context.post( strands[ i ].wrap( [ i, &mutex ]()
			{
				std::cout << "User : " << i << " Post 3 Thread Id : " << std::this_thread::get_id() << " start\n";
				boost::this_thread::sleep_for( boost::chrono::milliseconds( 100 ) );
				std::cout << "User : " << i << " Post 3 Thread Id : " << std::this_thread::get_id() << " end\n";
			} ) );
	}

	work.reset();
	for ( auto& thread : threads )
	{
		thread.join();
	}

	std::cout << "==============================================" << "\n";
    
    return 0;
}

 

멀티 쓰레드에서 구현과 달라진 점은 유저 수 만큼 strand 를 사용한것, post 시 strand.wrap 함수를 사용한 것이다.

테스트 결과는 다음과 같다.

 

아까와 달리 유저가 요청한 task 의 순서가 변하지 않고, 동시에 다른 쓰레드에서 돌아가지 않는 것을 확인 할 수 있다. (start - end 사이에 실행되지 않음으로 확인)

 

그림으로 표현하면 다음과 같다

 

 

위 그림을 설명하면 strands 를 통해 등록된 task 는 순서를 가지고 있고, 이전 task 가 끝났을 때 처리가 가능하다.

 

이때 할당 되는 쓰레드는 이전에 사용된 쓰레드일 수도 있고 ( 3번 그림의 쓰레드 1에 등록된 2번 테스크 ) 혹은 다른 쓰레드 일 수 도 있다 ( 3번 그림의 쓰레드 1에 등록된 3번 테스크 )

 

할당되는 쓰레드와 별개로, 호출 시점이 이전 task 를 끝내야만 호출 할 수 있기 때문에 동기화를 맞출 필요가 없다.

 

4. 결론

 

Strand 를 사용하면 멀티 쓰레드 환경에서 직렬화를 시켜주기 때문에 동기화를 개발자가 관리하지 않아도 된다.

 

또 다른 장점으로는, 같은 쓰레드에 할당 될 확률이 높아 캐싱에 유리하다고 한다.

 

전체 코드

https://github.com/clucle/boost-strand-sample

GitHub - clucle/boost-strand-sample: boost strand sample code