Asynchronous Agents Library
– message block 2. ( unbounded_buffer )

작성자: 임준환( mumbi at daum dot net )

 

시작하는 글

 이전 글에서 message block 의 인터페이스인 ISourceITarget 인터페이스에 대해서 알아보았습니다. 이번 글부터 그 인터페이스들을 상속받아 구현한 message block 에 대해 알아보겠습니다.

 Message block 은 버퍼( buffer ) 를 가질 수도 있고, 상태만 가질 수도 있고, 기능만 가질 수도 있습니다. 그러므로 각 message block 들의 특징을 잘 파악하고, 언제 필요한지 알아야 합니다.

 이번 글에서는 가장 범용적인 유용한 unbounded_buffer 에 대해서 알아보도록 하겠습니다.

 

unbounded_buffer< _Type >

 unbounded_buffer 는 message block 중 가장 많이 사용될 것입니다. unbounded_buffer 는 내부적으로 큐( queue )를 구현하고 있어 message 저장소 역할을 합니다. 누군가 unbounded_buffer 에 message 를 보내면 unbounded_buffer 에 순서대로 차곡차곡 쌓이고, 쌓인 순서대로 꺼내서 쓸 수 있습니다. 꺼낸 message 는 unbounded_buffer 에서 제거됩니다. 그렇기 때문에 여러 곳에서 같은 message 를 꺼내 받을 수 없습니다.

 이런 작업들은 비 동기 agent 들과 사용할 때, 빛을 발합니다.

 unbounded_buffer 는 스레드에 안전하므로 직접 lock 을 하지 않아도 됩니다.

 

생성자

 

unbounded_buffer() – 기본 생성자

 빈 unbounded_buffer 를 생성합니다.

 

unbounded_buffer( filter_method const& _Filter )

 빈 unbounded_buffer 를 생성합니다. 하지만 필터 함수를 지정하여 받을 수 있는 메시지를 거를 수 있습니다.

 이 필터 함수는 bool (_Type const &) 형의 시그니처( signature )를 갖습니다.

 

멤버 함수

 

bool enqueue( _Type const& _Item )

 하나의 message 를 unbounded_buffer 에 보냅니다.

 message 전송이 성공이면 true, 아니면 false 를 반환합니다.

 내부적으로 이 함수는 message 전달 함수인 send() 를 사용합니다. 그리고 send() 의 결과를 반환합니다.

 

_Type dequeue()

 unbounded_buffer 에서 하나의 message 를 꺼냅니다. 꺼낸 message 는 큐에서 제거됩니다.

 꺼내진 message 를 반환합니다. 그리고 꺼내진 message 는 unbounded_buffer 내부에서 제거됩니다.

 enqueue() 와 마찬가지로 내부적으로 message 전달 함수인 receive() 를 사용합니다. receive() 가 반환한 값을 반환합니다.

 

예제

 unbounded_buffer 를 사용하여 작은 시나리오를 구현해보도록 하겠습니다.

시나리오

 윈도우즈 OS 는 사용자의 이벤트들을 메시지 큐에 담고, 큐에 들어온 메시지들을 순차적으로 꺼내서 처리하는 메커니즘을 사용합니다. 이 시나리오를 agent 와 unbounded_buffer 를 이용하여 간단하게 구현해보겠습니다.

코드

#include <iostream>
#include <string>
#include <agents.h>

using namespace std;
using namespace Concurrency;

// 메시지 객체
class Message
{
	wstring		message;

public:
	Message( const wstring& message )
		: message( message ) { }

	const wstring& GetMessage() const
	{
		return this->message;
	}
};

// 메시지를 발생하는 사용자 agent
class User
	: public agent
{
	ITarget< Message >&	messageQueue;

public:
	User( ITarget< Message >& target )
		: messageQueue( target ) { }

	void ClickMouseLButton()
	{
		send( this->messageQueue, Message( L"WM_LBUTTONDOWN" ) );
		send( this->messageQueue, Message( L"WM_LBUTTONUP" ) );		
	}

	void DragMouseLButton()
	{
		send( this->messageQueue, Message( L"WM_LBUTTONDOWN" ) );
		send( this->messageQueue, Message( L"WM_MOUSEMOVE" ) );
		send( this->messageQueue, Message( L"WM_LBUTTONUP" ) );
	}

	virtual void run()
	{
		this->ClickMouseLButton();
		
		Concurrency::wait( 1000 );

		this->DragMouseLButton();

		this->done();
	}
};

// 발생한 메시지들을 처리하는 메시지 펌프 agent
class MessagePump
	: public agent
{
	ISource< Message >&	messageQueue;

public:
	MessagePump( ISource< Message >& source )
		: messageQueue( source ) { }

	void ProcessMessage( const Message& message )
	{
		wcout << message.GetMessage() << endl;
	}

	virtual void run()
	{
		while( true )
		{
			Message message = receive( this->messageQueue );

			this->ProcessMessage( message );		
		}

		this->done();
	}
};

int main()
{
	// 메시지 큐
	unbounded_buffer< Message >	messageQueue;

	// 메시지를 발생하는 사용자와 메시지 펌프
	User user( messageQueue );
	MessagePump messagePump( messageQueue );

	// agent 시작.
	user.start();
	messagePump.start();	

	// agent 의 작업이 모두 끝날 때까지 대기
	agent* agents[] = { &user, &messagePump };
	agent::wait_for_all( 2, agents );
}

[ 코드1. agent 와 unbounded_buffer 를 이용한 메시지 펌프 간략 구현 ]

 Message 클래스는 단순히 문자열을 래핑( wrapping ) 하는 클래스로 큐에 저장되는 메시지를 나타냅니다.

 agent 로 2개를 정의하였는데 하나는 사용자가 이벤트 메시지를 발생하는 것을 흉내 낸 User 클래스이고, 다른 하나는 메시지 펌프를 간략화한 MessagePump 클래스입니다.

 사용자 agent( User 객체 )는 약간의 시간차를 두고 이벤트 메시지를 발생합니다. 발생된 메시지는 메시지 큐에 저장됩니다.

 메시지 펌프 agent 는 메시지가 저장될 때까지 대기하다가 메시지가 저장되면 그 메시지를 받아서 처리합니다. 처리된 메시지는 메시지 큐에서 제거됩니다.

 agent 들의 start() 를 사용하여 작업을 시작하고, 모든 agent 의 작업이 끝날 때까지 기다립니다.

 실제로는 하나의 agent 가 무한 루프를 수행하므로 프로그램이 종료되지 않습니다.

 위의 예제처럼 데이터를 보내고, 순차적으로 받아서 처리하고 싶을 때, 내부적으로 큐가 필요할 때 유용한 message block 이 바로 unbounded_ buffer 입니다.

 멀티 스레드 프로그래밍 시 자료 구조 중 큐가 많이 사용되기 때문에 unbounded_buffer 도 많이 사용하게 될 것입니다.

 

[ 그림1. agent 와 unbounded_buffer 를 이용한 메시지 펌프 간략 구현 결과 ]

[ 그림1. agent 와 unbounded_buffer 를 이용한 메시지 펌프 간략 구현 결과 ]

 

마치는 글

 이번 글에서는 message block 중 가장 사용도가 높은 unbounded_buffer 에 대해서 알아보았습니다. unbounded_buffer 이외에도 다양한 message block 이 있습니다.

 다음 글에서는 overwrite_buffer 라는 message block 에 대해서 알아보도록 하겠습니다.

이 글은 MSDN 글, "Solving The Dining Philosophers Problem With Asynchronous Agents"를 참고하여 작성되었습니다.

Asynchronous Agents Library로 Dining Philosophers 문제 해결하기 - 1

자, 이제 본격적으로 코드를 살펴보기 전에 메시지 블록이 무엇인지 먼저 짚고 넘어가겠습니다. AAL액터모형을 사용한다고 말씀드렸습니다. 또한, 액터모형에서 액터들은 메시지만으로 통신한다고 말씀드렸죠. 이 때 메시지를 받는 대상 혹은 메시지의 출처의 역할을 하는 것이 메시지 블록입니다. 전자의 경우 목적(target) 블록이라 하고, 후자는 원천(source) 블록이 됩니다.

전회에서 이번 예제에 쓰이는 네가지 메시지 블록을 소개했었는데요. unbounded_buffer는 목적 및 원천으로 쓰이며 큐와 같이 여럿의 메시지를 담고 있을 수 있는 놈입니다. overwrite_buffer는 하나의 변수처럼 값 하나만을 지니며, 새로 메시지가 올 경우 기존 값은 덮어씌여집니다. 역시 원천으로도 쓰일 수 있으며, 이 경우 사본을 보냅니다. 반면, call목적 블록으로만 쓰여 메시지 도착 시 특정 함수개체를 불러주는 기능을 합니다. join은 이번 예제에서 핵심 역할을 하는 블록으로서 여러 메시지를 동시에 기다려 하나로 묶어 출력하는 기능을 합니다.

먼저 가장 간단한 Chopstick 클래스를 살펴보죠.

   22 class Chopstick{

   23     const std::string m_Id;

   24 public:

   25     Chopstick(std::string && Id):m_Id(Id){};

   26     const std::string GetID()

   27     {

   28         return m_Id;

   29     };

   30 };


이와 같이 젓가락 식별용의 문자열을 가질뿐입니다. 생성자에서 r-value 참조를 쓰고 있다는 것 정도가 주목할만한 사항이겠군요.

다음은 ChopstickProvider로 다음과 같이 단순히 typedef입니다.

   34 typedef Concurrency::unbounded_buffer<Chopstick*> ChopstickProvider;


unbounded_buffer 메시지 블록을 이용해 메시지로 젓가락을 받으면 담고 있다가 철학자의 요청이 있으면 제공하는 역할을 합니다. 물록 철학자가 한입 먹고 나선 다시 젓가락을 놓으면 다시 받아놓는 역할도 합니다. 이 예제에서는 unbounded_buffer의 개수무제한(unbounded) 특성이 사실 굳이 필요 없습니다만 그래도 unbounded_buffer의 move semantic이 필요하기에(이 점에서 사본을 보내는 overwrite_buffer와는 다르죠) 이를 쓰는 것입니다.

다음이 대망의 Philosopher 클래스가 되겠습니다. 먼저, Concurrency::agent에서 public 상속을 받고 있는 것을 확인할 수 있습니다. 말씀드린 것처럼 각 철학자가 액터가 되어 독립적으로 동작하기 (즉, 별도 스레드로) 위함입니다.

   35 class Philosopher : public Concurrency::agent

   36 {

   37     ChopstickProvider* m_LeftChopstickProvider;

   38     ChopstickProvider* m_RightChopstickProvider;

   39 

   40 public:

   41     const std::string m_Name;

   42     const size_t  m_Bites;

   43     Philosopher(const std::string&& name, size_t bites=10):m_Name(name),m_Bites(bites){};

   44     Concurrency::unbounded_buffer<ChopstickProvider*> LeftChopstickProviderBuffer;

   45     Concurrency::unbounded_buffer<ChopstickProvider*> RightChopstickProviderBuffer;

   46     Concurrency::overwrite_buffer<PhilosopherState> CurrentState;

   47     void run()

   48     {

   49 

   50         //run에서 제일 먼저 해야하는 것은 ChopstickProvider를 초기화하는 것입니다. 여기서는 receive를 통해 public 변수에 메시지가 도착하기를 기다리게 하는 방식으로 처리합니다:

   51 

   52         //ChopstickProvider들을 초기화합니다.

   53         m_LeftChopstickProvider  = Concurrency::receive(LeftChopstickProviderBuffer);

   54         m_RightChopstickProvider = Concurrency::receive(RightChopstickProviderBuffer);

   55 

   56         //이제 생각하다가 먹기를 반복해야 합니다. 그를 위해 아직 등장하지 않은 두 함수(PickupChopsticks과 PutDownChopsticks)를 이용하려고 합니다:

   57 

   58         for(size_t i = 0; i < m_Bites;++i)

   59         {

   60             Think();

   61             std::vector<Chopstick*> chopsticks(PickupChopsticks());

   62             Eat();

   63             PutDownChopsticks(chopsticks);

   64         }

   65 

   66         //남은 일은 run 메소드를 나가기 전에 정리 작업을 하는 것인데, 다른 곳에 쓰일 수 있도록 ChopstickProvider를 반환하고 에이전트의 상태를 완료로 설정하고 있습니다.

   67         Concurrency::send(LeftChopstickProviderBufferm_LeftChopstickProvider);

   68         Concurrency::send(RightChopstickProviderBuffer, m_RightChopstickProvider);

   69 

   70         this->done(Concurrency::agent_done);

   71     }

   72 

   73     std::vector<Chopstick*> PickupChopsticks()

   74     {

   75         //join 생성

   76         Concurrency::join<Chopstick*,Concurrency::non_greedy> j(2);

   77         m_LeftChopstickProvider->link_target(&j);

   78         m_RightChopstickProvider->link_target(&j);

   79 

   80         //젓가락을 듭니다.

   81         return Concurrency::receive (j);

   82     } 

   83     void PutDownChopsticks(std::vector<Chopstick*>& v)

   84     {

   85         Concurrency::asend(m_LeftChopstickProvider,v[0]);

   86         Concurrency::asend(m_RightChopstickProvider,v[1]);

   87     }

   88 private:

   89     void Eat()

   90     {

   91         send(&CurrentState,Eating);

   92         RandomSpin();

   93     };

   94     void Think()

   95     {

   96         send(&CurrentState,Thinking);

   97         RandomSpin();

   98     };

   99 };


그 다음으로 한쌍의 젓가락을 위한 두 ChopstickProvider 포인터 변수(m_LeftChopstickProvider, m_RightChopstickProvider)가 보입니다. 철학자 이름(m_Name)과 몇번 먹을지를 나타내는 변수(m_Bites), 생성자까지는 파악하시는데 어려움이 없을 겁니다.

ChopstickProvider (이 자체도 unbounded_buffer인데) 포인터를 템플릿 인자로 가지는 unbounded_buffer 변수 한쌍이 등장하는데요. (44,45줄) 철학자가 젓가락을 소유하고 있는 상황이 아니고 철학자와는 별개로 젓가락들이 존재하는 상황이기에 필요한 변수들입니다. 이 두 public 변수들을 통해, 나중에 철학자들에게 필요할 때 젓가락을 제공해주는 ChopstickProvider를, 어딘가에서 받을 수 있습니다. 이들을 갖추고 나면 그 후부터 생각하다가 먹다가 할 수 있겠죠.

그 뒤로 run 메소드가 나옵니다. 실제 액터가 구동되면 수행될 함수입니다. 먼저, 전술한 두 변수를 통해 ChopstickProvider가 제공되기를 기다립니다. 이 때 Concurrency::receive 함수를 쓰고 있습니다. (이의 비동기 버전인 Concurrency::try_receive도 있습니다.)

58줄부터는 생각하다 먹기를 반복하는 반복문이 나옵니다. ThinkEat 함수는 89줄 이하에서 확인할 수 있는 것처럼 철학자의 현재 상태를 나타내는 overwrite_buffer 형의 변수 CurrentState를 설정하는 것 이외에는 특별히 하는 일이 없습니다. 그냥 시간을 좀 지체할 뿐입니다.

그리고 이 두 함수 호출 사이에 PickupChopsticksPutDownChopsticks 함수를 써서 실제 가장 중요한 젓가락 한 쌍을 안전하게 획득하고 다시 내려놓는 일을 합니다.


이에 대한 설명은 다음 회를 기대해주세요~ ^^
이 글은 MSDN 글, "Solving The Dining Philosophers Problem With Asynchronous Agents"를 참고하여 작성되었습니다.

오늘은 AAL(Asynchronous Agents Library)의 액터기반프로그래밍을 사용하여, 동기화 개체들로는 해법이 상당히 골치아프기로 유명한 "철학자들의 식사(Dining Philosophers) 문제"를 풀어보겠습니다. 내용이 길어질듯 하여 3회의 연재글로 구성하려 합니다.

먼저 간단히 철학자들의 식사 문제를 소개하면,


간단히 위 그림과 같은 상황입니다. 철학자 다섯명이 식사를 하는데 젓가락(그림에는 포크지만 상관없습니다;)이 보시는바와 같이 역시 다섯개뿐입니다. 그들은 철학자답게 생각하다가 한입 먹다가를 반복합니다. 한입 먹으려면 젓가락 한쌍이 필요해서 옆사람이 사이에 놓인 젓가락을 이미 선점해 먹고 있다면 기다려야 하는 것이죠. 공유 상태를 고려하지 않고 구현하면 데드락 등으로 철학자가 굶는(starvation) 상황이 발생할 수 있습니다. 이 문제는 저명한 컴퓨터과학자 다익스트라가 처음 제시하였습니다. 모니터 등의 동기화 개체를 사용하여 해결하는 방법이 기존에 많이 설명되어 있습니다만... 솔직히 이해하기가 쉽지 않고 구현도 어렵습니다.

이때 AAL이 제공하는 액터모형을 이용하면 그러한 난해함이나 복잡함 없이 이 문제를 해결할 수 있습니다. 액터모형은 독립적으로 동작하며 서로간에는 오로지 메시지만으로 소통하는(즉, 공유 상태를 가지지 않는) 액터들로 시스템을 모델링하는 방법이라 하겠습니다.

본 예제에서는 철학자를 액터(AAL 용어로는 에이전트)로 보고 메시지 전달을 위해 AAL에서 제공하는 몇몇 메시지 블록(message block)들을 사용하여 철학자들의 식사 문제를 해결합니다.

다음과 같은 다섯 클래스들을 작성하게 됩니다.

  • Chopstick 클래스
  • 식탁 위의 젓가락을 실제 소유하며 요청에 따라 철학자에게 제공하는 역할을 하는 ChopstickProvider 
  • 생각하고 먹는 에이전트 역할의 Philosopher 클래스. 이 클래스는 한쌍의 ChopstickProvider와만 소통합니다.
  • 생각하고 먹는 상태를 나타내는 PhilosopherState 열거형
  • 젓가락들과 철학자들이 배치될 Table 클래스

이 과정에서 다음과 같은 AAL의 클래스 및 함수들을 이용합니다.
  • Concurrency::agent - 에이전트 기반 클래스
  • 이하는 메시지 블록에 속하는 여러 타입들
    • Concurrency::unbounded_buffer
    • Concurrency::overwrite_buffer
    • Concurrency::join
    • Concurrency::call
  • 이상의 메시지 블록들에 메시지를 주고 받는데 사용하는 함수들
    • Concurrency::send
    • Concurrency::asend - 위의 비동기 버전으로, 받음 여부를 확인하지 않고 바로 리턴
    • Concurrency::receive

본격적인 구현 과정은 다음 회에 계속됩니다~ ^^