首页 » 开发 » Lock-Free数据结构

无锁程序设计

以下笔记主要来自:Coolshell - 无锁队列的实现

CAS操作

所谓CAS指Compare and Set(或Compare and Swap)。现在几乎所有CPU指令都支持CAS,如X86的CMPXCHG汇编指令。CAS通常被视为无锁(lock free)数据结构的基础。CAS的C语言描述如下:

int compare_and_swap(int* reg, int oldv, int newv) 
{
    int old_reg_v = *reg;    
    if(old_reg_v == oldv) {
        *reg = newv;
    }
    return old_reg_v;
}

bool compare_and_swap(int* accum, int* dest, int newv)
{
    if(*accum == *dest) {
        *dest = newv;
        return true;
    }
    return false;
}

返回bool的好处在于调用者可以知道是否成功更新。与CAS类似的其他原子操作如:Fetch and Add(+1操作)、Test-and-set(写值到指定内存位置并返回旧值,汇编指令BST)、Test and Test-and-set。

无锁队列的链表实现。入队列:

push(x)
{
    q = new Record();
    q->value = x;
    q->next = NULL;

    do {
        p = tail;       /* fetch tail pointer */
    } while(!CAS(p->next, NULL, q)); 
    
    CAS(tail, p, q);    /* set tail point to q (the new record) */
}

当while循环结束时,q已经加入队列成为最后一个元素,但tail指针还没有更新,最后只需将tail指向q即可。分析可能存在的竞争:当thread1将q加入队列,但还未更新tail时,切换到thread2运行,此时thread2获取了tail的值,但tail->next != NULL,因此CAS失败,thread2继续在while循环,直到thread1更新tail值后(tail->next == NULL),thread2才能跳出while循环。

这里有一个潜在问题:如果thread1在把q加入队列后,更新tail前停止运行了,其他线程将进入死循环。为防止这种异常,可以在do-while循环中,直接搜索tail指针:

push(x)
{
    q = new Record();
    q->value = x;
    q->next = NULL;

    oldp = p = tail;
    do {
        /* fetch tail pointer directly */
        while(p->next) 
            p = p->next;
    } while(!CAS(p->next, NULL, q)); 
    
    /* TODO: bug here! */
    CAS(tail, oldp, q);    /* set tail point to q (the new record) */
}

出队列:

pop()
{
    do {
        p = head;
        if(!p->next) {
            return false;
        }
    } while(!CAS(head, p, p->next));
    return p->next->value;
}

注意出队列的操作对象是header->next而不是head本身,这样设计的目的是防止链表中只有一个元素,head和tail指向同一节点的问题。

ABA问题

ABA问题:

  1. P1读取指定内存的值为A
  2. P1被挂起P2运行
  3. P2将指定内存的值从A修改为B,再改回A。
  4. 再次调度到P1
  5. P1发现指定内存的值没有变,于是继续执行。

lock free算法中容易出现ABA问题,特别是用指针的实现。下面有一个很好的比喻:

你拿着一个装满钱的手提箱在机场,迎面走来一个辣妹,然后她很暧昧地挑逗着你,并趁你不注意时把你的手提箱掉包。辣妹之前就准备了多个跟你同款的手提包,当她拿到你手提包时,把钱掏空,并从手头所有的皮箱中(包括刚被掏空那个)任选一个掉包给你。如果你在皮箱上留了记号,大部分情况下,还是能觉察到被调包了;然而如果她恰好选择“物归原主”,光看皮箱你是无法作出判断的。

以下笔记来自Herb Sutter的Writing Lock-Free Code: A Corrected Queue

Sutter介绍了什么叫ordered atomic变量(lock-free-safe):原子性(Atomicity)表示对其他所有读、写线程而言,当前线程的读写操作是原子的,要么成功、要么失败,不会处于一个中间状态(all-or-nothing)。原子性变量通常是指针(C/C++)或对象引用(Java、.NET)或整型变量,这些变量通常都等同于CPU的一个字长(word)。有序(order)表示读、写操作都按源代码次序进行,编译器、CPU、缓存都会遵循源代码的次序。所谓CAS,可以这样理解:

variable.compare_exchange(expect_value, new_value);

如果variable的值等于expect_value,就用new_value替换它。我们常把它放到if中:

if(variable.compare_exchange(x,y)) {}

Ordered atomic变量在C++11中就是atomic。如果语言没有提供Ordered atomic变量,可以自己模拟:使用CPU字长的变量;要满足有序(order),则用系统提供的API(如Win32的InterlockedCompareExchange)或内存屏障(memory fences/barriers;如Linux的mb)。

下面Sutter给了一段代码:实现无锁队列。这里假设只有1个生产者和1个消费者。队列类(链表实现):

template
class LockFreeQueue 
{
private:
    struct Node {
        Node(T v) : value(v), next(nullptr) {}
        T value;
        Node* next;
    };  

    Node* first;        /* for producer only */
    atomic<Node*> divider, last;    /* shared */

接下来是构造(构造一个dummy指针)、析构函数(释放未消费的节点):

public:
    LockFreeQueue() {   /* add dummy seperator */
        first = divider = last = new Node(T());
    }   
    ~LockFreeQueue() {
        while(first != nullptr) {   /* release list */
            Node* t = first;
            first = t->next;
            delete t;
        }   
    }   

生产者函数(只被生产者线程调用):

    /* Produce is called by producer thread only */
    void Produce(const T& t) {
        last->next = new Node(t);   /* append new item; private for producer */
        last = last->next;          /* update last; publish it */
        while(first != divider) {   /* trim unused nodes */
            Node* t = first;
            first = first->next;
            delete t;
        }
    }

生产者函数第1行:last->next = new Node(t); 这里新元素只有生产者可见,因为还没更新last指针。

生产者函数第2行:last = last->next; 原子性地更新last指针。

生产者函数接下来是通过循环清除没有用的节点。不必担心divider在consumer中被修改,因为我们总能读取到最新的divider值。

消费者函数(只被消费者线程调用):

    /* Consume is called by consumer thread only */
    bool Consume(T& result) {
        if(divider != last) {
            result = divider->next->value;
            divider = divider->next;
            return true;
        }
        return false;
    }
};

再回头看队列的成员:first、divider、last。生产者独占first,读divider、写last;消费者写divider、读last,因此它们的类型分别为:Node*、atomic<Node*>、atomic<Node*>。

多线程内存模型

以下笔记主要来自:刘未鹏 - C++0x漫谈 - 多线程内存模型

分享

0