标签: lock-free

我读的越多,我变得越困惑......我会认为找到一个在c ++中实现的正式正确的mpsc队列是微不足道的.

每当我发现另一次刺痛时,进一步的研究似乎表明存在诸如ABA或其他微妙的竞争条件之类的问题.

许多人都在谈论垃圾收集的必要性.这是我想要避免的.

那里有一个公认的正确的开源实现吗？

要执行无锁和无等待的延迟初始化,请执行以下操作:

private AtomicReference<Foo> instance = new AtomicReference<>(null);  

public Foo getInstance() {
   Foo foo = instance.get();
   if (foo == null) {
       foo = new Foo();                       // create and initialize actual instance
       if (instance.compareAndSet(null, foo)) // CAS succeeded
           return foo;
       else                                   // CAS failed: other thread set an object 
           return instance.get();             
   } else {
       return foo;
   }
}

除了一件事之外,它的效果非常好:如果两个线程看到实例null,它们都会创建一个新对象,只有一个幸运的是通过CAS操作设置它,这会浪费资源.

有没有人建议另一种无锁延迟初始化模式,这会降低两个并发线程创建两个昂贵对象的可能性？

从理论上讲,应该至少可以强制验证无锁算法(只有很多函数调用的组合相交).是否有任何工具或形式推理过程可用于实际证明无锁算法是正确的(理想情况下它还应该能够检查竞争条件和ABA问题)？

注意:如果您知道一种方法来证明一点(例如,只证明它对ABA问题是安全的)或者我没有提到的问题,那么无论如何都要发布解决方案.在最坏的情况下,可以依次完成每种方法以完全验证它.

我读到某处(无法找到页面)锁定免费数据结构对于"某些工作负载"更有效,这似乎意味着有时它们实际上更慢或者在某些情况下它们的增益可能为零.对锁定指令进行大约100次循环命中来执行原子操作听起来要快得多,而不是等待调度程序将进程重新唤醒,所以对于我来说,在什么情况下无锁数据结构不是很明显比旧式的互斥体更不可取.如果锁定在99%的时间内可用且进程无需进入休眠状态,那么互斥锁会更快吗？假设有合适的无锁数据结构,是否有一个很好的经验法则可以知道哪条路可用？

我需要实现一个无锁的跳过列表.我试图找文件.不幸的是,我发现的是无锁单链表(多种口味).但是如何实现无锁跳过列表？

我有类似的东西:

if (f = acquire_load() == ) {
   ... use Foo
}

和:

auto f = new Foo();
release_store(f)

您可以很容易地想象使用atomic with load(memory_order_acquire)和store(memory_order_release)的acquire_load和release_store的实现.但是现在如果release_store是用_mm_stream_si64实现的,这是一个非临时写入,而不是针对x64上的其他商店进行排序的？如何获得相同的语义？

我认为以下是最低要求:

atomic<Foo*> gFoo;

Foo* acquire_load() {
    return gFoo.load(memory_order_relaxed);
}

void release_store(Foo* f) {
   _mm_stream_si64(*(Foo**)&gFoo, f);
}

并使用它:

// thread 1
if (f = acquire_load() == ) {
   _mm_lfence(); 
   ... use Foo
}

和:

// thread 2
auto f = new Foo();
_mm_sfence(); // ensures Foo is constructed by the time f is published to gFoo
release_store(f)

那是对的吗？我非常肯定这里绝对需要sfence.但是那个lfence怎么样？是否需要或者简单的编译器障碍对于x64是否足够？例如asm volatile("":::"memory").根据x86内存模型,负载不会与其他负载重新排序.所以根据我的理解,只要存在编译器障碍,acquire_load()必须在if语句中的任何加载之前发生.

有趣的是,我发现很多程序员错误地认为"无锁"只意味着"没有互斥的并发编程".通常,还存在一个相关的误解,即编写无锁代码的目的是为了获得更好的并发性能.当然,无锁的正确定义实际上是关于进度保证.无锁算法保证至少一个线程能够前进,无论其他线程正在做什么.

这意味着无锁算法永远不会有一个代码,其中一个线程依赖于另一个线程才能继续.例如,无锁代码不能具有线程A设置标志的情况,然后线程B在等待线程A取消设置标志时保持循环.这样的代码基本上实现了一个锁(或者我称之为伪装的互斥锁).

然而,其他情况更微妙,在某些情况下我真的无法确定算法是否符合无锁定的要求,因为"取得进步"的概念有时对我来说似乎是主观的.

其中一个例子是(备受好评的,afaik)并发库liblfds.我正在研究liblfds中多生产者/多消费者有界队列的实现 - 实现非常简单,但我无法确定它是否应该符合无锁定条件.

相关算法在lfds711_queue_bmm_enqueue.c.Liblfds使用自定义原子和内存障碍,但算法很简单,我可以用段落左右来描述.

队列本身是一个有界的连续数组(ringbuffer).共享read_index和write_index.队列中的每个插槽都包含一个用户数据字段和一个sequence_number值,它基本上类似于一个纪元计数器.(这避免了ABA问题).

PUSH算法如下:

1. 原子上负载 write_index
2. 尝试write_index % queue_size使用试图设置write_index为的CompareAndSwap循环在队列中保留一个插槽write_index + 1.
3. 如果CompareAndSwap成功,则将用户数据复制到保留的插槽中.
4. 最后,sequence_index通过使其等于来更新插槽write_index + 1.

实际的源代码使用自定义原子和内存障碍,因此为了进一步清楚这个算法,我简要地将它翻译成(未经测试的)标准C++原子以获得更好的可读性,如下所示:

bool mcmp_queue::enqueue(void* data)
{
    int write_index = m_write_index.load(std::memory_order_relaxed);

    for (;;)
    {
        slot& s = m_slots[write_index % m_num_slots];
        int sequence_number = s.sequence_number.load(std::memory_order_acquire);
        int difference = sequence_number - write_index;

        if (difference == 0)
        {
            if (m_write_index.compare_exchange_weak(
                write_index,
                write_index + …