::NPTEAM:: Network Programer Team

검색 :
RSS 구독 : 글 / 댓글 / 트랙백 / 글+트랙백

std::for_each와 Concurrency::parallel_for_each의 속도 비교

std:for_eachConcurrency::parallel_for_each 중에 어떤것이 더 빠른 속도를 낼 것인가?

가정 1. CPU Core가 늘어남에 따라 Single Thread보다는 Multi Thread를 활용하여 CPU 자원을 최대한 활용한다.

가정 2. 컨테이너를 읽기 전용으로 접근할 경우 Lock 객체를 사용하지 않아도 된다.
          (Race Condition, 스레드간 공유 자원 접근 문제)

가정 3. Thread 갯수가 과도하게 늘어나서 빈번하게 Context Switching이 발생할 경우 속도 저하가 발생한다.
           (Concurrency::parallel_for_each는 Thread 갯수를 스스로 조절한다.)

가정 4. 암달의 법칙(Amdahl's law)으로 직렬화된 작업을 병렬화 작업으로 바꾸더라도 속도 향상에는 한계가 있을 것이다.

위 4가지 가정으로 봤을때, 컨테이너를 이터레이팅 할때 많은 도움은 안되더라도 약간의 속도 향상을 기대하였다.

int _tmain(int argc, _TCHAR* argv[])
{
	setlocale( LC_ALL, "Korean" );

	std::vector< int > vecContainer;

	for( int i = 0; i < 10000; ++i )
	{
		vecContainer.push_back( i );
	}

	class fntorPrintElem : public std::unary_function< int, void >
	{
	public:
		explicit fntorPrintElem() {};
		
		void operator() ( const int& elem ) const
		{
			std::wcout << "";
		}
	};

	// std::for_each 속도 체크 코드 생략...
	std::for_each( vecContainer.begin(), vecContainer.end(), fntorPrintElem() );
	
	// Concurrency::parallel_for_each 속도 체크 코드 생략...
	Concurrency::parallel_for_each( vecContainer.begin(), vecContainer.end(), fntorPrintElem() );
	
	return 0;
}
사용자 삽입 이미지




그러나 위 4가지 가정에서 예측한 것과 다른 값이 도출되었다.
std::for_each 수행 시간 : 0.004950
Concurrency::parallel_for_each 수행시간 : 0.005819

parallel_for_each를 사용하면, std::for_each를 사용했을때 보다 오히려 성능이 떨어진다.
일반적인 상황에서 Concurrency::parallel_for_each를 사용했을 경우 성능 향상을 보장 받을 수 없다.

그렇다면 parallel_for_each가 더 빠르게 하려면 무엇을 해야 할까?

functor에서 Sleep(1)을 넣어서 측정한 결과는 다음과 같다.
사용자 삽입 이미지





std::for_each 수행 시간 : 10.019599
Concurrency::parallel_for_each 수행시간 : 3.753167

측정 결과 parallel_for_eachstd::for_each보다 상대적으로 약 2.6배 빠르지만,
std::for_each는 2504배 느려지고, parallel_for_each는 750배 느려졌다.

일반적인 컨테이너를 이터레이팅 하는 상황에서 Concurrency::parallel_for_each대체하기만 하면,
속도 향상이 있을 것이라는 기대는 깨졌지만,

네트워크 I/O와 같은 비동기 대기가 발생하는 상황에서
코딩 난이도가 높지 않은 parallel_for_each를 활용하면, 속도 향상에 도움이 될 것으로 예측한다.
2011/07/28 16:45 2011/07/28 16:45

맨 위로

std::numeric_limits 관련 에러가 발생할때 대처 방법

std::numeric_limits는 Template Type을 체크하여 min, max 상수값을 알려주는 유용한 함수이다.
그러나 windows.h 헤더에 정의된
#define min(a, b)
#define max(a, b)
2가지 정의가 전처리기에서 우선으로 처리되기 때문에 컴파일 에러가 발생한다.

따라서 windows.h 헤더위에 #define NOMINMAX를 선언하여 매크로를 비활성화 시키는 방법을 사용한다.

위 내용이 포함된 링크를 추가한다.
2011/02/07 21:54 2011/02/07 21:54

맨 위로

[STL] std::copy_if, std::remove_copy_if

STL 알고리즘에는 std::copy와 std::remove_copy가 정의되어 있습니다.

이름에서 추측할 수 있듯이 std::copy는 두 컨테이너간 element 복사를 합니다.
또한 std::remove_copy는 특정한 element를 제외하고 복사 합니다.

std::copy_if 는 STL 알고리즘에 포함되어 있지 않습니다.
그 대신 std::remove_copy_if 가 정의되어 있습니다.

왜 그럴까요?

C++ 표준화 위원회에서 살짝 빠트린(?) 걸까요?

그렇게 생각하기엔 뭔가 이상하다고 생각되시죠. 저도 그렇습니다. ㅎㅎ

그렇다면, std::remove_copy_if는 어떻게 생겼을까요?
template< class _InIt,
	class _OutIt,
	class _Pr > inline
	_OutIt _Remove_copy_if(_InIt _First, _InIt _Last, _OutIt _Dest, _Pr _Pred, _Range_checked_iterator_tag)
	{	// copy omitting each element satisfying _Pred
	_DEBUG_RANGE(_First, _Last);
	_DEBUG_POINTER(_Dest);
	_DEBUG_POINTER(_Pred);
	for (; _First != _Last; ++_First)
		if (!_Pred(*_First))
			*_Dest++ = *_First;
	return (_Dest);
	}
위와 같이 생겼습니다. 무슨 뜻인지 모르시겠다구요.

딱! 보면...
저도 잘 모르겠습니다.(딱 봐서 알기엔 아직 실력이 부족해서요. ㅎㅎ)

아무래도 첫번째 컨테이너의 First Iterator에서 Last Iterator까지
반복문을 돌면서 조건식에 일치하지 않는 iterator만
두번째 컨테이너에 하나씩 element 복사를 하는 로직으로 보입니다.

헉. 조건에 일치하지 않는? 뭔가 이상하죠. ㅎㅎ

제거하면서 복사해야 하니까 조건에 일치하지 않는 것들만 복사하는게 맞는데요.
뭐가 이상하다는거죠?

눈치 빠르신 분은 느끼셨겠지만 조건에 일치하는 값만 복사하게 한다면?

조건에 일치하는 값만 복사하게 한다는 것은 결국 std::copy_if로 사용한다는 것입니다.

오호~ 그렇다면 조건에 일치하지 않는 것을 조건에 일치하게 만들어서 쓰면 된다는 것이군요.

그래서 아래와 같이 std::remove_copy_if를 -> std::copy_if 처럼 사용할 수 있습니다.
그 방법은 바로 조건문을 반전시키는 std::not1, std::not2 입니다.
	std::vector< int > vecInteger;
	vecInteger.push_back( 1 );
	vecInteger.push_back( 2 );
	vecInteger.push_back( 3 );
	vecInteger.push_back( 2 );
	vecInteger.push_back( 5 );
		
	std::vector< int > vecOutput;
	std::remove_copy_if( vecInteger.begin(), vecInteger.end(), std::back_inserter( vecOutput ), std::bind2nd( std::equal_to< int >(), 2 ) );
사용자 삽입 이미지
이 결과는 이렇습니다.

 그렇다면 아래와 같이 std::not1을 활용하면,
	std::vector< int > vecInteger;
	vecInteger.push_back( 1 );
	vecInteger.push_back( 2 );
	vecInteger.push_back( 3 );
	vecInteger.push_back( 2 );
	vecInteger.push_back( 5 );
		
	std::vector< int > vecOutput;
	std::remove_copy_if( vecInteger.begin(), vecInteger.end(), std::back_inserter( vecOutput ), std::not1( std::bind2nd( std::equal_to< int >(), 2 ) ) );
사용자 삽입 이미지
우리가 원한 결과입니다.
조건자에 std::not 시리즈를 사용하기 위해서는 술어 함수, 즉 Logical 함수가 필요합니다.
술어함수가 되기 위해선 조건식 함수자가 std::unary_function, std::binary_function을 상속 받아야 합니다.
그 이유는 result_type이 필요하다고 합니다. 사실 bool을 리턴하긴 해야 하니까요.

따라서 std::copy_if를 쓰시려면 std::remove_copy_if와 std::not 조합으로 쓰시면 되구요.
혹시 std::remove_copy_if가 보이신다면 std::not1, std::not2 함수가 보이는지 살펴보세요.

좀더 명시적으로 사용하려면 std namespace 안에 copy_if를 정의하는 것도 방법이겠지만,
표준화 위원회에서는 std namespace안에 다른 함수를 정의하는 것을 추천하지 않는다고 합니다.

라이브러리 사용자 입장에서는 일단 이렇게 써야겠습니다.
혹시 이 글을 보시고 더 좋은 방법을 알고 계신분은 댓글로 알려주시면 감사드리겠습니다.
2010/10/10 22:42 2010/10/10 22:42

맨 위로

[STL] STL Container 자료구조에서 특정 값 제거를 위한 코드

STL Container 자료구조에서 특정 값 제거를 위한 코드 

컨테이너가 표준 시퀀스 컨테이너이면, container::erase + std::remove 를 사용합니다.
(단, list 컨테이너는 list container::remove를 사용합니다.)

컨테이너가 표준 연관 컨테이너이면, container::erase 를 사용합니다.

#include < vector >
#include < algorithm >

// std::vector 삭제 코드
typedef std::vector< int > VEC_Container;
VEC_Container vecContainer;

vecContainer.push_back( 1 );
vecContainer.push_back( 2 );
vecContainer.push_back( 1 );
vecContainer.push_back( 4 );
vecContainer.push_back( 1 );

const int eraseValue = 1;
vecContainer.erase( std::remove( vecContainer.begin(), vecContainer.end(), eraseValue ), vecContainer.end() );
#include < list >

// std::list 삭제 코드
typedef std::list< int > LIST_Container;
LIST_Container listContainer;

listContainer.push_back( 1 );
listContainer.push_back( 2 );
listContainer.push_back( 1 );
listContainer.push_back( 4 );
listContainer.push_back( 1 );

const int eraseValue = 1;
listContainer.remove( eraseValue );
#include < set >

// std::multiset 삭제 코드
typedef std::multiset< int > MULTISET_Container;
MULTISET_Container multisetContainer;

multisetContainer.push_back( 1 );
multisetContainer.push_back( 2 );
multisetContainer.push_back( 1 );
multisetContainer.push_back( 4 );
multisetContainer.push_back( 1 );

const int eraseValue = 1;
multisetContainer.erase( eraseValue );
2010/09/09 20:13 2010/09/09 20:13

맨 위로

Set Container Iterating

Map Container에 이어서 Set Container Iterating에 대해서 측정해 보았다.

Map Container와 마찬가지로 Functor Iterating을 이용하여 측정하였다.

측정 환경 : Visual Studio 2008
측정 컨테이너 :
    1. std::set
    2. stdext::hash_set
    3. boost::unordered_set
측정 방법 : Functor


boost::unordered_set의 경우 컨테이너의 element가 1천개가 되더라도 bucket이 1:1.5로 생기는 것을 확인하였다.
(약 1.5배의 Bucket이 생성되었다.)

Hash 알고리즘에서 Collision이 발생하지 않도록 Bucket을 생성하는 것이 핵심인데,
VisualStudio 2008의 std::hash_set에서는 Type을 고려하지 않고 동일한 해쉬 함수를 사용하는데,
boost::unordered_set은
    1. std::numeric_limits로 해당 타입의 크기
    2. signed / unsigned
를 고려하여 해쉬 Seed 값을 생성하여 사용하기 때문에 각각의 Type 특성에 맞는 해쉬 키를 사용한다.


2010/04/11 23:41 2010/04/11 23:41

맨 위로

Map Container Iterating

이번엔 Map Container Iterating에 대해서 측정해 보았다.

지난번 블로깅에서 Functor가 가장 빠른 Iterating 속도를 보여줬으므로,
모든 Map Container에 대해서 Functor Iterating을 이용하여 측정하였다.

측정 환경 : Visual Studio 2008
측정 컨테이너 :
    1. std::map
    2. stdext::hash_map
    3. boost::unordered_map
측정 방법 : Functor

std::map과 boost::unordered_map은 약 2:1의 측정 속도 차이를 보여준다.
map data의 순서 정렬 보장이 필요없는 경우에는 boost::unordered_map을 사용하는 것이 더 빠른 속도를 얻을 수 있다.
map container의 경우에는 균형 탐색 트리(balanced search tree)로 구현되어 있기 때문에, 로그(Log) 시간 복잡도를 보장한다.
unordered_map은 Hash 알고리즘으로 Bucket을 찾아가기 때문에 검색속도가 빠르고 향상된 검색 속도만큼 Iterating 시간이 짧이졌다.
그에 비해 hash_map의 경우에는 초기에 STL에 포함되지 않아서 각 컴파일러 제작 회사마다 각각 구현하여 포함되었으므로, Hash 알고리즘이 최적화되어 있지 않아서 Iterating 및 검색속도가 떨어지는 문제점이 있다.
2010/04/11 21:20 2010/04/11 21:20

맨 위로

[STL] Vector Container Iterating 속도 비교

std::vector container를 Iterating하는 방법은 여러가지가 존재합니다.

평소 컨테이너 Iterating은 크게 생각하지 않고 사용하던 중
for_each에 대한 속도 향상 관련 글을 읽고 도대체 얼마나 빨라지는지 테스트를 해 보았습니다.

방법 1. None Const Iterating
std::vector< int >::iterator it     = vecInteger.begin();
std::vector< int >::iterator it_end = vecInteger.end();

for( ; it != it_end; ++it )
{
	const int& nValue = (*it);
	/* Do Something */
}
방법 2. Const Iterating
std::vector< int >::const_iterator it     = vecInteger.begin();
std::vector< int >::const_iterator it_end = vecInteger.end();

for( ; it != it_end; ++it )
{
	const int& nValue = (*it);
	/* Do Something */
}
방법 3. std::for_each( Functor )를 사용함
struct stVecInteger
{ 
	void operator() (int nElement)
	{
		const int& nValue = nElement;
		/* Do Something */
	};
} stVecIntergerFunctor;

std::for_each( vecInteger.begin(), vecInteger.end(), stVecIntergerFunctor );
방법 4. BOOST_FOREACH
BOOST_FOREACH( int nElement, vecInteger )
{
	const int& nValue = nElement;
	/* Do Something */
};
방법 5. for each(element in Container)
for each( int nElement in vecInteger )
{
	const int& nValue = nElement;
	/* Do Something */
};

Vector 컨테이너에 int 자료형을 1억개 넣고, Iterating 했을때의 결과(Release 컴파일)

Release 모드로 컴파일 했을때,
Functor가 최적화되어 어셈블리 코드가 없을때에는 0의 실행시간을 보여주는 엄청난 결과를 알게 되었다.

Functor에 별도의 수행 코드가 추가되더라도 가장 빠른 수행 성능을 보여주었다.
2010/03/01 21:32 2010/03/01 21:32

맨 위로

[STL] STL Container 자료구조에서 순환 삭제를 위한 코드

STL Container 자료구조에서 순환 삭제를 위한 코드 

컨테이너가 표준 시퀀스 컨테이너이면, 컨테이너 요소를 하나씩 사용하는 루프를 작성합니다.
(erase를 호출할 때마다 그 함수의 반환값으로 반복자를 업데이트 하는 일을 꼭 해야 합니다.)

컨테이너가 표준 연관 컨테이너이면, 컨테이너 요소를 하나씩 사용하는 루프를 작성합니다.
(erase를 호출하면서 erase에 넘기는 반복자를 후위 증가 연산자로 증가시킵니다.)

(출처 : Effective STL p88)

주의사항 : erase에 continue가 붙어서 작성되는 점이 중요하다.
이유 : erase가 ++it 역할을 하기 때문에 ++it 없이 continue를 하면 무한루프가 발생하기 때문이다.

팁 : erase() 함수의 리턴값이 있으면 반환값을 받고, 없으면 후위 증가 연산자로 증가시킵니다.

// std::vector 순환 삭제 코드
typedef std::vector< int > VEC_Container;
VEC_Container vecContainer;

vecContainer.push_back( 1 );
vecContainer.push_back( 2 );
vecContainer.push_back( 3 );
vecContainer.push_back( 4 );
vecContainer.push_back( 5 );

VEC_Container::iterator it = vecContainer.begin();
for( ; it != vecContainer.end(); )
{
	// erase 코드
	if( TRUE /*FALSE*/ )
	{
		it = vecContainer.erase( it );
		continue;
	}
	
	++it;
}


// std::list 순환 삭제 코드
typedef std::list< int > LIST_Container;
LIST_Container listContainer;

listContainer.push_back( 1 );
listContainer.push_back( 2 );
listContainer.push_back( 3 );
listContainer.push_back( 4 );
listContainer.push_back( 5 );

LIST_Container::iterator it = listContainer.begin();
for( ; it != listContainer.end(); )
{
	// erase 코드
	if( TRUE /*FALSE*/ )
	{
		it = listContainer.erase( it );
		continue;
	}
	
	++it;
}


// std::map 순환 삭제 코드
typedef std::map< int, int > MAP_Container;
MAP_Container mapContainer;

mapContainer.insert( std::make_pair(1, 5) );
mapContainer.insert( std::make_pair(2, 4) );
mapContainer.insert( std::make_pair(3, 3) );
mapContainer.insert( std::make_pair(4, 2) );
mapContainer.insert( std::make_pair(5, 1) );

MAP_Container::iterator it = mapContainer.begin();
for( ; it != mapContainer.end(); )
{
	// erase 코드
	if( TRUE /*FALSE*/ )
	{
		mapContainer.erase( it++ );
		continue;
	}
	
	++it;
}
2009/11/30 21:52 2009/11/30 21:52

맨 위로