[c++] CAS 구현 및 ABA 문제 해결 :: ABA 해결_3. Counter 구현

 

목차

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

Counter

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

using namespace std;

class Node {
public:
    int value;
    Node* next;
    Node() : value(0) { next = NULL; }

    Node(int k_value) {
        next = NULL;
        value = k_value;
    }
};

int node_n;

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

//출력을 위한 mutex
mutex mut;

class LFStack {
private:
    long long CountedPtr;

public:
    void printNodes() {
        int size = 0;
        Node* head = reinterpret_cast<Node*>(CountedPtr);
        while (head != nullptr) {
            size++;
            head = head->next;
        }
        cout << size << endl;
    }

    long long getCount(long long cp) {
        return ((unsigned long long)cp >> 52); // 왼쪽 비트 0으로 채우기
    }
    long long setCount(long long count) {
        return count << 52;
    }

    void push(Node* newNode) {
        long long oldCP, newCP;
        unsigned long long newCount;
        do {
            oldCP = this->CountedPtr;
            newCount = getCount(oldCP) + 1; // count 1 증가
            newNode->next = reinterpret_cast<Node*>(oldCP); // list의 head 노드를 가리키게 하기
            newCP = setCount(newCount) + reinterpret_cast<long long>(newNode); // 바꾼 카운터와 node 결합
        } while (!(_InterlockedCompareExchange64(&(this->CountedPtr),newCP,oldCP)==oldCP)); //CAS
    }
    Node* pop() {
        long long oldCP, newCP;
        do {
            oldCP = this->CountedPtr;
            Node* popNode = reinterpret_cast<Node*>(oldCP);  // 맨 앞 노드 구하기
            if (popNode == nullptr)
                return nullptr;
            newCP = (long long)(setCount(getCount(oldCP))) + reinterpret_cast<long long>(popNode->next); // counter는 그대로, node는 다음 노드
        } while (!(_InterlockedCompareExchange64(&(this->CountedPtr),newCP,oldCP)==oldCP)); // CAS
        return reinterpret_cast<Node*>(oldCP);
    }
};

LFStack* FreeList;
LFStack* HeadList;

void subThread() {
    // random n, m
    int n = dis(gen);
    int m = dis(gen);

    int num = 10000; // while문 반복 횟수
    while (num--) {
        for (int i = 0; i < n; i++)
        {
            // free_list에 있으면 head_list에 옮기는 작업
            Node* tmp = FreeList->pop();
            if (tmp == nullptr)
                continue;
            HeadList->push(tmp);

        }
        for (int i = 0; i<m; i++) {
            // head_list에 있으면 free_list에 옮기는 작업
            Node* tmp = HeadList->pop();
            if (tmp == nullptr)
                continue;
            FreeList->push(tmp);
        }
    }

}

int main() {
    // free, head list 생성
    FreeList = new LFStack();
    HeadList = new LFStack();

    // 노드 생성 및 free_list에 삽입
    node_n = 100000; // 노드 갯수
    for (int i = node_n - 1; i >= 0; i--) {
        Node* node = new Node(i);
        FreeList->push(node);
    }

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

    FreeList->printNodes();
    HeadList->printNodes();
    cin >> n;
}

 대부분의 64bit CPU에서 가상 주소를 52bit 밖에 사용하지 않는다는 점을 이용한 방법이다.

 

 각 List는 long long 형 CountedPtr이라는 자료형 하나로 이루어져있다. long long은 8byte 자료형이므로 64bit의 주소를 모두 표현할 수 있다. CountedPtr의 하위 52bit는 주소를 나타내는데에 쓰고, 상위 12bit는 Count를 나타내는 데에 쓰기로 하자.

 

 List가 가리키는 첫번째 노드의 주소와 List에 push가 사용된 횟수(Count)를 각각 하위 52bit, 상위 12bit에 할당하여 CountedPtr을 만든다. ABA 문제는 push가 발생해야지만 일어날 가능성이 있으므로 push함수 시행 시에만 Count를 증가시켜주었다. push 함수가 실행될 때마다 Count는 1씩 증가하므로 _InterlockedCompareExchange64 에서 비교 시 Count가 달라지면 연산이 수행되지 않는다. 따라서 ABA 문제를 해결할 수 있다.

 

 pop 함수에서 List의 첫번째 노드가 가리키는 다음 노드를 받아오려면 CountedPtr이라는 정수형에서 Node의 주소인 Node*를 추출할 필요가 있다. 이 때는 reinterpret_cast<Node*>(대상) 이라는 캐스팅 연산을 사용하여 Node를 불러온다.

 

 또한 Count도 1 늘리기 위해 추출하는 과정이 필요한데, Count는 getCount와 setCount 함수를 따로 구현하여 다루었다. Count 추출 시 유의할 점은 CPU마다 right shift 연산 시 0과 1 중에 어떤 값이 채워지는 지가 다르다는 것이다. 따라서 unsigned 자료형을 이용하여 음수가 되지 않도록(0으로 채워지도록) 만들어주어야한다. 아니면 0으로 채워주는 >>> 연산을 사용해도 되는데 c++에는 해당 연산이 존재하지 않는다.

 

=> ABA를 해결하는 기타 방법들은 다음 포스팅에