Concurrency Runtime
– Task Scheduler 4. ( ScheduleTask )

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

시작하는 글

 Concurrency Runtime 에서는 기존의 멀티스레드 프로그래밍할 때, 필수적인 스레드를 생성하는 함수( CreateThread(), _beginthread() 등 )와 같은 기능을 제공합니다.

이 글에서는 어떻게 위와 같은 기능을 제공하고, 어떻게 사용하면 되는지 알아보도록 하겠습니다.

ScheduleTask

 이 전에 설명했던 Parallel Patterns Library( 이하, PPL ) 이나 Asynchronous Agents Library( 이하, AAL ) 을 사용하게 되면 암묵적으로 스레드가 생성되고, 관리되기 때문에, 아주 간단한 작업을 처리할 때에는 비교적 높은 오버헤드( overhead )를 갖게 됩니다.

 그래서 Concurrency Runtime 에서는 간단한 작업들을 처리하기에 비교적 적은 오버헤드를 갖는 기능을 제공합니다.

 이 간단한 작업을 처리하는 방법은 기존의 스레드를 생성하는 방법과 유사합니다. 즉, 직접 스레드 코드를 제어하고 싶을 때, 사용할 수 있습니다.

 아쉬운 점은 간단한 작업의 처리를 완료했을 때, 알려주지 않습니다. 스레드 핸들을 사용하지 않기 때문에 WaitForSingleObject() 와 같은 함수도 사용할 수 없습니다.

 그렇기 때문에 스레드로 수행될 함수의 인자를 넘길 때, 스레드 종료를 알리는 메커니즘에 필요한 데이터를 포함해주어야 합니다.( 또는 전역적으로.. )

 Concurrency Runtime 에서는 위와 같이 스레드를 생성하는 기능을 제공하지만, 간단한 작업이 아니라면 PPL 또는 AAL 을 사용하는 것을 권하고 있습니다.

 이러한 기능은 ScheduleGroup::ScheduleTask(), CurrentScheduler::ScheduleTask(), Scheduler::ScheduleTask() 를 호출하여 사용합니다.

ScheduleGroup::ScheduleTask( TaskProc _Proc, void* _Data );

 일정 그룹 내에 간단한 작업을 추가합니다.

 첫 번째 매개변수의 타입인 TaskProc 는 다음과 같이 정의되어 있습니다.

typedef void (__cdecl * TaskProc)(void *)

[ 코드1. TaskProc 의 정의 ]

 그렇기 때문에 위와 같은 모양( signature ) 를 갖는 함수 포인터를 인자로 넘기면 됩니다.

두 번째 매개변수는 위에서 인자로 넘긴 함수 포인터의 인자로 넘길 데이터입니다.

CurrentScheduleer::ScheduleTask( TaskProc _Proc, void* _Data );

 호출하는 컨텍스트와 연결된( attached ) 스케줄러에 간단한 작업을 추가합니다.

매개변수의 내용은 위의 ScheduleGroup::ScheduleTask() 와 같습니다.

만약 호출하는 컨텍스트에 연결된 스케줄러가 없을 경우에는 기본 스케줄러에 추가합니다.

Scheduler::ScheduleTask( TaskProc _Proc, void* _Data );

 해당 스케줄러에 간단한 작업을 추가합니다.

매개변수의 내용은 위의 ScheduleGroup::ScheduleTask() 와 같습니다.

예제

 위에 언급된 내용과 함수들을 어떻게 사용하는 알아보도록 하겠습니다.

시나리오

 새로운 스레드를 생성하고, 그 스레드에서 인자로 전달 받은 구조체의 내용을 출력하는 내용입니다.

코드

#include <iostream>
#include <concrt.h>

using namespace std;
using namespace Concurrency;

void __cdecl MyThreadFunction( void* pParam );

// 스레드에서 사용할 데이터
struct MyData
{
	int val1;
    int val2;
    event signal;
};

int main()
{
	MyData* pData = new MyData;

	if( nullptr == pData )
		return 1;
	
	pData->val1 = 50;
	pData->val2 = 100;
	
	// _beginthreadex() 처럼 스레드를 생성하고 인자로 넘어간 함수를 생성된 스레드에서 수행한다.
	CurrentScheduler::ScheduleTask( MyThreadFunction, pData );

	// 수행 중인 작업이 끝날 때까지 대기.
	pData->signal.wait();

	// 작업이 끝난 후, 자원 해제.
	delete pData;	

	return 0;
}

[ 코드2. CurrentScheduler::ScheduleTask() 를 사용하여 스레드를 생성하는 예제 ]

 간단한 구조체를 생성하고, 스레드에서 수행될 함수에 전달합니다.

 스레드에서는 구조체의 정보를 출력합니다.

 스레드의 수행이 종료되기 전에 구조체를 제거하면 안되므로, event 객체를 이용해 스레드가 종료될 때까지 기다린 후, 제거합니다.

마치는 글

 Concurrency Runtime 을 사용 중에 간단하게 스레드를 생성해야 한다면, 기존의 스레드 생성 함수를 사용하는 것보다 일관성 있게 ScheduleTask 를 사용하는 것이 좋습니다.

 하지만 이 글에서 언급했듯이 간단하지 않은 작업이라면 PPL 이나 AAL 을 사용하는 것이 좋습니다.

 다음 글에서는 Scheduler 에서 제공하는 Context 에 대해서 알아보도록 하겠습니다.

Concurrency Runtime
– Task Scheduler 3. ( ScheduleGroup )

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

시작하는 글

 Scheduler 는 항상 하나 이상의 스케줄 그룹을 가지고 있습니다. 스케줄 그룹이 무엇인지 알아보고 동시성 프로그래밍에 어떤 영향을 주는지 알아보도록 하겠습니다.

SchedulerGroup

 SchedulerGroup 이란 Scheduler 가 스케줄링 해야 할 작업들을 모아 놓은 스케줄 그룹을 대표하는 클래스입니다.

정책

 이전 글에서 본 스케줄러 정책들 중 SchedulingProtocol 정책은 스케줄링 순서를 변경하여 작업들의 처리 순서를 변경할 수 있습니다. ScheduleringProtocol 정책이 EnhanseScheduleGroupLocality 로 설정된 경우에는 스케줄링 중인 스케줄 그룹을 변경하지 않고, 현재 스케줄링 중인 스케줄 그룹을 다른 스케줄 그룹보다 먼저 처리합니다. 반면에 SchedulingProtocol 정책이 EnhanseForwardProgress 로 설정된 경우에는 다른 스케줄링 그룹의 작업을 처리합니다. 스케줄 그룹 간에 작업을 공평하게 처리하게 됩니다.

생성

 CurrentScheduler::CreateScheduleGroup() 또는 Scheduler::CreateScheduleGroup() 을 통해 ScheduleGroup 객체를 생성할 수 있습니다.

 Scheduler 가 참조 개수를 사용하여 수명을 관리하는 것처럼 ScheduleGroup 객체 또한 참조 개수를 사용합니다. 생성되는 순간 참조 개수는 1이 됩니다. SchedulerGroup::Reference() 는 참조 개수는 1이 증가하고, SchedulerGroup::Release() 는 참조 개수가 1이 감소하게 됩니다.

동작 및 사용

 Scheduler 객체에는 기본적으로 기본 ScheduleGroup 을 가지고 있습니다. 명시적으로 ScheduleGroup 을 생성하지 않는다면 스케줄링 해야 할 작업들은 기본 ScheduleGroup 에 추가됩니다.

 ScheduleGroup 은 Concurrency Runtime 중 Asynchronous Agents Library( 이하, AAL ) 에서 사용할 수 있습니다. agent 클래스나 message block 들의 생성자로 ScheduleGroup 을 전달하여 사용할 수 있습니다.

예제

 ScheduleGroup 을 어떻게 사용하는지 이해를 도울 예제를 살펴보겠습니다.

시나리오

 단순하게 어떤 스케줄 그룹의 작업들이 어떤 순서로 처리되는지 알아보는 예제입니다.

코드

#include <agents.h>
#include <vector>
#include <algorithm>
#include <iostream>
#include <sstream>

using namespace std;
using namespace Concurrency;

#pragma optimize( "", off )

void spin_loop()
{
	for( unsigned int i = 0; i < 500000000; ++i )
	{
	}
}

#pragma optimize( "", on )

class work_yield_agent
	: public agent
{
public:
	explicit work_yield_agent( unsigned int group_number, unsigned int task_number )
		: group_number( group_number )
		, task_number( task_number ) { }

	explicit work_yield_agent( ScheduleGroup& scheduleGroup, unsigned int group_number, unsigned int task_number )
		: agent( scheduleGroup )
		, group_number( group_number )
		, task_number( task_number ) { }

protected:
	void run()
	{
		wstringstream header, ss;

		header << L"agent no. " << this->group_number << L"-" << this->task_number << L": ";

		ss << header.str() << L"first loop..." << endl;
		wcout << ss.str();
		spin_loop();

		ss = wstringstream();
		ss << header.str() << L"waiting..." << endl;
		wcout << ss.str();
		Concurrency::wait( 0 );

		ss = wstringstream();
		ss<< header.str() << L"second loop..." << endl;
		wcout << ss.str();
		spin_loop();

		ss = wstringstream();
		ss << header.str() << L"finished..." << endl;
		wcout << ss.str();

		this->done();
	}

private:
	unsigned int	group_number;
	unsigned int	task_number;
};

void run_agents()
{
	const unsigned int group_count = 2;
	const unsigned int tasks_per_group = 2;

	vector< ScheduleGroup* > groups;
	vector< agent* > agents;

	for( unsigned int group_index = 0; group_index < group_count; ++group_index )
	{
		groups.push_back( CurrentScheduler::CreateScheduleGroup() );

		for( unsigned int task_index = 0; task_index < tasks_per_group; ++task_index )
		{
			agents.push_back( new work_yield_agent( *groups.back(), group_index, task_index ) );
		}
	}

	for_each( agents.begin(), agents.end(), []( agent* pAgent )
	{
		pAgent->start();
	} );

	agent::wait_for_all( agents.size(), &agents[0] );

	for_each( agents.begin(), agents.end(), []( agent* pAgent )
	{
		delete pAgent;
	} );

	for_each( groups.begin(), groups.end(), []( ScheduleGroup* pScheduleGroup )
	{
		pScheduleGroup->Release();
	} );
}

int main()
{
	wcout << L"Using EnhanceScheduleGroupLocality..." << endl;
	CurrentScheduler::Create( SchedulerPolicy( 3,
		MinConcurrency, 1,
		MaxConcurrency, 2,
		SchedulingProtocol, EnhanceScheduleGroupLocality ) );

	run_agents();
	CurrentScheduler::Detach();

	wcout << endl << endl;

	wcout << L"Using EnhanceForwardProgress..." << endl;
	CurrentScheduler::Create( SchedulerPolicy( 3,
		MinConcurrency, 1, 
		MaxConcurrency, 2,
		SchedulingProtocol, EnhanceForwardProgress ) );

	run_agents();
	CurrentScheduler::Detach();
}

[ 코드1. 처리되는 스케줄 그룹의 순서를 확인하는 예제 ]

 최대 동시성 리소스( computing core ) 의 개수를 2개로 하고, 한 번은 SchedulingProtocol 정책을 EnhanceScheduleGroupLocality 로 하고, 한 번은 EnhanceForwardProgress 으로 하여 작업의 순서를 알아보았습니다.

 2개의 스케줄 그룹을 생성하고, 하나의 그룹에 2개의 작업을 추가하여, 총 4개의 작업이 수행됩니다.

 예제의 spin_loop() 는 어떤 작업을 처리하는 것을 의미하고 wait() 는 양보를 뜻합니다. wait() 가 호출되면 다른 작업이 스케줄링 되는데 이 때 결정되는 작업이 어떤 작업인지는 앞서 설정한 SchedulingProtocol 정책에 따릅니다.

 EnhanceScheduleGroupLocality 로 설정된 경우에는 같은 스케줄 그룹 내의 작업이 처리되는 반면에, EnhanceForwardProgress 로 설정된 경우에는 다른 스케줄 그룹의 작업이 처리되게 됩니다.

[ 그림1. 처리되는 스케줄 그룹의 순서를 확인하는 예제 실행 결과 ]

마치는 글

 이렇게 스케줄링 정책과 스케줄 그룹을 통해서 스케줄링 순서를 결정하는데 관여할 수 있다는 것을 알아보았습니다.

다음 글에서는 스레드를 직접 생성하여 사용할 수 있는 ScheduleTask 라는 것에 대해서 알아보도록 하겠습니다.