DeadLock
- '교착 상태'라고도 하며, 두 개 이상의 작업이 서로 상대방의 작업이 끝나기 만을 기다리고 있기 때문에 결과적으로 아무것도 완료되지 못하는 상태
- 시스템적으로 한정된 자원을 여러 곳에서 사용하려고 할 때 발생
- 데드락이 발생할 수 있는 경우:
Process 1과 Process2가 Resource 1, 2를 둘 다 얻어야 한다고 가정할 때, 서로 원하는 Resource가 상대 Process에게 할당되어 있기 때문에 두 프로세스가 무한정 기다리게 된다.
교착 상태의 조건
| 조건 | 설명 |
| 상호 배제 (Mutual exclusion) |
- 자원은 한 번에 하나의 프로세스만이 사용할 수 있어야 한다. - 프로세스들이 필요로 하는 자원에 대해 배타적인 통제권을 요구한다. - 상호 배제를 제거하는 것은 가장 확실한 교착 상태 제거 방법이지만, 프로세스나 스레드의 동시성 제어 외에는 잘 사용하지 않는다. (Mutex) |
| 점유 대기 (Hold and wait) |
- 최소한 하나의 자원을 점유하고 있으면서 다른 프로세스에 할당되어 사용하고 있는 자원을 추가로 점유하기 위해 대기하는 프로세스가 있어야 한다. - 프로세스가 할당된 자원을 가진 상태에서 다른 자원을 기다린다. ex) Skype가 마이크와 카메라를 쓴다고 가정. 이 때, 마이크는 연결에 성공했지만 다른 앱이 카메라를 잡고 있어서 연결할 수 없는 경우, Skype가 카메라를 계속 대기하게 되므로 다른 앱이 마이크를 사용하지 못하는 상황 |
| 비선점 (No preemption) |
- 다른 프로세스에 할당된 자원은 사용이 끝날 때까지 강제로 빼앗을 수 없어야 한다. - 프로세스가 어떤 자원의 사용을 끝낼 때까지 그 자원을 뺏을 수 없다. |
| 순환 대기 (Circular wait) |
- 프로세스의 집합 {P0, P1, ... , Pn}에서 P0은 P1이 점유한 자원을 대기하고 P1은 P2가 점유한 자원을 대기하고 P2 ... Pn-1은 Pn이 점유한 자원을 대기하며 Pn은 P0가 점유한 자원을 요구해야 한다. - 각 프로세스는 순환적으로 다음 프로세스가 요구하는 자원을 가지고 있다. |
교착 상태(DeadLock)에 대해 어느 정도 알아보았으니, 이제 교착 상태를 탐지하는 법을 알아보자.
그래프 사이클 판별
우리가 사용하는 스레드가 락을 거는 순서를 그래프로 표현했을 때 위와 같은 그래프가 나왔다고 해보자.
(클래스가 6개있고, 내부적으로 각각 락을 1개씩 잡고있는 상황)
0번 락을 잡은 상태에서 1번 락을 잡게 되면 아래 그래프처럼 나타낼 수 있다.
여기서 2번 락까지 잡게 된다면 그래프의 모습은 아래와 같이 될 것이다.
만약 0번에서 1번락을 잡으려고 하고 다른 코드에서
1번락을 잡고 0번락을 잡으려고 하는 상황이 발생한다면, 데드락이 발생하게 된다.
꼭 2개를 대상으로만 사이클이 일어나는 것은 아니다.
만약 3번에서 1번락과 0번락을 잡고있었다면, 삼각형 모양으로 사이클이 발생하게 될 것이다.
락을 잡고 있는 구조를 그래프로 표현해서 사이클이 일어나는 지를 판단하면 데드락을 탐지하여 예방 할 수 있게된다.
그렇다고 항상 탐지를 하는게 아닌 디버그 모드에서 개발을 할 때만 탐지를 하면 성능 부하도 없고 사전에 버그를 잡을 수 있게 된다.
DFS
그럼 그래프 사이클을 어떻게 판별을 할까?
길찾기 알고리즘에서도 사용했던, DFS 알고리즘을 이용하면 된다.
DFS 알고리즘을 이용하여 간선을 판별했을 때, 역방향 간선이 판별 된다면 사이클이 판별됐다는 것이다.
어떤 방향 그래프를 깊이 우선 탐색(DFS)했을 때, 탐색이 따라간 간선들을 모으면 트리 형태를 가지는데, 이때 생성된 트리를 DFS 스패닝 트리(DFS Spanning Tree)라고 부른다.
1. 트리 간선 (Tree Edge)
- DFS 스패닝 트리에 포함된 간선
2. 순방향 간선 (Forward Edge)
- DFS 스패닝 트리의 선조(ancestor)에서 자손(descendant)으로 가는 간선이면서, 트리에 포함되지 않은 간선
3. 역방향 간선 (Back Edge)
- DFS 스패닝 트리의 자손에서 선조로 가는 간선이면서, 트리에 포함되지 않은 간선
4. 교차 간선(Cross Edge)
- 위의 세 분류에 포함되지 않은 간선. 즉, 트리의 선조와 자손의 관계가 아니면서 트리에 포함되지 않은 간선
물론 DFS 과정에서 정점을 어떤 순서로 방문할지에 따라서 트리의 구조나 간선의 분류가 달라질 수 있다.
무방향 그래프에서는 모든 간선이 양방향으로 통행이 가능하기 때문에 교차 간선은 생길 수 없다. 그리고 순방향과 역방향의 구분또한 없다. 따라서 트리에 포함되거나, 포함되지 않는 간선 두 가지로 나뉘게 된다.
그렇다면, 실제로 그래프가 주어졌을 때, 간선들을 어떻게 분류할 수 있을까?
우선, 트리 간선은 간단하게 확인할 수 있다. dfs(u) 내에서 간선 (u, v)를 확인할 때, v가 아직 방문하지 않은 정점이라면 (u, v)는 트리를 구성하게 된다.
반면, v가 이미 방문한 정점이라면 v가 u의 부모인지 자손인지 둘 다 아닌지 판별하기가 어렵다. 따라서 이를 판별하기 위해서 'dfs 정점 방문 순서'를 기록해둔다.
1. (u, v)가 순방향 간선
- v가 u의 자손이어야 한다. 따라서 v는 u보다 더 늦게 방문했어야 한다.
2. (u, v)가 역방향 간선
- v가 u의 선조이어야 한다. 따라서 v는 u보다 더 먼저 방문했어야 한다.
3. (u, v)가 교차 간선
- dfs(v)가 dfs(u)보다 먼저 종료되어야 하므로 v가 u보다 더 먼저 방문했어야 한다.
따라서, u를 먼저 방문했다면 (u, v)는 순방향 간선이고, v를 먼저 방문했고, dfs(v)가 아직 종료되지 않았다면 역방향 간선, v를 먼저 방문했고 dfs(v)가 종료되었다면 교차 간선이다.
DeadLockProfiler
//DeadLockProfiler.h
#pragma once
#include <stack>
#include <map>
#include <vector>
/*-----------------------
*
* DeadLockProfiler
*
-----------------------*/
class DeadLockProfiler
{
public :
void PushLock(const char* name);
void PopLock(const char* name);
void CheckCycle();
private:
void Dfs(int32 here);
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;
};
- unordered_map<const char*, int32> _nameToId : 이름과 해당 락의 번호를 관리.
- unordered_map<int32, const char*> _idToName : 락의 번호와 이름을 관리
- stack<int32> _lockStack : 락을 추적하기 위한 스택
- map<int32, set<int32>> _lockHistory : 어떠한 락이 어떤 락을 잡았는지 히스토리를 만들어줌. (간선 정보)
- Mutex _lock : 멀티스레드 환경에서 안전하게 동작하기 위한 뮤텍스
//DeadLockProfiler.cpp
#include "pch.h"
#include "DeadLockProfiler.h"
/*-----------------------
*
* 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];
//새로운 간선이 발견된 경우에만 CheckCycle 해줌.
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)
{
//_discoveredOrder[here]가 -1이 아니면 방문한적 있다는 뜻이므로 return
if (_discoveredOrder[here] != -1)
return;
//방문했으니까 1증가 시켜줌.
_discoveredOrder[here] = _discoveredCount++;
//모든 인접한 정점을 순회한다.
auto findIt = _lockHistory.find(here);
//_lockHistory를 체크했는데 없다고 하면
if (findIt == _lockHistory.end())
{
//해당 락을 잡은 상태에서 다른 락을 잡은 적이 없다는 뜻이기 때문에
_finished[here] = true;
return;
}
//_lockHistory를 체크했는데 있다고 하면 어떤 락을 잡은 상태에서 새로운 락을 잡은적이 있다는 뜻이기 때문에
//사이클이 있는지 판별하기 위해 set을 추출해준다.
set<int32>& nextSet = findIt->second;
for (int32 there : nextSet)
{
//아직 방문한 적이 없다면 방문한다.
if (_discoveredOrder[there] == -1)
{
//부모(선조)와 자식 관계 형성.
//there는 here 때문에 방문.(here -> there)
_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;
while (true)
{
printf("%s -> %s\n", _idToName[here], _idToName[there]);
now = _parent[now];
if (now == there)
break;
}
CRASH("DEADLOCK_DETECTED");
}
}
_finished[here] = true;
}
void PushLock(const char* name);
int32 lockId = 0;
auto findIt = _nameToId.find(name);
if (...) else ...
아이디를 찾거나 발급한다.
_nameToId에서 name으로 find를 했는데 찾을 수 없어서 end 이터레이터를 반환 받았을 경우,
현재 _nameToId의 size로 아이디를 발급해준다.
만약 find 했을 경우에는 lockId에다가 findIt의 second(id)를 넣어준다.
if (_lockStack.empty() == false) {...}
잡고 있는 락이 있는지 판별한다.
잡고 있는 락이 있을 경우 기존에 발견된적 있는 케이스인지 판별하기 위해 _lockStack의 top을 prevId에 저장한다.
우리는 락을 만들 때 재귀적으로 똑같은 락을 여러번 잡을 수 있도록 하였기 때문에 lockId와 prevId가 같은 경우는 패스해주고 다른 경우에만 데드락 여부를 다시 확인하기 위해 사이클을 체크해준다.
사이클을 체크하기 위해 _lockHistory[prevId]를 참조해준다.
새로운 간선이 발견된 경우에만 history에 lockId를 넣어주고 checkCycle을 해준다.
(접근 하려는 lockId가 처음 발견된 경우, 즉 새로운 락을 잡는 경우)
void PopLock(const char* name);
_lockStack에서 해당 락을 pop해준다.
void CheckCycle();
지금까지 발견된 lockCount는 _nameToId의 size를 가져오면 된다.
_discoveredOrder, _discoveredCount, _finished, _parent 사이클에 사용할 변수들을 초기화 해준다.
그리고 lockCount 개수 만큼 DFS를 돌려준다.
연산이 끝났으면 변수들을 정리해준다.
void Dfs(int32 here);
//주석 참조.
'스레드 프로그래밍' 카테고리의 다른 글
| Reader-Writer Lock (0) | 2023.02.02 |
|---|---|
| ThreadManager (0) | 2023.02.01 |
| Lock-Free Stack(3) (0) | 2023.01.27 |
| Lock-Free Stack(2) (0) | 2023.01.27 |
| Lock-Free Stack (1) (0) | 2023.01.26 |