java - 无锁并发栈实现

Question

我一直在研究用 Java 实现无锁堆栈的简单方法。

编辑:请参阅下面的固定/工作版本

---

您发现此实现有任何问题吗？

本地语言中的类似实现似乎受到 the ABA problem 的影响，但我不确定这是否是个问题；显然没有直接在 Java 中完成指针处理，并且考虑到我所关心的是弹出和推送的堆栈末尾，我不明白堆栈的任何非尾元素中的任何更改“丢失”如何会导致问题.

public class LockFreeStack<T extends LockFreeStack.StackItem<T>>
{
    public abstract static class StackItem<SELF extends StackItem<SELF>>
    {
        volatile SELF next;
        // .. data ..
    }

    final AtomicReference<T> top = new AtomicReference<T>(null);

    public void push(T item)
    {
        T localTop;

        do {
            localTop = top.get();
            item.next = localTop;
        } while(!top.compareAndSet(localTop, item));
    }

    public T pop()
    {
        T localTop;

        do {
            localTop = top.get();
        } while(localTop != null && !top.compareAndSet(localTop, localTop.next));

        return localTop;
    }
}

但是，这是我不明白的。我写了一个启动几个线程的简单测试；每个都从预先存在的 LockFreeStack 弹出项目，然后(稍后，从弹出它的同一个线程)将它们推回。在它弹出后，我增加了一个原子计数器，在推回它之前，我减少了它。所以我总是希望计数器为 0(在递减之后/在推回堆栈之前)或 1(在弹出和递增之后)。

但是，事实并非如此......

public class QueueTest {
    static class TestStackItem extends LockFreeStack.StackItem<TestStackItem>
    {
        final AtomicInteger usageCount = new AtomicInteger(0);

        public void inc() throws Exception
        {
            int c = usageCount.incrementAndGet();

            if(c != 1)
                throw new Exception(String.format("Usage count is %d; expected %d", c, 1));
        }

        public void dec() throws Exception
        {
            int c = usageCount.decrementAndGet();

            if(c != 0)
                throw new Exception(String.format("Usage count is %d; expected %d", c, 0));
        }
    }

    public final LockFreeStack<TestStackItem> testStack = new LockFreeStack<TestStackItem>();

    public void test()
    {
        final int NUM_THREADS = 4;

        for(int i = 0; i < 10; i++)
        {
            TestStackItem item = new TestStackItem();
            testStack.push(item);
        }

        Thread[] threads = new Thread[NUM_THREADS];
        for(int i = 0; i < NUM_THREADS; i++)
        {
            threads[i] = new Thread(new TestRunner());
            threads[i].setDaemon(true);
            threads[i].setName("Thread"+i);
            threads[i].start();
        }

        while(true)
        {
            Thread.yield();
        }

    }

    class TestRunner implements  Runnable
    {
        @Override
        public void run() {
            try {
                boolean pop = false;
                TestStackItem lastItem = null;
                while (true) {
                    pop = !pop;

                    if (pop) {
                        TestStackItem item = testStack.pop();
                        item.inc();
                        lastItem = item;
                    } else {
                        lastItem.dec();
                        testStack.push(lastItem);
                        lastItem = null;
                    }
                }
            } catch (Exception ex)
            {
                System.out.println("exception: " + ex.toString());
            }
        }
    }
}

抛出不确定的异常，例如

exception: java.lang.Exception: Usage count is 1; expected 0
exception: java.lang.Exception: Usage count is 2; expected 1

或来自另一次运行

exception: java.lang.Exception: Usage count is 2; expected 0
exception: java.lang.Exception: Usage count is 3; expected 1
exception: java.lang.Exception: Usage count is 3; expected 1
exception: java.lang.Exception: Usage count is 2; expected 1

所以一定是这里发生了类似竞争条件的问题。

这里有什么问题 - 这确实与 ABA 相关(如果是，究竟是如何相关的？)还是我遗漏了其他任何东西？

谢谢!

---

注意:这可行，但它似乎不是一个很好的解决方案。它既不是无垃圾的(StampedAtomicReference 在内部创建对象)，也不是无锁的好处似乎真的得到返回；在我的基准测试中，这在单线程环境中并没有真正更快，并且在同时测试 6 个线程时，它明显落后于仅在 push/pop 函数周围加锁

根据下面建议的解决方案，这确实是一个 ABA 问题，这个小改动将解决这个问题:

public class LockFreeStack<T extends LockFreeStack.StackItem<T>>
{
    public abstract static class StackItem<SELF extends StackItem<SELF>>
    {
        volatile SELF next;
        // .. data ..
    }

    private final AtomicStampedReference<T> top = new AtomicStampedReference<T>(null, 0);

    public void push(T item)
    {
        int[] stampHolder = new int[1];

        T localTop;

        do {
            localTop = top.get(stampHolder);
            item.next = localTop;
        } while(!top.compareAndSet(localTop, item, stampHolder[0], stampHolder[0]+1));
    }

    public T pop()
    {
        T localTop;
        int[] stampHolder = new int[1];

        do {
            localTop = top.get(stampHolder);
        } while(localTop != null && !top.compareAndSet(localTop, localTop.next, stampHolder[0], stampHolder[0]+1));

        return localTop;
    }
}

Answer 1

在你的测试中你真的不需要这个带有“if condition”和“lastItem”的奇怪循环，你可以通过简单地弹出和推送相同的节点来重现这个错误。

要解决上述问题，您可以在将其插入堆栈时创建新的 TestStackItem(并将现有计数器传递给新创建的节点)，或者您可以使用 AtomicStampedReference 查看节点是否已被修改。