c++ - 原子线程计数器

Question

我正在尝试使用 C++11 原子原语来实现各种原子“线程计数器”。基本上，我有一个代码的关键部分。在这个代码块中，任何线程都可以自由地从内存中读取。然而，有时，我想做一个重置或清除操作，将所有共享内存重置为默认的初始化值。

这似乎是一个使用读写锁的好机会。 C++11 不包括开箱即用的读写互斥锁，但也许可以做一些更简单的事情。我认为这个问题将是一个很好的机会来更加熟悉 C++11 原子原语。

所以我想了一会儿这个问题，在我看来我所要做的就是:

1. 每当线程进入临界区时，递增一个原子计数器变量

2. 每当线程离开临界区时，递减原子计数器变量

3. 如果一个线程希望重置所有变量为默认值，它必须原子地等待计数器为 0，然后自动将其设置为一些特殊的“清除标志”值，执行清除，然后将计数器重置为 0。

4. 当然，希望递增和递减计数器的线程还必须检查清除标志。

所以，我刚才描述的算法可以用三个函数来实现。第一个函数，increment_thread_counter()必须始终在进入临界区之前调用。第二个功能，decrement_thread_counter() , 必须始终在离开关键部分之前立即调用。最后，函数 clear()可以从临界区外部调用仅当且仅当线程计数器 == 0。

这是我想出的:

给定:

1. 线程计数器变量，std::atomic<std::size_t> thread_counter
2. 常数clearing_flag设置为 std::numeric_limits<std::size_t>::max()

...

void increment_thread_counter()
{
    std::size_t expected = 0;
    while (!std::atomic_compare_exchange_strong(&thread_counter, &expected, 1))
    {
        if (expected != clearing_flag)
        {
            thread_counter.fetch_add(1);
            break;
        }
        expected = 0;
    }
}

void decrement_thread_counter()
{
    thread_counter.fetch_sub(1);
}

void clear()
{
    std::size_t expected = 0;
    while (!thread_counter.compare_exchange_strong(expected, clearing_flag)) expected = 0;

    /* PERFORM WRITES WHICH WRITE TO ALL SHARED VARIABLES */

    thread_counter.store(0);
}

据我所知，这应该是线程安全的。请注意 decrement_thread_counter函数不应该需要任何同步逻辑，因为它是给定的 increment()总是在 decrement() 之前调用.所以，当我们到达 decrement() , thread_counter 永远不能等于 0 或 clearing_flag .

无论如何，由于 THREADING IS HARD™，而且我不是无锁算法方面的专家，所以我不完全确定该算法是无竞争条件的。

问题:这段代码线程安全吗？这里可能存在任何竞争条件吗？

Answer 1

您有竞争条件；如果另一个线程更改 increment_thread_counter() 对 clearing_flag 的测试和 fetch_add 之间的计数器，就会发生不好的事情。

我认为这个经典的 CAS 循环应该工作得更好:

void increment_thread_counter()
{
    std::size_t expected = 0;
    std::size_t updated;
    do {
        if (expected == clearing_flag) {     // don't want to succeed while clearing, 
             expected = 0;      //take a chance that clearing completes before CMPEXC
        }

        updated = expected + 1;
        // if (updated == clearing_flag) TOO MANY READERS!
    } while (!std::atomic_compare_exchange_weak(&thread_counter, &expected, updated));
}