Lock-based Concurrent Data Structure

Cuncurrent Counters

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;
}

Traditional 방식

락을 사용하지 않는 카운터: 레이스 컨디션에서 부정확함

 

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

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

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);
}

락을 사용한 경우 -> 정확하긴 한데 성능에 문제가 있음

 

=> Approximate Counter (= Sloppy Counter)

예시) 코어 개수가 4개 -> 로컬 카운터 4개와 글로벌 카운터 1개

 

typedef struct __counter_t {
	int gloval;						// 글로벌 변수 값
    pthread_mutex_t glock;			// 글로벌 락
    int local[NUMCPUS];				// 로컬 변수 값
    pthread_mutex_t llock[NUMCPUS];	// 로컬 락
    int threshold;
} counter_t;

void init(counter_t *c, int threshold) {
	c->threshold = threshold;
    c->global = 0;
    int i;
    
    int i;
    for( i = 0; i < NUMCPUS; i++ ) {
    	c->local[i] = 0;
        pthread_mutex_init(&c->llock[i], NULL);
    }
}

void update(counter_t *c, int threadID, int amt) {
	int cpu = threadID % NUMCPUS;
    // 로컬 락
    pthread_mutex_lock(&c->llock[cpu]);
    
    c->local[cpu] += amt;
    
    if( c->local[cpu] >= threshold ) {
    	pthread_mutex_lock(&c-<glock);		// 글록벌 락
        c->global += c->local[cpu];
        pthread_mutex_unlock(&c->glock);	// 글로벌 언락
        c->local[cpu] = 0;
    }
    // 로컬 언락
    phtread_mutex_unlock(&c->glock);
}

// 리턴은 글로벌 값으로
int get(counter_t *c) {
	pthread_mutex_lock(&c->glock);
    int val = c->global;
    pthread_mutex_unlock(&c->glock);
    
    return val;
}

코어 개수만큼 로컬 락과 로컬 카운터를 두고, 로컬 카운터가 임계치(threshold)를 넘으면 글로벌 변수에 반영.

100% 정확하진 않지만, 충돌이 적게 발생하여 훨씬 빠름.

 

* 로컬 변수에 락을 두는 이유

코어 수보다 스레드 수가 많은 경우, 로컬 변수에 대한 상호 배제가 이루어지지 않음

---

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_Iint(list_t *L) {
	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;
}

이 경우, 임계 영역을 너무 넓게 잡아서 비효율적임 (그만큼 다른 스레드에서 접근 못하는 기간이 길어짐)

 

 

* 효율 개선

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

임계영역은 리스트에 접근하는 경우만 해줘야 함

 

 

* 디버깅이 힘든 Lookup

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;
        }
    }
    pthread_mutex_unlock(&L->lock);
    
    return -1;
}

리턴 문은 가능하다면 하나만 두는 것이 디버깅할 때 편함

 

* 개선형 Lookup

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;
        }
    }
    pthread_mutex_unlock(&L->lock);
    
    return rv;
}

---

Cuncurrent 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;
}

void Queue_Init(queue_t *q) {
	node_t *tmp = malloc(sizeof(node_t));
    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 = temp;
    pthread_mutex_unlock(&q->lock);
}

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;
    }
    
    *value = newHead->value;
    q->head = newHead;
    pthread_mutex_unlock(&q->headLock);
    free(tmp);
    return 0;
}

enqueue는 테일에서, dequeue는 헤드에서 하니 같은 락으로 하기 보다는 분리해서 효율성을 높일 수 있음

---

Cuncurrent Hash Table

#define BUCKETS (101)

typedef struct __hast_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]);
}

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

해시는 사실상 멀티플 리스트, Cuncurrent List로 구현 가능

한개의 락으로 관리하는 것보다 성능이 더 좋음 (fine-grained lock)