DeadLockProfiler

데드 락 방지

멀티 쓰레드에서 lock을 사용하다 보면 프로그래머의 실수로 데드 락이 발생할 수 있다. 이런 데드 락을 피하기 위해서는 프로그래머가 lock을 사용할 때 순서를 주의해서 사용해야한다. 그러나 사람은 실수를 하기 때문에 이를 보완해줄 방법이 있으면 좋을 것 같다. 

데드 락을 방지하는 방법중 하나로 lock을 걸 때 현재 걸린 lock들의 상황을 파악해 이 lock이 데드락을 유발할 가능성이 있는지 검사하는 것이다. 이것이 이 프로젝트에 적용된 DeadLockProfiler 이다.

---

원리

데드 락은 lock 내부에서 서로의 lock이 풀릴 때까지 무한히 대기하는 현상이다. 이 서로를 기다린다는 것의 의미를 잘 생각해보자. 이 말은 lock들의 관계를 그래프로 나타내었을 때 사이클을 이루게 되면 데드 락이 발생할 수 있는 상황이라는 뜻이다. 즉 우리는 lock을 해줄 때마다 lock들의 관계를 그래프로 나타내고, 그 그래프가 사이클을 이루는지 검사하면 된다.

 

그래프로 나타내기

그럼 어떻게 lock들의 관계를 그래프로 나타낼 수 있을까? 우선 그래프는 정점과 간선으로 나타내어진다. 정점은 lock, 간선은 어떤 lock이 걸려있는 상태에서 또 다른 lock을 거는 관계 이다. 

예를 들어 어떤 클래스 A가 Lock을 걸어 둔 상태에서 클래스 B도 Lock을 걸었다면, A -> B 라는 정점과 간선 정보를 생성하는 것이다. 

 

사이클을 이루는지 검사하기

그럼 이 그래프에서 사이클이 존재하는지 어떻게 알 수 있을까? 방법은 여러 가지 있을 수 있다. 여기선 DFS를 기초로 하고, 노드들이 추가된 순서를 활용하는 방법으로 검사해보자. 

알고리즘은 대략 아래와 같다. 

1. 우선 그래프를 만들 때 각 노드가 추가된 순서를 기록해둔다.
2. 단방향 그래프가 완성되면 DFS 를 시작한다.
3. DFS 탐색 중 인접 노드를 방문했는데 이미 방문했던 노드라면
4. 인접 노드가 더 먼저 추가된 노드인지 확인한다.
5. 인접 노드가 더 먼저 추가된 노드라면 (역방향이라면) 사이클이 발생한 것이므로 CRASH
6. 인접 노드가 더 나중에 추가된 노드라면 (순방향이라면) 사이클이 아니므로 탐색을 계속한다.
7. 모든 노드를 방문할 때까지 계속한다.

간단히 말하면 DFS 탐색을 하는 중에 방문하는 노드가 이미 방문했던 노드이고, 역방향이라면 사이클이라는 뜻이다.

 

---

코드

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

	int32 lockId = 0;

	//아이디를 찾거나 새로 만듦
	auto foundIt = _nameToId.find(name);
	if (foundIt == _nameToId.end())
	{
		lockId = _nameToId.size();
		_nameToId[name] = lockId;
		_idToName[lockId] = name;
	}
	else 
	{
		lockId = foundIt->second;
	}

	//아이디를 lockStack에 넣음

	//하려는 lock이 사이클을 만드는지 확인

	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);
	_discoverdCount = 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] = _discoverdCount++;

	//정점 순회 시작
	//연결된 노드가 없으면 finish
	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;
		}

		//연결된 노드가 이미 발견한 노드면 
		//연결된 노드가 더 나중에 발견된 노드면 (순방향이면) 넘어감
		if (_discoveredOrder[here] < _discoveredOrder[there])
			continue;

		//연결된 노드가 더 먼저 발견된 노드면 (역방향이면) 사이클 발생
		if (_finished[there] == false)
		{
			printf("%s->%s\n", _idToName[here], _idToName[there]);
			int32 now = here;
			while (true)
			{
				printf("%s->%s\n", _idToName[_parent[now]], _idToName[now]);
				now = _parent[now];
				if (now == there)
					break;
			}
			CRASH("DEADLOCK_DETECTED");
		}

	}
	_finished[here] = true;

}