消费者和生产者问题的双缓冲

Question

所以我正在尝试为典型的生产者和消费者问题实现双缓冲区。

1.get_items() 基本上一次产生10个项目。

2.producer 基本上将 10 个项目推送到写入队列中。假设目前我们只有一个生产者。

3.consumers 将消费队列中的一项。有很多消费者。

所以我将我的代码分享如下。实现思路很简单，从 readq 消费直到它为空，然后交换队列指针，readq 现在将指向 writeq，而 writeq 现在将指向已清空的队列并开始再次填充它。因此生产者和消费者可以独立工作而不会相互停止。这种以空间换时间的方式。

但是，我的代码不适用于多个消费者案例。在我的代码中，我启动了 10 个消费者线程，它总是卡在 .join() 处。

所以我认为我的代码肯定有问题。但是，通过仔细检查，我没有找到那个错误在哪里。并且似乎代码在lk1.unlock()之后卡住了，所以它没有卡在一段时间或其他明显的地方。

mutex m1;
mutex m2; // using 2 mutex, so when producer is locked, consumer can still run


condition_variable put;
condition_variable fetch;

queue<int> q1;
queue<int> q2;

queue<int>* readq = &q1;
queue<int>* writeq = &q2;

bool flag{ true };
vector<int> get_items() {
    vector<int> res;
    for (int i = 0; i < 10; i++) {
        res.push_back(i);
    }
    return res;
}

void producer_mul() {
    unique_lock<mutex> lk2(m2);
    put.wait(lk2, [&]() {return flag == false; }); //producer waits for consumer signal
    vector<int> items = get_items();
    for (auto it : items) {
        writeq->push(it);
    }
    flag = true; //signal queue is filled
    fetch.notify_one();
    lk2.unlock();
}


int get_one_item_mul() {
    unique_lock<mutex> lk1(m1);
    int res;
    if (!(*readq).empty()) {   
        res = (*readq).front(); (*readq).pop();  
        if ((*writeq).empty() && flag == true) { //if writeq is empty
            flag = false;
            put.notify_one();
        }
    }
    else {
        readq = writeq; // swap queue pointer
        while ((*readq).empty()) { // not yet write
            if (flag) {
                flag = false;  
                put.notify_one();//start filling process
            }
            //if (readq->empty()) {       //upadted due to race. readq now points to writeq, so if producer finished, readq is not empty and flag = true.
                fetch.wait(lk1, [&]() {return flag == true; }); 
            //}
        }
        if (flag) {
            writeq = writeq == &q1 ? &q2 : &q1; //swap the writeq to the alternative queue and fill it again
            flag = false;
            //put.notify_one(); //fill that queue again if needed. but in my case, 10 item is produced and consumed, so no need to use the 2nd round, plus the code does not working in this simple case..so commented out for now.
        }
        res = readq->front(); readq->pop();
    }
    lk1.unlock();
    this_thread::sleep_for(10ms);
    return res;
}



int main()
{
    std::vector<std::thread> threads;
    std::packaged_task<void(void)> job1(producer_mul);
    vector<std::future<int>> res;

    for (int i = 0; i < 10; i++) {
        std::packaged_task<int(void)> job2(get_one_item_mul);
        res.push_back(job2.get_future());
        threads.push_back(std::thread(std::move(job2)));
    }

    threads.push_back(std::thread(std::move(job1)));

    for (auto& t : threads) {
        t.join();
    }

    for (auto& a : res) {
        cout << a.get() << endl;
    }

    return 0;

}

我添加了一些 cmets，但想法和代码非常简单且不言自明。

我正在尝试找出问题出在我的代码中的位置。它适用于多个消费者吗？此外，如果这里有多个生产者，它是否有效？我没有看到问题，因为基本上在代码中锁不是细粒度的。 Producer 和 Consumer 从头到尾都被锁定。

期待讨论，感谢任何帮助。

更新

根据其中一个答案更新了竞态条件。 该程序仍然无法运行。

Answer 1

您的程序包含数据竞争，因此表现出未定义的行为。我至少看到两个：

• producer_mul 访问和修改flag，同时持有m2 互斥锁，但不持有m1。 get_one_item_mul 访问和修改 flag，同时持有 m1 互斥锁，但不持有 m2。所以flag 实际上并没有防止并发访问。

• 类似地，producer_mul 访问 writeq 指针，同时持有 m2 互斥体，但不持有 m1。 get_one_item_mul 修改 writeq 同时持有 m1 mutex 但不修改 m2。

---

队列本身也存在数据竞争。最初，两个队列都是空的。 producer_mul 被阻止等待 flag。然后发生以下顺序（P 为生产者线程，C 为消费者线程）：

C: readq = writeq;  // Both now point to the same queue
C: flag = false; put.notify_one();  // This wakes up producer
  **P: writeq->push(it);
  **C: if (readq->empty())

最后两行同时发生，没有针对并发访问的保护。一个线程修改std::queue 实例，而另一个线程访问同一实例。这是一场数据竞赛。

---

设计的核心是数据竞赛。假设只有一个生产者P 和两个消费者C1 和C2。最初，P 等待put 直到flag == false。 C1抢m1； C2 在 m1 上被阻止。

C1 设置readq = writeq，然后解除阻止P1，然后调用fetch.wait(lk1, [&]() {return flag == true; });。 这将解锁m1，允许C2 继续。所以现在P 正忙着写信给writeq 而C2 正忙着从readq 读取——这是同一个队列。