Multi Thread 공부 5일차 (List - 낙천적인 동기화)

12 Jun 2021

Reading time ~4 minutes

List

저번 세밀한 동기화는 lock을 너무 많이 호출했었기 때문에 원하는 성능보다 더욱 안좋은 결과가 나왔다.

이번엔 낙천적인 동기화를 알아보자.

낙천적인 동기화

세밀한 동기화의 경우 잠금의 획득과 해제가 너무 빈번하다.
- 리스트가 길어지는 경우 성능이 매우 떨어진다.
해결
- 이동 시 잠금을 하지 않난다.
- 잠금을 획득하지 않고 검색한 후, pred와 curr를 잠그고, 잠긴 노드가 정확한지 확인한다.
  - 드문 경우이지만, 만약 엉뚱한 노드가 잠길 경우 잠금을 해제하고 다시 시작한다.
  - 2개의 노드를 잠금으로 dead lock을 주의하자.

그러니 성긴 동기화는 Lock을 리스트에 사용했지만, 이번에는 Node에 Lock을 거는 방법을 사용한다.

따라서 노드를 새로 넣을때도, 읽을때도 Lock을 해야 한다.

주의점
- add와 remove 시점의 pred, curr가 가리키는 노드는 locking이 되어 있어야 한다.
- head부터 node 이동을 할 때 lock을 잠그면서 이동해야 한다.
  - 예를 들어 a의 잠금을 푸고 나서 b (a->next 였던)의 잠금을 한다면 그 사이에 다른 스레드에 의해 b가 제거될 수 있기 때문이다.
  - 즉, 이동 시 pred이 잠금상태 일 때 동안 curr의 잠금을 획득한다.

이 전에는 중간에 삭제를 하거나 추가를 하게 되면 지금까지 읽던 데이터랑 맞지 않았기 때문에 lock을 자주하거나 list를 lock 하는 방법을 사용했다.

그런데 만약 읽던 도중 삭제를 하게 된다면 큰 일이 난다.

해결책
- 제거된 노드를 통해서 이동하는 것을 허용하라.
  - 제거 : remove 삭제 : delete
- remove시 Node를 delete하지 않는다.
  - delete하면 next가 어떤 값으로 바뀔지 알 수 없다.
  - 정확성은 보장할 수 있지만 안정성은 보장할 수 있다.
- 이동 종료 후, pred와 next가 제대로 자리를 잡았는지 검사해야 한다.
  - remove 된 Node를 거친 이동은 잘못된 검색 결과를 야기할 수 있다.

해결책을 보면 삭제하지 않고 제거를 하게 된다면, 탐색할 때 아무런 문제 없이 탐색이 가능하다.

그리고 탐색을 완료 하면, 유효성 검사를 한다.

Validate() : 유효성 검사
- 다시 처음부터 이동해서 원래 pred, curr로 다시 올 수 있는지 확인한다.
- pred->next == cur 인것을 확인한다.
위의 검사로 충분한가?
- pred와 curr가 lock되어있으므로 충분하다.

검사를 하는 도중에는 그 값이 lock이 되어있으니 안전하게 검사를 할 수 있다.

문제정
- 낙천적인 동기화 알고리즘은 기아를 겪을 수 있다.
  - 스레드는 새로운 노드가 반복해서 추가되거나 제거되면 영원히 지연될 수 있다.
- 기아상태를 겪는 경우는 흔치 않은 경우익기 대문에 실제로는 잘 동작할 가능성이 크다.

문제점 마지막에 진짜 낙천적이다. 닉값한다.

왜냐하면 실패할 경우 다시 처음부터 확인한다.

여러번 실패할 수 있지만 영원히 실패하는 확률은 매우 적다.

#include <iostream>
#include <thread>
#include <mutex>
#include <vector>

#define MAX 10000

using namespace std;
using namespace std::chrono;

class Node
{
public:
	int key;
	Node* next;
	mutex m;

	Node() { key = 0; next = nullptr; }

	Node(int key)
	{
		this->key = key;
		next = nullptr;
	}
	~Node() = default;

	void lock()
	{
		m.lock();
	}

	void unlock()
	{
		m.unlock();
	}
};

class List
{
public:
	Node head, tail;
	mutex lock;

	List()
	{
		head.key = 0x80000000;
		tail.key = 0x7FFFFFFF;
		head.next = &tail;
	}
	~List() {}

	void Clear()
	{
		Node* ptr;
		while (head.next != &tail)
		{
			ptr = head.next;
			head.next = head.next->next;
			delete ptr;
		}
	}

	bool Add(int key)
	{
		Node* pred;
		Node* curr;

		pred = &head;
		curr = pred->next;
		while (curr != &tail)
		{
			pred = curr;
			curr = curr->next;
		}
		curr->lock();
		pred->lock();
		if (Validate(pred, curr))
		{
			if (key == curr->key)
			{
				curr->unlock();
				pred->unlock();
				return false;
			}
			else
			{
				Node* node = new Node(key);
				node->next = curr;
				pred->next = node;
				curr->unlock();
				pred->unlock();
				return true;
			}
		}
		curr->unlock();
		pred->unlock();
	}

	bool Validate(Node* pred, Node* curr)
	{
		Node* search_node = &head;
		while (search_node != &tail)
		{
			if (search_node == pred)
			{
				return pred->next == curr;
			}
			search_node = search_node->next;
		}
		return false;
	}

	void Print()
	{
		Node* curr = head.next;

		while (curr != &tail)
		{
			cout << curr->key << " ";
			curr = curr->next;
		}
		cout << endl;
	}

	void Print(int count)
	{
		Node* curr = head.next;

		for (int i = 0; i < count; i++)
		{
			cout << curr->key << " ";
			curr = curr->next;
		}
		cout << endl;
	}
};

List temp_list;


void Thread(int thread_count)
{
	for (int i = 0; i < MAX / thread_count; i++)
	{
		temp_list.Add(i);
	}
}

int main()
{
	for (int i = 1; i <= 8; i *= 2)
	{
		temp_list.Clear();
		steady_clock::time_point start_time = high_resolution_clock::now();
		vector<thread> v;
		for (int j = 0; j < i; j++)
		{
			v.emplace_back(thread(Thread, i));
		}

		for (int j = 0; j < v.size(); j++)
		{
			v[j].join();
		}
		steady_clock::time_point end_time = high_resolution_clock::now();
		auto time = end_time - start_time;

		int result_time = duration_cast<milliseconds>(time).count();

		temp_list.Print(20);
		cout << "스레드 갯수 : " << i << endl;

		cout << "걸린 시간 : " << result_time << endl;
	}
}

이 전까지 했던건과 다른 함수가 있다

bool Validate(Node* pred, Node* curr)
{
	Node* search_node = &head;
	while (search_node != &tail)
	{
		if (search_node == pred)
		{
			return pred->next == curr;
		}
		search_node = search_node->next;
	}
	return false;
}

지금 넣는 key가 맞는지 아닌지 확인하는 함수이다.

쭉쭉가면서 맞는지 검사한다.

이렇게 하면,

0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19
스레드 갯수 : 1
걸린 시간 : 370
0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19
스레드 갯수 : 2
걸린 시간 : 90
0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19
스레드 갯수 : 4
걸린 시간 : 72
0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19
스레드 갯수 : 8
걸린 시간 : 58

와! 진짜 엄청나게 많이 빨라졌다.

이번에는 멀티 스레드를 쓰는 이유가 있을 정도로 빨라졌다.

하지만!!, 한번 더 리스트를 순회한다는 문제가 있다.

출처

Multi Thread