[c++] CAS 구현 및 ABA 문제 해결 :: ABA 해결_2. hazard pointer

 

목차

1. 문제 정의
2. lock free 구현
3. ABA 해결
   1. int형 구현(+ Hazard pointer)
   2. Counter
   3. 그 외의 방법들
4. mutex lock(spin lock)과의 비교

 

Hazard Pointer

https://m.blog.naver.com/PostView.nhn?blogId=jjoommnn&logNo=130127286459&proxyReferer=https:%2F%2Fwww.google.com%2F 를 참고하여 작성하였다.

 

문제점

먼저 delete를 이용하여 객체를 삭제하면,

1. 해당 객체를 참조하고 있던 스레드에 문제가 생길 수도 있고
2. 객체가 삭제된 주소에 다시 같은 객체가 할당되어 ABA 문제가 일어날 가능성이 있다.
3. 새로 객체를 할당하는 것은 사용자가 관리하는 list가 아니다.

1번 문제를 해결하기 위해서 Hazard pointer를 구현해보았으나, 2, 3번 문제 때문에 결국 새로운 객체를 만드는 방법은 ABA 문제의 온전한 해결책이 되지 못한다. 그래도 헤저드 포인터의 원리를 공부하고 코드를 구현해보았다.

 

원리

우선 각 쓰레드마다 삭제가 필요한 노드들을 담아두는 list를 가지고 있어야한다. 이를 RetireList라고 하자. 쓰레드들은 pop을 할 때마다 필요없어진 노드들을 삭제하기 위해 RetireList에 담아둔다. 하지만 이 노드들이 참조되지 않을 때 삭제해야하므로 모든 쓰레드가 공유하는 Hazard pointer를 만든다.

 Hazard pointer는 연결리스트 형태로 구성되어 있으며 어떤 노드가 참조되고 있는지를 나타내는 역할을 한다. Hazard pointer의 각 공간은 현재 해당 공간이 사용되고 있는지를 나타내는 active 변수와 사용하고 있는 노드를 가리키는 node 변수, 그리고 다음 공간을 가리키는 next 변수로 구성되어있다.

 먼저 각 스레드는 pop 시 Hazard pointer에 공간을 하나 부여받는다. 공간을 부여 받을 때에는 현재 active가 아닌 공간이 있는 지 확인하고, 있다면 해당 공간을 재사용한다. (이렇게 하지 않으면 공간이 계속 늘어나 수행시간이 매우 오래 걸린다.) 공간을 할당 받은 후에는 해당 공간의 active 변수를 1로 만들어 공간을 사용하겠다고 표시하고 사용할 노드(list에서 pop할 노드)를 node 변수에 넣는다. 이러한 코드는 다른 스레드에게 이 node는 현재 사용 중이니 삭제하지 말라고 표시하는 역할을 한다.

 tmp_next를 할당하고 pop할 data를 저장한 뒤 CAS를 시행한다. CAS가 성공하고 나서는 할당 받은 Hazard pointer의 공간을 반납하고 tmp가 가리키고 있는 노드를 현재 쓰레드의 RetireList에 담는다.

 RetireList에 노드가 담길 때마다 Hazard pointer에 있는 모든 사용 중인 노드들을 확인하고 지운다면 너무 오래걸리므로 RetireList에 담긴 노드가 특정 수 이상이 될 때 한꺼번에 검사하고 삭제하기로 하자. (실제 GC도 언제 실행될 지 사용자가 알 수 없다.) 여기서 RetireList는 쓰레드마다 하나씩 가지고 있기 때문에 동기화할 필요가 없다. 따라서 코드에서는 vector 자료구조로 구현하였다.

 

의문점

 그렇다면 Hazard Pointer는 쓰레드가 공유하는 자원인데 상호배제가 필요없을까? atomic 변수를 이용해서 CAS연산으로 active setting을 하기 때문에 lock-free로 구현가능하다.

 그렇다면 ABA 문제는 안 생길까? Hazard Pointer는 push하는 코드만 있지 pop하는 코드가 없다. 쓰레드들은 빈공간을 찾아서 할당하고 없으면 공간을 새로 하나 만든다. 할당받은 메모리를 반납하는 것도 사실은 Hazard Pointer의 공간 자체를 없애는 것이 아니라 해당 공간의 active 변수를 0으로 만들고 node setting을 초기화하는 것 뿐이다. 따라서 ABA 문제도 안 생긴다.

 

코드

절차지향 방식으로 구현되어있다

#define _CRT_SECURE_NO_WARNINGS
#define NULL_INT -1

#include <iostream>
#include <thread>
#include <vector>
#include <algorithm>
#include <atomic>
#include <mutex>
// #include <ctime>
#include <random>
#include <intrin.h>


using namespace std;

struct Node {
    int data;
    Node* next_node;
};

int node_n;
atomic<Node*> free_list;
atomic<Node*> head_list;

//출력을 위한 mutex
mutex mut;

//random 구현
random_device rd;
mt19937 gen(rd());
uniform_int_distribution<> dis(0, 10);


// Hazard pointer 구현
struct HzPRec {
    atomic<int> active;
    Node* node;
    HzPRec* next;
};
atomic<HzPRec*> pHpHead;
vector<Node*> RetireList[3];
atomic<int> DWORD = 0;

// Hazard pointer 함수
HzPRec* AllocHp() {
    HzPRec* p;

    for (p = pHpHead.load(); p != nullptr; p = p->next) { // 현재 빈공간이 있는지 확인
        int active;
        active = p->active.load();
        if (active == 0) {
            if (atomic_compare_exchange_weak(&(p->active), &active, 1))
                return p;
        }
    }

    // 없으면 새로 HP 노드 하나 만들기
    p = new HzPRec();
    p->active = 1;
    p->node = nullptr;

    // 삽입 => aba 안일어나는 이유는 삭제가 없기 때문
    HzPRec* oldHead;
    do {
        oldHead = pHpHead.load();
        p->next = oldHead;
    } while (!atomic_compare_exchange_weak(&pHpHead, &oldHead, p));

    return p;
}

void ReleaseHp(HzPRec* p) { // 미사용중으로 만들기
    p->active = 0;
    p->node = nullptr;
}

void Scan(int t_n) {
    vector<Node*> hpList;
    for (HzPRec* p = pHpHead.load(); p != nullptr; p = p->next) {
        if (p->node != NULL)
            hpList.push_back(p->node);
    }

    sort(hpList.begin(), hpList.end());

    Node* p = RetireList[t_n].back();
    RetireList[t_n].pop_back();
    // thread마다 각각 가지고 있는 자료구조이기 때문에 RetireList는 상호배제 필요없음

    Node* tmp;
    int index = 0;
    while (p != nullptr) { // RetireList를 하나씩 꺼내면서 확인
        int flag = 0;
        for (int i = 0; i < hpList.size(); i++) {
            if (hpList[i] == NULL)
                continue;
            if (hpList[i]->data == p->data) { // 현재 사용중인 노드인가?
                flag = 1;
                break;
            }
        }
        if (!flag) {
            // 노드가 사용 중이 아니라면
            delete p; // 메모리 반납
            if (index < RetireList[t_n].size() - 1) {
                p = RetireList[t_n].back(); // 다음 노드 검사
                RetireList[t_n].pop_back();
            }
            else
                p = nullptr;
        }
        else {
            // 노드가 사용 중이라면
            if (index < RetireList[t_n].size() - 1) {
                p = RetireList[t_n].back();
                RetireList[t_n].pop_back();
            }
            else
                p = nullptr;
        }
    }
}

void RetireNode(Node* node, int t_n) {
    RetireList[t_n].push_back(node);

    if (RetireList[t_n].size() > 200) { // 노드가 200개 이상일 때 메모리 반납
        Scan(t_n);
    }
}

// push, pop 함수, print 함수


void printNodes(int type) { // list 출력 함수
    Node* tmp;
    if (type == 0) {
        cout << "head_list ";
        tmp = head_list.load();
    }
    else {
        cout << "free_list ";
        tmp = free_list.load();
    }
    int num = 0;
    while (tmp != nullptr) {
        //cout << tmp->data << " ";
        tmp = tmp->next_node;
        num++;

        if (num > 100000) { // error 처리
            cout << "node is more than 100000" << endl;
            return;
        }
    }
    cout << endl;
    cout << num << endl;
}

void push(int data) {
    Node* tmp_node = new Node();
    tmp_node->data = data;
    Node* tmp;
    do {
        tmp = head_list.load();
        tmp_node->next_node = tmp;
    } while (!atomic_compare_exchange_weak(&head_list, &tmp, tmp_node));
}

void push_f(int data) {
    Node* tmp_node = new Node();
    tmp_node->data = data;
    Node* tmp;
    do {
        tmp = free_list.load();
        tmp_node->next_node = tmp;
    } while (!atomic_compare_exchange_weak(&free_list, &tmp, tmp_node));
}

int pop(int t_n) {
    Node* tmp; Node* tmp_next;
    HzPRec* hp = AllocHp(); // 메모리 할당
    while (true) {
        if ((tmp = head_list.load()) == nullptr) {
            ReleaseHp(hp); // 메모리 해제
            return NULL_INT;
        }
        // head_list에 data가 들어올 때까지 기다린다.
        hp->node = tmp;

        if (tmp != head_list.load()) // tmp_next 전에 다시 한 번 확인
            continue;

        tmp_next = tmp->next_node;
        int data = tmp->data;

        if (atomic_compare_exchange_weak(&head_list, &tmp, tmp_next)) {
            ReleaseHp(hp); // 메모리 해제
            RetireNode(tmp, t_n); // 삭제할 노드에 등록
            return data;
        }
    }
}

int pop_f(int t_n) {
    Node* tmp; Node* tmp_next;
    HzPRec* hp = AllocHp();
    while (true) {
        if ((tmp = free_list.load()) == nullptr) {
            ReleaseHp(hp);
            return NULL_INT;
        }
        // head_list에 data가 들어올 때까지 기다린다.
        hp->node = tmp;

        if (tmp != free_list.load())
            continue;

        tmp_next = tmp->next_node;
        int data = tmp->data;

        if (atomic_compare_exchange_weak(&free_list, &tmp, tmp_next)) {
            ReleaseHp(hp);
            RetireNode(tmp, t_n);
            return data;
        }
    }
}

void subThread() {
    // random n, m
    int n = dis(gen);
    int m = dis(gen);
    int thread_num = DWORD;
    DWORD.fetch_add(1);

    mut.lock();
    cout << n << " " << m << endl;
    mut.unlock();

    int num = 10000;
    while (num--) {
        for (int i = 0; i < n; i++)
        {
            // free_list에 있으면 head_list에 옮기는 작업
            int tmp = pop_f(thread_num);
            if (tmp == NULL_INT)
                continue;
            push(tmp);
        }
        for (int i = 0; i<m; i++) {
            // head_list에 있으면 free_list에 옮기는 작업
            int tmp = pop(thread_num);
            if (tmp == NULL_INT)
                continue;
            push_f(tmp);
        }
    }
}

int main() {
    // 노드 생성 및 free_list 삽입
    node_n = 100000; // 노드 갯수
    for (int i = node_n - 1; i >= 0; i--) {
        push_f(i);
    }


    // 스레드 생성 및 동작 시작 
    vector<thread> threads;
    int n = 3; // 스레드 갯수
    for (int i = 0; i < n; i++) {
        RetireList[i] = (vector<Node*>());
        threads.emplace_back(subThread);
    }
    for (auto &t : threads)
        t.join();
    threads.clear();

    printNodes(0);
    printNodes(1);
    cin >> n;

}

 

 

결과

• 실행 조건

노드 갯수	thread	while문 수행횟수
100,000개	3개	10,000회

 

• 결과

구분	free_list 노드 수	head_list 노드 수
1	100,000	0
2	50,004	49,996
3	40,000	60,000

돌려본 결과 잘 동작하지만 위에서 언급했던 문제들 때문에 Hazard pointer는 온전한 해결책이 되지 못한다. 다른 방법을 찾아보자.

 

=> Counter 방법은 다음 포스팅에