我最近用 C++ 实现了一个公平的读写器票证自旋锁。代码相当简单，我认为它运行良好。我已经将自旋锁集成到一个更大的应用程序中，我注意到在极少数情况下，代码运行非常缓慢，而大多数时候，它运行得非常快。我知道这是由于自旋锁引起的，因为如果我立即用一个简单的读写器自旋锁替换它(不公平且没有票)，代码突然运行得更快。它在不同的机器上发生了几次。我知道如果你用比内核更多的线程运行这些锁，它们会运行得很慢，但我在一台有 48 个内核的机器上用 16 个线程运行它。我无法在具有 4 个线程和 4 个内核的笔记本电脑上重现该问题。这是代码:

    inline size_t rndup(size_t v) {

        v--;
        v |= v >> 1;
        v |= v >> 2;
        v |= v >> 4;
        v |= v >> 8;
        v |= v >> 16;
        v |= v >> 32;
        v++;

        return v;
    }    

    class SpinLockRW_MCS {

        public:

            SpinLockRW_MCS(const size_t nb_readers) :   writer(nullptr), lock_pool(nullptr), it_lock_pool(0),
                                                        load_lock_pool(0), mask_it(rndup(2 * nb_readers + 1) - 1),
                                                        padding1{0}, padding2{0}, padding3{0}, padding4{0} {

                if (nb_readers <= std::thread::hardware_concurrency()){

                    lock_pool = new Lock[mask_it + 1];
                    lock_pool[0].is_locked = false;
                }
            }

            ~SpinLockRW_MCS() {

                clear();
            }

            inline void clear() {

                if (lock_pool != nullptr){

                    delete[] lock_pool;
                    lock_pool = nullptr;
                }

                writer = nullptr;

                it_lock_pool = 0;
                load_lock_pool = 0;
            }

            inline void acquire_reader() {

                uint_fast32_t retry = 0;

                const size_t prev_reader_id = it_lock_pool.fetch_add(1) & mask_it;
                const size_t new_reader_id = (prev_reader_id + 1) & mask_it;

                while (lock_pool[prev_reader_id].is_locked){

                    if (++retry > 100) this_thread::yield();
                }

                ++load_lock_pool;

                lock_pool[prev_reader_id].is_locked = true;
                lock_pool[new_reader_id].is_locked = false;
            }

            inline void release_reader() {

                --load_lock_pool;
            }

            inline void acquire_writer() {

                uint_fast32_t retry = 0;

                const size_t prev_reader_id = it_lock_pool.fetch_add(1) & mask_it;
                const size_t new_reader_id = (prev_reader_id + 1) & mask_it;

                while (lock_pool[prev_reader_id].is_locked){

                    if (++retry > 100) this_thread::yield();
                }

                while (load_lock_pool){

                    if (++retry > 100) this_thread::yield();
                }

                lock_pool[prev_reader_id].is_locked = true;

                writer = &lock_pool[new_reader_id];
            }

            inline void release_writer() {

                writer->is_locked = false;
            }

            inline void release_writer_acquire_reader() {

                ++load_lock_pool;

                writer->is_locked = false;
            }

        private:

            struct Lock {

                std::atomic<bool> is_locked;
                const int padding[15];

                Lock() : is_locked(true), padding{0} {}
            };

            Lock* writer;
            const int padding1[14];
            Lock* lock_pool;
            const int padding2[14];
            const size_t mask_it;
            const int padding3[14];
            std::atomic<size_t> it_lock_pool;
            const int padding4[14];
            std::atomic<size_t> load_lock_pool;
    };

如有任何建议，我们将不胜感激!谢谢!

最佳答案

在没有更多细节的情况下评估问题有点困难，但这是我在黑暗中拍摄的照片:我怀疑在您的场景中读者需要非常频繁地获取锁(否则，您可能最好使用传统锁) .这是您的问题:

任何一个线程都能够饿死所有其他线程。

这对读者和作者都是正确的，而在非公平算法中通常只对作者有效。当您有多个读者排队等待读取访问时，您的情况就会出现问题。每个线程将等待前面的锁变得可用 (while (lock_pool[prev_reader_id].is_locked) ...)。如果他们能得到那个锁，一切都很好，但是一旦一个线程不能得到它，你就会遇到麻烦。所有阅读器线程都排队等待看到它们的前任线程翻转到 false。他们每个人都依赖于他们的直接前任。

现在想象第一个读者无法获得锁。它会继续旋转一段时间并最终 yield()。这实际上意味着您的线程现在不再运行。操作系统将其从调度队列中删除，并且它不会运行很长的时间(它们的其余时间片，与完成 100 次旋转所花费的时间相比要长) .因此，整个等待线程链很可能会进入 yield。

最终，第一个线程等待的标志将翻转为 false。但是您的调度程序现在陷入了困境。它周围有一堆线程，但它们在再次进入 yield 之前只旋转了很短的时间。这里的期望是，除了链中的第一个线程之外的所有线程，如果它们被选中，它们几乎肯定注定要休眠一个完整的时间片。因此，如果等待线程链中较早的线程发生这种情况，您也会谴责链中的所有其他线程至少等待同样长的时间。

您在这里玩的是概率游戏，随着队列中读者数量的增加，您获胜的几率会显着降低。这就是为什么当从 4 线程移动到 16 线程时问题变得更糟的原因。特别是，一旦达到新读者到达队列所需的平均时间大致与线程在队列中移动所需时间的顺序相同，您将很难返回再次进入空队列。这并非不可能，因为我们在这里讨论的是多个时间片，这会将您带入数十到数百毫秒的数量级。

这是公平调度算法中的典型权衡。公平是有代价的，在这种情况下，单个读者可以阻止所有人。因为如果你设法先接到获取电话，我的读者永远不会超过你的读者，如果你不继续前进，我将不得不永远等待。此问题的一种解决方案是向调度程序提供每个线程正在等待的额外信息，以便它更有可能以正确的顺序唤醒它们。另一种方法是选择更适合您的特定场景的不同算法。

关于c++ - C++ 中的公平读写器票证自旋锁运行缓慢，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/55697595/