14. Lock-Free Stack #2

bool TryPop(T& value)
{
	Node* oldHead = _head;

	while (oldHead && (_head.compare_exchange_strong(oldHead, oldHead->next) == false))
	{
	}

	if (oldHead == nullptr)
		return false;

	value = oldHead->data;

	//문제 발생
	delete oldHead;
	return true;
}

결국은 oldHead가 가르키고 있는 노드를 아무 스레드도 참조하고 있지 않을 때 비로소 메모리에서 해제를 시킬 수 있다. 

 

bool TryPop(T& value)
{
	Node* oldHead = _head;

	while (oldHead && (_head.compare_exchange_strong(oldHead, oldHead->next) == false))
	{
	}

	if (oldHead == nullptr)
		return false;

	value = oldHead->data;
	
    //노드 삭제 시도
	TryDelete(oldHead);
	return true;
}

따라서 위와 같이 노드를 삭제하는 함수를 새롭게 구현하여 만들어야한다.

 

template<typename T>
class LockFreeStack
{
	//생략
public:
	bool TryPop(T& value)
	{
		++_popCnt;

		Node* oldHead = _head;

		while (oldHead && (_head.compare_exchange_strong(oldHead, oldHead->next) == false))
		{
		}

		if (oldHead == nullptr)
		{
			--_popCnt;
			return false;
		}

		value = oldHead->data;

		TryDelete(oldHead);

		return true;
	}

	void TryDelete(Note* oldHead)
	{
		if (_popCnt == 1)
		{
			// 자신의 데이터 삭제
			delete oldHead;
		}
	}

private:
	atomic<Node*> _head;

	atomic<int> _popCnt = 0; // pop을 실행중인 스레드 개수
	atomic<Node*> _pendingList; // 삭제되어야할 노드
};

 

변수 하나를 만들어 pop을 하는 스레드의 개수를 체크하고 그 노드를 리스트에 넣은 후(예약) pop을 하는 스레드가 1개일 때 delete를 시킨다.

 

함수의 호출 순서를 보면 노트를 먼저 분리 시킨 후 popCnt를 체크한다. 따라서 delete를 시도하는 시점에는 이미 노드가 분리된 상태이다. 그렇기 때문에 popCnt가 1이라면 이후 다른 스레드가 접근을 시도해도 이미 head는 다음 노드를 가르키고 있기 때문에 삭제하고자하는 노드(oldHead)에는 접근을 할 수 없다.

 

이제 모든 상황에서 작동할 수 있도록 TryDelete함수를 완성해야한다.

template<typename T>
class LockFreeStack
{
// 생략
public:
	bool TryPop(T& value)
	{
		++_popCnt;

		Node* oldHead = _head;

		// 노드 분리
		while (oldHead && (_head.compare_exchange_strong(oldHead, oldHead->next) == false))
		{
		}

		if (oldHead == nullptr)
		{
			--_popCnt;
			return false;
		}

		value = oldHead->data;

		TryDelete(oldHead);

		return true;
	}

	void TryDelete(Note* oldHead)
	{
		if (_popCnt == 1)
		{
			// 어차피 혼자이니 삭제 예약된 리스트 삭제
			Node* node = _pendingList.exchange(nullptr);

			if (--_popCnt == 0)
			{
				// 중간에 다른 스레드가 pop을 시도하지 않았을 때(노드가 참조되지 않음)
				// 이미 분리가 완료되었기 때문에 이 때 접근해도 문제 X
				DeleteNode(node);
			}
			else if(node)
			{
				// 다른 스레드가 접근 함(삭제 불가)
				ChainPendingNodeList(node);
			}


			// 자신의 데이터 삭제
			delete oldHead;
		}
		else
		{
			// 다른 스레드가 존재함(pop을 시도)
			// 삭제 예약
			ChainPendingNode(oldHead);
			--_popCnt;
		}
	}

	// 리스트 이어붙이는 함수
	void ChainPendingNodeList(Node* first, Node* last)
	{
		last->next = _pendingList;

		while(_pendingList.compare_exchange_strong(last->next, first) == false))
		{
		}
	}
	
	void ChainPendingNodeList(Node* node)
	{
		Node* last = node;

		while (last->next)
		{
			last = last->next;
		}

		ChainPendingNodeList(node, last);
	}

	// 노드 1개만 추가
	void ChainPendingNode(Node* node)
	{
		ChainPendingNodeList(node, node);
	}

	// 노드 삭제
	static void DeleteNode(Node* node)
	{
		while (node)
		{
			Node* next = node->next;
			delete node;
			node = next;
		}
	}

private:
	atomic<Node*> _head;

	atomic<int> _popCnt = 0; // pop을 실행중인 스레드 개수
	atomic<Node*> _pendingList; // 삭제되어야할 노드
};

Chain___함수는 _pendingList에 이어붙이는 함수이다.

 

TryPop에서 _popCnt가 1일 때 자신만 pop을 시도하고 있다는 뜻이므로 삭제 예약된 모든 노드의 삭제를 시도한다. 이때 _popCnt를 1 줄이고 정상적으로 0이 되면 모든 노드를 삭제한다. 하지만 _popCnt가 0이 아니라는 뜻은 다른 스레드에서 pop을 시도하면서 노드를 참조하고 있다는 뜻이기 때문에 다시 원래대로 돌려놓는다. 이후 다른 스레드에서 삭제해 줄 것이기 때문이다.

 

동시에 여러 스레드가 pop을 시도하여 _popCnt가 1을 넘었을 때에는 바로 삭제하지 않고 _pendingList(삭제 예약)에 추가한 후 _popCnt를 줄인다.

 

코드의 흐름은 이렇다.

Lock-Free Pop

 

완성된 코드이다.

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

		T data;
		Node* next;
	};
public:
	void Push(const T& value)
	{
		Node* node = new Node(value);
		node->next = _head;

		while (_head.compare_exchange_strong(node->next, node) == false)
		{
		}
	}

	bool TryPop(T& value)
	{
		++_popCnt;

		Node* oldHead = _head;

		// 노드 분리
		while (oldHead && (_head.compare_exchange_strong(oldHead, oldHead->next) == false))
		{
		}

		if (oldHead == nullptr)
		{
			--_popCnt;
			return false;
		}

		value = oldHead->data;

		TryDelete(oldHead);

		return true;
	}

	void TryDelete(Note* oldHead)
	{
		if (_popCnt == 1)
		{
			// 어차피 혼자이니 삭제 예약된 리스트 삭제
			Node* node = _pendingList.exchange(nullptr);

			if (--_popCnt == 0)
			{
				// 중간에 다른 스레드가 pop을 시도하지 않았을 때(노드가 참조되지 않음)
				// 이미 분리가 완료되었기 때문에 이 때 접근해도 문제 X
				DeleteNode(node);
			}
			else if(node)
			{
				// 다른 스레드가 접근 함(삭제 불가)
				ChainPendingNodeList(node);
			}


			// 자신의 데이터 삭제
			delete oldHead;
		}
		else
		{
			// 다른 스레드가 존재함(pop을 시도)
			// 삭제 예약
			ChainPendingNode(oldHead);
			--_popCnt;
		}
	}

	void ChainPendingNodeList(Node* first, Node* last)
	{
		last->next = _pendingList;

		while(_pendingList.compare_exchange_strong(last->next, first) == false))
		{
		}
	}
	
	void ChainPendingNodeList(Node* node)
	{
		Node* last = node;

		while (last->next)
		{
			last = last->next;
		}

		ChainPendingNodeList(node, last);
	}

	void ChainPendingNode(Node* node)
	{
		ChainPendingNodeList(node, node);
	}

	static void DeleteNode(Node* node)
	{
		while (node)
		{
			Node* next = node->next;
			delete node;
			node = next;
		}
	}

private:
	atomic<Node*> _head;

	atomic<int> _popCnt = 0; // pop을 실행중인 스레드 개수
	atomic<Node*> _pendingList; // 삭제되어야할 노드
};

 

꽤나 어렵기 때문에 많이 들여다보도록 하자