[OSTEP 운영체제 정리] - Lock based Concurrent Data Structures

 

진짜 어렵다... 밑에 있는 책을 꼭 읽어보길 추천...책이 정말 잘 설명해준다.. 근데 난 이해 못하겠다..ㅎㅎ

자료구조에 락을 추가하는 방법?

자료구조에 락을 추가하여 thread safe하게 만들 수 있따.

  Correctness : 어떻게 락을 추가하고 올바르게 추가하려면 어떻게 해야할까?

  Concurrency : 데이터 구조가 높은 성능을 제공하고, 많은 스레드가 동시에 구조에 접근할 수 있도록 잠금을 추가하는 방법은 무엇인가?

Simple Counter (Without Locks)

typedef struct __counter_t {
int value;
} counter_t;
void init(counter_t *c) {
c->value = 0;
}
void increment(counter_t *c) {
c->value++;
}
void decrement(counter_t *c) {
c->value--;
}
int get(counter_t *c) {
return c->value;
}

-> 락이 없는 카운터

typedef struct __counter_t {
	int value;
	pthread_mutex_t lock;
} counter_t;

void init(counter_t *c) {
	c->value = 0;
	pthread_mutex_init(&c->lock, NULL);
}

void increment(counter_t *c) {
	pthread_mutex_lock(&c->lock);
	c->value++;
	pthread_mutex_unlock(&c->lock);
}

void decrement(counter_t *c) {
	pthread_mutex_lock(&c->lock);
	c->value--;
	pthread_mutex_unlock(&c->lock);
}

int get(counter_t *c) {
	pthread_mutex_lock(&c->lock);
	int rc = c->value;
	pthread_mutex_unlock(&c->lock);
	return rc;
}

-> 락이 있는 카운터

Perfect Scaling

• 더 많은 작업이 수행되더라도, 병렬로 수행된다
• 작업을 완료하는 데 걸리는 시간이 증가하지 않는다.

Scalable Counting, 확장성 있는 카운팅

[ Sloppy counter ]

엉성한 카운터는 하나의 논리적 카운터로 표현된다.

CPU코어마다 존재하는 하나의 물리적인 지역카운터와 하나의 전역카운터로 구성 그외에 지역카운터를 위한 락, 전역 카운터의 락이 존재

기본개념
쓰레드는 지역 카운터를 증가시킨다. 이 지역 카운터는 지역 락에 의해 보호된다. 각 CPU는 각자 카운터를 갖기 때문에 CPU들에 분산되어 있는 쓰레드들은 지역 카운터를 경쟁 없이 갱신할 수 있다. ➡️ 카운터의 갱신은 확장성 있다.
최신 값으로 갱신하기 위해서 지역 카운터의 값은 주기적으로 전역 카운터로 전달된다, 이때 전역 락을 사용하여 지역 카운터의 값을 전역 카운터의 값에 더하고 그 지역 카운터의 값은 0으로 초기화
따라서 얼마나 local-to-global transfer가 발생하느냐에 따라서 counter의 상한선인 S(sloppiness)가 결정된다.

 

 

• 임계값 S는 5로 설정됩니다.
• 네 개의 CPU에 스레드가 있습니다:
  • 각 스레드는 자신의 로컬 카운터(L1, L2, L3, L4)를 업데이트합니다.

 

Importance of the threshold value S

네 개의 스레드가 네 개의 CPU에서 카운터를 100만 번 증가하는 경우에 한계치 설정 (S)의 중요성을 보여준다. S가 너무 낮아

⇒ 성능이 안좋지만, 전역 카운터값 정확해진다.

S가 너무 크다

⇒ 성능이 좋다, 전역 카운터 값 안좋아짐..

typedef struct __counter_t {
	int global; // 전역 카운터
	pthread_mutex_t glock; // 전역 카운터
	int local[NUMCPUS]; // 지역 카운터
	pthread_mutex_t llock[NUMCPUS]; //그리고 락들
	int threshold; // 갱신 빈도
} counter_t;

//init: 한계치를 기록하고 락과 지역카운터 그리고 전역 카운터의 값들을 초기화함
void init(counter_t *c, int threshold) {
	c->threshold = threshold;
	c->global = 0;
	pthread_mutex_init(&c->glock, NULL);
	int i;
	for (i = 0; i < NUMCPUS; i++) {
		c->local[i] = 0;
		pthread_mutex_init(&c->llock[i], NULL);
	}
}

//update : 보통은 지역 락을 획득한 후 지역 값을 갱신함
// 한계치까지 지역 카운터의 값이 커지면, 전역 락을 획득하고 지역 값을 전역 카운터에 전달함
void update(counter_t *c, int threadID, int amt) {
	int cpu = threadID % NUMCPUS;
	pthread_mutex_lock(&c->llock[cpu]);
	c->local[cpu] += amt; // amt > 0 가정
	if (c->local[cpu] >= c->threshold) { // 전역으로 전달
		pthread_mutex_lock(&c->glock);
		c->global += c->local[cpu];
		pthread_mutex_unlock(&c->glock);
		c->local[cpu] = 0;
	}
	pthread_mutex_unlock(&c->llock[cpu]);
}

//get: 전역 카운터의 값을 리턴(정확하기 지 않을 수 있음)
int get(counter_t *c) {
	pthread_mutex_lock(&c->glock);
	int val = c->global;
	pthread_mutex_unlock(&c->glock);
	return val; // 대략의 값임
}

Performance of the concurrent counter

Concurrent Linked Lists

typedef struct __node_t {
	int key;
	struct __node_t *next;
} node_t;

typedef struct __list_t {
	node_t *head;
	pthread_mutex_t lock;
} list_t;

void List_Init(list_t *L) {
	L->head = NULL;
	pthread_mutex_init(&L->lock, NULL);
}

int List_Insert(list_t *L, int key) {
	pthread_mutex_lock(&L->lock);
	node_t *new = malloc(sizeof(node_t));
	if (new == NULL) {
		perror("malloc");
		pthread_mutex_unlock(&L->lock);
		return -1; // fail
	}
	new->key = key;
	new->next = L->head;
	L->head = new;
	pthread_mutex_unlock(&L->lock);
	return 0; // success
}

int List_Lookup(list_t *L, int key) {
	pthread_mutex_lock(&L->lock);
	node_t *curr = L->head;
	while (curr) {
		if (curr->key == key) {
			pthread_mutex_unlock(&L->lock);
			return 0; // success
		}
		curr = curr->next;
	}
	pthread_mutex_unlock(&L->lock);
	return -1; // failure
}

Scaling Linked Lists

void List_Init(list_t *L) {
	L->head = NULL;
	pthread_mutex_init(&L->lock, NULL);
}

void List_Insert(list_t *L, int key) {
	// synchronization not needed
	node_t *new = malloc(sizeof(node_t));
	if (new == NULL) {
		perror("malloc");
		return;
	}
	new->key = key;
	// just lock critical section
	pthread_mutex_lock(&L->lock);
	new->next = L->head;
	L->head = new;
	pthread_mutex_unlock(&L->lock);
}

int List_Lookup(list_t *L, int key) {
	int rv = -1;
	pthread_mutex_lock(&L->lock);
	node_t *curr = L->head;
	while (curr) {
		if (curr->key == key) {
			rv = 0;
			break;
		}
		curr = curr->next;
	}
	pthread_mutex_unlock(&L->lock);
	return rv; // bug pruning
}

More Scalable Linked List?

병행성을 개선하는 방법 : Hand-over-hand locking (lock coupling)

전체 리스트에 하나의 락이 있는 것이 아니라 개별 노드마다 락을 추가하는 것이다. 리스트를 순회할 때 다음 노드의 락을 먼저 획득하고 지금 노드의 락을 해제하도록 한다.

하지만 실제로 간단한 락 방법에 비해 속도를 기대하기 쉽지 않다.

리스트를 순회할 때 각 노드에 락을 획득하고 해제하는 오버해드가 너무 크다

Concurrent Queues

typedef struct __node_t {
	int value;
	struct __node_t *next;
} node_t;

typedef struct __queue_t {
	node_t *head;
	node_t *tail;
	pthread_mutex_t headLock;
	pthread_mutex_t tailLock;
} queue_t;

void Queue_Init(queue_t *q) {
	node_t *tmp = malloc(sizeof(node_t)); // dummy node
	tmp->next = NULL;
	q->head = q->tail = tmp;
	pthread_mutex_init(&q->headLock, NULL);
	pthread_mutex_init(&q->tailLock, NULL);
}

void Queue_Enqueue(queue_t *q, int value) {
	node_t *tmp = malloc(sizeof(node_t));
	assert(tmp != NULL);
	tmp->value = value;
	tmp->next = NULL;
	pthread_mutex_lock(&q->tailLock);
	q->tail->next = tmp;
	q->tail = tmp;
	pthread_mutex_unlock(&q->tailLock);
}

int Queue_Dequeue(queue_t *q, int *value) {
	pthread_mutex_lock(&q->headLock);
	node_t *tmp = q->head;
	node_t *newHead = tmp->next;
	if (newHead == NULL) {
		pthread_mutex_unlock(&q->headLock);
		return -1; // queue was empty
	}
	*value = newHead->value;
	q->head = newHead;
	pthread_mutex_unlock(&q->headLock);
	free(tmp);
	return 0;
}

Concurrent Hash Table

#define BUCKETS (101)
typedef struct __hash_t {
	list_t lists[BUCKETS];
} hash_t;

void Hash_Init(hash_t *H) {
	int i;
	for (i = 0; i < BUCKETS; i++)
		List_Init(&H->lists[i]);
}

int Hash_Insert(hash_t *H, int key) {
	int bucket = key % BUCKETS;
	return List_Insert(&H->lists[bucket], key);
}

int Hash_Lookup(hash_t *H, int key) {
	int bucket = key % BUCKETS;
	return List_Lookup(&H->lists[bucket], key);
}

Performance of Concurrent Hash Table자료구조에 락을 추가하는 방법?

자료구조에 락을 추가하여 thread safe하게 만들 수 있따.

Correctness : 어떻게 락을 추가하고 올바르게 추가하려면 어떻게 해야할까?

Concurrency : 데이터 구조가 높은 성능을 제공하고, 많은 스레드가 동시에 구조에 접근할 수 있도록 잠금을 추가하는 방법은 무엇인가?

Simple Counter (Without Locks)

typedef struct __counter_t {
int value;
} counter_t;
void init(counter_t *c) {
c->value = 0;
}
void increment(counter_t *c) {
c->value++;
}
void decrement(counter_t *c) {
c->value--;
}
int get(counter_t *c) {
return c->value;
}

typedef struct __counter_t {
	int value;
	pthread_mutex_t lock;
} counter_t;

void init(counter_t *c) {
	c->value = 0;
	pthread_mutex_init(&c->lock, NULL);
}

void increment(counter_t *c) {
	pthread_mutex_lock(&c->lock);
	c->value++;
	pthread_mutex_unlock(&c->lock);
}

void decrement(counter_t *c) {
	pthread_mutex_lock(&c->lock);
	c->value--;
	pthread_mutex_unlock(&c->lock);
}

int get(counter_t *c) {
	pthread_mutex_lock(&c->lock);
	int rc = c->value;
	pthread_mutex_unlock(&c->lock);
	return rc;
}

Perfect Scaling

• 더 많은 작업이 수행되더라도, 병렬로 수행된다
• 작업을 완료하는 데 걸리는 시간이 증가하지 않는다.

Scalable Counting, 확장성 있는 카운팅

[ Sloppy counter ]

엉성한 카운터는 하나의 논리적 카운터로 표현된다.

CPU코어마다 존재하는 하나의 물리적인 지역카운터와 하나의 전역카운터로 구성 그외에 지역카운터를 위한 락, 전역 카운터의 락이 존재

기본개념

쓰레드는 지역 카운터를 증가시킨다. 이 지역 카운터는 지역 락에 의해 보호된다. 각 CPU는 각자 카운터를 갖기 때문에 CPU들에 분산되어 있는 쓰레드들은 지역 카운터를 경쟁 없이 갱신할 수 있다. ➡️ 카운터의 갱신은 확장성 있다.

최신 값으로 갱신하기 위해서 지역 카운터의 값은 주기적으로 전역 카운터로 전달된다, 이때 전역 락을 사용하여 지역 카운터의 값을 전역 카운터의 값에 더하고 그 지역 카운터의 값은 0으로 초기화

따라서 얼마나 local-to-global transfer가 발생하느냐에 따라서 counter의 상한선인 **S(sloppiness)**가 결정된다.

• 임계값 S는 5로 설정됩니다.
• 네 개의 CPU에 스레드가 있습니다:
  • 각 스레드는 자신의 로컬 카운터(L1, L2, L3, L4)를 업데이트합니다.

Importance of the threshold value S

 

네 개의 스레드가 네 개의 CPU에서 카운터를 100만 번 증가하는 경우에 한계치 설정 (S)의 중요성을 보여준다. S가 너무 낮아

⇒ 성능이 안좋지만, 전역 카운터값 정확해진다.

S가 너무 크다

⇒ 성능이 좋다, 전역 카운터 값 안좋아짐..

typedef struct __counter_t {
	int global; // 전역 카운터
	pthread_mutex_t glock; // 전역 카운터
	int local[NUMCPUS]; // 지역 카운터
	pthread_mutex_t llock[NUMCPUS]; //그리고 락들
	int threshold; // 갱신 빈도
} counter_t;

//init: 한계치를 기록하고 락과 지역카운터 그리고 전역 카운터의 값들을 초기화함
void init(counter_t *c, int threshold) {
	c->threshold = threshold;
	c->global = 0;
	pthread_mutex_init(&c->glock, NULL);
	int i;
	for (i = 0; i < NUMCPUS; i++) {
		c->local[i] = 0;
		pthread_mutex_init(&c->llock[i], NULL);
	}
}

//update : 보통은 지역 락을 획득한 후 지역 값을 갱신함
// 한계치까지 지역 카운터의 값이 커지면, 전역 락을 획득하고 지역 값을 전역 카운터에 전달함
void update(counter_t *c, int threadID, int amt) {
	int cpu = threadID % NUMCPUS;
	pthread_mutex_lock(&c->llock[cpu]);
	c->local[cpu] += amt; // amt > 0 가정
	if (c->local[cpu] >= c->threshold) { // 전역으로 전달
		pthread_mutex_lock(&c->glock);
		c->global += c->local[cpu];
		pthread_mutex_unlock(&c->glock);
		c->local[cpu] = 0;
	}
	pthread_mutex_unlock(&c->llock[cpu]);
}

//get: 전역 카운터의 값을 리턴(정확하기 지 않을 수 있음)
int get(counter_t *c) {
	pthread_mutex_lock(&c->glock);
	int val = c->global;
	pthread_mutex_unlock(&c->glock);
	return val; // 대략의 값임
}

Performance of the concurrent counter

Concurrent Linked Lists

typedef struct __node_t {
	int key;
	struct __node_t *next;
} node_t;

typedef struct __list_t {
	node_t *head;
	pthread_mutex_t lock;
} list_t;

void List_Init(list_t *L) {
	L->head = NULL;
	pthread_mutex_init(&L->lock, NULL);
}

int List_Insert(list_t *L, int key) {
	pthread_mutex_lock(&L->lock);
	node_t *new = malloc(sizeof(node_t));
	if (new == NULL) {
		perror("malloc");
		pthread_mutex_unlock(&L->lock);
		return -1; // fail
	}
	new->key = key;
	new->next = L->head;
	L->head = new;
	pthread_mutex_unlock(&L->lock);
	return 0; // success
}

int List_Lookup(list_t *L, int key) {
	pthread_mutex_lock(&L->lock);
	node_t *curr = L->head;
	while (curr) {
		if (curr->key == key) {
			pthread_mutex_unlock(&L->lock);
			return 0; // success
		}
		curr = curr->next;
	}
	pthread_mutex_unlock(&L->lock);
	return -1; // failure
}

Scaling Linked Lists

void List_Init(list_t *L) {
	L->head = NULL;
	pthread_mutex_init(&L->lock, NULL);
}

void List_Insert(list_t *L, int key) {
	// synchronization not needed
	node_t *new = malloc(sizeof(node_t));
	if (new == NULL) {
		perror("malloc");
		return;
	}
	new->key = key;
	// just lock critical section
	pthread_mutex_lock(&L->lock);
	new->next = L->head;
	L->head = new;
	pthread_mutex_unlock(&L->lock);
}

int List_Lookup(list_t *L, int key) {
	int rv = -1;
	pthread_mutex_lock(&L->lock);
	node_t *curr = L->head;
	while (curr) {
		if (curr->key == key) {
			rv = 0;
			break;
		}
		curr = curr->next;
	}
	pthread_mutex_unlock(&L->lock);
	return rv; // bug pruning
}

More Scalable Linked List?

병행성을 개선하는 방법 : Hand-over-hand locking (lock coupling)

전체 리스트에 하나의 락이 있는 것이 아니라 개별 노드마다 락을 추가하는 것이다. 리스트를 순회할 때 다음 노드의 락을 먼저 획득하고 지금 노드의 락을 해제하도록 한다.

하지만 실제로 간단한 락 방법에 비해 속도를 기대하기 쉽지 않다.

리스트를 순회할 때 각 노드에 락을 획득하고 해제하는 오버해드가 너무 크다

Concurrent Queues

typedef struct __node_t {
	int value;
	struct __node_t *next;
} node_t;

typedef struct __queue_t {
	node_t *head;
	node_t *tail;
	pthread_mutex_t headLock;
	pthread_mutex_t tailLock;
} queue_t;

void Queue_Init(queue_t *q) {
	node_t *tmp = malloc(sizeof(node_t)); // dummy node
	tmp->next = NULL;
	q->head = q->tail = tmp;
	pthread_mutex_init(&q->headLock, NULL);
	pthread_mutex_init(&q->tailLock, NULL);
}

void Queue_Enqueue(queue_t *q, int value) {
	node_t *tmp = malloc(sizeof(node_t));
	assert(tmp != NULL);
	tmp->value = value;
	tmp->next = NULL;
	pthread_mutex_lock(&q->tailLock);
	q->tail->next = tmp;
	q->tail = tmp;
	pthread_mutex_unlock(&q->tailLock);
}

int Queue_Dequeue(queue_t *q, int *value) {
	pthread_mutex_lock(&q->headLock);
	node_t *tmp = q->head;
	node_t *newHead = tmp->next;
	if (newHead == NULL) {
		pthread_mutex_unlock(&q->headLock);
		return -1; // queue was empty
	}
	*value = newHead->value;
	q->head = newHead;
	pthread_mutex_unlock(&q->headLock);
	free(tmp);
	return 0;
}

Concurrent Hash Table

#define BUCKETS (101)
typedef struct __hash_t {
	list_t lists[BUCKETS];
} hash_t;

void Hash_Init(hash_t *H) {
	int i;
	for (i = 0; i < BUCKETS; i++)
		List_Init(&H->lists[i]);
}

int Hash_Insert(hash_t *H, int key) {
	int bucket = key % BUCKETS;
	return List_Insert(&H->lists[bucket], key);
}

int Hash_Lookup(hash_t *H, int key) {
	int bucket = key % BUCKETS;
	return List_Lookup(&H->lists[bucket], key);
}

Performance of Concurrent Hash Table

---

[출처]
https://github.com/remzi-arpacidusseau/ostep-translations/blob/master/korean/README.md

ostep-translations/korean/README.md at master · remzi-arpacidusseau/ostep-translations

해당 책을 기반으로 정리하였습니다.