c++ - 生产者-消费者模型

Question

我正在尝试模拟关于多线程的生产者-消费者模型。

我们假设要遵守三个规则:

当桶装满产品时，生产者不能将产品添加到桶中。

当桶为空时，消费者无法从桶中获取产品。

生产和消费不能同时进行。换句话说，这两个 Action 是异步的。

现在我有什么:

一个 int存储桶中产品数量的变量

一个 const int存储桶容量的变量，在我的代码中值为 5。

一个 int互斥量变量，初始值为1

一个 vector<HANDLE>挂起句柄的变量，并且会有一个函数调度这些挂起的线程。

结果是:有时它工作得很好，但有时它变成了死锁。代码和结果如下:

代码:

#include <iostream>
#include <Windows.h>
#include <vector>

using namespace std;

// the count of product, the initial value is 0
int product_count = 0;

const int product_capacity = 5;

int mutex = 1;

vector<HANDLE> suspendedHandleVector;

HANDLE GetCurrentRealHandle() {
    HANDLE realHandle = 0;
    return OpenThread(THREAD_ALL_ACCESS, TRUE, GetCurrentThreadId());
}

void ThreadScheduling() {
    if (suspendedHandleVector.size() > 0) {
        HANDLE handle = suspendedHandleVector[0];
        suspendedHandleVector.erase(suspendedHandleVector.begin());
        ResumeThread(handle);
    }
}

void P() {
    --mutex;
    if (mutex < 0) {
        auto handle = GetCurrentRealHandle();
        suspendedHandleVector.push_back(handle);
        SuspendThread(handle);
    }
}

void V() {
    ++mutex;
    if (mutex >= 0) {
        ThreadScheduling();
    }
}

DWORD WINAPI ProducerThread(LPVOID param) {
    while (true) {
        P();
        if (product_count == product_capacity) {
            V();
            continue;
        }
        ++product_count;
        cout << "I'm producer, and there are " << product_count << " products now" << endl;
        V();
        Sleep(100);
    }
    return 0;
}

DWORD WINAPI ConsumerThread(LPVOID param) {
    while (true) {
        P();
        if (product_count == 0) {
            V();
            continue;
        }
        --product_count;
        cout << "I'm consumer, and there are " << product_count << " products rest now" << endl;
        V();
        Sleep(150);
    }
    return 0;
}

void main() {
    auto producer_handle = CreateThread(nullptr, 0, ProducerThread, nullptr, 0, nullptr);
    auto consumer_handle = CreateThread(nullptr, 0, ConsumerThread, nullptr, 0, nullptr);
    while (true) {
        cout << suspendedHandleVector.size() << endl; // This is for debugging
        Sleep(100);
    }
}

按预期工作时的结果:

0
I'm producer, and there are 1 products now
I'm consumer, and there are 0 products rest now
0
I'm producer, and there are 1 products now
I'm consumer, and there are 0 products rest now
0
I'm producer, and there are 1 products now
0
I'm consumer, and there are 0 products rest now
I'm producer, and there are 1 products now
0
I'm producer, and there are 2 products now
I'm consumer, and there are 1 products rest now

这是一个预期的无限循环。产品计数将在 4 和 5 处流动，因为我将生产者的 sleep 时间设置为比消费者多一点。

但这是意想不到的结果:

I'm producer, and there are 5 products now
I'm consumer, and there are 4 products rest now
0
I'm consumer, and there are 4 products rest now
I'm producer, and there are 5 products now
0
0
I'm consumer, and there are 4 products rest now
I'm producer, and there are 5 products now
0
2
2
2
2
2
2

如我们所见，挂起的线程 vector 的大小从 0 到了 2，其中遗漏了 1。

这只是两个线程同时唤醒然后发生冲突的巧合吗？

我认为我的代码有问题，需要您的帮助。

在测试过程中我也遇到了一个问题:如何获取线程，然后将其存储在 vector 中。

我不确定使用 OpenThread去做这个。我的类(class)要求我使用系统调用，所以我没有包含 thread头文件。

感谢您的帮助!

Answer 1

在生产者-消费者问题中，存在三个参与者:生产者、消费者和用于存储生产数据的缓冲区。生产者和消费者通常生活在一个或多个线程中(多生产者单消费者单生产者多消费者多生产者多消费者)。

现在这种模式非常常用，因为它是避免显式同步的方法之一。

正如我所说，消费者和消费者通常生活在一个或多个线程中。缓冲区通常是一个同步队列(可以是锁定队列或无锁队列)。

一个非常简短的片段:

SyncronizedQueue queue;

void consumer()
{
  while(1) {
    queue.push(std::rand());
  }
}

void producer()
{
    while(1)
    {
      auto product = queue.pop();   // blocks until there are elements in the 
    }
}

void main()
{
  std::thread producerThread{producer};
  std::thread consumerThread{consumer};

  producerThread.join();
  consumerThread.join();
}

所以任务是写队列，这样是安全的 push元素和解锁 pop当一个新元素被推送时。

一种简单的实现方法是使用条件变量，因此 pop函数被阻塞，直到触发条件并且 push 方法将元素添加到队列并触发条件。