Lock free stack #3

쓰지 않는 포인터를 모아 놨다가 한 번에 소멸시키는 로직을 shared_ptr ref count를 이용해 더 이상 참조하는 곳이 없으면 메모리 free하는 로직으로 바꿔보자.

 

template<typename T>
class LockFreeStack
{
	struct Node
	{
		Node(const T& value) : data(make_shared<T>(value)), next(nullptr)
		{

		}

		shared_ptr<T> data;
		shared_ptr<Node> next;
	};

public:
	void Push(const T& value)
	{
		shared_ptr<Node> node = make_shared<Node>(value);
		node->next = std::atomic_load(&_head);
		while (std::atomic_compare_exchange_weak(&_head, &node->next, node) == false)
		{

		}

	}

	shared_ptr<T> TryPop()
	{
		shared_ptr<Node> oldHead = std::atomic_load(&_head); // 원자적으로 ref count 증가

		while (oldHead && std::atomic_compare_exchange_weak(&_head, &oldHead, oldHead->next) == false)
		{

		}

		if (oldHead == nullptr)
			return shared_ptr<T>();

		return oldHead->data;
	}

private:
	shared_ptr<Node> _head;
	
};

 

 

하지만 shared_ptr을 원자적으로 바꾸는 동작이 lock없는 방식인지는 확인해보아야 한다.

 

shared_ptr<int32> ptr;
bool value = std::atomic_is_lock_free(&ptr);

 

cpu마다 다르지만 내 컴퓨터에서는 false가 나온다. lock을 사용하여 원자적동작을 구현했다는 의미다.

때문에 직접 ref count를 구현하여 사용하여야 한다.

 

아래 코드는 CountedNodePtr이 원자적으로 동작한다는 전제가 있다.

template<typename T>
class LockFreeStack
{
	struct Node;

	struct CountedNodePtr
	{
		int32 externalCount = 0;
		Node* ptr = nullptr;
	};

	struct Node
	{
		Node(const T& value) : data(make_shared<T>(value))
		{

		}

		shared_ptr<T> data;
		atomic<int32> internalCount = 0;
		CountedNodePtr next;
	};

public:
	void Push(const T& value)
	{
		CountedNodePtr node;
		node.ptr = new Node(value);
		node.externalCount = 1;
		// [!]
		node.ptr->next = _head;
		while (_head.compare_exchange_weak(node.ptr->next, node) == false)
		{

		}


	}

	shared_ptr<T> TryPop()
	{
		CountedNodePtr oldHead = _head;
		while (true)
		{
			// 참조권 획득 (externalCount를 현 시점 기준 +1 한 애가 이김)
			IncreaseHeadCount(oldHead);
			// 최소한 externalCount >= 2 일테니 삭제 x (안전하게 접근할 수 있음)
			Node* ptr = oldHead.ptr;
			
			// 데이터 없음
			if (ptr == nullptr)
				return shared_ptr<T>();

			// 소유권 획득 (ptr->next로 head를 바꿔치기 한 애가 이김)
			if (_head.compare_exchange_strong(oldHead, ptr->next))
			{
				shared_ptr<T> res;
				res.swap(ptr->data);

				// external : 1 -> 2(+1) -> 4(나+1, 남+2)
				// internal : 0

				// 나 말고 또 누가 있는가?
				const int32 countIncrease = oldHead.externalCount - 2;

				// fetch_add는 실행 후, 이전 값 리턴함
				if (ptr->internalCount.fetch_add(countIncrease) == -countIncrease)
					delete ptr;

				return res;
			}
			else if (ptr->internalCount.fetch_sub(1) == 1)
			{
				// 참조권은 얻었으나, 소유권은 실패 -> 뒷수습은 내가 한다.
				delete ptr;
			}
		}
	}

private:
	void IncreaseHeadCount(CountedNodePtr& oldCounter)
	{
		while (true)
		{
			CountedNodePtr newCounter = oldCounter;
			newCounter.externalCount++;
			if (_head.compare_exchange_strong(oldCounter, newCounter))
			{
				oldCounter.externalCount = newCounter.externalCount;
				break;
			}
		}
	}

private:
	atomic<CountedNodePtr> _head;

};

 

핵심은 externalCount로 참조권을, internalCount로 소유권을 파악하는 것이고, 경합 중인 쓰레드 개수를 파악하는 부분이다.

 

lock_free programming은 어렵다. 검증이 안된 코드를 사용하는 것은 프로그래밍 디버깅을 매우 어렵게 만드는 원인이 된다. 믿을 수 있는 코드만 사용하자.

 

참조 : [C++과 언리얼로 만드는 MMORPG 게임 개발 시리즈] Part4: 게임 서버 - 인프런 | 강의 (inflearn.com)