生产者-消费者模式

Question

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

我们假设有三个规则需要遵守。

1. 生产者不能在桶里添加产品 当桶里装满了产品的时候
2. 当桶里的产品空了的时候，消费者不能从桶里获取产品。
3. 生产和消费不能同时进行。换句话说，这两个动作是异步的。

现在我有什么。

1. a int 变量，用于存储桶中产品的数量。
2. a const int 变量，表示桶的容量，在我的代码中，它的值是5。
3. a int 缄默变量，初始值为1。
4. a 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

这是一个预期的无限循环。product-count会在4和5的时候流转，因为我设置了producer的睡眠时间比consumer多一点。

但这是意想不到的结果。

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

正如我们所看到的, 暂停的线程向量的大小从0到了2, 这就排除了1.

这是否只是一个巧合，两个线程同时被唤醒然后发生冲突？

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

我在测试的时候也遇到了一个问题。我怎样才能得到线程，然后把它存储在向量中呢？

我不知道用 OpenThread 来做这件事。我的课程要求我使用系统调用，所以我没有包括以下内容 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方法向队列添加一个元素并触发条件。