c++ - 确定删除并发队列的安全性

Question

我正在用 C++ 为实时系统编写无锁单消费者单生产者可增长队列。内部队列可以工作，但它需要是可增长的。生产者线程是实时的，因此任何操作都需要是确定性的(因此没有等待、锁、内存分配)，而消费者线程则不是。

因此，如果需要，消费者线程偶尔会增加队列的大小。队列的实现使得消费者端无法增长。因此，实际队列间接包装在一个调度调用的对象中，实际增长是通过将对内部队列的引用交换到一个新队列来实现的，同时保持旧队列在生产者线程使用它的情况下挂起。

然而，问题是，我不知道如何证明生产者线程何时停止使用旧队列，因此可以安全地删除而不必求助于锁。这是代码的伪表示:

template<typename T>
class queue
{
public:

    queue()
        : old(nullptr)
    {
        current.store(nullptr);
        grow();
    }

    bool produce(const T & data)
    {
        qimpl * q = current.load();
        return q->produce(data);
    }

    bool consume(T & data)
    {
        // the queue has grown? if so, a new and an old queue exists. consume the old firstly.
        if (old)
        {
            // here is the problem. we never really know when the producer thread stops using 
            // the old queue and starts using the new. it could be concurrently halfway-through inserting items
            // now, while the following consume call fails meanwhile.
            // thus, it is not safe yet to delete the old queue.
            // we know however, that it will take at most one call to produce() after we called grow()
            // before the producer thread starts using the new queue.

            if (old->consume(data))
            {
                return true;
            }
            else
            {
                delete old;
                old = nullptr;
            }
        }

        if (current.load()->consume(data))
        {
            return true;
        }
        return false;
    }
    // consumer only as well
    void grow()
    {
        old = current.load();
        current.store(new qimlp());
    }
private:

    class qimpl
    {
    public:
        bool produce(const T & data);
        bool consume(const T & data);
    };

    std::atomic<qimpl *> current;
    qimpl * old;
};

请注意，ATOMIC_POINTER_LOCK_FREE == 2 是代码编译的条件。我看到的唯一可证明的条件是，如果调用 grow()，下一个 produce() 调用将使用新的内部队列。因此，如果每次调用都会增加 produce 中的原子计数，那么删除 N + 1 处的旧队列是安全的，其中 N 是调用 grow() 时的计数。然而，问题是您随后需要自动交换新指针并存储计数，这似乎是不可能的。

欢迎任何想法，作为引用，系统将如何工作:

queue<int> q;

void consumer()
{
    while (true)
    {
        int data;

        if (q.consume(data))
        {
            // ..
        }
    }

}

void producer()
{
    while (true)
    {
        q.produce(std::rand());
    }
}

int main()
{
    std::thread p(producer); std::thread c(consumer);
    p.detach(); c.detach();
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
}

编辑:好的，现在问题解决了。我突然意识到，当一个项目被推送到新队列时，旧队列已经过时了。因此，该片段现在看起来像这样:

bool pop(T & data)
{
    if (old)
    {
        if (old->consume(data))
        {
            return true;
        }

    }
    // note that if the old queue is empty, and the new has an enqueued element, we can conclusively
    // prove that it is safe to delete the old queue since it is (a) empty and (b) the thread state 
    // for the producer is updated such that it uses all the new entities and will never use the old again.
    // if we successfully dequeue an element, we can delete the old (if it exists).
    if (current.load()->consume(data))
    {
        if (old)
        {
            delete old;
            old = nullptr;
        }
        return true;
    }
    return false;
}

Answer 1

我不完全理解 grow() 在你的算法中的用法，但你似乎需要某种读取-复制-更新 (RCU) 机制来安全删除队列不需要。

这article描述了与 Linux 相关的这种机制的不同风格，但您可以谷歌 RCU 风格，适用于其他平台。