Concurrency Runtime – Task Scheduler 5. ( Context blocking )

Concurrency Runtime
– Task Scheduler 5. ( Context blocking )

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

 

시작하는 글

 이번 글에서는 컨텍스트를 제어할 수 있도록 도와주는 wait() 와 Context 클래스에 대해서 알아보도록 하겠습니다.

 

void Concurrency::wait( unsigned int _Milliseconds )

 현재 컨텍스트를 지정된 시간 동안 지연시킵니다.

 Windows API 의 Sleep() 과 같은 기능을 합니다. 하지만 Sleep() 과 달리 협조적 스케줄링을 한다는 것이 다릅니다. 컨텍스트가 지연되는 동안 컴퓨팅 리소스( core ) 는 다른 작업을 수행합니다.

 매개변수로 0을 전달하면 다른 모든 활성화된 컨텍스트들이 작업을 수행할 수 있을 때까지 현재 컨텍스트를 지연시킵니다. 다른 작업들에게 양보한다는 뜻입니다.

 

Context

 Context 클래스는 현재 컨텍스트를 나타냅니다. Context 클래스는 크게 2 가지의 기능을 가지고 있는데 그 중 첫 번째는 컨텍스트의 블러킹( blocking ) 이고, 두 번째는 초과 스레드 생성입니다.

이번 글에서는 컨텍스트의 블러킹에 대해서 알아보록 하고, 초과 스레드 생성은 다음 글에서 살펴보도록 하겠습니다.

 

Block & Unblock

 현재 컨텍스트를 블럭킹하려면 Context::Block() 을 호출하면 됩니다. 이렇게 되면 현재 컨텍스트는 수행을 멈추고 다른 작업들에게 리소스를 양보하게 됩니다.

 블러킹을 해제하려면 Context::Unblock() 을 호출하면 됩니다. 하지만 블러킹된 컨텍스트에서 스스로 블러킹을 해제할 수 없습니다. 블러킹된 컨텍스트에서 자기의 컨텍스트의 블러킹을 해제하려고 하면 context_self_unblock 예외가 던져집니다.

 실제로 Context 객체를 사용해 블러킹과 해제를 하려면 Context::CurrentContext() 를 호출하여 현재 스레드에 연결된 Context 객체를 참조하여 블러킹하고, 다른 컨텍스트에 그 참조를 전달하여, 나중에 참조를 사용하여 블러킹을 해제해야 합니다.

 Block() 과 Unblock() 은 항상 쌍을 이루어 호출되어야 합니다. 만약 같은 함수가 연속으로 호출되면 context_unblock_unbalanced 예외를 던집니다.

 또한 Context 객체는 스케줄링 양보를 위한  Context::Yield() 를 제공합니다. Context::Block() 을 사용하면 스케줄링되지 않습니다.

 

예제

 Context 클래스를 이용한 세마포어( semaphore ) 의 구현을 보도록 하겠습니다.

 

시나리오

 구현된 세마포어를 사용한 parallel_for() 를 사용하여 작업이 어떻게 진행되는지 살펴보겠습니다.

 

코드

#include <Windows.h>
#include <concrt.h>
#include <concurrent_queue.h>
#include <iostream>
#include <sstream>
#include <ppl.h>

using namespace std;
using namespace Concurrency;

class semaphore
{
public:
	explicit semaphore( long capacity )
		: semaphore_count( capacity ) { }

	void acquire()
	{
		if( InterlockedDecrement( &this->semaphore_count ) < 0 )
		{
			this->waiting_contexts.push( Context::CurrentContext() );
			Context::Block();
		}
	}

	void release()
	{
		if( InterlockedIncrement( &this->semaphore_count ) <= 0 )
		{
			Context* pWaitingContext = nullptr;
			while( !this->waiting_contexts.try_pop( pWaitingContext ) )
				Context::Yield();

			pWaitingContext->Unblock();
		}
	}

	class scoped_lock
	{
	public:
		~scoped_lock()
		{
			this->s.release();
		}

		scoped_lock( semaphore& s )
			: s( s )
		{
			this->s.acquire();
		}

	private:
		semaphore&	s;
	};

private:
	long							semaphore_count;
	concurrent_queue< Context* >	waiting_contexts;
};

int main()
{
	semaphore s( 3 );

	parallel_for( 0, 10, [&]( int i )
	{
		semaphore::scoped_lock lock( s );

		wstringstream ss;
		ss << L"In loop interation " << i << L"..." << endl;
		wcout << ss.str();

		Concurrency::wait( 2000 );		
	} );
}

[ 코드1. Context 객체를 이용한 세마포어 구현 예제 ]

 Parallel Patterns Library( 이하, PPL ) 의 acquire() 호출 시, 남은 자원이 없다면 해당 Context 객체를 블러킹하고 concurrent_queue 를 사용하여 관리합니다.

 release() 호출 시, 남는 자원이 생기면 관리 중인 Context 객체를 꺼내서 블러킹을 해제합니다.

 parallel_for() 를 사용해 10개의 작업을 수행하는데 세마포어에 의해 3 개의 작업 단위로 수행하게 됩니다.

 

[ 그림1. Context 객체를 이용한 세마포어 구현 예제 실행 결과 ]

 

마치는 글

 Context 클래스를 이용해서 어떻게 컨텍스트를 제어할 수 있는지 알아보았습니다.

 다음 글에서는 Context 클래스의 두 번째 기능인 초과 스레드 생성이란 무엇인지 알아보겠습니다.