原子线程计数器

Question

我正在尝试使用C++ 11原子基元来实现各种原子" 线程计数器 ".基本上,我有一个关键的代码部分.在此代码块中,任何线程都可以从内存中自由读取.但是,有时,我想执行重置或清除操作,这会将所有共享内存重置为默认初始化值.

这似乎是一个使用读写锁的好机会.C++ 11不包含开箱即用的读写互斥锁,但可能更简单.我认为这个问题是一个很好的机会,可以更熟悉C++ 11原子基元.

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

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

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

3. 如果线程希望将所有变量重置为默认值,则它必须原子等待计数器为0,然后原子地将其设置为某个特殊的"清除标志"值,执行清除,然后将计数器重置为0.

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

因此,我刚才描述的算法可以用三个函数实现.increment_thread_counter()必须始终在进入临界区之前调用第一个函数.decrement_thread_counter()在离开临界区之前,必须始终调用第二个函数.最后,只有当线程计数器== 0时,clear()才能从临界区外部调用该函数.

这就是我想出的:

鉴于:

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是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));
}