[C++ 게임 서버] 1-23. DeadLock 탐지

Rookiss 님의 '[C++과 언리얼로 만드는 MMORPG 게임 개발 시리즈] Part4: 게임 서버' 를 들으며 배운 내용을 정리하고 있습니다. 관심이 있으신 분은 꼭 한 번 들어보시기를 추천합니다!

[C++ 게임 서버] 1-23. DeadLock 탐지

핵심 :

1. 데드락을 잡는다는 것은, 결국 락을 잡는 방식에 있어 사이클이 형성되는지를 판단하는 것과 동일하다.

2. DFS 를 응용하여, 사이클이 생기는지 여부를 검사하면 DeadLock 상황을 미리 검출할 수 있다.

---

DeadLock 의 경우, 개발 단계에서는 나오지 않으나, 막상 라이브 서비스를 하면 검출되는 경우가 많다.

구조적으로는 데드락이 걸릴 로직이 있음에도, 접속자 수가 많지 않아 데드락이 걸리지 않을 수 있다. 이런 타이밍 적인 이슈에 의존하지 않고, DFS 방식으로 사이클이 있는지를 검사하여, 데드락 문제를 미리 방지할 수 있다.

데드락을 잡는다는 것은, 결국 락을 잡는 방식에 있어 사이클이 형성되는지를 판단하는 것과 동일하기 때문이다!

 

자세한 것은 코드를 통해 설명하겠다.

 

일단 main 함수에서는, 다음과 같이 무조건 데드락이 걸리는 구조로 간단하게 코드를 짠다(이전의 Reader-Writer Lock 클래스는 그대로 활용).

int main()
{
	GThreadManager->Launch([=]
		{
			while (true)
			{
				cout << "PlayerThenAccount" << endl;
				GPlayerManager.PlayerThenAccount();
				this_thread::sleep_for(100ms);
			}
		});

	GThreadManager->Launch([=]
		{
			while (true)
			{
				cout << "AccountThenPlayer" << endl;
				GAccountManager.AccountThenPlayer();
				this_thread::sleep_for(100ms);
			}
		});

	GThreadManager->Join();
}

PlayerThenAccount 와 AccountThenPlayer 는 그냥 더미 클래스로, 아래와 같은 역할을 하는 녀석이다.

 

void AccountManager::AccountThenPlayer() 
{
	WRITE_LOCK;
	GPlayerManager.Lock();
}

void AccountManager::Lock()
{
	WRITE_LOCK;
}

 

중요한 것은, DeadLockProfiler 라는 클래스다. 이 녀석이 DFS 를 수행하며 사이클을 검사할 것이다.

DeadLockProfiler.h

/*--------------------
	DeadLockProfiler
---------------------*/

class DeadLockProfiler
{
public:
	void PushLock(const char* name);
	void PopLock(const char* name);
	void CheckCycle();

private:
	void Dfs(int32 index);

private:
	unordered_map<const char*, int32>	_nameToId;
	unordered_map<int32, const char*>	_idToName;
	stack<int32>						_lockStack;
	map<int32, set<int32>>				_lockHistory;

	Mutex _lock;

private:
	vector<int32>	_discoveredOrder; // 노드가 발견된 순서를 기록하는 배열
	int32			_discoveredCount = 0; // 노드가 발견된 순서
	vector<bool>	_finished; // Dfs(i)가 종료 되었는지 여부
	vector<int32>	_parent;
};

 

DeadLockProfiler.cpp

/*--------------------
	DeadLockProfiler
---------------------*/

void DeadLockProfiler::PushLock(const char* name)
{
	LockGuard guard(_lock);

	// 아이디를 찾거나 발급한다.
	int32 lockId = 0;

	auto findIt = _nameToId.find(name);
	if (findIt == _nameToId.end())
	{
		lockId = static_cast<int32>(_nameToId.size());
		_nameToId[name] = lockId;
		_idToName[lockId] = name;
	}
	else
	{
		lockId = findIt->second;
	}

	// 잡고 있는 락이 있었다면
	if (_lockStack.empty() == false)
	{
		// 기존에 발견되지 않은 케이스라면 데드락 여부 다시 확인한다.
		const int32 prevId = _lockStack.top();
		if (lockId != prevId)
		{
			// 현재 잡고 있었던 락의 히스토리를 참고. 즉, 현재 잡고 있었던 락에 대해, 새로 추가하려는 락을 넣고,
			set<int32>& history = _lockHistory[prevId];
			if (history.find(lockId) == history.end())
			{
				history.insert(lockId);
				CheckCycle();
			}
		}
	}

	_lockStack.push(lockId);
}

void DeadLockProfiler::PopLock(const char* name)
{
	LockGuard guard(_lock);

	if (_lockStack.empty())
		CRASH("MULTIPLE_UNLOCK");

	int32 lockId = _nameToId[name];
	if (_lockStack.top() != lockId)
		CRASH("INVALID_UNLOCK");

	_lockStack.pop();
}

void DeadLockProfiler::CheckCycle()
{
	const int32 lockCount = static_cast<int32>(_nameToId.size());
	_discoveredOrder = vector<int32>(lockCount, -1);
	_discoveredCount = 0;
	_finished = vector<bool>(lockCount, false);
	_parent = vector<int32>(lockCount, -1);

	for (int32 lockId = 0; lockId < lockCount; lockId++)
		Dfs(lockId);

	// 연산이 끝났으면 정리한다.
	_discoveredOrder.clear();
	_finished.clear();
	_parent.clear();
}

void DeadLockProfiler::Dfs(int32 here)
{
	if (_discoveredOrder[here] != -1)
		return;

	_discoveredOrder[here] = _discoveredCount++;

	// 모든 인접한 정점을 순회한다.
	auto findIt = _lockHistory.find(here);
	if (findIt == _lockHistory.end())
	{
		_finished[here] = true;
		return;
	}

	set<int32>& nextSet = findIt->second;
	for (int32 there : nextSet)
	{
		// 아직 방문한 적이 없다면 방문한다.
		if (_discoveredOrder[there] == -1)
		{
			_parent[there] = here;
			Dfs(there);
			continue;
		}

		// here가 there보다 먼저 발견되었다면, there는 here의 후손이다. (순방향 간선)
		if (_discoveredOrder[here] < _discoveredOrder[there])
			continue;

		// 순방향이 아니고, Dfs(there)가 아직 종료하지 않았다면, there는 here의 선조이다. (역방향 간선)
		if (_finished[there] == false)
		{
			printf("%s -> %s\n", _idToName[here], _idToName[there]);

			int32 now = here;

			// 이 루프는 실제 역방향 간선의 이름을 출력하기 위해 구현된 것이다 (사이클 여부는 이미 위의 if 문에서 판단됨)
			// 부모를 계속 타고 올라가면서, 사이클을 일으키는 간선을 검출함
			while (true)
			{
				printf("%s -> %s\n", _idToName[_parent[now]], _idToName[now]);
				now = _parent[now];
				if (now == there)
					break;
			}

			CRASH("DEADLOCK_DETECTED");
		}
	}

	_finished[here] = true;
}

핵심은 CheckCyle 함수이다.

 

간단히 설명하자면, 새로운 락을 잡으려고 할때, 이전 락이 있을 경우 사이클을 검사한다.

기존에 잡혀 있던 lock 의 history 에 해당하는 Set 에 현재 락 Id 를 넣고, DFS 를 돌려서 해당 락을 이미 방문하였는지(해당 락을 이미 잡았던 적이 있는지)를 체크하는 것이다.

자세한 건... 코드를 보자 😅

 

그리고, 이전의 Reader-Writer Lock 에서 Lock 과 Unlock 함수 각각에 PushLock, PopLock 을 삽입한다.

#if _DEBUG
	GDeadLockProfiler->PushLock(name);
#endif

#if _DEBUG
	GDeadLockProfiler->PopLock(name);
#endif

 

그럼 타이밍 이슈와 관계없이, 데드락이 있을 경우가 잘 검출되게 된다 😉