理解与使用Treiber Stack

Posted 2020-12-14 mrcharleshu

目录

• 背景
• 名称由来
• CompletableFuture源码实现
• FutureTask实现
• Treiber Stack抽象实现
  • 入栈
  • 出栈
• 示例
• 参考

背景

最近在很多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设计