0

我有一个生产者/消费者的尝试

制片人

#pragma once
#ifndef PRODUCER_H
#define PRODUCER_H

#include <thread>
#include "Mailbox.h"

class Producer
{
private:
    std::thread producer;
    Mailbox& mailbox;
public:
    Producer(Mailbox& newmailbox);
    ~Producer();
    void start();
    void run();
};

Producer::Producer(Mailbox& newMailbox) : mailbox(newMailbox) {}

Producer::~Producer() {}

void Producer::start()
{
    producer = std::thread(&Producer::run, this);
}

void Producer::run()
{
    mailbox.inc();
}

#endif

消费者

#pragma once
#ifndef CONSUMER_H
#define CONSUMER_H

#include "Mailbox.h"
#include <thread>
#include <iostream>

class Consumer
{
private:
    Mailbox& mailbox;
    std::thread consumer;
public:
    Consumer(Mailbox& newMailbox);
    ~Consumer();
    void start();
    void run();
};

Consumer::Consumer(Mailbox& newMailbox) : mailbox(newMailbox) {}


Consumer::~Consumer() {}

void Consumer::start()
{
    consumer = std::thread(&Consumer::run, this);
}

void Consumer::run()
{
    mailbox.read();
}

#endif

邮箱

#pragma once
#ifndef MAILBOX_H
#define MAILBOX_H

#include <mutex>
#include <iostream>

class Mailbox
{
private:
    int& mailbox;
    int init_val;
    std::mutex mmutex;
    std::condition_variable condition;
public:
    Mailbox();
    ~Mailbox();
    void inc();
    void read();
};

Mailbox::Mailbox() : mailbox(init_val), init_val(0) {}

Mailbox::~Mailbox()
{

}

void Mailbox::inc()
{   
    int count = 0;
    while (count < 10)
    {

        std::unique_lock<std::mutex> lock(mmutex);
        std::cout << "Producer increment\n";
        mailbox += 1;
        lock.unlock();

        count += 1;
    }

}

void Mailbox::read()
{
    int count = 0;
    while (count < 10)
    {

        std::unique_lock<std::mutex> lock(mmutex);

        condition.wait(lock, [this](){return get_cflag(); });

        condition.notify_one();

        count += 1;

    }

}

#endif

主要的

int main()
{

    Mailbox* mailbox = new Mailbox();
    Consumer* consumer = new Consumer(*mailbox);
    Producer* producer = new Producer(*mailbox);

    consumer->start();
    producer->start();

    return 0;
}

互斥锁虽然是异步工作的，因为我无法控制 astd::thread何时开始，所以我决定使用std::unique_lock除了std::mutex.

问题是，消费者等待，生产者在没有通知的情况下继续前进，至少这是调试器告诉我的，最后一次生产者迭代结果是 abort() 所以这里出了点问题。

4

2 回答 2

0

根据 David Schwartz 的评论、Mike Strobel 的见解以及其他研究，我更改了生产者和消费者函数

制片人

void Mailbox::inc()
{   
    int count = 0;
    while (count < 10)
    {

        std::unique_lock<std::mutex> lock(mmutex);
        std::cout << "Producer increment\n";
        mailbox += 1;
        lock.unlock();
        set_cflag(true); // signal to the consumer data is ready
        condition.notify_one();

        {
            std::unique_lock<std::mutex> lock(mmutex);
            condition.wait(lock, [this]() {return get_pflag(); });
        }

        set_pflag(false);

        count += 1;
    }

}

消费者

 void Mailbox::read()
{
    int count = 0;
    while (count < 10)
    {

        std::unique_lock<std::mutex> lock(mmutex);

        condition.wait(lock, [this](){return get_cflag(); });

        std::cout << "Consumer: " << mailbox << "\n";
        lock.unlock();

        set_pflag(true);
        condition.notify_one();

        count += 1;

        set_cflag(false);

    }

}

邮箱

class Mailbox
{
private:
    int& mailbox;
    int cflag, pflag;
    int init_val;
    std::mutex mmutex;
    std::condition_variable condition;
public:
    Mailbox();
    ~Mailbox();
    int get_cflag() { return cflag; }
    void set_cflag(int newFlag) { cflag = newFlag; }
    int get_pflag() { return pflag; }
    void set_pflag(int newFlag) { pflag = newFlag; }
    void inc();
    void read();
};

Mailbox::Mailbox() : mailbox(init_val), init_val(0), cflag(0), pflag(0) {}

Mailbox::~Mailbox()
{

}

执行时的输出是我想要的

int main()
{

    Mailbox* mailbox = new Mailbox();
    Consumer* consumer = new Consumer(*mailbox);
    Producer* producer = new Producer(*mailbox);

    consumer->start();
    producer->start();

    fgetc(stdin);

    return 0;
}

生产者增量

消费者：1

生产者增量

消费者：2

生产者增量

消费者：3

生产者增量

消费者：4

生产者增量

消费者：5

生产者增量

消费者：6

生产者增量

消费者：7

生产者增量

消费者：8

生产者增量

消费者：9

生产者增量

消费者：10

于 2017-10-14T16:57:33.397 回答
0

我不是 C++ 人，但如果这些条件变量按照我认为的方式工作，那么只有在您等待时信号到达时您才会收到通知。如果信号在您开始等待之前到达，您将无限期地阻塞。

在“Mailbox::read”中获得锁后，您应该检查项目是否可用，如果没有，则仅等待条件变量。如果有，请继续并采取它：

int Mailbox::read()
{
    std::unique_lock<std::mutex> lock(m);
    while (mailbox <= 0)
        condition.wait(lock);
    return mailbox--;
}
于 2017-10-13T00:01:55.120 回答