c ++中的细粒度锁定队列

Question

这是Anthony Williams在6.2.3 C++ Concurrency in Action中引入的细粒度锁定队列.

/*
    pop only need lock head_mutex and a small section of tail_mutex,push only need
    tail_mutex mutex.maximum container concurrency.
*/
template<typename T> class threadsafe_queue
{
    private:
    struct node
    {
        std::shared_ptr<T> data;
        std::unique_ptr<node> next;
    }
    std::mutex head_mutex;   //when change the head lock it.
    std::unique_ptr<node> head;  
    std::mutex tail_mutex;   //when change the tail lock it.
    node* tail;
    std::condition_variable data_cond;

    node* get_tail()
    {
        std::lock_guard<std::mutex> tail_lock(tail_mutex);
        return tail;
    }

    public:
    /* 
        create a dummy node
    */
    threadsafe_queue():
        head(new node),tail(head.get())
    {}

    std::shared_ptr<T> wait_and_pop()
    {
        std::unique_lock<std::mutex> head_lock;
        data_cond.wait(head_lock,[&]{return head.get()!=get_tail();}); //#1
        std::unique_ptr<node> old_head=std::move(head);
        head=std::move(old_head->next);
        return old_head;
    }

    void push(T new_value)
    {
        std::shared_ptr<T> new_data(
        std::make_shared<T>(std::move(new_value)));
        std::unique_ptr<node> p(new node);
        {
            std::lock_guard<std::mutex> tail_lock(tail_mutex);
            tail->data=new_data;
            node* const new_tail=p.get();
            tail->next=std::move(p);
            tail=new_tail;
        }
        data_cond.notify_one();
    }
}

情况如下:有两个线程(thread1和thread2).thread1正在做wait_and_pop和thread2正在做push.队列是空的.

thread1在#2中,之前已经检查head.get()!=get_tail()过了data_cond.wait().此时它的CPU周期已用完.thread2开始.

thread2完成了push功能并做到了data_cond.notify_one().thread1再次开始.

现在thread1开始data_cond.wait(),但它永远等待.

这种情况可能会发生吗？如果是这样,如何修复这个容器？

Answer 1

是的,OP中描述的情况是可能的,并且会导致通知丢失.在谓词函数中注入一个很好的大时间延迟可以很容易地触发.这是Coliru的演示.注意程序如何完成10秒(超时的长度wait_for)而不是100毫秒(生产者在队列中插入项目的时间).通知丢失.

在条件变量的设计中隐含的假设是条件的状态(谓词的返回值)在关联的互斥锁被锁定时不能改变.对于此队列实现,情况并非如此,因为push可以在不保持的情况下更改队列的"空" head_mutex.

§30.5p3指定wait有三个原子部分:

1. 释放互斥锁,进入等待状态;
2. 解除等待; 和
3. 重新获取锁定.

请注意,这些都没有提到检查谓词(如果有的话)wait.wait使用谓词的行为在§30.5.1p15中描述:

功效:

while (!pred())
      wait(lock);

请注意,这里不能保证谓词检查和wait原子执行.有是一个先决条件lock被锁定,它相关的调用线程持有互斥.

至于修复容器以避免丢失通知,我会将其更改为单个互斥实现并完成它.这是一个有点夸张的称之为细粒度锁时,push和pop两个最终以相同的互斥(tail_mutex)反正.