如何在 C 中实现无锁共享标志？

Posted 2023-02-16

【中文标题】如何在 C 中实现无锁共享标志？

【英文标题】：How to implement a lock-free shared flag in C?

【发布时间】：2012-10-08 19:45:50

【问题描述】：

我有一个生产者一消费者模型，我需要生产者在数据可用时设置一个标志。我怀疑我可以在没有锁定共享标志的情况下逃脱，因为：

生产者在设置之前从不检查值 偶尔错过标志更新不是问题（尽管我也可以使用原子操作来避免这种情况？）。

所以我的问题是，我该如何实现呢？我对 volatile 关键字的理解，以及 __sync_synchronize() 之类的东西充其量是微不足道的，所以假设我知道的很少。具体来说，我希望能够确保及时在其他线程中看到对标志的更改。

编辑：我在 Linux 上使用 GCC。

【问题讨论】：

测试和设置指令？ penguin.cz/~literakl/intel/b.html#BTS 使用一点内联汇编应该不会太难。

不知道。这是否允许我假设其他线程中的可见性？还是允许编译器在生产者线程中使用标志的本地副本？此外，GCC 具有用于原子操作的内置函数。

它是为此目的而设计的，所以它应该很快。理论上。我从来不需要编写自己使用它的软件，我只知道它存在。

在 C11 中，使用 atomic_flag 变量类型和 atomic_flag_test_and_set 函数。详见 7.17.8.1。

@KerrekSB，太好了，我以前没见过。谢谢。有没有办法在 C99 中模拟它，我认为我使用的不是最新版本的 GCC。

【参考方案1】：

使用两个变量：

volatile size_t elements_produced; // producer increments it when data is available
volatile size_t elements_consumed; // consumer increments it

新数据恰好在elements_produced != elements_consumed 时可用。 如果您需要无限量，那么在更新时添加它。

produce(...) 
    elements_produced = (elements_produced + 1) % (max_elements_in_queue + 1);


consume(...) 
    elements_consumed = (elements_consumed + 1) % (max_elements_in_queue + 1);

不需要锁或原子。

这是单生产者、单消费者循环环形缓冲区的标准实现。

【讨论】：

"no need for locks or atomics" 这是不正确的。当同时从两个线程调用produce 时，这里仍然存在并发问题。每个线程都可以为elements_produced 读取相同的初始值。第一个线程的输出将被第二个线程覆盖，最终结果是变量增加一次而不是两次。这只有在您可以确保一次只有一个线程调用该函数时才有效，这将需要某种锁定。

@bta 除了在OP的情况下，只有一个生产者和一个消费者。

@bta 没错，我的错。除非 OP 可以指定准确性，否则我通常不在乎我是否因“检查周期”而落后于此类通知。只有读取是并发执行的。

@loan- 我们知道生产者和消费者都是单线程的吗？

不需要模组。由于size_t 是无符号的，溢出的行为是明确定义的（免费模组！）。只要“队列中”的项目数小于SIZE_MAX，那么elements_produced - elements_produced 将是队列中的项目数，无论是否发生回绕（即溢出）。

【参考方案2】：

原子操作意味着（非常短暂地）阻塞其他原子操作，直到第一个操作完成。

我假设我们正在谈论线程，因此我们应该考虑互斥体（但这也适用于进程和信号量）。无需尝试获取锁即可检查（读取）互斥锁（或信号量）。

如果互斥体（信号量）的状态是已锁定，则继续其他操作，稍后再试。

【参考方案3】：

实际上不可能以便携的方式做到这一点。

但是，您可以使用各种编译器内部函数来实现它。

例如，对于 x86(-64) 上的 gcc，可能至少 ARM：

static int queued_work;

static void inc_queued_work()

    (void)__sync_add_and_fetch( &queued_work, 1 );


/*
  Decrement queued_work if > 0.
  Returns 1 if queued_work was non-equal to 0 before
  this function was called.
*/
static int dec_queued_work()

    /* Read current value and subtract 1.
       If the result is equal to -1, add 1 back and return 0.
    */
    if( __sync_sub_and_fetch( &queued_work, 1 ) == -1 )
    
        __sync_fetch_and_add( &queued_work, 1 );
        return 0;
    
    return 1;

一些 CPU:s 只支持 compare_and_swap。 您也可以使用该 intrisic 来实现它：

/* Alternative solution using compare_and_swap  */
void inc_queued_work()

    do 
        int queued = queued_work;
        /* Try to write queued-1 to the variable. */
        if( __sync_bool_compare_and_swap( &queued_work,
                                         queued, queued+1 ) )
            return;
     while( 1 );


int dec_queued_work()

    do 
        int queued = queued_work;
        if( !queued ) return 0;
        /* Try to write queued-1 to the variable. */
        if( __sync_bool_compare_and_swap( &queued_work, 
                                         queued, queued-1 ) )
            return queued;
     while( 1 );

即使您有多个写入者和读取者，这些功能也应该可以工作。 和朋友一起用google 'sync_add_and_fetch' 会给你很多参考文档

【参考方案4】：

如果没有锁，两个线程将永远不会同步，消费者将永远等待一个永远不会改变的值（因为该值被缓存，因此永远不会发生内存 put/fetch）。

总而言之，您必须 a) 确保内存是从生产者那里写入的，并且 b) 确保内存是由消费者读取的。这正是锁的作用，这就是你应该使用它们的原因。如果你对锁死了，你可以使用像 sleep 这样的函数，它可以保证唤醒后的缓存状态。