논 블로킹 동기화 방식으로 구현된 Set# 논 블로킹 동기화 방식으로 구현된 Set 만들기

게으른 동기화방식으로 구현된 Set으로도 우리는 멀티스레드 프로그래밍의 병렬성으로 인한 성능 향상에 대해 충분히 느낄 수 있었다. 그러나 여전히 Add(...)와 Remove(...)에서는 잠금을 걸고 있어 잠금으로 인한 문제들과 오버헤드에 자유롭지 못하다.

동기화 방식

• 대단위 동기화 Coarse-Grained Synchronization
• 세밀한 동기화 Fine-Grained Synchronization
• 낙관적 동기화 Optimistic Synchronization
• 게으른 동기화 Lazy Synchronization
• 논 블로킹 동기화 Non-Blocking Synchronization

잠금을 사용하지 않고 원자성을 유지하기 위해서 우리는 CAS (Compare and Swap)를 사용하면 된다는 것을 익히 알고 있다. 이 장에서는 잠금을 사용하지 않은 Add(...)와 Remove(...)를 구현하여 알고리즘 전체를 논 블로킹 동기화방식으로 구현할 것이다. 그로 인한 성능 향상은 물론 논 블로킹 알고리즘의 어려움 또한 깨닫게 된다.

전체 소스코드

소스코드가 앞서 살펴봤던 것들과는 사뭇 다르게 느껴진다. 그러나 기본적인 동작은 4백만번 동안 렌덤하게에 Add, Remove, Contains 동작을 수행하며 처리속도를 측정하는 프로그램으로 동일하다.

class Node_nb의 맴버 필드

class Node_nb
{
public:
    int key;
    Node_nb* pNext = nullptr;

    Node_nb() {}

    Node_nb(int value)
        : key(value)
    {
    }

    bool CAS(int oldValue, int newValue)
    {
        return atomic_compare_exchange_strong(
            reinterpret_cast<atomic_int*>(&pNext),
            &oldValue,
            newValue
            );
    }

    Node_nb* GetNext()
    {
        int target = reinterpret_cast<int>(pNext);
        return reinterpret_cast<Node_nb*>(target & 0xFFFFFFFE);
    }
    /*...*/
}

기능이 많이 달라졌기 때문에 이름 뒤에 _nb(non-blocking)을 덧붙였을 뿐 큰 의미는 없다. 지금까지는 Node를 설명하는데 한 단락이면 충분했지만 이제는 내용이 조금 길어졌기에 나눠서 설명한다.

기본적으로 key와 pNext를 가지는건 동일하다. 락프리 알고리즘으로 개발될 것이므로 mutex변수나 잠금 관련 메서드가 없는 것을 볼 수 있다.

bool CAS(...)는 인자로 들어온 oldValue의 값과 pNext 자체의 주소값을 비교하여 같다면 newValue로 교체한다. 포인터에 대한 기본적인 이해가 있어도 다소 혼동할 수 있는 부분이니 주의해야 한다.

Node_nb* GetNext()는 pNext를 0xFFFFFFFE로 마스킹 하여 32비트 값의 가장 낮은 자릿수를 제거한 뒤 Node_nb로 캐스팅하여 돌려준다. 그냥 pNext를 돌려주지 않는 이유에 대해서는 다음 단락에서 함께 설명하겠다.

Node_nb의 mark관련 메서드

bool GetMark()
{
    int target = reinterpret_cast<int>(pNext);
    return (target & 0x1) != 0;
}

Node_nb* GetNextMark(bool* out_isDeleted)
{
    int target = reinterpret_cast<int>(pNext);
    *out_isDeleted = (target & 0x1) != 0;
    return reinterpret_cast<Node_nb*>(target & 0xFFFFFFFE);
}

bool GetMark()를 보면 pNext의 정체가 조금씩 드러나기 시작한다. pNext를 int로 변경하여 0x1로 마스크 연산을 한다는 건 32비트 int 값에서 가장 낮은 자릿수만 떼어내겠다는 의미이다. 이 값이 0이면 false를, 1이면 true를 반환하는 메서드이다. Node_nb의 멤버에서 사라진 isDeleted가 바로 여기 붙어있는 것이다.

pNext의 구조가 포인터 + 마스크필드라는 것을 알게 되었으니 Node_nb* GetNextMark(...)도 어렵지 않게 이해할 수 있을 것이다. 반환 값은 pNext의 앞 부분. 즉 주소 값이며 out_isDeleted인자값을 통해 별도로 삭제 여부를 반환하고 있다. 그렇다면 pNext는 왜 굳이 다음 노드의 포인터와 삭제 여부가 합쳐진 형태가 되었을까? 다음 장을 보도록 하자.

Node_nb의 노드단위 CAS와 마크시도 메서드

bool CAS(Node_nb* pOldNext, Node_nb* pNewNext, bool oldMark, bool newMark)
{
    int oldValue = reinterpret_cast<int>(pOldNext);
    oldValue = oldMark ? oldValue | 0x1 : oldValue & 0xFFFFFFFE;

    int newValue = reinterpret_cast<int>(pNewNext);
    newValue = newMark ? newValue | 0x1 : newValue & 0xFFFFFFFE;

    return CAS(oldValue, newValue);
}

bool AttemptMark(Node_nb* pOldNode, bool newMark)
{
    int target = reinterpret_cast<int>(pOldNode);
    target &= 0xFFFFFFFE;

    int newValue = target;
    if(newMark)
    {
        newValue |= 0x1;
    }

    return CAS(target, newValue);
}

먼저 노드 단위 CAS연산을 수행하는 메서드를 보자. 인자로 들어온 두 노드 다 삭제되지 않았고, this->pNext가 다른 스레드에 의해 변경되지 않았다면 pNewNext로 교체하는 일을 한다. 이 과정에서 마스크 필드를 이용해 Node_nb*에게 값 끼워 넣는 부분이 있지만 이제는 이해할 수 있을 것이다.

Node_nb의 마지막 메서드인 AttemptMark는 삭제됨 마크를 시도한다. 노드를 잠그지 않는 락프리 알고리즘의 특성 상 마킹 도중에도 다른 스레드가 노드의 값을 변경할 수 있다. 따라서 원래 있을 것으로 생각되는 값과 마스크 값을 변경한 값으로 CAS를 시도한다. 노드단위 CAS와 다른 점은 이 메서드는 포인터 값은 그대로 이며 마스크 필드 값만 바꾸려고 시도한다는 점이다.

Set_NonBlocking의 맴버 필드

class Set_NonBlocking
{
private:
    Node_nb head;
    Node_nb tail;

public:
    Set_NonBlocking()
    {
        head.key = 0x80000000;
        tail.key = 0x7FFFFFFF;
        head.pNext = &tail;
    }

    void Init()
    {
        while(head.pNext != &tail)
        {
            Node_nb* pNode = head.pNext;
            head.pNext = head.pNext->GetNext();
            delete pNode;
        }
    }
    // ...
}

지금까지와 동일하게 보초 노드인 head와 tail이 있는 방식이므로 자세한 설명은 생략한다.

Set_NonBlocking.Find

    void Find(int value, Node_nb** outPred, Node_nb** outCurr)
    {
        bool isDeleted = false;
retry:
        *outPred = &head;
        *outCurr = (*outPred)->pNext;
        while(true)
        {
            Node_nb* pTarget = (*outCurr)->GetNextMark(&isDeleted);
            while(isDeleted)
            {
                if(!(*outPred)->CAS((*outCurr), pTarget, false, false))
                    goto retry;
            }

            if((*outCurr)->key >= value)
                return;

            *outPred = *outCurr;
            *outCurr = pTarget;
        }
    }

Set_NonBlocking에서 처음 등장한 Find는 해당 value가 들어가야 하는 위치의 앞 노드와 현재 노드를 반환한다. 만약 value가 이미 있다면 outCurr->key == value로 판단할 수 있다. 이 메서드는 goto문을 사용하여 코드를 간단히 하였다. 인자값 value를 19로 넣었을때의 동작을 단계별로 살펴보자.

1. while문 안에서 pTarget과 지워진 여부 &isDeleted를 얻어옴

이 단계에서는 지워졌다면 oldPred와 outCurr(안 지워짐)을 비교하여 같다면 oldPred를 pTarget(안 지워짐)으로 교체하는 CAS를 시도한다. 현재 그림에서는 발생하지 않으므로 실패하게 되고, goto retry로 돌아가 처음부터 수행하게 된다.

지워지지 않은 경우라면 while(isDeleted)를 넘어가고 outCurr->key(3) >= value(19)를 검사하게 된다. 이 경우에서는 참이 아니므로 넘어가게 되며 outPred와 outCurr를 한칸씩 이동한다.

실행 결과

에;; 왜 게으른보다 느림요?