多线程队列的算法优化 | Parallel Labs

多线程队列的算法优化 | Parallel Labs
多线程队列（Concurrent Queue）的使用场合非常多，高性能服务器中的消息队列，并行算法中的Work Stealing等都离不开它。对于一个队列来说有两个最主要的动作：添加（enqueue）和删除（dequeue）节点。在一个（或多个）线程在对一个队列进行enqueue操作的同时可能会有一个（或多个）线程对这个队列进行dequeue操作。因为enqueue和dequeue都是对同一个队列里的节点进行操作，为了保证线程安全，一般在实现中都会在队列的结构体中加入一个队列锁（典型的如pthread_mutex_t q_lock），在进行enqueue和dequeue时都会先锁住这个锁以锁住整个队列然后再进行相关的操作。这样的设计如果实现的好的话一般性能就会很不错了。以链表实现的队列的结构体一般是这样的：

01 02 03 04 05

struct queue_t {     node_t *head;     node_t *tail;     pthread_mutex_t q_lock; };

但是，这其中其实有一个潜在的性能瓶颈：enqueue和dequeue操作都要锁住整个队列，这在线程少的时候可能没什么问题，但是只要线程数一多，这个锁竞争所产生的性能瓶颈就会越来越严重。那么我们可不可以想办法优化一下这个算法呢？当然可以！如果我们仔细想一想enqueue和dequeue的具体操作就会发现他们的操作其实不一定是冲突的。例如：如果所有的enqueue操作都是往队列的尾部插入新节点，而所有的dequeue操作都是从队列的头部删除节点，那么enqueue和dequeue大部分时候都是相互独立的，我们大部分时候根本不需要锁住整个队列，白白损失性能！那么一个很自然就能想到的算法优化方案就呼之欲出了：我们可以把那个队列锁拆成两个：一个队列头部锁（head lock)和一个队列尾部锁(tail lock)。这样这样的设计思路是对了，但是如果再仔细思考一下它的实现的话我们会发现其实不太容易，因为有两个特殊情况非常的tricky（难搞）：第一种就是往空队列里插入第一个节点的时候，第二种就是从只剩最后一个节点的队列中删除那个"最后的果实"的时候。
为什么难搞呢？当我们向空队列中插入第一个节点的时候，我们需要同时修改队列的head和tail指针，使他们同时指向这个新插入的节点，换句话说，我们此时即需要拿到head lock又需要拿到tail lock。而另一种情况是对只剩一个节点的队列进行dequeue的时候，我们也是需要同时修改head和tail指针使他们指向NULL，亦即我们需要同时获得head和tail lock。有经验的同学会立刻发现我们进入危险区了！是什么危险呢？死锁！多线程编程中最臭名昭著的一种bug就是死锁了。例如，如果线程A在锁住了资源1后还想要获取资源2，而线程B在锁住了资源2后还想要获取资源1，这时两个线程谁都不能获得自己想要的那个资源，两个线程就死锁了。所以我们要小心奕奕的设计这个算法以避免死锁，例如保证enqueue和dequeue对head lock和tail lock的请求顺序（lock ordering）是一致的等等。但是这样设计出来的算法很容易就会包含多次的加锁/解锁操作，这些都会造成不必要的开销，尤其是在线程数很多的情况下反而可能导致性能的下降。我的亲身经历就是在32线程时这个思路设计出来的算法性能反而下降了10%左右，原因就是加锁/解锁的开销增加了。
好在有聪明人早在96年就想到了一个更妙的算法。这个算法也是用了head和tail两个锁，但是它有一个关键的地方是它在队列初始化的时候head和tail指针不为空，而是指向一个空节点。在enqueue的时候只要向队列尾部添加新节点就好了。而dequeue的情况稍微复杂点，它要返回的不是头节点，而是head->next，即头节点的下一个节点。先来看伪代码：

01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43

typedef struct node_t {     TYPE value;     node_t *next } NODE; typedef struct queue_t {     NODE *head;     NODE *tail;     LOCK q_h_lock;     LOCK q_t_lock; } Q; initialize(Q *q) {    node = new_node()   // Allocate a free node    node->next = NULL   // Make it the only node in the linked list    q->head = q->tail = node   // Both head and tail point to it    q->q_h_lock = q->q_t_lock = FREE   // Locks are initially free } enqueue(Q *q, TYPE value) {    node = new_node()       // Allocate a new node from the free list    node->value = value     // Copy enqueued value into node    node->next = NULL       // Set next pointer of node to NULL    lock(&q->q_t_lock)      // Acquire t_lock in order to access Tail       q->tail->next = node // Link node at the end of the queue       q->tail = node       // Swing Tail to node    unlock(&q->q_t_lock)    // Release t_lock ｝ dequeue(Q *q, TYPE *pvalue) {    lock(&q->q_h_lock)   // Acquire h_lock in order to access Head       node = q->head    // Read Head       new_head = node->next       // Read next pointer       if new_head == NULL         // Is queue empty?          unlock(&q->q_h_lock)     // Release h_lock before return          return FALSE             // Queue was empty       endif       *pvalue = new_head->value   // Queue not empty, read value       q->head = new_head  // Swing Head to next node    unlock(&q->q_h_lock)   // Release h_lock    free(node)             // Free node    return TRUE            // Queue was not empty, dequeue succeeded }

发现玄机了么？是的，这个算法中队列总会包含至少一个节点。dequeue每次返回的不是头节点，而是头节点的下一个节点中的数据：如果head->next不为空的话就把这个节点的数据取出来作为返回值，同时再把head指针指向这个节点，此时旧的头节点就可以被free掉了。这个在队列初始化时插入空节点的技巧使得enqueue和dequeue彻底相互独立了。但是，还有一个小地方在实现的时候需要注意：对第一个空节点的next指针的读写。想象一下，当一个线程对一个空队列进行第一次enqueue操作时刚刚运行完第25行的代码（对该空节点的next指针进行写操作）；而此时另一个线程对这个队列进行第一次dequeue操作时恰好运行到第33行（对该空节点的next指针进行读操作），它们其实还是有冲突！不过，好在一般来讲next指针是32位数据，而现代的CPU已经能保证多线程程序中内存对齐了的32位数据读写操作的原子性，而一般来讲编译器会自动帮你对齐32位数据，所以这个不是问题。唯一需要注意的是我们要确保enqueue线程是先让要添加的新节点包含好数据再把新节点插入链表（也就是不能先插入空节点，再往节点中填入数据），那么dequeue线程就不会拿到空的节点。其实我们也可以把q_t_lock理解成生产者的锁，q_h_lock理解成消费者的锁，这样生产者（们）和消费者（们）的操作就相互独立了，只有在多个生产者对同一队列进行添加操作时，以及多个消费者对同一队列进行删除操作时才需要加锁以使访问互斥。
通过使用这个算法，我成功的把一个32线程程序的性能提升了11%！可见多线程中的锁竞争对性能影响之大！此算法出自一篇著名的论文：M. Michael and M. Scott. Simple, Fast, and Practical Non-Blocking and Blocking Concurren Queue Algorithms. 如果还想做更多优化的话可以参考这篇论文实现相应的Non Blocking版本的算法，性能还能有更多提升。当然了，这个算法早已被集成到java.util.concurrent里了（即LinkedBlockingQueue），其他的并行库例如Intel的TBB多半也有类似的算法，如果大家能用上现成的库的话就不要再重复造轮子了。为什么别造并行算法的轮子呢？因为高性能的并行算法实在太难正确地实现了，尤其是Non Blocking，Lock Free之类的"火箭工程"。有多难呢？Doug Lea提到java.util.concurrent中一个Non Blocking的算法的实现大概需要1年的时间，总共约500行代码。所以，对最广大的程序员来说，别去写Non Blocking, Lock Free的代码，只管用就行了，我看见网上很多的Non Blocking阿，无锁编程的算法实现啊什么的都非常地害怕，谁敢去用他们贴出来的这些代码啊？我之所以推荐这个two lock的算法是因为它的实现相对Non Blocking之类的来说容易多了，非常具备实用价值。虽然这篇论文出现的很早，但是我在看了几个开源软件中多线程队列的实现之后发现他们很多还是用的本文最开始提到的那种一个锁的算法。如果你想要实现更高性能的多线程队列的话，试试这个算法吧！
Update: 多线程队列算法有很多种，大家应根据不同的应用场合选取最优算法（例如是CPU密集型还是IO密集型）。本文所列的算法应用在这样一个多线程程序中：每个线程都拥有一个队列，每个队列可能被本线程进行dequeue操作，也可以被其他线程进行dequeue（即work stealing），线程数不超过CPU核心数，是一个典型的CPU/MEM密集型客户端单写者多读者场景。

java.util.concurrent.LinkedBlockingQueue

http://grepcode.com/file/repository.grepcode.com/java/root/jdk/openjdk/8u40-b25/java/util/concurrent/LinkedBlockingQueue.java

    /** Lock held by take, poll, etc */
    private final ReentrantLock takeLock = new ReentrantLock();

    /** Wait queue for waiting takes */
    private final Condition notEmpty = takeLock.newCondition();

    /** Lock held by put, offer, etc */
    private final ReentrantLock putLock = new ReentrantLock();

    /** Wait queue for waiting puts */
    private final Condition notFull = putLock.newCondition();

    /**
     * Signals a waiting take. Called only from put/offer (which do not
     * otherwise ordinarily lock takeLock.)
     */
    private void signalNotEmpty() {
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();
        try {
            notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
    }

    /**
     * Signals a waiting put. Called only from take/poll.
     */
    private void signalNotFull() {
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            notFull.signal();
        } finally {
            putLock.unlock();
        }
    }

    public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        // Note: convention in all put/take/etc is to preset local var
        // holding count negative to indicate failure unless set.
        int c = -1;
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();
        try {
            /*
             * Note that count is used in wait guard even though it is
             * not protected by lock. This works because count can
             * only decrease at this point (all other puts are shut
             * out by lock), and we (or some other waiting put) are
             * signalled if it ever changes from capacity. Similarly
             * for all other uses of count in other wait guards.
             */
            while (count.get() == capacity) {
                notFull.await();
            }
            enqueue(node);
            c = count.getAndIncrement();
            if (c + 1 < capacity)
                notFull.signal();
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
    }

   public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        E x = null;
        int c = -1;
        long nanos = unit.toNanos(timeout);
        final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly();
        try {
            while (count.get() == 0) {
                if (nanos <= 0)
                    return null;
                nanos = notEmpty.awaitNanos(nanos);
            }
            x = dequeue();
            c = count.getAndDecrement();
            if (c > 1)
                notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)
            signalNotFull();
        return x;
    }

ArrayBlockingQueue
/** Main lock guarding all access */
final ReentrantLock lock;

/** Condition for waiting takes */
private final Condition notEmpty;

/** Condition for waiting puts */
private final Condition notFull;
/** The queued items */
final Object[] items;

/** items index for next take, poll, peek or remove */
int takeIndex;

/** items index for next put, offer, or add */
int putIndex;

/** Number of elements in the queue */
int count;

public boolean offer(E e, long timeout, TimeUnit unit)
    throws InterruptedException {

    checkNotNull(e);    long nanos = unit.toNanos(timeout);    final ReentrantLock lock = this.lock;    lock.lockInterruptibly();    try {
        while (count == items.length) {
            if (nanos <= 0)
                return false;            nanos = notFull.awaitNanos(nanos);        }
        enqueue(e);        return true;    } finally {
        lock.unlock();    }
}

private void enqueue(E x) {
    // assert lock.getHoldCount() == 1;    // assert items[putIndex] == null;    final Object[] items = this.items;    items[putIndex] = x;    if (++putIndex == items.length)
        putIndex = 0;    count++;    notEmpty.signal();}

/** * Extracts element at current take position, advances, and signals. * Call only when holding lock. */private E dequeue() {
    // assert lock.getHoldCount() == 1;    // assert items[takeIndex] != null;    final Object[] items = this.items;    @SuppressWarnings("unchecked")
    E x = (E) items[takeIndex];    items[takeIndex] = null;    if (++takeIndex == items.length)
        takeIndex = 0;    count--;    if (itrs != null)
        itrs.elementDequeued();    notFull.signal();    return x;}

http://www.cs.rochester.edu/research/synchronization/pseudocode/queues.html

Two-Lock Concurrent Queue Algorithm

  structure node_t {value: data type, next: pointer to node_t}
  structure queue_t {Head: pointer to node_t, Tail: pointer to node_t,
                        H_lock: lock type, T_lock: lock type}
  
  initialize(Q: pointer to queue_t)
     node = new_node()  // Allocate a free node
     node->next = NULL          // Make it the only node in the linked list
     Q->Head = Q->Tail = node // Both Head and Tail point to it
     Q->H_lock = Q->T_lock = FREE // Locks are initially free
  
  enqueue(Q: pointer to queue_t, value: data type)
     node = new_node()         // Allocate a new node from the free list
     node->value = value  // Copy enqueued value into node
     node->next = NULL          // Set next pointer of node to NULL
     lock(&Q->T_lock)  // Acquire T_lock in order to access Tail
        Q->Tail->next = node // Link node at the end of the linked list
        Q->Tail = node  // Swing Tail to node
     unlock(&Q->T_lock)  // Release T_lock
  
  dequeue(Q: pointer to queue_t, pvalue: pointer to data type): boolean
     lock(&Q->H_lock)         // Acquire H_lock in order to access Head
        node = Q->Head  // Read Head
        new_head = node->next // Read next pointer
        if new_head == NULL // Is queue empty?
           unlock(&Q->H_lock) // Release H_lock before return
           return FALSE  // Queue was empty
        endif
        *pvalue = new_head->value // Queue not empty.  Read value before release
        Q->Head = new_head // Swing Head to next node
     unlock(&Q->H_lock)  // Release H_lock
     free(node)   // Free node
     return} TRUE  // Queue was not empty, dequeue succeeded

Read full article from 多线程队列的算法优化 | Parallel Labs