스레드 동기화가 필요한 이유

 스레드에 관련된 글에서 설명 한 것 처럼 프로세스에 속한 스레드들은 프로세스의 리소스 자원을 공유합니다. 

 또한 시스템의 힙, 파일, 윈도우등등 많은 시스템 리소스에도 접근해야 합니다. 

 이러한 리소스에 다수의 스레드이 동시에 접근한다면 어떻게 될까요?

 https://jungwoong.tistory.com/6

 

[동기화] 데이터 레이스

"데이터레이스"란? 멀티 쓰레드 환경에서 여러 쓰레드가 공유자원에 동시에 접근할 때 쓰레드의 경쟁에 유발되는 문제입니다. 아래의 예제를 통해서 "데이터 레이스"에 대해서 설명을 진행합니다. <c++로 된="" 예제..<="" p=""> </c++로>

jungwoong.tistory.com

 위의 내용은 두개의 스레드가 전역변수에 접근할 때 발생하는 데이터 레이스에 대해서 설명한 글입니다. 

 스레드간에 공유 자원에 접근 할 때는 배타적인 접근을 지원해야합니다. 

InterLocked 함수들

 윈도우에는 다양한 방법으로 스레드간의 공유 자원을 배타적으로 사용할 수 있도록 지원합니다.

 그중에서 InterLocked 함수들을 하나의 변수의 값을 원자적으로 다룰 수 있도록 지원합니다. 

InterLocked Functions
장 점 사용 방법이 간단 최소한의 접근을 지원하기 때문에 효율적입니다.(유저모드 동작)
단 점  단일 변수에 대한 배타적 접근만을 지원하기 때문에 사용 범위가 제한적

다양한 InterLocked함수 사용방법

// 이 함수의 모든 작업은 원자적으로 진행됩니다.

// Addend인자 값을 1씩 증가시킵니다.
LONG InterlockedIncrement(
  LONG volatile *Addend
);

// Addend인자 값을 1씩 감소 시킵니다.
LONG InterlockedDecrement(
  LONG volatile *Addend
);

// Addend인자 값에 Value값을 더해줍니다.
LONG InterlockedAdd(
  LONG volatile *Addend,
  LONG          Value
);

// Addend인자 값에 Value값을 변경해줍니다.
LONG InterlockedExchange(
  LONG volatile *Target,
  LONG          Value
);

// 1. Destination인자 값과 Comperand인자 값을 비교합니다
// 2. 만약 값이 동일하다면 Destination와 ExChange값을 변경합니다. 
// 3. 만약 값이 다르다면 아무 작업도 진행하지 않습니다.
LONG InterlockedCompareExchange(
  LONG volatile *Destination,
  LONG          ExChange,
  LONG          Comperand
);

 더 많은 Interlocked 함수들이 있지만 위의 함수들만 설명을 진행합니다. 기본 함수들은 LONG(int)형식의 값을

 지원하고 해당함수 뒤에 64를 붙이면 LONGLONG(int64) 지원하고 16을 붙이면 SHORT형 변수를 지원합니다

volatile LONG* pValue = (LONG*)_aligned_malloc(sizeof(LONG),MEMORY_ALLOCATION_ALIGNMENT);
if (pValue)
{
    *pValue = 10;
    // 리소스 값을 1증가 시킵니다.
    InterlockedIncrement(pValue);
    
    // 리소스 값을 1감소 시킵니다.
    InterlockedDecrement(pValue);
    
    // 리소스 값을 두번째 인자값을 증가합니다.
    InterlockedAdd(pValue, 10);
    
    // 리소스 값을 두번째 인자 값과 교환합니다.
    InterlockedExchange(pValue, 40);
    
    // pValue는 40이고 Compare인자 값과 다르기 때문에 아무 작업도 하지 않습니다.
    InterlockedCompareExchange(pValue, 100, 30);
    
    // pValue는 40이고 Compare인자 값과 같기 때문에 pValue는 100으로 변경됩니다. 
    InterlockedCompareExchange(pValue, 100, 40);
}

_aligned_free((void *)pValue);

해당 로직의 결과값

 위의 간단한 예제를 통해서 InterLocked 함수를 사용하는 방법을 확인해볼 수 있습니다. 

InterLocked 함수의 내부 동작

 수행중인 CPU 플랫폼마다 다르지만 x86계열 CPU에서는 버스에 하드웨어 시그널을 통해서 다른 CPU가 동일 메모리

 주소에 접근하지 못하도록 하며 InterLocked 내부 수행 도중에 인터럽트가 발생되지 않도록 해서 해당 자원에 대한

 배타적인 접근을 지원합니다. 유저모드에서 동작 되기 때문에 매우 빠르게 동작합니다. 

 InterLocked 함수를 사용시 주의할 점은 공유 자원으로 사용할 값은 반드시 정렬되어 있어야 합니다. 

 _aligned_malloc 함수를 사용하면 정렬된 메모리 블록을 할당 할 수 있습니다. 

void * _aligned_mallock(size_t size, size_t alignment);
void _aligned_free(void * _Block);

 

InterLocked 함수 실제 속도체크

volatile LONG* pValue = (LONG*)_aligned_malloc(sizeof(LONG), MEMORY_ALLOCATION_ALIGNMENT);
volatile LONG* pValue2 = new LONG;
LONG* pValue3 = new LONG;
*pValue2 = 200;
*pValue3 = 100;
*pValue = 10;

	// 일반적인 LONG 변수 값 더하기
	{
		ElapsedTimeOutput elapsed("LONG");
		for (int i = 0; i < 10000; i++)
			*pValue3 += 1;
		elapsed.nanoSecPrint();
	}
	
	// volatile LONG 변수 값 더하기
	{
		ElapsedTimeOutput elapsed("volatile LONG");
		for (int i = 0; i < 10000; i++)
			*pValue2 += 1;
		elapsed.nanoSecPrint();
	}

	{
		ElapsedTimeOutput elapsed("InterlockedIncrement");
		for(int i = 0; i <10000 ; i++)
			InterlockedIncrement(pValue);
		elapsed.nanoSecPrint();
	}
		
	{
		ElapsedTimeOutput elapsed("InterlockedExchangeAdd");
		for (int i = 0; i < 10000; i++)
			InterlockedExchangeAdd(pValue, 10);
		elapsed.nanoSecPrint();
	}

	{
		ElapsedTimeOutput elapsed("InterlockedExchange");
		for (int i = 0; i < 10000; i++)
			InterlockedExchange(pValue, 40);
		elapsed.nanoSecPrint();
	}

	{
		ElapsedTimeOutput elapsed("InterlockedCompareExchange Fail");
		for (int i = 0; i < 10000; i++)
			InterlockedCompareExchange(pValue, 100, 30);
		elapsed.nanoSecPrint();
			
		ElapsedTimeOutput elapsed2("InterlockedCompareExchange Success");
		for (int i = 0; i < 10000; i++)
			InterlockedCompareExchange(pValue, 100, *pValue);
		elapsed2.nanoSecPrint();
	}

 위와 같은 코드를 실행해서 InterLocked 함수의 수행속도를 측정해 보았습니다.

 수행 횟수는 10000회로 진행되었으며 수행 머신의 CPU는 "I7-4790 3.6GHz" 입니다.

 64비트 Release 환경에서 측정하였습니다.

수행속도 결과

InterLockedIncrement 함수는 volatile Long보다 3배 느리며 일반 Long보다 6~7배 정도 느린 것으로

판명되었습니다. InterLockedCompareExchange함수의 경우 성공시 조금 더 느린 모습을 보여주었습니다.  

멀티 스레드 환경에서 속도차이

DWORD WINAPI ThreadFuncLongAdd(PVOID pvParam)
{
	LONG * value = (LONG *)pvParam;

	for (int i = 0; i < 100000; i++)
	{
		*value += 1;
	}
	

	return (DWORD)(value);
}

DWORD WINAPI ThreadFuncVolatileLongAdd(PVOID pvParam)
{
	volatile LONG* volatile_LONG = (volatile LONG*)pvParam;

	for (int i = 0; i < 100000; i++)
	{
		*volatile_LONG += 1;
	}


	return (DWORD)(volatile_LONG);
}

DWORD WINAPI ThreadFuncInterLockedAdd(PVOID pvParam)
{
	volatile LONG* volatile_LONG = (volatile LONG*)pvParam;

	for (int i = 0; i < 100000; i++)
		InterlockedIncrement(volatile_LONG);

	return (DWORD)(volatile_LONG);
}

 멀티 스레드 환경에서의 속도차이를 확인하기 위해 위의 스레드 함수 로직을 실행 시켰습니다.

 스레드 수가 증가 할 수록 동기화 작업 때문에 InterLockedIncrement함수의 수행시간이

 지연되는 것을 확인 할 수 있었습니다. 동기화 하지 않은 LONG, volatile LONG의 속도는 빨랐지만 

 결과 값은 데이터레이스로 인해서 비 정상적인 값이 설정되었습니다.

 (LONG, volatile LONG의 시간 차이가 많이 나지 않는건 아마도 스레드 생성시간이 추가되서 그런 듯)

스레드 수 LONG volatile Long InterLockedIncrement
2 30 마이크로초 33 마이크로초 229 마이크로초
4 47 57 678
8 60 96 1464
16 116 150 2940
32 210 287 5945

std::atomic

 c++의 std에서는 <atomic> 헤더 파일을 통해서 std::atomic을 지원합니다. 

 window에서는 interlock함수를 통해서 구현되며 std::atomic 또한 변수에 대한 원자적인 접근을

 지원합니다.

Interlocked Singly Linked Lists

 WINDOWS에서는 Interlocked 함수로 구현한 멀티 스레드 환경에서 안전한 스택(이름은 리스트인데..) 컨테이너를

 지원합니다. 

 윈도우 8 이전에는 32비트에서만 지원이 가능했는데 윈도우 8부터는 InterlockedCompare64Exchange128. 함수가

 지원되면서 64비트에서 동작 가능합니다.

함수명 설명
InitializeSListHead 인터락 싱글 링크 리스트의 헤더를 초기화합니다.(사용준비작업)
InterlockedFlushSList 인터락 싱글 링크드 리스트의 내부를 비웁니다.
InterlockedPopEntrySList 맨 앞의 아이템을 삭제합니다. 
InterlockedPushEntrySList 맨 앞에 아이템을 추가합니다.
InterlockedPushListSList 맨 앞에 다른 인터락 싱글 링크드 리스트를 추가합니다.
QueryDepthSList 현재 아이템의 갯수를 리턴합니다. 

MSDN에 올라 온 간단한 예제를 통해서 구현한 로직입니다.

#include <windows.h>
#include <iostream>

using namespace std;
// 주의!!!
// Single Linked List 에서 사용하려는 구조체에는
// SLIST_ENTRY 변수가 추가되어야 하며, 
// 이 변수를 첫번째 인자로 사용해야 합니다.
typedef struct _TESTDATA
{
	SLIST_ENTRY ItemEntry;
	DWORD dwIdx;
} TESTDATA, * PTESTDATA;


int _tmain(int argc, _TCHAR* argv[])
{
	DWORD dwCount;
	PSLIST_ENTRY pFirstEntry;
	PSLIST_HEADER pListHead;
	PTESTDATA pTestData;

	// 내부에서 InterLocked 함수를 사용, 정렬된 메모리를 할당합니다.
	pListHead = (PSLIST_HEADER)_aligned_malloc(sizeof(PSLIST_HEADER), MEMORY_ALLOCATION_ALIGNMENT);

	if (NULL == pListHead)
	{
		cout << "pListHead Memeory Allocate Fail _aligned_malloc" << endl;
		return -1;
	}
	InitializeSListHead(pListHead);

	// 데이터 10개를 Stack에 추가합니다. 
	for (dwCount = 1; dwCount < 10; dwCount++)
	{
		pTestData = (PTESTDATA)_aligned_malloc(sizeof(TESTDATA), MEMORY_ALLOCATION_ALIGNMENT);
		if (NULL == pTestData)
		{
			cout << "pTestData Memeory Allocate Fail _aligned_malloc" << endl;
			return -2;
		}
		pTestData->dwIdx = dwCount;
		pFirstEntry = InterlockedPushEntrySList(pListHead, &(pTestData->ItemEntry));
	}

	// 데이터 10개를 Stack에서 삭제합니다. 
	PSLIST_ENTRY pListEntry;
	for (dwCount = 1; dwCount < 10; dwCount++)
	{
		pListEntry = InterlockedPopEntrySList(pListHead);

		if (NULL == pListEntry)
		{
			cout << "pListEntry NULL" << endl;
			return -3;
		}

		pTestData = (PTESTDATA)pListEntry;
		cout << "Stack pop" << pTestData->dwIdx << endl;

		// Stack의 크기를 리턴합니다. 
		cout << "Stack Count" << QueryDepthSList(pListHead) << endl;

		// 사용예제에서도 보면서 느끼시겟지만
		// 구조체의 첫번째 인자는 PSLIST_ENTRY이어야 합니다. 
		_aligned_free(pListEntry);

		// 예제에서는 메모리를 해제했지만실제 사용시에는 메모리 풀을 사용하면 
		// 더욱 효율적으로 작업할 수 있습니다. 
	}


	// 스택을 비우는 작업을 실행합니다. 
	pListEntry = InterlockedFlushSList(pListHead);
	pFirstEntry = InterlockedPopEntrySList(pListHead);
	if (NULL != pFirstEntry)
	{
		cout << "Stack is not empty!!!??" << endl;
		return -4;
	}

	_aligned_free(pListHead);

	return 0;
}

 소스코드를 설명을 드리면 우선 InterLocked 함수를 내부적으로 사용하기 때문에 정렬된 메모리를 사용해야 합니다.

 또 한가지 사용 시 주의할 점이 있습니다.

 컨테이너의 아이템을 선언 할 때 아이템 객체의 가장 상위 변수를 SLIST_ENTRY로 선언해야 합니다.

typedef struct DECLSPEC_ALIGN(16) _SLIST_ENTRY {
    struct _SLIST_ENTRY *Next;
} SLIST_ENTRY, *PSLIST_ENTRY;

 SLIST_ENTRY는 구조는 위와 같은데 링크드 리스트를 위한 Next 포인터 정보를 담고 있습니다.

 왜 이렇게 선언해야 할까요? 그 이유는 내부적으로 SLIST_ENTRY의 포인터 형태로 리스트를 관리하기 때문입니다. 

 SLIST_ENTRY의 포인터 형태로 데이터를 가져와서 컨테이너 아이템 객체로 형변환을 해서 사용하도록 

 구현되어 있습니다.

 예제를 보면 InterLockedPopEntrySList 통해서 front에 있는 아이템의 PSLIST_ENTRY 전달받은 다음에 형변환을 통해서 

 PTESTDATA변환하여 아이템를 사용합니다.  (예제의 TESTDATA 구조체는 가장 상위 변수값이 ItemEntry입니다.)

 이렇게 설계되어 있기 때문에 아이템 객체에 최상위에 SLIST_ENTRY가 존재하지 않으면 잘못된 메모리 변환으로

 인해서 비정상적인 동작이 발생합니다. 

 사용방법이 비직관적이여서 엄청나게 성능이 좋지 않다면 사용하지 않을 것 같습니다....

Interlocked Singly Linked Lists VS Using Critical Section Stack

// Critical Section을 사용한 Stack 구현
unsigned __stdcall LOCKPUSHPOP(void* arg)
{
	SLIST_ARGUMENT* slistArg = (SLIST_ARGUMENT*)arg;

	if (slistArg)
	{
		DWORD dwCount = 0;
		std::for_each(slistArg->ContainerTestData.begin(), slistArg->ContainerTestData.end(),
			[&dwCount](PTESTDATA pTestData)
		{
			CFSScopeLock ScopeLock(&g_cs);
			g_StackData.push(pTestData);
			++dwCount;
		});

		// 데이터 삭제
		for (DWORD i = 0; i < dwCount - 1; ++i)
		{
			CFSScopeLock ScopeLock(&g_cs);
			g_StackData.pop();
		}
	}

	return 0;
}

// InterLocked Singly linged list 을 사용한 Stack 구현
unsigned __stdcall SLISTPUSHPOP(void* arg)
{
	SLIST_ARGUMENT* slistArg = (SLIST_ARGUMENT*)arg;
	if (slistArg)
	{

		DWORD dwCount = 0;
		// 데이터 삽입
		std::for_each(slistArg->ContainerTestData.begin(), slistArg->ContainerTestData.end(),
			[&slistArg, &dwCount](PTESTDATA pTestData)
		{
			InterlockedPushEntrySList(slistArg->pListHead, &(pTestData->ItemEntry));
			++dwCount;
		});

		// 데이터 삭제
		for (DWORD i = 0; i < dwCount - 1; ++i)
		{
			InterlockedPopEntrySList(slistArg->pListHead);
		}
	}

	return 0;
}

//===== [ 수행 로직] =====

{
	ElapsedTimeOutput elapsed("InterLocked Stack Thread Run ");
	for (DWORD dwThreadIdx = 0; dwThreadIdx < kTEST_THREAD_COUNT; ++dwThreadIdx)
	{
		unsigned ThreadId = 0;
		hSListThreads[dwThreadIdx] = (HANDLE)_beginthreadex(NULL, NULL, SLISTPUSHPOP, vecArgument[dwThreadIdx], NULL, &ThreadId);
	}
	DWORD dwResult = WaitForMultipleObjects(kTEST_THREAD_COUNT, hSListThreads, TRUE, INFINITE);
	elapsed.microSecPrint();
	if (WAIT_OBJECT_0 == dwResult)
		cout << "Run SLISTPUSHPOP Thread = " << kTEST_THREAD_COUNT << " pListHead Depth = " << QueryDepthSList(pListHead) << endl;
	else
		cout << "SLISTPUSHPOP Fail WaitForMultipleObjects" << endl;
}


{
	ElapsedTimeOutput elapsed("Critical Section Stack Thread Run ");
	for (DWORD dwThreadIdx = 0; dwThreadIdx < kTEST_THREAD_COUNT; ++dwThreadIdx)
	{
		unsigned ThreadId = 0;
		hSListThreads[dwThreadIdx] = (HANDLE)_beginthreadex(NULL, NULL, LOCKPUSHPOP, vecArgument[dwThreadIdx], NULL, &ThreadId);
	}
	DWORD dwResult = WaitForMultipleObjects(kTEST_THREAD_COUNT, hSListThreads, TRUE, INFINITE);
	elapsed.microSecPrint();
	if (WAIT_OBJECT_0 == dwResult)
		cout << "Run LOCKPUSHPOP Thread = " << kTEST_THREAD_COUNT << " g_StackData Count = " << g_StackData.size() << endl;
	else
		cout << "LOCKPUSHPOP Fail WaitForMultipleObjects" << endl;
}

 Lock을 사용하는 것보다 InterLocked Singly Linked lists를 사용하면 얼마나 효율적인지 테스트를 위한 로직을 구현 후

 수행시켜 보았습니다. terLockelinked list

스레드수 오브젝트수 InterLocked Singly linked lists Critical Section's Stack
1 5000 1221 마이크로초 422 마이크로초
2 1000 1108 652
2 5000 2069 2531
4 1000 1263 1424
4 5000 3478 7179
8 1000 1301 3571
16 1000 2185 8124
32 1000 7191 14525
32 5000 38984 89848

32스레드 결과

 결과를 보면 수행 되는 스레드가 많아 지거나 오브젝트가 많아 질수록 InterLocked Singly Linked lists의 속도가

 빨라졌습니다. 스레드나 오브젝트 수가 작을 땐 오히려 Critical Section이 더 빠르게 동작합니다. 

 결과가 예상한 것보다 훨씬 재미있게 나왔습니다. 상황에 따라 InterLocked Singly Linked lists을 사용하는 것도

 괜찮은 방법인 것 같습니다.( 사용하기 쉽도록 Wrapper 클래스를 만들어서 사용해야 할 듯합니다.) In

저작자표시 동일조건

'c++ > windows' 카테고리의 다른 글

[Window c++] 크리티컬 섹션  (0) 2020.02.16
[windows] 멀티 프로세서 환경에서의 CPU 캐시라인  (1) 2020.02.15
[window c++] 스레드 스케줄링  (0) 2020.02.09
[window c++] 스레드  (0) 2020.02.04
[windows c++] 프로세스  (1) 2020.01.31

+ Recent posts

티스토리툴바