[C++/게임서버] Lock 기반 Stack/Queue

기본적으로 STL에서 제공하는 Stack이나 Queue는 Thread-Safe한 물건은 아니다 그렇기 때문에, 데이터를 push/pop하는 과정에서 락이나 그에 상응하는 방식으로 데이터의 원자성을 보장을 해줘야 한다. 다음과 같은 예시가 있다고 가정하자.

 

queue<int32> q;

void Push()
{
	while (true)
	{
		int32 value = rand() % 100 + 1;
		q.push(value);

		this_thread::sleep_for(10ms);
	}
}

void Pop()
{
	while (true)
	{
		if (q.empty())
			continue;

		int32 value = q.front();
		cout << value << endl;
		q.pop();

		this_thread::sleep_for(10ms);
	}
}

int main()
{
	thread t1(Push);
	thread t2(Pop);
	thread t3(Pop);
    
    t1.join();
    t2.join();
    t3.join();

	return 0;
}

다음과 같은 경우 경합이 일어나서 pop 부분에서 에러가 날 가능성이 굉장히 높은 편이다. 그렇기 때문에 Lock 기능이 존재하는 Stack과 Queue를 개발해 둘 필요가 존재해보인다. 이번 포스트에서는 Mutex 기반의 선형 자료구조를 소개한다.

 

template <typename T>
class ConcurrentStack
{
public : 
	ConcurrentStack() {}
	~ConcurrentStack() {}
	ConcurrentStack(const ConcurrentStack& rhs) = delete;
	ConcurrentStack& operator=(const ConcurrentStack& rhs) = delete;

	void Push(T value)
	{
		lock_guard<mutex> lock(_mutex);
		_st.push(value);
		_cv.notify_one();
	}

	bool TryPop(T& value)
	{
		lock_guard<mutex> lock(_mutex);
		if (_st.empty())
			return false;
		
		// top과 pop이 분리되어있는 이유
		// 한 번에 데이터를 뽑아오는 경우
		// 뽑아오는 순간 Exception이 발생할 수 있다.
		// (ex. 메모리 부족 등)
		// 즉 자료 구조의 붕괴를 막기 위해서 분리를 한 것
		// 사실 게임의 경우 이런 붕괴를 견디는 것보다
		// 차라리 프로그램이 종료되게 하는 편이 더 나은 편
		value = std::move(_st.top());
		_st.pop();

		return true;
	}

	void WaitPop(T& value)
	{
		// unique_lock : cv를 위해서는 내부적으로 락을 풀어아되기 때문에
		unique_lock<mutex> lock(_mutex);
		_cv.wait(lock, [this] { return _st.empty() == false; });

		value = std::move(_st.top());
		_st.pop();
	}

private : 
	mutex _mutex;
	stack<T> _st;
	condition_variable _cv;
};

먼저 풀 코드를 보면서 Stack 구조부터 확인해보자.

 

사실 제일 문제는 empty이다. 지금 당장 관측하는 시점에서는 데이터가 있었다 치더라도, 연산을 하려는 사이에 다른 스레드가 개입해서 유일한 데이터를 빼가면, 결국 에러가 발생할 것이고 결국 의미가 없어지는 것이기 때문이다. 즉 일반적인 STL 라이브러리와는 여기서부터 개념이 달라지는 것이다.

사실 여기서 top()과 pop() 메서드가 분리되어있는 이유는 만약 한 번에 데이터를 뽑아오는 경우 뽑아오는 순간 Exception이 발생할 수 있다. (ex. 메모리 부족 등) 즉 자료 구조의 붕괴를 막기 위해서 분리를 한 것이라 볼 수 있는 것이다. 사실 이런 경우 게임의 경우 이런 붕괴를 견디는 것보다 차라리 프로그램이 종료되게 하는 편이 더 나은 편이다.

 

하지만 저런 경우 계속 empty한 경우 계속 기다려야 하는 코드가 되버린다. 즉 무한 루프를 유발할 수 있는 것이다. 데이터가 차기 전까지는 다른 일을 할 수 있게 만들어주는 편이 더 나아보인다. 이런 경우를 위해 condition_variable을 이용해 WaitPop을 만들어주었다.

 

이를 기반으로 Queue를 만들면 다음과 같다.

template <typename T>
class ConcurrentQueue
{
public:
	ConcurrentQueue() {}
	~ConcurrentQueue() {}
	ConcurrentQueue(const ConcurrentQueue& rhs) = delete;
	ConcurrentQueue& operator=(const ConcurrentQueue& rhs) = delete;

	void Push(T value)
	{
		lock_guard<mutex> lock(_mutex);
		_qu.push(value);
		_cv.notify_one();
	}

	bool TryPop(T& value)
	{
		lock_guard<mutex> lock(_mutex);
		if (_qu.empty())
			return false;

		value = std::move(_qu.front());
		_qu.pop();

		return true;
	}

	void WaitPop(T& value)
	{
		// unique_lock : cv를 위해서는 내부적으로 락을 풀어아되기 때문에
		unique_lock<mutex> lock(_mutex);
		_cv.wait(lock, [this] { return _qu.empty() == false; });

		value = std::move(_qu.front());
		_qu.pop();
	}

private:
	mutex _mutex;
	queue<T> _qu;
	condition_variable _cv;
};