12. Lock-Based Queue/Stack

지금까지 배운 내용들로 간단하게 멀티스레드 환경에서 사용할 수 있는 큐와 스택을 알아보자.

 

#include <iostream>
#include <thread>

queue<int> q;
stack<int> s;

void Push()
{
	while (true)
	{
		int value = rand() % 100;

		q.push(value);

		this_thread::sleep_for(10ms);
	}
}

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

		int data = q.front();
		q.pop();

		cout << data << endl;
	}
}

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

	t1.join();
	t2.join();
	t3.join();
}

데이터를 push하는 스레드와 pop하는 스레드를 만들었다. 당연하게도 위의 코드는 크래시가 나버린다. 같은 공용 데이터에 동시에 접근하기 때문에 생기는 문제이다. 그렇다면 Push와 Pop을 할 때 락을 걸어주면 되지 않을까? 물론 그렇게 할 수도 있겠지만 매번 락을 걸어주어야 한다면 코드를 작성할 때 실수를 할 수도 있으니 이번에는 클래스를 만들어서 활용을 해보자.

 

ConcurrentQueue.h

#pragma once

#include <mutex>

template<typename T>
class LockQueue
{
public:
	LockQueue() {}

	//복사 방지
	LockQueue(const LockQueue&) = delete;
	LockQueue& operator=(const LockQueue&) = delete;

	void Push(T value)
	{
		lock_guard<mutex> lock(_mutex);
		_queue.push(move(value));

		_cv.notify_one();
	}

	//무한루프 돌면서 검사
	bool TryPop(T& value)
	{
		lock_guard<mutex> lock(_mutex);

		if (_queue.empty())
			return false;

		value = move(_queue.front());
		_queue.pop();

		return true;
	}

	//Condotional Variable 활용
	void WaitPop(T& value)
	{
		unique_lock<mutex> lock(_mutex);
		_cv.wait(lock, _queue.empty() == false);

		value = move(_queue.front());
		_queue.pop();
	}

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

내부적으로 락을 잡아 멀티스레드 환경에서 문제없이 작동하는 큐를 구현해 보았다. 멀티스레드 환경에서 복사는 치명적일 수 있으므로 복사생성자와 대입연산자는 사용할 수 없도록 하였다. empty와 pop을 따로 구현하지 않는 이유는 따로 구현하게 되면 어차피 검사를 하고 pop을 하는 도중 락이 풀리게 되므로 다른 스레드가 접근할 수 있기 때문이다. 그래서 정말 큐가 비었는지 확인해야되는 경우가 있으면 구현을 해야겠지만 우리가 하고싶어 하는 일은 그와 관련이 없기 때문에 TryPop이라는 함수에서 큐가 비었을 때 false를 리턴하고 데이터가 큐에 있을 때는 인자로 데이터를 건내준 후 true를 리턴하도록 만들었다.

 

WaitPop은 조건변수를 활용하여 무한정 기다리는 것이 아니라 데이터가 들어왔을 때에만 작동하는 함수로 2개의 pop함수를 필요에 따라 사용하면 된다.

 

ConcurrentStack.h

#pragma once

#include <mutex>

template<typename T>
class LockStack
{
public:
	LockStack() {}

	LockStack(const LockStack&) = delete;
	LockStack& operator=(const LockStack&) = delete;

	void Push(T value)
	{
		lock_guard<mutex> lock(_mutex);
		_stack.push(move(value));

		_cv.notify_one();
	}

	bool TryPop(T& value)
	{
		lock_guard<mutex> lock(_mutex);

		if (_stack.empty())
			return false;

		value = move(_stack.top());
		_stack.pop();

		return true;
	}

	void WaitPop(T& value)
	{
		unique_lock<mutex> lock(_mutex);
		_cv.wait(lock, _stack.empty() == false);

		value = move(_stack.top());
		_stack.pop();
	}

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

스택도 마찬가지로 똑같이 구현할 수 있다.

 

main.cpp

#include <iostream>
#include <thread>
#include "ConcurrentQueue.h"
#include "ConcurrentStack.h"

LockQueue<int> q;
LockStack<int> s;

void Push()
{
	while (true)
	{
		int value = rand() % 100;

		q.Push(value);

		this_thread::sleep_for(10ms);
	}
}

void Pop()
{
	while (true)
	{
		int data = 0;

		if (q.TryPop(data))
			cout << data << endl;
	}
}

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

	t1.join();
	t2.join();
	t3.join();
}

새롭게 만든 클래스를 이용하여 메인함수를 수정한 모습이다. 상황에 맞춰 TryPop과 WaitPop을 사용하면 된다.

출력 화면

정상적으로 잘 작동하는 것을 볼 수 있다.