Asynchronous Agents Library
– message block 4. ( call )

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

시작하는 글

 이번 글에서는 call 이라는 message block 에 대해서 알아보겠습니다. 이번에 알아볼 call 또한 이미 알아본 message block 들과는 다른 특징이 있으니 특징을 이해하고, 필요할 때 바로 사용할 수 있어야 합니다.

call< _Type >

 call 은 함수 포인터와 같은 역할을 합니다. call 에 message 를 보내면 그 message 는 생성자에서 지정한 함수 포인터의 인자로 사용됩니다. 그래서 call 에 message 를 보낸다는 것은 함수를 호출하는 것과 같은 효과를 얻을 수 있습니다.

특징

 call 과 함수 호출의 다른 점이 있습니다. call 에 의해서 수행되는 함수를 동기 또는 비 동기로 수행되도록 선택할 수 있습니다.

 동기 전달 함수인 send() 를 사용하여 message 를 전달하면 지정된 함수의 수행이 종료될 때까지 기다립니다. 그렇기 때문에 지정된 함수의 수행이 종료되어야 해당 context 가 진행됩니다.

 반면에 비 동기 전달 함수인 asend() 를 사용하여 message 를 전달하면 지정된 함수의 수행이 종료될 때까지 기다리지 않고 호출한 context 는 계속 진행됩니다. 즉, 호출한 스레드와 call 의 함수가 동시에 진행됩니다.

 이처럼 동기와 비 동기 처리를 적절히 선택하여 구현할 수 있습니다.

 또 하나 기억해 두어야 할 특징은 하나의 call 객체에 message 를 여러 차례 보내더라도 동시에 수행되지 않습니다. 하나의 call 객체는 내부적으로 작업 큐( queue ) 를 가지고 있어 순차적으로 수행됩니다.

 하지만 하나의 call 객체가 아니라 서로 다른 call 객체는 asend() 를 이용해 동시에 수행될 수 있습니다. 서로 각각의 작업 큐를 가지고 있기 때문입니다.

 call 의 안타까운 점은 함수의 반환을 처리하지 못한다는 점입니다. 만약 지정된 함수의 결과를 얻어내야 한다면 해당 함수 내부에서 다른 message block 으로 결과를 message 로 보내야 합니다.

 call 은 생성자 이외에 public 인 멤버 함수가 없습니다.

선언

 사실 call 이라는 message block 의 타입은

template < class _Type, class _FunctorType = std::tr1::function< void( _Type const & ) > >
class call

[ 코드1. call 의 선언 ]

입니다.

 tr1 의 function 은 일반 함수 포인터와 멤버 함수 포인터, 함수 객체를 모두 가질 수 있으므로 기본 템플릿 매개변수인 _FunctorType 을 지정하지 않고 사용하는 것이 좋습니다. 

 위의 선언이 보여주듯이 _Type 은 수행될 함수의 매개변수의 타입입니다. bind 와 같은 어댑터 함수들과 함께 사용하여 여러 인자들을 바인딩할 수 있습니다.

예제

 call 의 특징들을 쉽게 알아볼 수 있는 예제를 구현해보았습니다.

시나리오

 Loader 라는 agent 클래스가 작업을 진행하고 call 을 이용해 진행 상황을 화면에 출력해주는 예제입니다.

코드

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

using namespace std;
using namespace Concurrency;

class Loader
	: public agent
{
public:
	Loader( call< unsigned int >& displayDelegate )
		: displayDelegate( displayDelegate ) { }

protected:
	void run()
	{
		for( unsigned int percent = 0; percent <= 100; percent += 10 )
		{
			Concurrency::wait( 1000 );
			asend( this->displayDelegate, percent );
		}

		this->done();
	}

private:
	call< unsigned int >&		displayDelegate;
};

int main()
{
	call< unsigned int >	showLoading( []( unsigned int percent )
	{
		switch( percent )
		{
		case 0:		wcout << L"loading..";					break;
		case 100:	wcout << endl << L"complete!" << endl;	break;
		default:	wcout << L"..";							break;
		}		
	} );

	Loader loader( showLoading );

	loader.start();

	agent::wait( &loader );
}

[ 코드2. call 의 비 동기 처리를 이용한 예제 ]

 Loader agent 의 작업 진행 상황을 화면에 출력합니다. 진행률이 0 일 때는 “loading.." 을 출력하고, 이후, 진행이 완료되기 전까지 “..” 을 추가하여 진행 상황을 알리고, 완료되면 “complete!” 를 출력하여, 작업의 완료를 알립니다.

[ 그림1. call 의 비 동기 처리를 이용한 예제 ]

마치는 글

 이미 알아본 message block 들과는 많이 다른 call 에 대해서 알아보았습니다. call 은 함수의 수행을 동기 및 비 동기 처리로 할 수 있도록 도와주는 반면 결과를 얻기는 쉽지 않습니다.

 이러한 call 의 단점의 보완과 더 많은 특징을 가지고 있는 transformer 라는 message block 을 제공합니다.

 다음 글에서는 transformer 를 이용해 더욱 더 신나는 멀티 코어 프로그래밍을 해보도록 해보겠습니다.

1. 비관리 클래스에서 관리 클래스를 멤버로

비관리 클래스에서 관리 클래스를 멤버로 가지고 싶을 때는 ‘gcroot’라는 템플릿을 사용합니다.

‘gcroot’를 사용하기 위해서는

를 포함해야 합니다.

비관리 클래스에서 관리 클래스인 String을 다음과 같이 멤버로 선언합니다.

그리고 사용하기 위해서는 할당을 합니다.

2. 관리 클래스에서 비관리 클래스를 멤버로

관리 클래스에서 비관리 클래스를 멤버로 가질 때는 비관리 클래스를 포인터로 선언하여 비관리 힙에 동적할당을 합니다.

참고

http://msdn.microsoft.com/ko-kr/library/481fa11f%28v=VS.80%29.aspx

http://blog.naver.com/scor7910/40048083284

Asynchronous Agents Library
– message block 3. ( overwrite_buffer & single_assignment )

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

시작하는 글

 지난 글 unbounded_buffer 에 이어 또 다른 message block 인 overwrite_buffer 와 single_assignment 에 대해서 알아보도록 하겠습니다.

 Message block 들은 특징들이 모두 다르기 때문에 지난 unbounded_buffer 을 생각하시면 이해가 어려울 수 있습니다. message block 하나 하나의 쓰임새가 다르므로 새로운 것을 알아본다고 생각하시는 것이 좋을 것 같습니다.

overwrite_buffer< _Type >

 지금부터 설명드릴 overwrite_buffer 은 unbounded_buffer 와는 달리 하나의 변수라고 생각하시면 이해하기 쉬울 것입니다.

 Concurrency runtime 을 사용하지 않고, 스레드 간의 상태나 정보를 공유하려면 전역 변수나 힙( heap ) 에 할당된 변수에 락( lock ) 을 걸어 사용해야 합니다.

 overwrite_buffer 는 방금 언급한 번거로운 작업들을 알아서 해줍니다. 내부에서 힙에 메모리를 할당하고, 접근 시 락을 겁니다. 하지만 사용하는 우리는 그런 것들을 신경 쓰지 않고 마치 지역 변수처럼 사용할 수 있습니다.

 unbounded_buffer 는 외부에서 message 를 받아가면 내부에서 해당 message 가 제거되는 반면에, overwrite_buffer 는 제거되지 않습니다. 또한 하나의 변수와도 같기 때문에 외부에서 message 를 보내면 이 전의 message 를 덮어쓰고 새 message 가 저장됩니다.

 결국 overwrite_buffer 는 단 하나의 message 만을 갖게 됩니다.

 그럼 overwrite_buffer 의 멤버 함수에 대해서 알아보도록 하겠습니다.

멤버 함수

 생성자와 소멸자를 제외한 public 인 멤버 함수들입니다.

bool has_value() const

 현재 message 를 가지고 있는지 반환합니다.

 어떠한 message 도 갖지 않을 경우에 false 를 반환합니다. 만약, 한번이라도 overwrite_buffer 에 message 가 전달된다면 그 후부터는 true 를 반환합니다. overwrite_buffer 는 외부에서 message 를 받아가도 내부의 message 가 제거되지 않기 때문입니다.

 Message 를 갖고 있지 않을 때, 외부에서 동기 함수인 receive() 를 사용해 message 를 얻기를 원한다면 overwrite_buffer 에 message 가 들어올 때까지 기다립니다. message 가 제거되지 않기 때문에 한번이라도 overwrite_buffer 가 message 를 받으면 receive() 가 기다리는 일은 없을 것입니다.

_Type value();

 현재 가지고 있는지 message 를 반환합니다.

 내부적으로 동기 전달 함수인 receive() 를 사용하므로 message 를 가지고 있지 않다면 message 를 갖게 될 때까지 기다립니다. 만약 이 때, has_value() 를 호출했다면 false 를 반환할 것입니다.

 Message 가 제거되지 않기 때문에 전달 함수를 이용해 message 를 받아갈 경우, 복사본이 전달됩니다.

예제

 overwrite_buffer 의 간단한 예제를 구현해보도록 하겠습니다.

시나리오

 네트워크 지연 시간을 갱신하고, 출력하는 프로그램을 작성할 것입니다.

 네트워크 지연 시간을 갱신하는 역할을 하는 agent 와 갱신된 정보를 출력하는 agent 가 하나의 overwrite_buffer 를 공유하여 사용하는 예제입니다.

코드

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

using namespace std;
using namespace Concurrency;

// 지연 시간을 얻어오는 agent.
class PingUpdater
	: public agent
{
public:
	PingUpdater( const array< unsigned int, 5 >& delayTimeSource, ITarget< unsigned int >& targetBlock )
		: delayTimeSource( delayTimeSource )
		, targetBlock( targetBlock ) { }

protected:
	// 2초마다 지연 시간을 얻어 옴.
	void run()
	{
		while( true )
		{
			asend( this->targetBlock, this->GetDelayTime() );

			Concurrency::wait( 2000 );
		}

		this->done();
	}

	// 지연 시간을 시뮬레이션하는 함수.
	unsigned int GetDelayTime()
	{
		static unsigned int index = 0;

		unsigned int delayTime = this->delayTimeSource[ index ];

		if( index + 1 < this->delayTimeSource.size() )
			++index;		
		else
			index = 0;

		return delayTime;
	}

private:
	const array< unsigned int, 5 >&	delayTimeSource;
	ITarget< unsigned int >&		targetBlock;
};

// 지연 시간을 출력하는 agent.
class PingDisplayer
	: public agent
{
public:
	PingDisplayer( ISource< unsigned int >& sourceBlock )
		: sourceBlock( sourceBlock ) { }
protected:
	// 1초마다 지연 시간을 출력한다.
	void run()
	{
		while( true )
		{
			this->Display( receive( this->sourceBlock ) );

			Concurrency::wait( 1000 );
		}

		this->done();
	}

	// 지연 시간을 출력하는 함수.
	void Display( unsigned int delayTime )
	{
		wcout << L"current delay time: " << delayTime << endl;
	}

private:
	ISource< unsigned int >&	sourceBlock;
};

int main()
{
	// 네트워크 지연 시간의 시뮬레이션 정보.
	array< unsigned int, 5 > delayTimeSource = { 210, 211, 261, 246, 223 };

	// 공유 버퍼
	overwrite_buffer< unsigned int > delayTimeBuffer;

	// 네트워크 지연 시간을 갱신하는 agent 와 출력하는 agent.
	PingUpdater updater( delayTimeSource, delayTimeBuffer );
	PingDisplayer displayer( delayTimeBuffer );

	// agent 시작.
	updater.start();
	displayer.start();

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

[ 코드1. overwrite_buffer 를 이용한 네트워크 지연 시간 갱신 및 출력 예제 ]

 PingUpdater 클래스는 agent 클래스로 네트워크 지연 시간을 갱신하는 역할을 합니다. 2초에 한 번씩 시뮬레이션을 위해 준비된 정보를 순회하며 얻어와서 overwrite_buffer 에 전달합니다.

 PingDisplayer 클래스도 agent 클래스로 갱신된 네트워크 지연 시간을 화면에 출력하는 역할을 합니다. 1초에 한번씩 overwrite_buffer 로부터 갱신된 정보를 가져와서 화면에 출력합니다.

 예제에서 사용된 Concurrency::wait() 는 Win32 API 의 Sleep() 과 같은 역할을 합니다. agent::wait() 과 혼동하지 않길 바랍니다.

 위 코드를 보시면 굉장히 직관적이고, 간단하게 멀티 스레드 프로그래밍을 할 수 있다는 것을 알 수 있을 것입니다.

[ 그림1. overwrite_buffer 를 이용한 네트워크 지연 시간 갱신 및 출력 예제 ]

single_assignment< _Type >

 single_assignment 는 위에서 설명한 overwrite_buffer 와 거의 흡사합니다.

 단지 다른 점이 있다면 message 를 한번만 받을 수 있다는 것입니다. 만약 두 번 이상 보낸 다면 두 번째부터는 무시됩니다.

 멤버 함수 또한 거의 같지만, 다른 점을 알아보겠습니다.

멤버 함수

 생성자와 소멸자를 제외한 public 인 함수들입니다.

bool has_value() const

 위에서 설명한 overwrite_buffer 의 has_value() 와 같습니다.

 Message 를 단 한번도 받지 않았다면 false 를 반환하고, 받았다면 true 를 반환합니다.

_Type const & value()

 overwrite_buffer 의 value() 와 같은 기능을 합니다.

 하지만 값을 반환하지 않고 const 참조를 반환한다는 것이 다릅니다.

 overwrite_buffer 의 value() 와 마찬가지로 message 를 갖고 있지 않다면 message 를 갖게 될 때까지 기다립니다.

마치는 글

 이번 글에서는 overwrite_buffer 와 single_assignment 에 대해서 알아보았습니다.

  single_assignment 는 overwrite_buffer 와 거의 흡사하기 때문에 예제는 생략하였습니다.

 지난 글에서 본 unbounded_buffer 와는 분명히 쓰임새가 다르므로 특징을 잘 파악해두시면 좋을 것입니다.

 다음 글에서 또 다른 message block 을 소개해드릴 것입니다. 그 message block 또한 쓰임새가 분명히 다르므로 Asynchronous Agents Library 의 활용도가 굉장히 넓다는 것을 아시게 될 것입니다.

3) String^을 C/C++ 문자열로 변환

1)번에서는 C/C++의 문자열을 String^로 변환하는 방법에 대해서 설명했습니다.

이번에는 String^을 char*나 wchr_t*로 변환하는 방법에 대해서 설명합니다.

아래의 예제 코드를 봐 주세요

String^을 char*나 wchr_t*로 변환하는 방법은 두 가지가 있습니다.

첫 번째 std::string 사용

가장 간단한 방법입니다만 불필요한 std::string을 사용해야 단점이 있습니다.

두 번째 marshal_context 사용

첫 번째 방법에서 std::string을 사용한 이유는 다름이 아니고 메모리 확보 때문입니다.

마샬링을 통해서 char*와 wchar_t*에 메모리 주소를 저장합니다. 문자열 그 자체를 복사하는 것이 아닙니다. 그래서 변환한 문자열을 저장할 메모리 주소를 확보하고 사용 후에는 해제를 해야 합니다. 메모리 확보와 해제를 위해서 marshal_context를 사용합니다.

marshal_context는 변환에 필요한 메모리를 확보하고, 스코프를 벗어날 때 메모리를 해제합니다.

String^을 C/C++ 문자열로 변환할 때는 std::string + marshal_as 나 marshal_context 둘 중 하나를 선택하여 사용합니다.

참고

http://msdn.microsoft.com/en-us/library/bb384865.aspx

http://msdn.microsoft.com/ko-kr/library/bb531313%28VS.90%29.aspx

http://codezine.jp/article/detail/4774

Asynchronous Agents Library
– message block 2. ( unbounded_buffer )

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

시작하는 글

 이전 글에서 message block 의 인터페이스인 ISource 와 ITarget 인터페이스에 대해서 알아보았습니다. 이번 글부터 그 인터페이스들을 상속받아 구현한 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 를 이용한 메시지 펌프 간략 구현 결과 ]

마치는 글

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

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

관리코드와 비관리코드를 혼합해서 사용할 때 서로간에 문자열을 주고 받아야 하는 경우가 종종 있을 것입니다. 관리코드와 비관리코드간에 문자열을 서로 어떻게 변환하여 주고 받는지 알아보겠습니다.

1) C/C++ 문자열을 String^으로 변환

먼저 아래의 변환 예제 코드를 봐 주세요

관리코드와 비관리코드 간의 문자열 변환에는 ‘msrshal_as’를 사용하여 마샬링합니다.

C/C++ 문자열을 마샬링하기 위해서는

파일을 포함하고,

네임스패이스를 선언합니다.

사용 방법은 아주 간단합니다.

ANSI 문자열을 관리코드의 문자열로 변환할 때는 아래와 같이 합니다.

유니코드를 관리코드의 문자열로 변환할 때는 아래와 같이 합니다.

2) STL의 string과 String^간의 변환

이것도 marshal_as를 사용합니다.

아래의 예제 코드를 봐 주세요

STL의 문자열과 변환하기 위해서는 다음의 헤더 파일을 포함해야 합니다.

마샬링하는 방법은 앞에 설명한 C/C++ 문자열 변환과 같습니다.

주제와 좀 관계 없는 것으로 콘솔창에 유니코드 문자열을 출력하는 방법은 아래와 같습니다.

 setlocale로 국가를 설정하고(직접 나라를 지정할 수도 있고, 아니면 위처럼 시스템 설정에 따라가도록 할 수도 있습니다), ‘cout’ 대신 ‘wcout’를 사용합니다.

관리코드 문자열과 비관리코드 문자열간의 변환에 따른 성능

C++로 만드는 프로그램은 보통 고성능을 원하는 프로그램이므로 보통 C++ 프로그래머는 성능에 민감합니다. 마샬링은 공짜가 아닙니다만 많은 양을 아주 빈번하게 마샬링 하는 것이 아니면 성능에 너무 신경 쓰지 않아도 됩니다. 다만 기본적으로 관리코드의 문자열은 유니코드입니다. 그래서 비관리코드의 문자열이 ANSI 코드라면 유니코드를 사용했을 때 보다 더 많은 시간이 걸립니다(정확한 수치는 잘 모르지만 ANSI가 유니코드보다 3배정도 더 걸린다고도 합니다). 그래서 관리코드와 비관리코드를 같이 사용할 때는 가능한 유니코드를 사용하는 것이 훨씬 좋습니다.

아직 설명할 것이 많이 남아 있습니다. 다음을 기다려주세요^^

Asynchronous Agents Library

– message block 1. ( 인터페이스 )

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

시작하는 글

 이전 글까지 Asynchronous Agents Library( 이하, AAL ) 의 일부인 agent 와 message 전달 함수에 대해 알아보았습니다. agent 만 알아도 어느 정도 비 동기 처리를 쉽게 구현할 수 있습니다.

 이번 글에서는 agent 간 소통을 할 수 있는 message block 들에 대해서 알아보겠습니다. message block 을 이용하면 agent 간 데이터 또는 상태 동기화를 할 수 있습니다.

 AAL 은 스레드로부터 안전한 방식으로 구현되었고, 추상화되었습니다. 그래서 agent 객체와 message block 을 이용한 동기화 로직이 직관적이고 쉽게 흐름을 파악할 수 있어 데드락( dead-lock ) 을 방지하기 용이합니다.

 그럼 지금부터 agent 를 이용한 비 동기 처리에 날개를 달아주는 message block 에 대해 알아보도록 하겠습니다.

Message 객체

 예전 글부터 message, message 메커니즘, message 전달 함수, message block 등을 언급하면서 항상 message 란 개념을 사용했습니다.

 이 개념은 실제 클래스로 존재합니다. 하지만 단지 message 를 래핑( wrapping ) 할 뿐, 전혀 다른 기능을 가지고 있지 않은 클래스입니다.

 한 가지 기능이 있다면 식별자( id )를 갖는다는 것입니다. message 클래스는 Concurrency Runtime 의 _Runtime_object 클래스를 상속 받습니다. 이 클래스는 Runtime 에 의해 생성될 때 자동으로 id 를 갖게 됩니다. 이 id 를 알아보는 함수는 msg_id() 입니다. 이 메서드의 접근자가 public 으로 되어 있어 message 클래스에서도 사용 가능합니다.

 이 msg_id() 가 반환한 값은 message block 에서 사용되는 runtime_object_identity 형입니다. 몇몇 message block 메서드의 runtime_object_identity 형의 매개변수에 인자로 사용할 수 있습니다.

 사실, 직접 message block 을 구현하지 않는 한, message 클래스는 직접 사용할 경우는 없을 것입니다. 우리는 보내고 받는 데이터를 공급하면 내부적으로 그 데이터를 message 클래스로 래핑하고 message block 내부에서 사용하게 되는 것입니다. 그러므로 크게 신경쓰지 않아도 됩니다.

Source 와 target

 Message block 은 크게 두 가지 종류로 나눌 수 있습니다. 하나는 source 이고 다른 하나는 target 입니다.

 Message block 에서의 source 는 message 를 보낼 message block 을 일컫습니다. 마찬가지로 target 은 message 를 받을 message block 을 뜻합니다.

ISource 인터페이스

 Source 는 AAL 의 하나의 개념이지만, 이것을 인터페이스로 추상화 하였습니다. 이것이 ISource 인터페이스입니다.

 그러므로 source 로 쓰일 message block 들은 ISource 인터페이스를 상속하여 구현되었습니다. 만약 직접 source 로 사용될 message block 을 구현하신다면 ISource 인터페이스를 상속해야 합니다.

- ISource 인터페이스의 선언

template<   class _Type>class ISource;

[ 코드1. ISource 인터페이스의 선언 ]

 템플릿 매개변수인 _Type 은 message 로 쓰일 데이터 형( type )입니다. _Type 은 public typedef 인 source_type 으로 사용할 수 있습니다.

- ISource 인터페이스의 메서드

virtual void link_target(ITarget<_Type> * _PTarget) = 0;

[ 코드2. ISource::link_target() ]

 link_target() 은 target 인 message block 과 연결합니다. 여기서 연결의 의미는 자동으로 전달된다는 의미로 생각하시면 되겠습니다.

 즉, 이 ISource 를 상속받은 message block 에 link_target() 으로 target message block 을 연결했을 경우, 이 message block 의 message 들은 직접 전달 함수를 사용하지 않아도 자동으로 target message block 으로 전달됩니다.

 연결할 target 은 여러 개일 수 있습니다. 그러나 ISource 를 상속한 message block 의 구현에 따라 첫 번째 target 만 동작할 수도 있습니다. 예로 unbounded_buffer 가 있습니다. unbounded_buffer 는 내부적으로 큐를 구현하고 있어 전달 후, message 가 큐에서 제거되므로 두 번째 연결된 target 이 있더라도 message 를 보낼 수 없습니다.

 매개변수인 _PTarget 은 연결할 target message block 입니다. _PTarget 의 데이터 형인 ITarget 은 target 을 추상화한 인터페이스입니다. 곧 설명하도록 하겠습니다.

 _PTarget 이 NULL 이라면 invalid_argument 예외가 발생합니다.

virtual void unlink_target(ITarget<_Type> * _PTarget) = 0;

[ 코드3. ISource::unlink_target() ]

 unlink_target() 은 link_target() 으로 연결된 target 들 중 매개변수인 _PTarget 에 지정된 target 의 연결을 해제합니다.

 _PTarget 이 NULL 이라면 invalid_argument  예외가 발생합니다. 또한 _PTarget 이 연결된 target 들 중에 없다면 아무 것도 하지 않습니다.

virtual void unlink_targets() = 0;

[ 코드4. ISource::unlink_targets() ]

 unlink_targets() 는 연결된 모든 target 들의 연결을 해제합니다.

virtual message<_Type> * accept(runtime_object_identity _MsgId, ITarget<_Type> * _PTarget) = 0;

[ 코드5. ISource::accept() ]

 accept() 는 target 에서 호출되지만, source 가 제공합니다. source 의 message 를 수락하고, 소유권이 이전됩니다.

 매개변수인 runtime_object_identity 는 message 객체의 msg_id() 로 얻을 수 있습니다. 실제로 runtime_object_identity 는 __int32 를 typedef 한 것이고, Concurrency Runtime 에서 객체를 생성할 때 지정되는 고유의 번호입니다.

 다른 매개변수인 _PTarget 은 message 를 수락하는 target 입니다.

 수락된 message 가 반환됩니다.

virtual bool reserve(runtime_object_identity _MsgId, ITarget<_Type> * _PTarget) = 0;

[ 코드6. ISource::reserve() ]

 reserve() 는 message 를 예약합니다. 예약을 성공한 message 는 message 를 얻기 위한 comsume() 이나 예약을 해제 위한 release() 를 호출해야 합니다.

 매개변수는 위의 accept() 와 같습니다.

 예약에 성공한 경우 true 를, 실패한 경우 false 를 반환합니다. 실패할 수 있는 이유는 다양합니다. 이미 예약되었거나, 구현한 message block 의 특징에 따라 예약에 실패할 수 있습니다.

virtual message<_Type> * consume(runtime_object_identity _MsgId, ITarget<_Type> * _PTarget) = 0;

[ 코드7. ISource::consume() ]

 consume() 은 위의 accept() 와 비슷합니다.

 다른 점이 있다면 reserve() 를 호출해 true 를 반환했을 때에만 consume() 을 호출해야 합니다. 보통 reserve() 를 호출하지 않았거나 _Ptarget 이 예약된 target 과 다를 경우 bad_target 예외가 발생합니다.

 매개변수는 위의 accept() 와 같습니다.

virtual void release(runtime_object_identity _MsgId, ITarget<_Type> * _PTarget) = 0;

[ 코드8. ISource::release() ]

 release() 는 예약된 것을 해제합니다.

 매개변수는 accept() 와 같습니다.

virtual void acquire_ref(ITarget<_Type> * _PTarget) = 0;

[ 코드9. ISource::acquire_ref() ]

 acquire_ref() 는 참조 개수를 증가시킵니다. 현재 link_target() 으로 연결된 target 에서 호출됩니다.

 매개변수인 _PTarget 은 link_target() 으로 연결된 target 입니다.

virtual void release_ref(ITarget<_Type> * _PTarget) = 0;

[ 코드10. ISource::release_ref() ]

 release_ref() 는 참조 개수를 감소시킵니다. 현재 link_target() 으로 연결된 target 에서 호출됩니다.

매개변수인 _PTarget 은 link_target() 으로 연결된 target 입니다.

ITarget 인터페이스

 Target 또한 source 와 마찬가지로 AAL 의 하나의 개념이지만, 이것을 추상화 하였습니다. 이것이 ITarget 인터페이스 입니다.

 Target 으로 사용할 message block 을 구현하신다면 ITarget 인터페이스를 상속해야 합니다.

- ITarget 인터페이스의 선언

template<   class _Type>class ITarget;

[ 코드11. ITarget 인터페이스의 선언 ]

 템플릿 매개변수인 _Type 은 message 로 사용될 데이터 형입니다. _Type 은 public typedef 인 type 으로 사용할 수 있습니다.

 Target 은 필터를 지정할 수 있습니다. 그래서 필터 함수의 시그니처( signature )인 bool ( _Type const & ) 를 typedef std::tr1::function<bool(_Type const&)> filter_method 로 정의되어 있습니다.

- ITarget 인터페이스의 메서드

virtual message_status propagate(message<_Type> * _PMessage, ISource<_Type> * _PSource) = 0;

[ 코드12. ITarget::propagate() ]

 propagate() 는 지정된 source 로부터 해당 message 를 비 동기 방식으로 가져옵니다.

 매개변수인 _PMessage 는 가져올 message 이고, _PSource 는 보내는 message block 입니다. _PMessage 나 _PSource 가 NULL 일 경우, invalid_argument 예외를 발생할 수 있습니다.

 Message 의 전달이 성공 또는 실패 등의 message 상태를 반환합니다.

virtual message_status send(message<_Type> * _PMessage, ISource<_Type> * _PSource) = 0;

[ 코드13. ITarget::send() ]

 send() 는 지정된 source 로부터 해당 message 를 동기 방식으로 가져옵니다.

 매개변수는 propagate() 와 같고, 예외 또한 같습니다.

 Message의 생성 이외에 네트워크와 함께 사용할 경우, 데드락( dead lock ) 을 초래할 수 있습니다.

virtual void link_source(ISource<_Source_type> * _PSource)

[ 코드14. ITarget::link_source() ]

 link_source() 는 source 의 link_target() 에 대응되는 함수로 지정된 source 를 연결합니다.

 하지만, 이 함수는 target 에서 호출하면 안되고, source 에서 link_target() 와 함께 호출하여 서로 연결되어야 합니다.

virtual void unlink_source(ISource<_Source_type> * _PSource)

[ 코드15. ITarget::unlink_target() ]

 unlink_source() 는 unlink_target() 에 대응되는 함수로 지정된 source 와의 연결을 해제합니다.

 하지만, link_source() 와 마찬가지로 target 에서 호출하면 안되고, source 에서 unlink_target()나 unlink_targets() 와 함께 호출하여 서로 연결을 해제해야 합니다.

virtual void unlink_sources()

[ 코드16. ITarget::unlink_sources() ]

 unlink_sources() 는 unlink_targets() 에 대응되는 함수로 지정된 source 들과의 연결을 모두 해제합니다.

마치는 글

 이번 글에서는 message block 구현 시, 상속해야 할 인터페이스인 ISource 와 ITarget 에 대해서 알아보았습니다.

 실제로 ISource 와 ITarget 의 메서드들을 직접 호출하는 경우는 거의 없으며, message block 내부에서 사용됩니다.

 Message 를 전달하기 위해서는 위 인터페이스들의 메서드들보다 message 전달 함수들을 많이 사용합니다.

 이번에 소개한 인터페이스들의 구현 클래스들에 대해 아직 소개하지 않았고, 이 인터페이스들의 메서드들이 내부적으로 사용되지만 사용자가 직접 호출할 경우가 드물기 때문에 예제를 작성하지 않았습니다.

 다음 글에서 위 인터페이스들을 구현한 구현 클래스들에 대해서 살펴보고 예제를 보도록 하겠습니다.

아마 C++ 프로그래머가 C++/CLI를 사용할 때 가장 신경 쓰이는 부분이 관리 코드를 어떻게 하면 비관리 코드와 연동하는 방법이라고 생각합니다.

그래서 아직 C++/CLI의 델리게이트 등 C++/CLI의 특징을 설명하지 않은 것이 많지만 이런 것은 C#를 공부하면 배울 수 있는 것이므로 급하지 않다고 생각합니다. 그래서 관리 코드와 비 관리 코드의 연동에 대해서 앞으로 몇 차례에 걸쳐서 설명하려고 합니다.

이번은 첫 번째로 간단하게 array로 만든 배열을 비관리 코드의 포인터로 어떻게 전달하는지 설명하겠습니다.

먼저 코드를 봐 주세요^^

위 코드의 핵심은

입니다.

앞서 설명한 pin_ptr을 사용하였습니다. pin_ptr을 사용하여 관리 힙에 할당된 객체가 이동하지 못하도록 고정합니다. 고정하는 이유는 비관리 코드로 메모리 주소를 넘기기 때문에 관리 힙에서 이동이 되면 안되기 때문입니다.

또 여기서 자세히 봐야 되는 것이 있습니다.

pNative에 &ArrNums이 아닌 &ArrNums[0]을 대입하였습니다. 이유는 관리 코드에서는 &ArrNums는 ArrNums의 요소가 아닌 ArrNums 오브젝트 자체를 가리키는 것이기 때문입니다.

&ArrNums[0]을 대입해야 ArrNums에 들어가 있는 요소의 첫 번째 주소를 비관리 코드에 주소를 넘길 수 있습니다.

제가 요즘 바빠서 이번은 아주 간단하게 이것으로 끝내겠습니다.^^;

다음에는 문자열 변환에 대해서 설명하겠습니다.

참조

http://mag.autumn.org/Content.modf?id=20050507224044

Asynchronous Agents Library
– message 전달 함수. 2 ( 수신 )

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

Message 수신

 Message 를 message block 에 전송할 수 있듯이, message block 으로부터 수신할 수도 있습니다. message 수신 함수에도 전송 함수와 마찬가지로 동기 함수인 receive() 와 비 동기 함수인 try_receive() 가 있습니다.

동기 함수 receive()

 동기 함수인 receive() 는 message block 으로부터 수신이 완료될 때 수신된 message 를 반환합니다. 만약 message block 에 어떠한 message 도 없다면 receive() 는 message block 에 수신할 message 가 있을 때까지 기다립니다.

 아래는 receive() 의 선언입니다.

template <
   class _Type
>
_Type receive(
   ISource<_Type> * _Src,
   unsigned int _Timeout = COOPERATIVE_TIMEOUT_INFINITE
);

template <
   class _Type
>
_Type receive(
   ISource<_Type> * _Src,
   filter_method const& _Filter_proc,
   unsigned int _Timeout = COOPERATIVE_TIMEOUT_INFINITE
);

template <
   class _Type
>
_Type receive(
   ISource<_Type> &_Src,
   unsigned int _Timeout = COOPERATIVE_TIMEOUT_INFINITE
);

template <
   class _Type
>
_Type receive(
   ISource<_Type> &_Src,
   filter_method const& _Filter_proc,
   unsigned int _Timeout = COOPERATIVE_TIMEOUT_INFINITE
);

[ 코드1. receive() 의 선언 ] 

템플릿 매개변수인 _Type 은 message 의 자료 형입니다.

 함수 매개변수 중 _Src 는 message block 의 인터페이스 중 하나인 ISource 를 상속한 message block 객체이며, 이 객체로부터 message 를 수신합니다.

 함수 매개변수 중 _Timeout 은 최대 대기 시간입니다. 이것은 receive() 가 동기 함수이기 때문에 영원히 기다릴 상황을 대비하는 방법입니다. 이 매개변수를 지정했을 때, 최대 대기 시간을 초과하였을 경우, agent::wait()( Asynchronous Agents Library – agent. 2 ( 기능 ) 참고 ) 와 마찬가지로 operation_timed_out 예외를 발생합니다. 그러므로 이 매개변수를 지정 시 반드시 해당 예외를 처리해주어야 합니다. 기본 인자로 COOPERATIVE_TIME_INFINITE 가 지정되어 있으며, 무한히 기다리는 것을 의미합니다.

 함수 매개변수 중 _Filter_proc 는 message 를 거부할 수 있는 필터입니다. message block 생성자로 지정할 수 있는 필터와 마찬가지로 std::tr1::function<bool(_Type const&)> 입니다.

 Message 의 수신이 완료되면 해당 message 를 반환합니다.

예제

- 수신할 message 가 있는 경우

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

using namespace std;
using namespace Concurrency;

int main()
{
	// message block
	unbounded_buffer< wstring > message_block;

	wstring send_message( L"first message!" );
	send( message_block, send_message );

	wstring receive_message;

	try
	{
		receive_message = receive( message_block, 2000 );
	}
	catch( operation_timed_out& e )
	{
		wcout << L"operation_timed_out exception" << e.what() << endl;
	}

	wcout << L"received message: " << receive_message << endl;
}

[ 코드2. receive() 의 수신할 message 가 있는 경우 예제 ]

 receive() 의 매개변수로 최대 대기 시간을 지정했지만 message block 에 message 가 존재하기 때문에 수신하고 바로 반환합니다.

[ 그림1. receive() 의 수신할 message 가 있는 경우 예제 실행 결과 ]

- 수신할 message 가 없는 경우

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

using namespace std;
using namespace Concurrency;

int main()
{
	// message block
	unbounded_buffer< wstring > message_block;

	// wstring send_message( L"first message!" );
	// send( message_block, send_message );

	wstring receive_message;

	try
	{
		receive_message = receive( message_block, 2000 );
	}
	catch( operation_timed_out& e )
	{
		wcout << L"operation_timed_out exception" << e.what() << endl;
	}

	wcout << L"received message: " << receive_message << endl;
}

[ 코드3. receive() 의 수신할 message 가 없는 경우 예제 ]

 send() 를 주석 처리하여 message block 에 전달한 message 가 없기 때문에 receive() 는 지정된 최대 대기 시간인 2초( 2000 milli second ) 동안 기다린 후, operation_timed_out 예외가 발생합니다.

 이 예외를 처리해야 정상적으로 프로그램이 진행됩니다.

 만약 최대 대기 시간을 지정하지 않았다면 무한 대기하게 됩니다.

[ 그림2. receive() 의 수신할 message 가 없는 경우 예제 실행 결과 ]

비 동기 함수 try_receive()

비 동기 함수인 try_receive() 는 message 가 수신될 때까지 기다리지 않습니다. 만약 수신할 message block 에 어떠한 message 도 없다고 하더라도 기다리지 않고, 바로 반환됩니다.

아래는 try_receive() 의 선언입니다.

template <
   class _Type
>
bool try_receive(
   ISource<_Type> * _Src,
      _Type & _value
);

template <
   class _Type
>
bool try_receive(
   ISource<_Type> * _Src,
      _Type & _value,
   filter_method const& _Filter_proc
);

template <
   class _Type
>
bool try_receive(
   ISource<_Type> & _Src,
      _Type & _value
);

template <
   class _Type
>
bool try_receive(
   ISource<_Type> & _Src,
      _Type & _value,
   filter_method const& _Filter_proc
);

[ 코드4. try_receive() 의 선언 ]

 템플릿 매개변수인 _Type 은 receive() 와 마찬가지로 message 의 자료 형입니다.

 함수 매개변수 중 _Src 도 receive() 와 마찬가지로 message block 의 인터페이스 중 하나인 ISource 를 상속한 message block 객체이며, 이 객체로부터 message 를 수신합니다.

 함수 매개변수 중 _value 는 수신한 message 를 저장할 변수의 참조입니다. 수신이 성공하면 message 는 이 참조가 가리키는 변수에 저장됩니다.

 함수 매개변수 중 _Filter_proc 는 receive() 와 마찬가지로 message 를 거부할 수 있는 필터입니다.

 try_receive() 는 수신의 완료를 기다리지 않기 때문에 수신을 시도했을 때( try_receive() 를 호출했을 때 ) message block 에 어떠한 message 도 없다면 false 를 반환해 알려줍니다. message 가 있다면 true 를 반환합니다.

 만약, 수신 시도를 하자마자 시도한 컨텍스트가 계속 진행되기를 원한다면 receive() 에 _Timeout 매개변수에 0 을 지정하기 보다는 try_receive() 를 사용하는게 바람직합니다.

예제

- 수신할 message 가 있는 경우

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

using namespace std;
using namespace Concurrency;

int main()
{
	// message block
	unbounded_buffer< wstring > message_block;

	send( message_block, wstring( L"first message!" ) );

	wstring received_message;
	if( try_receive( message_block, received_message ) )
	{
		wcout << L"receive success" << endl;
		wcout << L"received message: " << received_message << endl;
	}
	else
	{
		wcout << L"receive fail" << endl;
	}
}

[ 코드5. try_receive() 의 수신할 message 가 있는 경우 예제 ]

 try_receive() 는 비 동기 함수이기 때문에 수신할 message 가 있든 없든 먼저 반환됩니다. 수신할 message 가 있으면 true 를 반환하고 인자인 참조 변수에 수신한 message 를 저장합니다,

[ 그림3. try_receive() 의 수신할 message 가 있는 경우 예제 실행 결과 ]

- 수신할 message 가 없는 경우

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

using namespace std;
using namespace Concurrency;

int main()
{
	// message block
	unbounded_buffer< wstring > message_block;

	// send( message_block, wstring( L"first message!" ) );

	wstring received_message;
	if( try_receive( message_block, received_message ) )
	{
		wcout << L"receive success" << endl;
		wcout << L"received message: " << received_message << endl;
	}
	else
	{
		wcout << L"receive fail" << endl;
	}
}

[ 코드6. try_receive() 의 수신할 message 가 없는 경우 예제 ]

 동기 함수인 receive() 와는 달리 비 동기 함수인 try_receive() 는 수신할 message 가 없을 경우, 기다리지 않고 false 를 반환합니다.

 수신할 message 가 있든 없든 바로 반환해야 하는 경우라면 receive() 의 매개변수인 최대 대기 시간을 0 으로 지정하는 것보다는 try_receive() 를 권장합니다.

 receive() 의 최대 대기 시간은 예외 메커니즘을 사용하므로 try_receive() 에 비해 오버헤드가 있을 수 있습니다.

[ 그림4. try_receive() 의 수신할 message 가 없는 경우 예제 실행 결과 ]

Message 필터

 지난 글에서 message block 에 필터를 지정할 수 있다고 언급했습니다. message block 에 지정되는 필터는 message block 에 전송 시에 적용됩니다.

 마찬가지로 수신 함수들에도 필터를 지정할 수 있습니다. message block 으로부터 수신 시에 적용됩니다.

 동기 함수인 receive() 는 수신할 message 가 필터에 의해 수락될 때까지 대기합니다. 즉, 수신할 message 가 필터에 의해 거부된다면 message block 에 message 가 없을 때와 같습니다.

 비 동기 함수인 try_receive() 또한 message block 에 message 가 없을 때와 마찬가지로 false 를 반환합니다.

 다시 말해, message block 에 message 가 없는 경우에도 필터에 의해 거부된다는 말과 같습니다.

 그럼, 필터를 이용한 수신 함수에 대한 예제를 살펴보겠습니다.

예제

#include <iostream>
#include <vector>
#include <iterator>
#include <functional>
#include <agents.h>
#include <ppl.h>

using namespace std;
using namespace std::tr1;
using namespace Concurrency;

class number_collector
	: public agent
{	
public:
	number_collector( ISource< int >& source, vector< int >& result, function< bool ( int ) > filter )
		: source( source )
		, result( result )
		, filter( filter ) { }
	
protected:
	void run()
	{
		while( true )
		{
			int number = receive( this->source, this->filter );

			if( 0 == number )
				break;

			this->result.push_back( number );
		}

		this->done();
	}

private:
	ISource< int >&				source;
	vector< int >&				result;
	function< bool ( int ) >	filter;
};

int main()
{
	// message block
	unbounded_buffer< int > message_block;

	// send number 1 ~ 10.
	parallel_for( 1, 11, [&]( int number )
	{
		send( message_block, number );
	} );

	// send stop signal.
	send( message_block, 0 );	// for even.
	send( message_block, 0 );	// for odd.

	vector< int > even_number_array, odd_number_array;

	number_collector even_number_collector( message_block, even_number_array, []( int number ) -> bool
	{
		return 0 == number % 2;
	} );

	number_collector odd_number_collector( message_block, odd_number_array, []( int number ) -> bool
	{
		if( 0 == number )
			return true;

		return 0 != number % 2;
	} );

	even_number_collector.start();
	odd_number_collector.start();

	// wait for all agents.
	agent* number_collectors[2] = { &even_number_collector, &odd_number_collector };
	agent::wait_for_all( 2, number_collectors );

	// print
	wcout << L"odd numbers: ";
	copy( odd_number_array.begin(), odd_number_array.end(), ostream_iterator< int, wchar_t >( wcout, L" " ) );

	wcout << endl << L"even numbers: ";
	copy( even_number_array.begin(), even_number_array.end(), ostream_iterator< int, wchar_t >( wcout, L" " ) );

	wcout << endl;
}

[ 코드7. 필터를 이용한 숫자 고르기 예제 ]

 우선 message block 에 1 ~ 10 의 정수를 전송합니다. parallel_for() 를 사용하였는데 이 함수는 Concurrency Runtime 위에서 AAL 과 작동하는 돌아가는 Parallel Patterns Library( 이하, PPL ) 에서 제공하는 함수입니다. PPL 에 대한 자세항 사항은 visual studio 팀 블로그에서 확인하실 수 있습니다.

 parallel_for() 는 반복될 내용을 병렬로 처리하기 때문에 성능에 도움을 줍니다. 그러나 반복되는 순서를 보장하지 않습니다.

 그래서 1 ~ 10 의 정수가 전송되는 순서는 알 수 없습니다. 하지만 1 ~ 10 의 정수를 모두 전송한 뒤, 0을 보내서 마지막 message 라는 것을 알려주었습니다. 두 번 보낸 0 중 하나는 짝수를 수신하는 agent 를 위한 것이고, 하나는 홀수를 수신하는 agent 를 위한 것입니다.

 사실, 이런 처리 로직을 구성할 때에는 상태 변화 알림에 유용한 다른 message block 을 사용하는 것이 좋지만 아직 message block 에 대해서 설명하지 않았기 때문에 혼란을 줄이기 위해 간단한 unbounded_buffer 하나만으로 처리하였습니다.

 위 코드에 정의된 agent 인 number_collector 는 message block 으로부터 필터에 의해 필터링된 message 를 컨테이너에 저장합니다.

 동기 함수인 receive() 를 사용했기 때문에 원하는 message 가 올 때까지 기다립니다. 이로 인해 필요한 만큼의 최소의 반복을 하여 오버헤드가 줄어 듭니다.

 만약 비 동기 함수인 try_receive() 를 사용했다면 쓸모 없는 반복 오버헤드를 발생시킬 것입니다. 이 예제의 경우에는 동기 함수인 receive() 가 적합합니다.

 정의된 agent 를 짝수용과 홀수용을 선언하고 start() 를 사용하여 작업을 시작합니다. 그리고 wait_for_all() 을 사용하여 두 agent 가 모두 끝날 때까지 기다린 후, 모든 작업이 종료되면 화면에 수집한 정수들을 출력합니다.

  위 예제 코드는 Visual studio 2008 부터 지원하는 tr1 의 function 과 visual studio 2010 부터 지원하는 C++0x 의 람다를 사용하였습니다. Concurrency Runtime 은 tr1, C++0x 등의 visual studio 2010 의 새로운 feature 들을 사용하여 구현되었기 때문에 이것들에 대해 알아두는 것이 좋습니다.

[ 그림5. 필터를 이용한 숫자 고르기 예제 실행 결과 ]

마치는 글

 이 글에서는 message 전달 함수 중 수신 함수인 receive() 와 try_receive() 에 대해서 알아보았습니다.

 receive() 와 try_receive() 는 사용해야 할 상황이 분명히 다르니 상황에 따라 사용에 유의해야 합니다.

 다음 글에서는 message 가 저장되는 message block 에 대해서 알아보도록 하겠습니다.

Asynchronous Agents Library
– message 전달 함수. 1 ( 전송 )

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

Message 메커니즘

 Asynchronous Agents Library( 이하, AAL ) 에서 message 메커니즘이란 agent 들 간의 데이터 교환이나 동기화 등 상호 작용을 위해 사용되는 기능입니다.

 Message 메커니즘은 크게 message 를 보내고( send ) 받는( receive ) 전달 함수( passing function )와 message 들을 관리하거나 message 에 특별한 기능을 부여하는 message block 으로 구성되어 있습니다.

 실질적으로 message block 이 다루는 message 란 빌트인 자료 형( built-in type )이나 클래스와 같은 사용자 정의 자료 형( user define type ) 의 데이터입니다.

 이 글에서는 먼저 message 를 주고 받는 전달 함수 중 전송 함수에 대해서 알아보겠습니다.

Message 전송

 Message 전달 함수 중 보내는 기능을 하는 함수에는 동기 전송 함수 send() 와 비 동기 전송 함수 asend(), 이렇게 두 가지가 있습니다.

 동기와 비 동기라는 용어가 혼란스러울 수 있기 때문에 잠깐 언급하고 넘어가겠습니다.

 여기서 쓰이는 동기( synchronous )라는 용어는 병렬 처리에서 쓰이는 동기화( synchronization )라는 용어와는 약간은 다른 개념입니다.

 동기화는 다른 두 시간을 하나로 일치시킨다는 뜻으로 행위를 말합니다. 반면 동기는 이미 동기화되었다는 뜻으로 상태를 뜻합니다. 마찬가지로 비 동기는 동기화되지 않았다는 뜻입니다.

 즉, 보통 프로그래밍에서 동기 함수란 그 함수가 호출되고, 그 함수가 반환될 때까지 해당 컨텍스트가 진행되지 않고 기다리다가 반환되고 나서야 컨텍스트가 진행되는 함수를 말합니다. 이것은 사실 컨텍스트가 기다리는 것이 아니라, 해당 컨텍스트가 함수의 내용을 직접 처리하기 때문에 함수를 호출한 입장에서 보면 기다리는 것처럼 보이는 것입니다.

 마찬가지로 비 동기 함수는 함수를 호출한 컨텍스트가 직접 함수의 내용을 처리하지 않고 새로운 작업 스레드를 생성하고 생성된 스레드의 컨텍스트가 진행되기 때문에 함수를 호출한 컨텍스트는 함수를 호출하자마자 함수의 반환을 받고, 계속해서 진행되는 것입니다. 함수를 호출한 컨텍스트는 이러한 비 동기 함수가 언제 실제로 종료될지 모르기 때문에 함수의 반환이 아닌 다른 기법이 필요합니다. 보통 폴링( polling )이나 메시지 또는 콜백 함수와 같은 기법을 사용하여 함수의 종료를 알 수 있습니다.

 그럼 이제 본격적으로 두 message 전달 함수에 대해서 알아 보겠습니다.

동기 전송 함수 send()

 앞에서 설명한 것처럼 send() 는 동기 함수이기 때문에 message 가 전송에 대한 결과가 확실해 졌을 때 반환됩니다. 즉, 전송된 결과가 확실할 때까지 기다린다는 뜻입니다.

  send() 의 선언은 다음과 같습니다.

template <
   class _Type
>
bool send(
   ITarget<_Type> * _Trg,
   const _Type& _Data
);

template <
   class _Type
>
bool send(
   ITarget<_Type> &_Trg,
   const _Type &_Data
);

[ 코드1. send() 의 선언 ]

 템플릿 매개변수인 _Type 은 전송할 message 의 자료 형입니다.

 함수 매개변수 중 _Trg 는 message block 의 인터페이스 중 하나인 ITarget 을 상속받은 message block 객체이며, 전송될 message 를 받게 됩니다. 나중에 message block 에 대해서 자세히 언급할 예정입니다

 또 다른 함수 매개변수인 _Data 가 바로 전송할 message 입니다.

 send() 가 message 전송에 성공했으면 true 를, 그렇지 않으면 false 를 반환합니다.

예제

#include <agents.h>

using namespace Concurrency;

int main()
{
	// message block
	unbounded_buffer< int > message_block;

	send( message_block, 1 );	
}

[ 코드2. send() 예제 ]

 아직 message block 에 대해서 설명하지 않았지만 예제를 위해 message block 중 하나인 unbounded_buffer 를 사용하였습니다.

 message_block 에 1 을 message 로 전송하는 코드입니다. 아래의 캡처 그림을 통해 message 가 message block 에 전송된 것을 확인할 수 있을 것입니다.

 아직 수신 함수를 설명하지 않았기 때문에 코드의 실행 결과가 아닌 코드의 디버깅 화면을 캡처한 그림으로 대신하겠습니다.

[ 그림1. send() 예제 디버깅 화면 ]

비 동기 함수 asend()

 asend() 는 비 동기 함수입니다. 즉, 전송이 완료되기 전에 반환됩니다.

 send() 는 전송 결과를 반환하는 반면, asend() 는 전송 결과가 아닌 message 를 받는 message block 이 전송을 수락했는지 아닌지를 반환합니다.

 아래는 asend() 의 선언입니다.

template <
   class _Type
>
bool asend(
   ITarget<_Type> * _Trg,
   const _Type& _Data
);

template <
   class _Type
>
bool asend(
   ITarget<_Type> &_Trg,
   const _Type &_Data
);

[ 코드3. asend() 의 선언 ]

 템플릿 매개변수와 함수 매개변수는 모두 send() 와 같습니다.

예제

#include <agents.h>

using namespace Concurrency;

int main()
{
	// message block
	unbounded_buffer< int > message_block;

	asend( message_block, 1 );

	Concurrency::wait( 10 );
}

[ 코드4. asend() 예제 ]

 asend() 가 반환되었을 때에는 아직 message block 에 message 가 전송되지 않았습니다. 이것으로 asend() 가 비 동기 함수임을 확인할 수 있습니다.

 약간의 시간( 10 milli second ) 이 지난 후에는 message block 에 message 가 전송된 것을 확인할 수 있습니다.

[ 그림2. asend() 예제 디버깅 화면 - 호출 직 후 ]

[ 그림3. asend() 예제 디버깅 화면 - 약간의 시간이 지난 후 ]

Message 필터

 Message 전송 함수인 send() 의 반환 값이 전송 결과라고 하였고, 실패할 경우 false 를 반환한다고 하였습니다. 사실, 실패할 경우란 message 를 받는 message block 이 전송을 거부할 경우, 즉 필터링되었을 경우입니다.

 결론적으로 send() 와 asend() 의 반환 값은 모두 message block 의 수락 또는 거절 여부입니다.

 여기서 집고 넘어가야 할 부분이 언제 전송이 거부되는 것인가 하는 것입니다.

 message block 은 두 가지 경우에 message 전송을 거부합니다.

 첫째는 message block 이 파괴되어 소멸자가 처리되고 있을 때입니다. 당연한 상황입니다.

 둘째는 message block 의 필터에 의해 message 가 거부당했을 때입니다. 모든 message block 의 생성자 중에는 filter_method 형의 매개변수를 갖는 생성자가 있습니다. filter_method 형은 사실 std::tr1::function<bool(_Type const&)> 입니다. message block 을 생성하는 클라이언트는 임의의 message 필터를 적용할 수 있습니다. 이 필터 함수가 false 를 반환할 경우, message 전송은 거부됩니다.

예제

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

using namespace std;
using namespace Concurrency;

int main()
{
	// 필터( 짝수만 수락 )가 적용된 message block
	unbounded_buffer< int > buffer( []( int message ) -> bool
	{
		return 0 == ( message % 2 );

	} );

	wcout << L"send 1: " << ( send( buffer, 1 ) ? L"accepted" : L"declined" ) << endl;	// 거부
	wcout << L"send 2: " << ( send( buffer, 2 ) ? L"accepted" : L"declined" ) << endl;	// 수락

	wcout << L"asend 3: " << ( asend( buffer, 3 ) ? L"accepted" : L"declined" ) << endl;	// 거부
	wcout << L"asend 4: " << ( asend( buffer, 4 ) ? L"accepted" : L"declined" ) << endl;	// 수락
}

[ 코드5. 전송 거부 예제 ]

 이 예제에는 message block 의 필터로 Visual studio 2010 에서 지원하는 C++0x 의 람다를 사용하였습니다. 람다는 이 글의 논제에서 벗어나기 때문에 설명하지 않도록 하겠습니다. Visual studio 팀 블로그에서 람다에 대한 정보를 얻을 수 있습니다.

 간단히 람다에 대해서 설명하고 넘어가자면 익명의 함수 객체라고 보셔도 될 것입니다. 

 예제에 사용된 message block 의 필터는 짝수만 수락하는 필터입니다. 그래서 실행 결과로 send() 와 asend() 모두 홀수는 거부되었고, 짝수는 수락되는 것을 볼 수 있습니다.

[ 그림4. 전송 거부 예제 실행 결과 ]

마치는 글

 이 글에서는 message 전달 함수 중 전송 함수들에 대해서 알아보았습니다.

 이 함수들 중 어떤 것을 사용하는 것이 적절한지를 판단하기 위해서는 반드시 동기와 비 동기에 대한 개념의 이해가 필요합니다.

 상황에 따라 적절한 함수를 사용하시면 원하는 결과를 얻을 수 있을 것입니다.

 전송 함수들에 대해서 알아보았지만 아직 수신 함수들에 대해 알아보지 않았습니다. 다음 글에서는 message block 으로부터 message 를 수신하는 수신 함수들에 대해 알아보겠습니다.

nullptr

C/C++에서 포인터를 초기화 할 때 ‘NULL’을 사용합니다. 그러나 VC++ 10에는 C++0x에서는 포인터를 초기화 할 때 NULL 대신 새로 생긴 ‘nullptr’을 사용할 수 있게 되었습니다.

C++/CLI는 이전부터 nullptr이 있었습니다.

C++/CLI에서는 ref 클래스의 핸들을 초기화 할 때는 nullptr을 사용합니다.

C++/CLI, C++0x의 nullptr은 C/C++ 처럼 ‘0’이 아니라는 것을 잘 기억하시기 바랍니다.

interior_ptr

interior_ptr은 관리 힙(managed heap. 즉 GC겠죠) 상의 value type나 기본형을 가리키는 포인터라고 할 수 있습니다. interior_ptr는 value type나 기본형을 비관리 코드의 포인터처럼 사용하고 싶을 때 사용하면 좋습니다.

< 코드 1. >

위 코드를 빌드 해 보면 SetValue( &refClass->nValue ); 에서 빌드 에러가 발생합니다. 매니지드 힙에 있는 것은 그 위치가 변하므로 비 관리 코드의 포인터를 넘길 수가 없습니다. 그럼 <코드 1>를 정상적으로 빌드 하기 위해서 interior_ptr를 사용해 보겠습니다.

< 코드 2. >

<코드 2>의 SetValue의 파라미터로 비관리 코드의 참조나 포인터를 넘길 수도 있습니다.

< 코드 3. >

그리고 interior_ptr에 대신 C++/CLI의 참조(‘%’)를 사용하는 방법도 있습니다.

pin_ptr

pin_ptr은 관리 힙 상의 value type나 기본형을 비관리 코드에서 포인터로 사용하고 싶을 때 사용하는 기능입니다. 가장 필요한 경우가 C++/CLI에서 기존의 비관리 코드로 만들어 놓은 라이브러리를 사용할 때입니다.

< 코드 4. >

pin_ptr에 메모리 주소를 할당하는 것을 ‘pin’이라고 부르고 사용이 끝난 후 nullptr로 초기화 하는 것을 ‘unpin’ 이라고 부릅니다. pin_ptr 사용이 끝난 후 가능한 빨리 unpin 해주는 것이 좋습니다.

interior_ptr과 pin_ptr의 차이점

interipor_ptr과 pin_ptr은 둘 다 관리 힙 상의 value type이나 기본형을 가리키는 포인터로 사용되지만 interior_ptr은 관리 힙 상에서 인스턴스가 이동하여도 올바르게 추적할 수 있는 포인터로 런타임의 지배하에 있습니다(즉 인스턴스가 관리 힙 상에서 이동하여도 괜찮습니다).

pin_ptr은 관리 힙 상의 value type을 비관리 코드에서 사용하고 싶을 때 사용합니다. 당연히 이 때는 관리 힙에 있는 인스턴스가 이동하면 안되므로 인스턴스의 이동을 금지합니다.

interipor_ptr과 pin_ptr의 같은 점 : 포인터처럼 사용할 수 있다.

interipor_ptr과 pin_ptr 다른 점 : interipor_ptr은 관리 코드 상에서 포인터로 사용하고, pin_ptr는 비관리 코드에 포인터로 넘길 때 사용합니다.

interipor_ptr과 pin_ptr을 공부했으니 다음에는 C++/CLI에서 비관리 C++과 혼합해서 사용할 때 어떻게 해야 하는지 설명하겠습니다.

참고

http://cppcli.shacknet.nu/cli:interior_ptr

http://cppcli.shacknet.nu/cli:pin_ptr

http://cppcli.shacknet.nu/cli:interior_ptr%E3%81%A8pin_ptr%E3%81%AE%E9%81%95%E3%81%84

프로그래밍 할 때 가장 자주 사용하는 자료구조가 바로 배열입니다. 배열을 사용하지 않고 프로그래밍 하기는 힘들죠^^.

그래서 이번에는 C++/CLI에서의 배열에 대해서 이야기하려고 합니다.

C++/CLI에서의 배열은 ‘array’

비관리 C++에서는 배열은 ‘[]’을 사용합니다.

그러나 C++/CLI에서의 배열은 ‘array’라는 클래스를 사용합니다.

int 형의 3개의 요소를 가지는 배열은 아래와 같이 정의합니다.

다음은 간단한 사용 예입니다.

< 코드 1. >

array에 유저 정의형 사용하기

array에는 기본형(int, float 등)만이 아닌 유저 정의형도 사용할 수 있습니다. 다만 비관리 클래스는 안됩니다. 오직 관리 클래스(ref class)만 가능합니다. 또 그냥 ref 클래스를 그대로 넣을 수는 없는 클래스의 핸들을 사용해야 합니다(ref 클래스는 GC에 동적 할당을 하기 때문이겠죠).

for each 사용하기

앞서 <코드1>의 예제에서는 배열의 모든 요소를 순환하기 하기 위해 ‘for’문을 사용했습니다. 그러나 .NET에서는 for문 보다 ‘for each’문을 사용하는 것이 성능이나 안정성 등에서 더 좋습니다(다만 for each를 사용하면 내부에서 값을 변경할 수 없다는 문제는 있습니다).

< 코드 2. >

다 차원 배열

array는 너무 당연하게 다 차원 배열도 만들 수 있습니다.

< 코드 3. 2차원 배열의 예 >

array의 기능은 .NET 라이브러리의 array 클래스를 사용하므로 자세한 사용방법은 MSDN에서 .NET 라이브러리 부분을 참고 하시가 바랍니다.

다음에는 nullptr, interior_ptr, pin_ptr 설명과 관리코드의 타입을 비관리 코드로 넘길 때 어떤 작업을 하는지 이야기 하겠습니다.^^

Visual Studio 2010 이 출시된 이후 짧은 시간 동안 많은 변화가 있었습니다. 그리고 그 변화의 움직임을 직접 느끼고 있고요. 많은 기업과 고객, 그리고 여러분들께서 보여주신 관심은 기대 이상이었습니다. 

이에 질세라 저희 Visual Studio 2010 공식 팀에서 여러분들을 위해 많은 문서를 무료로 만들어 배포하고 있습니다.

아래의 링크에서 방대한 양의 문서를 PDF 로 무료로 다운로드하실 수 있습니다.
http://msdn.microsoft.com/ko-kr/vstudio/default.aspx

Visual Studio 2010

Visual Studio 2010이 필요한 10가지 이유-라인업 및 기능소개 브로셔
Visual Studio 2010 UPGRADE NOW 프로모션 브로셔
Visual Studio 2010 라이센스 가이드
Visual Studio 2010 을 활용한 ALM 백서 (.NETXPERT 재직 중 엄준일 저)

Team Foundation Server 2010

Team Foundation Server 2010 설치 가이드 (다중서버) (.NETXPERT 재직 중 엄준일 저)
Team Foundation Server 2010 활용가이드 (FQDN) (.NETXPERT 재직 중 엄준일 저)
Team Foundation Server 2010 설치 가이드 (단일서버) (.NETXPERT 재직 중 엄준일 저)
Team Foundation Server 2010 설치 가이드 (Lab 환경구성) (.NETXPERT 재직 중 엄준일 저)

Language

First Look C#4.0 백서 (강보람 저)
Visual Studio 2010 과 C++0x 백서 (최흥배 저)

Visual Studio 2010 단축키

Visual Studio 2010 Visual Basic 단축키 포스터 (방수철님 역)
Visual Studio 2010 Visual F# 단축키 포스터 (방수철님 역)
Visual Studio 2010 Visual C++ 단축키 포스터 (방수철님 역)
Visual Studio 2010 Visual C# 단축키 포스터 (방수철님 역)

Asynchronous Agents Library – agent. 2 ( 기능 )

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

void run();

Agent 클래스를 상속 받아 작업을 하는 agent 를 만들 때, 처음으로 해야 할 일은 run() 재정의입니다.

 run() 는 CPU 가 해당 agent 스레드의 컨텍스트를 처리할 때, 수행되는 메소드입니다. 즉, 바로 agent 가 책임지고 처리해야 할 작업( task )이고, run() 을 재정의하기 위해 agent class 가 존재한다고 해도 과언이 아닐 정도로 중요합니다.

 Asynchronous Agents Library( 이하 AAL )을 사용하지 않고 Win32 API 로 직접 스레드를 생성할 때, 지정하는 콜백 함수와 같은 역할을 합니다.

 run() 에 필요한 정보( 매개 변수 )가 있다면 agent 를 상속 받은 클래스의 생성자를 이용하여 전달하면 됩니다.

 run() 이 호출될 때, agent 의 상태는 agent_started 가 됩니다.

 run() 을 재정의할 때, 주의할 점은 run() 이 끝나기 전에 done() 을 호출해야 한다는 것입니다. 실제로 run() 이 끝났다는 것은 작업이 끝난 것이지만 상태는 여전히 agent_started 이기 때문에 계속 수행 중인 것으로 인식됩니다. 그러므로 agent 의 상태를 바꿔주기 위해 반드시 run() 이 끝나기 전에 done() 을 호출해야 합니다.

또한 run() 은 어떤 예외도 던지지 않습니다.

bool done();

Agent 의 작업이 완료되었음을 알립니다. 이것은 agent 의 상태를 agent_done 으로 바꾸는 것을 의미합니다.

제대로 agent_done 상태가 되면 true 를 반환합니다. cancel() 에 의해 agent_cancel 상태인 agent 는 agent_done 상태가 되지 않고 false 를 반환합니다.

 protected 로 지정되어 있어 메소드 내에서만 호출할 수 있습니다.

bool start();

 start() 를 호출함으로써 CPU 스케줄에 의해 run() 이 호출되는 것입니다. run() 이 호출되기 위해서는 반드시 start() 를 호출해야 합니다. 직접 run() 을 호출하면 병렬 처리 또는 비 동기 처리되지 않고, 호출한 스레드의 컨텍스트에서 일반 함수를 호출한 것과 같게 됩니다.

그러므로 직접 run() 을 호출하는 일은 없어야 하며, 꼭 start() 를 호출하도록 해야 합니다.

 start() 는 agent 의 상태를 agent_created 에서 agent_runnable 로 바꿉니다. 즉, 스케줄하여 컨텍스트 스위칭( context switching ) 의 대상이 되도록 합니다.

Agent 가 제대로 스케줄 되었다면 true 를 반환합니다. 스케줄 되기 전( start() 호출 전 )에 cancel() 을 호출하면 스케줄 되지 않고 false 를 반환합니다.

bool cancel();

Agent 객체의 작업을 취소할 때 사용합니다.

Agent 객체가 생성되어 agent_created 상태가 되거나, start() 에 의해 agent_runnable 상태일 때에 작업을 취소하고 종료된 상태인 agent_cancel 상태로 바꿉니다.

다시 말해, run() 이 호출되어 agent_started 상태에서는 agent_cancel 상태로 바뀌지 않고 실패하여 false 를 반환합니다. 제대로 agent_cancel 상태로 바뀌었다면 true 를 반환합니다.

agent_status status();

Agent 객체의 현재 상태를 반환합니다.

 agent_status 는 enum 형으로 agent_canceled( 취소됨 ), agent_created( 생성됨 ), agent_done( 작업 완료 ), agent_runnable( 스케줄 됨 ), agent_started( 실행 중 ) 를 나타냅니다.

반환된 상태는 동기화되어 정확한 상태를 반환하지만, 반환하자마자 agent 의 상태가 변할 수 있어 반환된 상태가 현재 agent 의 상태라고 확신하면 안 됩니다.

ISource<agent_status> * status_port();

 status() 는 동기화된 agent 의 상태를 반환하는 반면, status_port() 는 비 동기 메커니즘은 message 를 통해 반환됩니다.

반환형인 ISource 는 message 메커니즘의 interface 입니다. ISource interface 형은 receive() 로 내용을 꺼내올 수 있습니다.

아직 message 에 대해서 언급하지 않았기 때문에 이해가 안 될 수 있습니다. 곧 message 에 대해서 설명할 것인데 그 때, 이 함수가 어떻게 동작하는지 알 수 있을 것입니다.

static agent_status wait( agent * _PAgent, unsigned int _Timeout = COOPERATIVE_TIMEOUT_INFINITE );

Win32 API 의 WaitForSingleObject() 와 같은 기능을 합니다.

인자로 넘긴 agent 의 작업이 종료 상태가 될 때까지 기다립니다. 종료 상태란 agent_cancel, agent_done 상태를 말합니다.

기다릴 최대 시간( timeout )을 정할 수 있는데 COOPERATIVE_TIMEOUT_INFINITE 는 무한대를 의미하며, 기본 값으로 지정되어 있습니다. 이 때, 최대 시간은 밀리 초( millisecond ) 단위입니다.

최대 시간을 정했을 경우, 최대 시간까지 agent 의 작업이 종료되지 않으면, operation_timed_out 예외가 발생합니다. 그러므로 최대 시간을 고려한 프로그래밍을 할 경우, 이 예외를 처리함으로써 최대 시간 이 후를 제어해야 합니다.

Agent 의 상태를 반환하는데, 이 상태는 agent_cancel 또는 agent_done 입니다. 왜냐하면 앞의 두 상태 중 하나가 되어야만 반환을 하기 때문입니다.

static void wait_for_all( size_t _Count, __in_ecount(_Count) agent ** _PAgents, __out_ecount_opt(_Count) agent_status * _PStatus = NULL, unsigned int _Timeout = COOPERATIVE_TIMEOUT_INFINITE );

Win32 API 의 WaitForMultipleObjects() ( 3번째 인자가 TRUE 인 )와 같은 기능을 합니다.

인자로 전달되는 agent 배열이 모두 완료되었을 때까지 기다립니다.

인자를 살펴보면,

• _Count - 기다릴 agent 의 개수( 뒤의 인자인 agent 배열의 원소 수와 같아야 합니다. )
• _PAgents - 기다릴 agent 들의 배열
• _pStatus – 이 함수가 반환될 때, agent 들의 상태들을 저장할 배열( _Timedout 을 지정했을 때, operation_timed_out 예외가 발생한다면, 상태를 저장하는 작업은 수행되지 않습니다. ), 기본 값은 NULL 입니다.
• _Timeout – 기다릴 최대 시간, 기본 값은 COOPERATIVE_TIMEOUT_INFINITE 이다.

 wait() 와 마찬가지로 기다릴 최대 시간을 지정할 경우, operation_timed_out 예외가 발생되며, 최대 시간을 고려할 경우, 이 예외를 처리해야 합니다.

static void wait_for_one( size_t _Count, agent ** _PAgents, agent_status& _Status, size_t& _Index, unsigned int _Timeout = COOPERATIVE_TIMEOUT_INFINITE );

Win32 API 의 WaitForMultipleObjects() ( 3번째 인자가 FALSE 인 )와 같은 기능을 합니다.

인자로 전달되는 agent 배열 중 하나라도 완료가 될 때까지 기다립니다.

인자를 살펴보면,

• _Count – 기다릴 agent 의 개수( 뒤의 인자인 agent 배열의 원소 수와 같아야 합니다. )
• _PAgent – 기다릴 agent 들의 배열
• _Status – 반환될 때의 agent 상태를 저장할 agent_status 변수
• _Index – agent 배열 중 완료된 agent 의 인덱스를 저장할 변수
• _Timeout – 기다릴 최대 시간, 기본 값은 COOPERATIVE_TIMEOUT_INFINITE 입니다.

 wait() 와 마찬가지로 기다릴 최대 시간을 지정할 경우, operation_timed_out 예외가 발생하며, 최대 시간을 고려할 경우, 이 예외를 처리해야 합니다.

예제

아래의 코드는 위에 설명한 메소드들을 사용하는 상황을 보여줍니다. 특별한 시나리오는 없지만, 충분히 어떤 상황에 어떻게 사용하는지 알 수 있을 것입니다.

 TestAgent 의 run() 안에서 사용된 Concurrency::wait() 는 Win32 API 의 Sleep() 과 같은 기능을 합니다. agent::wait() 와 혼동하지 않기를 바랍니다.

예제를 위한 준비

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

using namespace std;
using namespace Concurrency;

const wchar_t* GetAgentStatusString( agent_status status )
{
	switch( status )
	{
	case agent_created:
		return L"agent_created";

	case agent_runnable:
		return L"agent_runnable";

	case agent_started:
		return L"agent_started";

	case agent_done:
		return L"agent_done";

	case agent_canceled:
		return L"agent_canceled";

	default:
		return L"unknown";
	}
}

class TestAgent
	: public agent
{
private:
	unsigned int	id;

protected:
	virtual void run()
	{
		Concurrency::wait( this->id * 500 );		

		wcout << L"agent id: " << this->id << L" completed." << endl;

		done();
	}
public:
	TestAgent( unsigned int id )
		: id( id ) { }
};

[ 코드1. 예제를 위한 코드 ]

status_port() 의 사용

int main()
{	
	TestAgent testAgent( 5 );

	wcout << GetAgentStatusString( receive( testAgent.status_port() ) ) << endl;

	testAgent.start();
	
	for( unsigned int i = 0; i < 10; ++i )
	{		
		wcout << GetAgentStatusString( receive( testAgent.status_port() ) ) << endl;
	}

	agent::wait( &testAgent );

	wcout << GetAgentStatusString( receive( testAgent.status_port() ) ) << endl;
}

[ 코드2. status_port() 사용 예제 ]

 status() 와 같은 기능을 하지만 비 동기 메커니즘인 message 을 사용한다는 것이 다릅니다.

이 말은 status() 가 수행되는 동안 호출한 컨텍스트가 멈추어 있지만, status_port() 는 호출한 컨텍스트는 계속 진행되고, 다른 work 스레드의 컨텍스트가 상태를 처리하고, 그 결과를 message 로 받는 다는 말입니다.

[ 그림1. status_port() 사용 예제 실행 결과 ]

wait()의 사용

int main()
{
	TestAgent testAgent1( 3 );
	TestAgent testAgent2( 2 );
	TestAgent testAgent3( 1 );
	TestAgent testAgent4( 5 );
	TestAgent testAgent5( 4 );

	testAgent1.start();
	testAgent2.start();
	testAgent3.start();
	testAgent4.start();
	testAgent5.start();

	agent::wait( &testAgent1 );
}

[ 코드3. wait() 사용 예제 ]

각기 다른 5 개의 작업들을 수행하는 agent 들을 생성합니다. 각각의 생성자의 인자는 각 agent 의 id 이기도 하지만, 해당 작업들의 가중치이기도 합니다.

testAgent1 의 작업이 끝날 때까지 기다립니다. 곧바로 프로그램이 종료되므로 작업이 끝나지 않은 agent 들은 비정상적으로 종료됩니다. 보통 작업이 진행 중인 모든 agent 들이 완료될 때가지 기다리는 것이 좋습니다.

[ 그림2. wait() 사용 예제 실행 결과 ]

wait() 와 timeout

int main()
{
	TestAgent testAgent1( 3 );
	TestAgent testAgent2( 2 );
	TestAgent testAgent3( 1 );
	TestAgent testAgent4( 5 );
	TestAgent testAgent5( 4 );

	testAgent1.start();
	testAgent2.start();
	testAgent3.start();
	testAgent4.start();
	testAgent5.start();

	try
	{
		agent::wait( &testAgent4, 1000 );
	}
	catch( operation_timed_out& )
	{
		wcout << L"operation timed out." << endl;
	}
}

[ 코드4. wait() 와 timeout 예제 ]

 wait() 의 timeout 매개변수를 사용하려면 operation_timed_out 예외를 처리해야 합니다.

[ 그림3. wait() 와 timeout 예제 실행 결과 ]

wait_for_all() 의 사용

int main()
{
	TestAgent testAgent1( 3 );
	TestAgent testAgent2( 2 );
	TestAgent testAgent3( 1 );
	TestAgent testAgent4( 5 );
	TestAgent testAgent5( 4 );

	testAgent1.start();
	testAgent2.start();
	testAgent3.start();
	testAgent4.start();
	testAgent5.start();

	agent* runningAgents[5] = {
		&testAgent1,
		&testAgent2,
		&testAgent3,
		&testAgent4,
		&testAgent5
	};

	unsigned int runningAgentCount = sizeof( runningAgents ) / sizeof( runningAgents[0] );	

	agent::wait_for_all( runningAgentCount, runningAgents );
}

[ 코드5. wait_for_all() 사용 예제 ]

배열에 포함된 모든 agent 들이 완료될 때까지 기다립니다.

[ 그림4. wait_for_all() 사용 예제 실행 결과 ]

wait_for_all() 과 상태 반환

int main()
{
	TestAgent testAgent1( 3 );
	TestAgent testAgent2( 2 );
	TestAgent testAgent3( 1 );
	TestAgent testAgent4( 5 );
	TestAgent testAgent5( 4 );

	testAgent1.start();
	testAgent2.start();
	testAgent3.start();
	testAgent4.start();
	testAgent5.start();

	agent* runningAgents[5] = {
		&testAgent1,
		&testAgent2,
		&testAgent3,
		&testAgent4,
		&testAgent5
	};

	agent_status statuses[5];

	unsigned int runningAgentCount = sizeof( runningAgents ) / sizeof( runningAgents[0] );
	agent::wait_for_all( runningAgentCount, runningAgents, statuses );

	for( unsigned int i = 0; i < runningAgentCount; ++i )
		wcout << GetAgentStatusString( statuses[ i ] ) << endl;
}

[ 코드6. wait_for_all() 과 상태 반환 예제 ]

완료된 agent 들의 상태를 저장하는 배열의 크기는 agent 들을 포함한 배열의 크기와 같아야 합니다다. 만약 timeout 예외를 사용한다면 상태는 저장되지 않습니다.

[ 그림5. wait_for_all() 과 상태 반환 예제 실행 결과 ]

wait_for_one() 의 사용

int main()
{
	TestAgent testAgent1( 3 );
	TestAgent testAgent2( 2 );
	TestAgent testAgent3( 1 );
	TestAgent testAgent4( 5 );
	TestAgent testAgent5( 4 );

	testAgent1.start();
	testAgent2.start();
	testAgent3.start();
	testAgent4.start();
	testAgent5.start();

	agent* runningAgents[5] = {
		&testAgent1,
		&testAgent2,
		&testAgent3,
		&testAgent4,
		&testAgent5
	};

	agent_status finalStatus;
	unsigned int indexOfcompletedAgentFirst;

	unsigned int runningAgentCount = sizeof( runningAgents ) / sizeof( runningAgents[0] );	
	agent::wait_for_one( runningAgentCount, runningAgents, finalStatus, indexOfcompletedAgentFirst );

	wcout << L"final status : " << GetAgentStatusString( finalStatus ) << endl;
	wcout << L"index of completed agent first : " << indexOfcompletedAgentFirst << endl;	
}

[ 코드7. wait_for_one() 사용 예제 ]

여러 agent 들 중 하나라도 완료 되었을 때 반환됩니다. 상태 저장을 위한 변수와 인덱스 저장을 위한 변수는 참조로 넘겨야 하므로 생략할 수 없습니다.

[ 그림6. wait_for_one() 사용 예제 실행 결과 ]

마치는 글

위의 내용을 모두 숙지했다면 agent 를 마치 하나의 work 스레드처럼 다룰 수 있을 것입니다. 이렇게 하여 비 동기 병렬 처리를 쉽게 처리할 수 있습니다.

하지만 이것이 끝이 아닙니다. agent 에 message 메커니즘을 적용한다면 더욱 더 지능적인 agent 를 쉽게 만들 수 있습니다.

다음 글에서는 message 메커니즘에 대해 소개해 보도록 하겠습니다.

gcnew로 생성하지 않기

C++/CLI는 클래스를 생성할 때 ‘gcnew’를 사용하지 않고 생성할 수도 있습니다.

< 리스트 1. ‘gcnew’를 사용하지 않고 클래스 생성하기 >

< 결과 >

<리스트 1>의 ManagedTest MTest; 는 비 관리 C++처럼 GC가 아닌 스택 영역에 생성하는 것으로 착각할 수도 있겠지만 전혀 아닙니다. 클래스를 생성하면 언제나 GC 영역에 만들어집니다.

위의 코드는 그냥 ‘gcnew’ 사용을 우리가 생략하고 컴파일러가 대신 써 준다고 생각하시면 됩니다.

은 컴파일러에 의해서 아래의 코드로 바뀝니다.

위의 코드를 보시면 아시듯이 gcnew를 사용하지 않고 클래스를 생성하면 우리가 직접 delete를 쓰지 않아도 되는 편리함을 얻을 수 있습니다.

gcnew를 사용하지 않는 것은 C#에서 ‘using’을 간략화 시킨 것으로 생각하면 좋습니다.

< C#의 using 문 사용 예 >

value 클래스

관리 클래스는 복사 생성자와 대입 연사자를 가지지 못하므로 아래의 코드는 컴파일 에러가 발생합니다.

< 리스트 2. >

그러나 클래스를  ‘ref’가 아닌 ‘value’ 클래스로 정의하면 위의 코드는 컴파일 할 수 있습니다.

value 클래스는 클래스간 복사를 할 수 있는 능력이 있습니다. 이 복사는 비트 단위의 복사로 이른바 ‘memcpy’와 같은 복사입니다.

value 클래스는 복사를 할 수 있으므로 value 클래스의 멤버는 절대 복사 가능한 멤버만 가질 수 있습니다. 그래서 아래와 같은 value 클래스 C는 컴파일 에러가 발생합니다.

또 value 클래스의 특징으로는 ref 클래스가 GC에서 만들어지는 것과 달리 스택에 만들 수 있습니다.

위 코드에서 C 클래스는 스택에 만들어집니다(물론 gcnew를 사용하면 GC에 만들어집니다)

value 클래스의 특징은 좀 더 있는데 위에 설명한 것들과 포함해서 아래와 같이 정리할 수 있습니다.

value 클래스의 특징

1. 기본 생성자를 가질 수 없다.

2. 복사 생성자를 가질 수 없다.

3. 대입 연산자를 가질 수 없다.

4. 소멸자를 가질 수 없다.

5. finalize를 가질 수 없다.

6. 클래스간 복사를 할 수 있다.

7. 복사 불가능한 것을 멤버로 가질 수 없다(ref 클래스 등).

8. 스택에 생성할 수 있다.

9. 다른 클래스를 계승할 수 없다(단 interface는 가능하다)

관리 클래스를 파라메터로 넘기기

함수 foo1의 파라미터 정의는 클래스 A가 value 클래스일 때만 사용할 수 있습니다. 그러므로 아래와 같이 foo1의 파라미터를 정의해야 합니다.

그 외….

C++/CLI에서 참조는 ‘%’을 사용합니다. 이것은 비 관리의 ‘&’와 구별이 됩니다.

%는 아래와 같은 경우에 유용하게 사용할 수 있습니다.

참조는 C#에서는 'ref', VB.NET에서는 'ByRef'와 같다고 생각하시면 됩니다.

// 값을 참조로 넘기는 경우

구조체

C++/CLI에서의 구조체가 클래스와 다른 점

1. 'value class'와 같습니다. 물론 비 관리코드에서는 기존의 C++과 같다

2. 구조체는 상속 받을 수 없다.

Asynchronous Agents Library – agent. 1 ( 상태 )

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

Agent 란 무엇인가

개념

 Agent 는 사전적 의미로 동작이나 행위를 행하는 사람이나 사물이라는 뜻을 가지고 있습니다. 그 말 그대로 Asynchronous Agents Library( 이하 AAL ) 의 agent 는 어떤 행위를 하는 객체를 나타냅니다.

 지난 글에서 AAL 이 actor-based programming 이라고 언급한 적이 있습니다. 여기에서 actor 가 바로 agent 를 일컫습니다.

 이해를 돕기 위해 비유를 하자면 agent 는 하나의 작업이라고 생각해도 좋습니다. 그런데 이 작업은 단순히 한 번 처리되는 작업을 말하는 것이 아니라, 어떤 책임이나 역할을 하는 작업입니다. 우리는 멀티 스레드 프로그래밍을 할 때, 이런 개념을 worker 라고 말하기도 합니다.

 예를 들어, 네트워크 프로그래밍을 할 때, 소켓에 연결 요청을 기다리고 요청이 들어오면 연결해주는 역할을 하는 스레드가 필요합니다. 이 스레드는 어떤 책임이나 역할을 하는 작업을 가지고 있습니다. 이 스레드를 객체로 표현하면 agent 라고 할 수 있습니다.

클래스

 실제 개발 시 필요한 agent 클래스는 Concurrency 네임스페이스 안에 존재하고, agents.h 파일 안에 정의되어 있습니다.

 agent 클래스는 추상 클래스로 agent 로 만들 클래스가 상속해야 합니다. agent 클래스의 추상 메소드는 void run() 이고, 반드시 구현해야 합니다.

- 예

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

using namespace std;
using namespace Concurrency;

class TestAgent
	: public agent
{
protected:
	void run()
	{
		wcout << L"running.." << endl;
		this->done();
	}
};

[ 코드1. agent 상속 예 ]

agent 의 상태

 agent 는 비 동기 처리를 목적으로 하기 때문에, 내부적으로 스레드를 사용할 수 밖에 없습니다. agent 의 상태란 agent 의 작업을 처리하는 스레드의 상태라고 볼 수 있습니다.

 agent 는 생명 주기를 갖는데, 이 주기는 상태로 표현됩니다.

[ 그림1. agent 생명 주기 ]

 처음 agent 상태는 크게 초기 상태, 활성화 상태, 종료 상태로 나눌 수 있습니다. agent 객체를 생성하면 그 agent 는 초기 상태인 created 상태가 됩니다. start() 를 호출하면 Concurrency Runtime 에 의해 해당 agent 가 스케줄 되고 runnable 상태가 됩니다.

 created 나 runnable 상태에서 cancel() 을 호출하여 작업을 취소하면 canceled 상태가 되는데 이는 종료 상태 중 하나입니다.

 위의 그림에서 점선으로 표시된 run() 은 우리가 명시적으로 호출하는 것이 아니라 runnable 상태인 agent 를 Concurrency Runtime 이 호출하는 것을 의미합니다. run() 이 호출되면 활성화 상태인 started 상태가 됩니다.

 만약 모든 작업의 수행이 완료되었으면 done() 함수를 호출하여 종료 상태인 done 상태로 바꾸어야 합니다. 동기화를 위한 wait() 등의 함수는 해당 agent 가 종료 상태 즉, done 이나 canceled 상태가 되어야 반환됩니다.

 agent 생명 주기에서 가장 중요한 것은 순환되지 않는다는 것입니다. 즉, 되돌아 갈 수 없습니다. 한 번 종료 상태를 갖은 agent 객체는 이미 쓸모 없는 객체가 되고 재사용할 수 없습니다. 그러므로 해당 작업을 다시 수행하기 위해서는 새로운 agent 객체를 생성해야 합니다.

 다음은 각 상태에 대한 정의입니다.

[ 표1. agent status ]

 이 상태들은 status() 메소드로 알아 낼 수 있습니다. status() 는 동기화되기 때문에 정확한 상태를 반환하지만, 그 상태가 현재의 상태라고 보장할 수 없습니다. 왜냐하면 status() 가 반환된 직 후 agent 의 상태가 바뀔 수 있기 때문입니다.

- 예

코드

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

using namespace std;
using namespace Concurrency;

class TestAgent
	: public agent
{
protected:
	void run()
	{
		wcout << L"running.." << endl;
		this->done();
	}
};

void print_agent_status( agent& a )
{
	wstring status;
	switch( a.status() )
	{
	case agent_status::agent_created:	status = L"agent_created";	break;
	case agent_status::agent_runnable:	status = L"agent_runnable";	break;
	case agent_status::agent_started:	status = L"agent_started";	break;
	case agent_status::agent_done:		status = L"agent_done";		break;
	case agent_status::agent_canceled:	status = L"agent_canceled";	break;
	}

	wcout << status.c_str() << endl;
}

int main()
{
	TestAgent testAgent;

	print_agent_status( testAgent );

	testAgent.start();

	for( int i = 0; i < 10; ++i )
	{
		print_agent_status( testAgent );
	}

	agent::wait( &testAgent );

	print_agent_status( testAgent );
}

[ 코드2. agent 생명 주기 ]

실행 결과 

[ 그림2. 코드2 실행 결과 ]

마치는 글

 이번 글에서는 agent 의 개념과 클래스, 그리고 상태에 대해서 알아보았습니다. agent 를 사용하기 위해서는 조금 더 알아야 할 것들이 있습니다.

 조금 더 알아야 할 내용들은 다음 글에 작성해 보도록 하겠습니다.

참고

• 그림1. agent 생명 주기 - http://i.msdn.microsoft.com/dynimg/IC338844.png

C와 C++의 큰 차이점의 하나가 바로 C++에만 있는 ‘클래스’입니다. 또 클래스는 객체 지향 프로그래밍 언어에서 자주 볼 수 있습니다.

C++/CLI에서 관리 클래스는 ‘ref’라는 키워드를 사용하여 만듭니다.

비 관리 클래스

관리 클래스

관리 클래스는 ‘ref’를 제외하고는 외관상으로는 비 관리 클래스와 비슷하지만 관리와 비 관리가 많이 다르듯이 관리 클래스는 비 관리 클래스와 다른점이 있습니다. 그러므로 이것을 잘 파악하고 있어야 합니다.

관리 클래스의 특징

1. 정적 할당은 안 된다. 무조건 가비지컬렉션(GC)에 동적으로 생성한다.

2. 복사 생성자를 만들 수 없다.

3. ‘^’(핸들이라고 부른다)과 ‘gcnew’를 사용하여 클래스를 생성한다. 당근 메모리 해제는 GC에서 관리한다.

4. 핸들은 네이티브의 ‘*’(포인터) 나 ‘&’(참조)와 비슷한 것으로 더 안정스럽다.

5. ‘delete’는 명시적으로 클래스에서 사용하고 있는 리소스를 해제할 때 사용하는 것으로 ‘소멸자’가 호출 되는 것이지 메모리에서 해제하는 것은 아니다. 즉 delete로 GC에 있는 메모리를 해제하는 것은 아니다.

6. delete로 소멸자를 호출하면 GC에 의해서 진짜 파괴될 때 소멸자는 호출되지 않는다.

(소멸자를 선언한 클래스는 자동적으로 IDisposable 인터페이스를 구현한다. 소멸자는 컴파일러에 의해서 Dispose() 메소드로 치환된다).

7. 소멸자 이외에 'finalize'를 선언할 수 있다. finalize는 GC에서 인스턴스가 파괴될 때 호출된다. 단 delete로 소멸자를 호출한 경우에는 finalize는 호출되지 않는다.

8. finalize는 ‘!’ 키워드를 사용한다.

관리 클래스 사용해 보기

< 코드 1. 관리 클래스 정의 및 사용 >

< 결과 >

<코드 1> 설명

< 그림 1. <코드 1>의 소스 코드와 결과 >

<그림 1>의 코드를 보면 3번의 ‘gcnew’, 4번의 ‘^’를 사용하여 ManagedTest를 생성하였습니다.

<코드 1>에는 없지만 핸들은 포인터와 같은 것으로 관리 클래스의 멤버를 사용할 때는 ‘->’를 사용합니다.

5번 delete를 사용하면 소멸자가 호출되어서 6번이 출력됩니다.

MTest2는 소멸자를 호출하지 않아서 8번에서 정의한 finalize가 프로그램이 종료할 때 호출됩니다(7번).

<코드 1>은 아주 짧고 단순한 코드이지만 관리 클래스의 대부분의 특징을 다 나타내고 있습니다. 관리 클래스에 대한 설명은 아직 남아 있습니다. 이것은 다음 회에 또 설명하겠습니다.

Asynchronous Agents Library 소개

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

Concurrency Runtime 의 컴포넌트

[ 그림1. Concurrency Runtime Architecture ]

 Asynchronous Agents Library 는 위 그림에서 보듯이 Concurrency Runtime 프레임워크 안에서 돌아가는 내부 컴포넌트 중 하나입니다.

 라이브러리라는 명칭에서 독립적으로 수행될 것 같은 오해를 가질 수 있으나, 사실은 Concurrency Runtime 의 Task Scheduler 와 Resource Manager 를 기반으로 만들어졌습니다.

 Concurrency Runtime 에서 Task Scheduler 와 Resource Manager 를 하위 레벨 컴포넌트라고 본다면, Asynchronouse Agent Library( 이하, AAL ) 와 Parallel Patterns Library( 이하, PPL ) 은 상위 레벨 컴포넌트라고 볼 수 있습니다. 다시 말해, AAL, PPL 모두 Task Scheduler 와 Resource Manager 를 사용한다는 말입니다.

 Concurrency Runtime 이 아닌 다른 프레임워크에서도 하위 레벨보다 상위 레벨 컴포넌트가 사용 용이성과 안정성이 높은 것처럼 Concurrency Runtime 에서도 마찬가지입니다.

 Concurrency Runtime 개발자들은 비 동기 작업들을 하기 위해서는 AAL, 단위 작업의 병렬 처리를 하기 위해서는 PPL 을 사용하기를 권하고, 좀 더 특별한 최적화나 하위 레벨 작업이 필요한 경우에만 직접 Task Scheduler 를 사용하기를 권하고 있습니다.

 즉, AAL 은 Concurrency Runtime 의 비 동기 작업을 위한 인터페이스라고 볼 수 있습니다.( PPL 은 병렬 처리를 위한 인터페이스라고 볼 수 있습니다. )

비 동기 작업

 비 동기 작업이란 작업이 끝날 때까지 기다리지 않는다는 말입니다. 즉, 어떤 함수가 있을 때, 그 함수는 바로 반환되어야 합니다. 바로 반환되면 호출한 스레드는 바로 다른 작업을 수행할 수 있습니다. 이 때, 해당 함수의 작업은 호출한 스레드가 아니라 다른 작업 스레드에서 수행되고, 작업이 완료되면 호출한 스레드에게 통지하거나, 콜백( call back ) 함수를 호출합니다.

 이러한 처리 방식을 비 동기 처리 방식이라고 합니다. AAL 은 이런 비 동기 작업을 용이하게 해주는 라이브러리입니다.

기능

 AAL 은 크게 두 가지 특징을 가지고 있습니다. 하나는 actor-based 프로그래밍 모델이고, 다른 하나는 message passing 프로그래밍 모델입니다.

Agent

 Agent 는 AAL 의 기능 중 하나인 actor-based 프로그래밍 모델의 actor 를 나타내는 개념입니다. 인공 지능 등과 같은 다른 분야에서 사용하는 개념과 마찬가지로 어떤 임무를 수행하는, 즉 어떤 역할을 하는 객체를 일컫습니다.

 예를 들어, 스마트 폰이 만들어 지려면 케이스( case ) 생산, 하드웨어 생산, 케이스와 하드웨어 조립, OS 및 소프트웨어 설치, 테스트와 같은 공정이 필요한데, 각 공정들을 수행하는 객체들을 agent 라고 볼 수 있습니다.

Message

 Message 는 agent 들 간의 통신을 위한 메커니즘입니다. AAL 에서의 message 란 상황이나 상태를 알리기 위한 수단이 될 수도 있고, 실제 데이터를 전송하기 위한 수단이 될 수도 있습니다.

 예를 들면, 스마트 폰 공정들 사이에 케이스가 생산되었다면 생산된 케이스를 조립하는 곳으로 전달해야 합니다. 이 때, 케이스를 전달하는 수단으로 message 가 사용될 수 있습니다. 또한 케이스와 하드웨어의 조립, OS 및 소프트웨어 설치가 완료되어야지 테스트를 진행할 수 있습니다. 이런 완료 상태와 같은 작업의 상태를 알리기 위해서 message 가 사용될 수 있습니다.

예제

 위에 언급한 스마트 폰 생산 과정을 AAL 을 이용하여 구현해보았습니다. 예제가 비교적 긴 편이지만 최대한 이해하기 쉽게 직관적인 코드를 작성했습니다. AAL 을 중점적으로 소개하기 위해 디자인이나 테크닉은 무시하고 작성하였습니다.

 전체적인 흐름을 이해하는 데에는 어려움이 없을 것 같으나 사용된 함수들이 어떤 역할을 하는지 궁금할 것입니다.

시나리오

1.  스마트 폰 케이스 생산자와 하드웨어 생산자가 각각 케이스와 하드웨어를 부품으로 생산한다.
2.  생산된 부품을 조립하는 객체에게 전달하면 조립하는 객체가 조립하게 된다.
3.  조립된 스마트 폰은 소프트웨어 설치 객체에게 전달되고, 그 객체는 소프트웨어를 설치한다.
4.  소프트웨어가 설치된 스마트 폰은 테스터에게 전달되고, 테스터는 테스트에 성공한 스마트 폰을 제품 컨테이너에 저장한다.
5.  제품 컨테이너의 제품들을 출력하고 종료한다.

[ 그림2. 예제 시나리오 ]

코드

// 스마트폰 생산 예제 코드
#include <iostream>
#include <agents.h>
#include <ppl.h>

using namespace std;
using namespace Concurrency;

// 케이스 클래스
class Case
{
};

// 하드웨어 클래스
class Hardware
{
};

// 소프트웨어 클래스
class Software
{	
};

// 스마트폰 클래스
class SmartPhone
{
public:
	int			uid;

	Case*		pCase;
	Hardware*	pHardware;
	Software*	pSoftware;

	~SmartPhone()
	{
		if( 0 != this->pCase )
		{
			delete this->pCase;
			this->pCase = 0;
		}

		if( 0 != this->pHardware )
		{
			delete this->pHardware;
			this->pCase = 0;
		}

		if( 0 != this->pSoftware )
		{
			delete this->pSoftware;
			this->pSoftware = 0;
		}		
	}

	SmartPhone( int _uid, Case* _pCase = 0, Hardware* _pHardware = 0, Software* _pSoftware = 0 )
		: uid( _uid )
		, pCase( _pCase )
		, pHardware( _pHardware )
		, pSoftware( 0 )
	{
		if( 0 != _pSoftware )
			this->pSoftware = new Software( *_pSoftware );		
	}		

	void SetSoftware( Software* _pSoftware )
	{
		if( 0 != this->pSoftware )
			delete this->pSoftware;

		this->pSoftware = new Software( *_pSoftware );
	}
};

// 메시지 버퍼 typedef
typedef unbounded_buffer< Case* >		CaseBuffer;
typedef unbounded_buffer< Hardware* >	HardwareBuffer;
typedef unbounded_buffer< SmartPhone* >	SmartPhoneBuffer;

typedef vector< SmartPhone* >			SmartPhoneVector;

// 케이스 생산자 agent
class CaseProducer
	: public agent
{
private:
	unsigned int	caseCountToCreate;
	CaseBuffer&		caseBuffer;	

protected:
	void run()
	{
		// 필요한 개수 만큼 케이스를 생산하고 케이스 버퍼로 보낸다.
		for( unsigned int i = 0; i < this->caseCountToCreate; ++i )
		{
			send( this->caseBuffer, new Case );
			wcout << i << L": created a case." << endl;
		}

		// 모두 보냈으면 생산을 완료했다는 메시지로 널( 0 ) 포인터를 보낸다.
		send( this->caseBuffer, static_cast< Case* >( 0 ) );

		done();
	}

public:	
	CaseProducer( unsigned int _caseCountToCreate, CaseBuffer& _caseBuffer )
		: caseBuffer( _caseBuffer )
		, caseCountToCreate( _caseCountToCreate ) { }	
};

// 하드웨어 생산자 agent
class HardwareProducer
	: public agent
{	
private:
	unsigned int		hardwareCountToCreate;
	HardwareBuffer&		hardwareBuffer;

protected:
	void run()
	{
		// 필요한 개수 만큼 하드웨어를 생산하고 하드웨어 버퍼로 보낸다.
		for( unsigned int i = 0; i < this->hardwareCountToCreate; ++i )
		{
			send( this->hardwareBuffer, new Hardware );
			wcout << i << L": created a hardware." << endl;
		}

		// 모두 보냈으면 생산을 완료했다는 메시지로 널( 0 ) 포인터를 보낸다.
		send( this->hardwareBuffer, static_cast< Hardware* >( 0 ) );

		done();		
	}

public:
	HardwareProducer( unsigned int _hardwareCountToCreate, HardwareBuffer& _hardwareBuffer )
		: hardwareCountToCreate( _hardwareCountToCreate )
		, hardwareBuffer( _hardwareBuffer ) { }
};

// 부품( 케이스와 하드웨어)을 조립하는 agent
class Assembler
	: public agent
{
private:
	CaseBuffer&			caseBuffer;
	HardwareBuffer&		hardwareBuffer;

	SmartPhoneBuffer&	incompletedSmartPhoneBuffer;

protected:
	void run()
	{
		unsigned int loopCount = 0;

		// 버퍼에 쌓인 부품( 케이스와 하드웨어 )을 꺼내 조립하여 스마트폰을 만든다.
		while( true )
		{
			Case* pCase = receive( this->caseBuffer );
			Hardware* pHardware = receive( this->hardwareBuffer );

			// 더 이상 조립할 부품( 케이스 또는 하드웨어 )들이 없으면, 
			// 스마트폰 생산이 완료되었음을 알리는 메시지로 널( 0 ) 포인터를 보낸다.
			if( 0 == pCase || 0 == pHardware )
			{
				// 남은 케이스를 파괴한다.
				if( 0 != pCase )
				{
					while( true )
					{
						Case* pGarbageCase = receive( this->caseBuffer );

						if( 0 == pGarbageCase )
							break;

						delete pGarbageCase;
					}					
				}

				// 남은 하드웨어를 파괴한다.
				if( 0 != pHardware )
				{
					while( true )
					{
						Hardware* pGarbageHardware = receive( this->hardwareBuffer );

						if( 0 == pGarbageHardware )
							break;

						delete pGarbageHardware;
					}
				}

				send( this->incompletedSmartPhoneBuffer, static_cast< SmartPhone* >( 0 ) );
				break;
			}

			send( this->incompletedSmartPhoneBuffer,
				new SmartPhone( loopCount + 1, pCase, pHardware ) );

			wcout << loopCount << L": created a smart phone." << endl;

			++loopCount;
		}
		
		done();
	}

public:
	Assembler( CaseBuffer& _caseBuffer, HardwareBuffer& _hardwareBuffer, 
		SmartPhoneBuffer& _incompletedSmartPhoneBuffer )
		: caseBuffer( _caseBuffer )
		, hardwareBuffer( _hardwareBuffer )
		, incompletedSmartPhoneBuffer( _incompletedSmartPhoneBuffer ) { }	
};

// 소프트웨어 설치 agent
class SoftwareInstaller
	: public agent
{	
private:
	Software&			software;	

	SmartPhoneBuffer&	incompletedSmartPhoneBuffer;
	SmartPhoneBuffer&	completedSmartPhoneBuffer;

	SmartPhoneVector&	faultySmartPhoneArray;

protected:
	void run()
	{
		unsigned int loopCount = 0;

		// 조립된 스마트폰에 소프트웨어를 설치한다.
		while( true )
		{
			SmartPhone* pSmartPhone = receive( this->incompletedSmartPhoneBuffer );

			// 더 이상 소프트웨어를 설치할 조립된 스마트폰이 없으면, 설치를 중단하고
			// 소프트웨어 설치도 모두 완료되었음을 알리는 메시지로 널( 0 ) 포인터를 보낸다.
			if( 0 == pSmartPhone )
			{
				send( this->completedSmartPhoneBuffer, static_cast< SmartPhone* >( 0 ) );
				break;
			}

			wcout << loopCount;

			// 제대로 조립되었는지 판단 후, 소프트웨어를 설치한다. 
			if( 0 != pSmartPhone->pCase && 0 != pSmartPhone->pHardware )
			{
				pSmartPhone->SetSoftware( &this->software );

				send( this->completedSmartPhoneBuffer, pSmartPhone );
				wcout << L": installed the software." << endl;
			}
			else
			{
				this->faultySmartPhoneArray.push_back( pSmartPhone );
				wcout << L": failed to install the software." << endl;
			}

			++loopCount;
		}		

		done();
	}

public:
	SoftwareInstaller( Software& _software, SmartPhoneBuffer& _incompletedSmartPhoneBuffer,
		SmartPhoneBuffer& _completedSmartPhoneBuffer, SmartPhoneVector& _faultySmartPhoneArray )
		: software( _software )
		, incompletedSmartPhoneBuffer( _incompletedSmartPhoneBuffer )
		, completedSmartPhoneBuffer( _completedSmartPhoneBuffer )
		, faultySmartPhoneArray( _faultySmartPhoneArray ) { }
};

// 테스트하는 agent
class Tester
	: public agent
{
private:
	SmartPhoneBuffer&	completedSmartPhoneBuffer;

	SmartPhoneVector&	productArray;
	SmartPhoneVector&	faultySmartPhoneArray;

protected:
	void run()
	{
		unsigned int loopCount = 0;

		// 조립된 스마트폰에 소프트웨어 설치가 완료된 스마트폰을 테스트한다.
		while( true )
		{
			SmartPhone* pSmartPhone = receive( this->completedSmartPhoneBuffer );

			// 더 이상 테스트할 스마트폰이 없으면 중단한다.
			if( 0 == pSmartPhone )				
				break;

			wcout << loopCount;

			if( this->Test( pSmartPhone ) )
			{
				this->productArray.push_back( pSmartPhone );
				wcout << L": succeeded in testing." << endl;
			}
			else
			{
				this->faultySmartPhoneArray.push_back( pSmartPhone );
				wcout << L": failed to test." << endl;
			}

			++loopCount;
		}
		
		done();
	}

public:
	Tester( SmartPhoneBuffer& _completedSmartPhoneBuffer,
		SmartPhoneVector& _productArray, SmartPhoneVector& _faultySmartPhoneArray )
		: completedSmartPhoneBuffer( _completedSmartPhoneBuffer )
		, productArray( _productArray )
		, faultySmartPhoneArray( _faultySmartPhoneArray ) { }

	bool Test( SmartPhone* _pSmartPhone )
	{
		return ( 0 != _pSmartPhone->pCase && 0 != _pSmartPhone->pHardware
			&& 0 != _pSmartPhone->pSoftware );		
	}	
};

int main()
{
	// 케이스 생산자 agent 생성.
	int caseCountToCreate = 10;
	CaseBuffer caseBuffer;
	CaseProducer caseProducer( caseCountToCreate, caseBuffer );

	// 하드웨어 생산자 agent 생성.
	int hardwareCountToCreate = 10;
	HardwareBuffer hardwareBuffer;
	HardwareProducer hardwareProducer( hardwareCountToCreate, hardwareBuffer );

	// 부품( 케이스와 하드웨어 )을 조립하는 agent 생성.
	SmartPhoneBuffer	incompletedSmartPhoneBuffer;
	Assembler assembler( caseBuffer, hardwareBuffer, incompletedSmartPhoneBuffer );	
	
	// 소프트웨어 설치 agent 생성.
	SmartPhoneVector	faultySmartPhoneArray;

	SmartPhoneBuffer	completedSmartPhoneBuffer;
	SmartPhoneBuffer	faultySmartPhoneBuffer;
	Software android;
	SoftwareInstaller softwareInstaller( android, incompletedSmartPhoneBuffer,
		completedSmartPhoneBuffer, faultySmartPhoneArray );

	// 테스트하는 agent 생성.
	SmartPhoneVector	productArray;

	SmartPhoneBuffer	productBuffer;	
	Tester tester( completedSmartPhoneBuffer, productArray, faultySmartPhoneArray );

	// 모든 생성된 agent 작업 시작.
	caseProducer.start();
	hardwareProducer.start();
	assembler.start();
	softwareInstaller.start();
	tester.start();

	// 모든 agent 의 작업이 끝날 때까지 대기.
	agent* watingAgents[] = { &caseProducer, &hardwareProducer, &assembler,
		&softwareInstaller, &tester };

	agent::wait_for_all( sizeof( watingAgents ) / sizeof( agent* ), watingAgents );
	agent::wait( &caseProducer );
	agent::wait( &hardwareProducer );
	agent::wait( &assembler );
	agent::wait( &softwareInstaller );
	agent::wait( &tester );	

	// 완성된 제품 출력.
	wcout << L"completed products: " << endl;

	parallel_for_each( productArray.begin(), productArray.end(),
		[] ( SmartPhone* _pSmartPhone )
	{
		wcout << L"product uid: " << _pSmartPhone->uid << endl;
	});

	// 자원 정리 - 완제품.
	parallel_for_each( productArray.begin(), productArray.end(),
		[] ( SmartPhone* _pSmartPhone )
	{
		if( 0 != _pSmartPhone )
			delete _pSmartPhone;
	});	

	// 자원 정리 - 불량품.
	parallel_for_each( faultySmartPhoneArray.begin(), faultySmartPhoneArray.end(),
		[] ( SmartPhone* _pSmartPhone )
	{
		if( 0 != _pSmartPhone )
			delete _pSmartPhone;
	});
}

[ 코드1. 예제 코드 ]

실행 결과

 모든 agent 가 수행하는 작업이 비 동기적으로 호출되었고 결과를 기다립니다. 비 동기적으로 호출된 agent 의 작업들은 각각 독립적인 스레드에서 동시에 수행되며, 작업을 수행하는데 데이터가 필요하다면 메시지에 의해 필요한 데이터가 도착할 때까지 동기화되어 수행됩니다.

 wcout 객체는 동기화되지 않기 때문에 작업의 넘버링이 엉켜있는데 실제로 세어보면 정확히 맞아 떨어집니다.

[ 그림3. 코드1 실행 결과 ]

메모리 누수

코드에 명시적인 메모리 누수가 없어도 Concurrency Runtime 에 의한 메모리 누수가 발견됩니다. 이것은 개발팀에서도 버그라고 인정하고 있습니다. 하지만 이번 2010 버전에는 수정되지 않을 것이고, 다음 버전인 서비스팩에 수정되어 포함될 것이라고 합니다.

참고: http://vsts2010.net/263

AAL 의 편리함

 위의 예제가 보여주다시피 AAL 을 사용하면 정말 간단하게 여러 객체의 작업을 비 동기적으로 동시에 수행시킬 수 있습니다.

만약 직접 Win32 API 를 사용하여 이를 구현하려 한다면 각 데이터를 보관하는 컨테이너에 대한 동기화 등과 같은 복잡한 메카니즘을 이벤트 등과 같은 동기화 객체를 사용하여 직접 구현해야 합니다.

물론, 결국은 OS의 커널( kernel ) 객체를 사용하여 구현되겠지만, 빠른 생산성의 이익과 동기화로 인한 스트레스로 인해 피폐해져 가는 인간성을 지킬 수 있다는 측면에서는 굉장히 편리한 도구임에 틀림없습니다.

마치는 글

 이번 글은 AAL 에 대한 개념과 어떻게 동작하는지 간단한 소개입니다. 다음 글에서 구현에 필요한 함수들과 사용법들을 이야기해 보겠습니다.

참고

• 그림1. Concurrency Runtime Architecture - http://i.msdn.microsoft.com/dynimg/IC315446.png

앞서 두 번을 걸쳐서 C++/CLI에 대해서 잘 모르는 분과 싫어하는 분을 위해서 제 생각이나 MSDN에 있는 글을 정리해서 포스팅 했습니다.

이제 본격적으로 C++/CLI에 대해서 설명해 나가겠습니다(이 글을 보는 분들은 C++을 알고 있다고 가정을 하겠습니다).

1. ‘C++/CLI가 뭐야?’

라고 질문을 하면 가장 초 간단한 답은 ‘.NET에서 C++를 사용하기 위한 언어’ 라고 말할 수 있습니다. 그런데 이 답은 너무 간단하고 없어 보이죠? ^^;

그래서 좀 유식하게 보일 수 있도록 고급스럽게 답해 보겠습니다(또는 복잡하게).

C++/CLI에서 CLI는 ‘Common Language Infrastructure’의 약자입니다.

C++/CLI는 CLI 환경에서 돌아가는 프로그램을 만들기 위한 언어입니다. C++/CLI는 마이크로소프트(이하 MS)가 만들었지만 공업 표준화 단체인 ECMA에 의해서 표준 언어로 제정 되어 있습니다.

 C++/CLI가 MS가 만들었기 때문에 현재까지는 실행 환경이 Windows의 .NET 플랫폼이지만 언어 사양 상으로는 Windows나 .NET 플랫폼에만 사용할 수 있는 것이 아닙니다. 이론적으로는 Windows 이외의 Unix나 Linux, Mac에서도 실행할 수 있습니다(누구라도 Windows 이외서도 사용할 수 있도록 언어 사양을 따라서 구현만 하면 됩니다).

C++/CLI는 C++로 만든 프로그램을 거의 그대로 컴파일 할 수 있습니다.

C++의 표준 기능에 CLI를 위한 추가 기능이 더해져 있습니다.

2. 가장 많은 프로그래밍 언어로 만드는 프로그램 만들기

가장 많은 프로그래밍언어로 만드는 프로그램은 무엇일까요? 제가 생각하기에는 그 유명한 ‘Hello World’라고 생각합니다. 제가 공부한 대부분의 프로그래밍 언어 책에는 첫 번째 예제 프로그램이 Hello World였습니다.

그래서 C++/CLI도 관례(?)에 맞추어서 ‘Hello World’ 프로그램을 만들어 보겠습니다.

(VS의 마법사 기능에 의해서 코딩을 하나도 하지 않고 프로그램이 만들어집니다.)

< 그림 1. CLR 콘솔 어플리케이션을 선택합니다 >

< 리스트 1. ‘Hello World’ >

<리스트 1>의 코드는 <그림 1>에서 ‘OK’ 버튼을 누른 후 자동으로 생성되는 코드입니다. 이것을 빌드 후 실행을 하며 아래와 같은 결과가 나옵니다.

그런데 문제는 실행과 동시에 종료되어서 결과를 볼 틈이 없습니다.

예전에 C++에서는 이런 경우 코드의 아래에 ‘getchar()’를 사용하여 결과를 보고 종료시켰습니다.

그런데 다들 아시겠지만 ‘getchar()’라는 함수는 .NET 라이브러리에 있는 함수가 아닌 네이티브의 함수입니다. C#이라면 바로 사용하지 못하겠지만 C++/CLI는 이런 네이티브의 함수를 바로 사용할 수 있습니다.

<리스트 2. getchar() 사용 >

<리스트 2>는 .NET 라이브러리와 네이티브가 자연스럽게 공존하고 있습니다.

이런 것이 C++/CLI이기 때문에 가능한 것입니다.

C++/CLI로 만들어진  'Hello World' 소스 코드를 보면 C++ 프로그래머라면 몇개 처음보는 것이 있지만 이름만 봐도 대충 어떤 의미를 가지고 있는지 쉽게 파악할 수 있어서 소소 코드가 전혀 어렵지 않을 것입니다.

C++/CLI는 기존의 C++에서 CLI가 더해진 것으로 간단하게 말하면 이 더해진 'CLI'만 공부하면 C++/CLI는 마스터합니다.

앞으로 더해진 'CLI' 부분에 대해서 설명해 나가겠습니다.

다음에는 C++의 트레이드마크인 ‘클래스’와 C에서부터 친숙한 struct가 C++/CLI에서는 어떤 의미를 갖고 어떻게 사용되는지 알아 보겠습니다.

월간 마이크로소프트웨어 'C++ 개발자와 함께하는 Visual Studio 2010' 세미나를 주최합니다.

특히 Visual Studio 2010 과 C++0x 을 중심으로 매우 많은 관심을 보이고 있는 분야입니다. C++ 언어를 이용하여 개발을 하신다면 반드시 참석해야 할 더 없는 좋은 세미나라고 생각합니다.

저희 VIsual Studio 2010 팀에 계시는 NCSoft 의 최성기님께서 윈도우 7 개발 부분에서 발표를 하게 되었습니다.

세미나 신청은 아래의 링크를 통해 신청하시기 바랍니다. 많은 참석 부탁 드립니다.
http://new.imaso.co.kr/seminars/vs2010/agenda/