Memory pool #2

기존 memory pool #1의 아쉬운 점

1. memory pool에서 메모리 가져오기, 반환 때 lock을 사용하기에 여러 쓰레드에서 사용할 경우 경합이 일어난다.

2. memory pool 사용하기 위해 memory header를 저장하는 별도의 queue가 필요하다.

 

lock free stack을 이용하여 위 문제를 해결해보자.

이전에 구현할 때 Node를 사용했는데

template<typename T>
struct Node
{
    T data;
    Node* node;
};

위 방식의 문제는 T data를 만들기 위해 Node라는 공간을 또 만드는 것이다.

 

struct SListEntry
{
	SListEntry* next;
};

class Data
{
public:
    SListEntry _entry;
    int32 _hp;
    int32 _mp;
}

SListEntry를 사용하고자 하는 클래스에 넣어 사용하면 Node를 만들지 않고 Node를 만든것 같이 사용할 수 있다.

먼저 싱글 쓰레드 사용만 신경써서 기초 구조를 만든다.

LockFreeStack.h

struct SListEntry
{
	SListEntry* next;
};

struct SListHeader
{
	SListEntry* next = nullptr;
};

// [data][  ][  ]
// Header[next]
void InitializeHead(SListHeader* header);
void PushEntrySList(SListHeader* header, SListEntry* entry);
SListEntry* PopEntrySList(SListHeader* header);

LockFreeStack.cpp

// [data][  ][  ]
// Header[next]

void InitializeHead(SListHeader* header)
{
	header->next = nullptr;
}

void PushEntrySList(SListHeader* header, SListEntry* entry)
{
	entry->next = header->next;
	header->next = entry;
}

SListEntry* PopEntrySList(SListHeader* header)
{
	SListEntry* first = header->next;

	if (first != nullptr)
		header->next = first->next;

	return first;
}

 

사용은 다음처럼 한다.

class Data
{
public:
	SListEntry _entry;

	int32 _hp;
	int32 _mp;
};

int main()
{
	SListHeader header;
	InitializeHead(&header);

	Data* data = new Data();
	data->_hp = 10;
	data->_mp = 20;
	PushEntrySList(&header, (SListEntry*)data);

	Data* popData = (Data*)PopEntrySList(&header);
}

 

이제 멀티쓰레드 환경에서 작동가능하게 만들자.

LockFreeStack.cpp

void InitializeHead(SListHeader* header)
{
	header->next = nullptr;
}

void PushEntrySList(SListHeader* header, SListEntry* entry)
{
	entry->next = header->next;
	while (::InterlockedCompareExchange64((int64*)&header->next, (int64)entry, (int64)entry->next) != (int64)entry)
	{

	}
}

SListEntry* PopEntrySList(SListHeader* header)
{
	SListEntry* expected = header->next;
    
    // ABA Problem

	// [5000]->[7000]

	// 만약에 Header가 5000이라면, Header에다 6000을 넣어줘!
	// [5000]->[6000]->[7000]
	// [Header]
	while (expected && ::InterlockedCompareExchange64((int64*)&header->next, (int64)expected->next, (int64)expected) == (int64)expected->next)
	{

	}

	return expected;
}

PopEntrySList에서 expected 초기화 한 다음 InterlockedCompareExchange 연산을 하기 전에 다른 쓰레드에서 expected를 삭제하면 문제가 생긴다. 이전에는 이 문제를 해결하기 위해 refcount, pop하고 있는 쓰레드 개수 count등의 방법을 이용했다.

여기서 위 문제가 발생하지 않는다고 해도 다른 문제가 발생한다.

ABA Problem이다. 데이터가 5000 6000 7000 있는 상황에서 CAS연산을 하기전 쓰레드에 의해 5000 7000이 되었다고 가정하자. CAS연산은 5000이면 6000을 넣어주는 명령이므로 header는 엉뚱한 곳을 가리키게 된다.

 

문제를 해결하기 위해 주소값을 그대로 이용하지 않고 플래그를 두어 사이에 정보가 바뀌었는지 비교한다.

LockFreeStack.h, LockFreeStack.cpp

DECLSPEC_ALIGN(16)
struct SListEntry
{
	SListEntry* next;
};

DECLSPEC_ALIGN(16)
struct SListHeader
{
	SListHeader()
	{
		alignment = 0;
		region = 0;
	}
	union
	{
		struct
		{
			uint64 alignment;
			uint64 region;
		} DUMMYSTRUCTNAME;
		struct
		{
			uint64 depth : 16;
			uint64 sequence : 48;
			uint64 reserved : 4;
			uint64 next : 60;
		} HeaderX64;
	};
};

void InitializeHead(SListHeader* header);
void PushEntrySList(SListHeader* header, SListEntry* entry);
SListEntry* PopEntrySList(SListHeader* header);

void InitializeHead(SListHeader* header)
{
	header->alignment = 0;
	header->region = 0;
}

void PushEntrySList(SListHeader* header, SListEntry* entry)
{
	SListHeader expected = {};
	SListHeader desired = {};
	
	// 16 바이트 정렬
	desired.HeaderX64.next = (((uint64)entry) >> 4);

	while (true)
	{
		expected = *header;
		// 이 사이에 변경될 수 있다
		entry->next = (SListEntry*)(((uint64)expected.HeaderX64.next) << 4);
		desired.HeaderX64.depth = expected.HeaderX64.depth + 1;
		desired.HeaderX64.sequence = expected.HeaderX64.sequence + 1;

		if (::InterlockedCompareExchange128((int64*)header, desired.region, desired.alignment, (int64*)&expected) == 1)
			break;
	}
}

SListEntry* PopEntrySList(SListHeader* header)
{
	SListHeader expected = {};
	SListHeader desired = {};
	SListEntry* entry = nullptr;

	while (true)
	{
		expected = *header;
		entry = (SListEntry*)(((uint64)expected.HeaderX64.next) << 4);
		if (entry == nullptr)
			break;

		// Use-After-Free
		desired.HeaderX64.next = ((uint64)entry->next) >> 4;
		desired.HeaderX64.depth = expected.HeaderX64.depth - 1;
		desired.HeaderX64.sequence = expected.HeaderX64.sequence + 1;

		if (::InterlockedCompareExchange128((int64*)header, desired.region, desired.alignment, (int64*)&expected) == 1)
			break;
	}

	return entry;
}

위 해결방법도 완벽하지 않은데, Use-After-Free 문제 즉, 다른 쓰레드가 pop을 하여 참조 주소를 날려버린 경우 크래시 또는 메모리 오염문제가 발생한다.

 

1. PushEntrySList 에서 InterlockedCompareExchange64가 하는 역할은?

- 원자적 연산, 3번째 파라메터가 1번째 파라메터가 가리키는 값과 같으면 2번째 파라미터를 1번째 파라미터가 가리키는 위치에 복사함.

2. 1에서 0을 반환하면 계속 시도하는 이유는?

- 잘못됨. 현재 고침.

- 추가로 while 문안에서 entry->next를 초기화 하는 코드가 있어야 할 것으로 보임. 정확하지 않아 수정하지 않음.

3. DECLSPEC_ALIGN(16)을 사용하는 이유는?

- 하위 4비트를 비워두기 위해

 

참조 : [C++과 언리얼로 만드는 MMORPG 게임 개발 시리즈] Part4: 게임 서버 - 인프런 | 강의 (inflearn.com)