背景

最近在很多JDK源码中都看到了Treiber stack这个单词。

比如CompletableFuture中的：

volatile Completion stack;    // Top of Treiber stack of dependent actions

比如FutureTask中的：

/** Treiber stack of waiting threads */

private volatile WaitNode waiters;

比如Phaser中的：

/**

 * Wait nodes for Treiber stack representing wait queue

 */

static final class QNode implements ForkJoinPool.ManagedBlocker {

    final Phaser phaser;

    final int phase;

    final boolean interruptible;

    final boolean timed;

    boolean wasInterrupted;

    long nanos;

    final long deadline;

    volatile Thread thread; // nulled to cancel wait

    QNode next;

还比如ForkJoinPool中的描述：

 * Bits and masks for field ctl, packed with 4 16 bit subfields:

 * AC: Number of active running workers minus target parallelism

 * TC: Number of total workers minus target parallelism

 * SS: version count and status of top waiting thread

 * ID: poolIndex of top of Treiber stack of waiters

感觉这种名词出现的频率有点高，需要深入了解一下。

名称由来

Treiber Stack在 R. Kent Treiber在1986年的论文Systems Programming: Coping with Parallelism中首次出现。它是一种无锁并发栈，其无锁的特性是基于CAS原子操作实现的。

CompletableFuture源码实现

CompletableFuture的Treiber stack实现感觉有点复杂，因为有其他逻辑掺杂，代码不容易阅读，其实抽象来看，Treiber stack首先是个单向链表，链表头部即栈顶元素，在入栈和出现过程中，需要对栈顶元素进行CAS控制，防止多线程情况下数据错乱。

// Either the result or boxed AltResult

volatile Object result;

// Top of Treiber stack of dependent actions（Treiber stack栈顶元素）

volatile Completion stack;

/** Returns true if successfully pushed c onto stack. */

final boolean tryPushStack(Completion c) {

    Completion h = stack;

    lazySetNext(c, h);

    return UNSAFE.compareAndSwapObject(this, STACK, h, c);

}

/** Unconditionally pushes c onto stack, retrying if necessary. */

final void pushStack(Completion c) {

    do {} while (!tryPushStack(c));

}

简单来看，入栈的步骤如下：

尝试入栈，利用CAS将新的节点作为栈顶元素，新节点next赋值为旧栈顶元素
尝试入栈成功，即结束；入栈失败，继续重试上面的操作

FutureTask实现

FutureTask用了Treiber Stack来维护等待任务完成的线程，在FutureTask的任务完成/取消/异常后在finishCompletion钩子方法中会唤醒栈中等待的线程。

Treiber Stack抽象实现

入栈

void push(Node new) {

  do {

  } while(!tryPush(new)) // 尝试入栈

}

boolean tryPush(node) {

    Node oldHead = head;

    node.next = oldHead; // 新节点next赋值为旧栈顶元素

    return CAS(oldHead, node); // 利用CAS将新的节点作为栈顶元素

}

出栈

对于出栈，要做的工作就是将原来的栈顶节点移除，等待垃圾回收；新栈顶元素CAS为第一个子元素。伪代码：

E pop() {

    Node<E> oldHead;

    Node<E> newHead;

    do {

        oldHead = top.get();

        // 判断栈是否为空，为空直接返回

        if (oldHead == null)

            return null;

        newHead = oldHead.next;

    } while (!CAS(oldHead, newHead));

    // 旧的节点删掉next引用，等待gc

    oldHead.item = null;

    return oldHead.item;

}

示例

import sun.misc.Unsafe;

import java.lang.reflect.Field;

import java.util.Objects;

import java.util.Optional;

import java.util.Random;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

/**

 * 基于Unsafe实现TreiberStack

 * @author Charles

 */

public class TreiberStack<E> {

    private volatile Node<E> head;

    public void push(E item) {

        Objects.requireNonNull(item);

        Node<E> newHead = new Node<>(item);

        Node<E> oldHead;

        int count = 0;

        do {

            oldHead = head;

            count++;

        } while (!tryPush(oldHead, newHead, count));

        newHead.next = oldHead;

    }

    private boolean tryPush(Node<E> oldHead, Node<E> newHead, int count) {

        boolean isSuccess = UNSAFE.compareAndSwapObject(this, HEAD, oldHead, newHead);

        System.out.println(currentThreadName() + " try push [" + count + "]," +

                " oldHead = " + getValue(oldHead) +

                " newHead = " + getValue(newHead) +

                " isSuccess = " + isSuccess);

        return isSuccess;

    }

    public E pop() {

        Node<E> oldHead;

        Node<E> newHead;

        do {

            oldHead = head;

            System.out.println(currentThreadName() + " do pop:" +

                    " oldHead = " + getValue(oldHead) +

                    " newHead = " + Optional.ofNullable(head).map(s -> s.next.item).orElse(null));

            if (oldHead == null) {

                return null;

            }

            newHead = oldHead.next;

        } while (!tryPop(oldHead, newHead));

        oldHead.next = null;

        return oldHead.item;

    }

    private boolean tryPop(Node<E> oldHead, Node<E> newHead) {

        boolean isSuccess = UNSAFE.compareAndSwapObject(this, HEAD, oldHead, newHead);

        System.out.println(currentThreadName() + " try pop:" +

                " oldHead = " + getValue(oldHead) +

                " currentHead = " + getValue(head) +

                " newHead = " + getValue(newHead) +

                " isSuccess: " + isSuccess);

        return isSuccess;

    }

    private E getValue(Node<E> n) {

        return Optional.ofNullable(n).map(t -> t.item).orElse(null);

    }

    private static class Node<E> {

        E item;

        Node<E> next;

        Node(E item) {

            this.item = item;

        }

    }

    // Unsafe mechanics

    private static final sun.misc.Unsafe UNSAFE;

    private static final long HEAD;

    private static final long NEXT;

    static {

        try {

            Field getUnsafe = sun.misc.Unsafe.class.getDeclaredField("theUnsafe");

            getUnsafe.setAccessible(true);

            UNSAFE = (Unsafe) getUnsafe.get(null);

            Class<?> k = TreiberStack.class;

            HEAD = UNSAFE.objectFieldOffset(k.getDeclaredField("head"));

            NEXT = UNSAFE.objectFieldOffset(TreiberStack.Node.class.getDeclaredField("next"));

        } catch (Exception x) {

            throw new Error(x);

        }

    }

    private static class RandomValue {

        private final Integer value;

        public RandomValue() {

            this.value = new Random().nextInt(Integer.MAX_VALUE);

        }

        public Integer getValue() {

            return value;

        }

        @Override

        public String toString() {

            return value.toString();

        }

    }

    private static String currentThreadName() {

        return System.nanoTime() + " / " + Thread.currentThread().getName();

    }

    public static void main(String[] args) throws InterruptedException {

        TreiberStack<RandomValue> ts = new TreiberStack<>();

        ExecutorService es = Executors.newFixedThreadPool(10);

        Thread.sleep(2000);

        for (int i = 0; i < 5; i++) {

            es.submit(() -> ts.push(new RandomValue()));

        }

        for (int i = 0; i < 50; i++) {

            es.submit((Runnable) ts::pop);

        }

    }

}

参考

Wiki Treiber Stack

Treiber Stack介绍

 Treiber stack设计

巴特西

理解与使用Treiber Stack

背景

名称由来

CompletableFuture源码实现

FutureTask实现

Treiber Stack抽象实现

入栈

出栈

示例

参考

最新文章

热门文章