使用boost线程和循环缓冲区挂起的生产者/消费者

Question

我想到了.愚蠢的错误,我实际上并没有从队列中删除元素,我只是阅读第一个元素.我修改了代码,下面的代码没有用.谢谢大家的帮助.

我正在尝试使用boost实现生产者消费者问题,这实际上是更大项目的一部分.我已经从互联网上的例子中实现了一个程序,甚至我在这里找到了一些帮助.但是目前我的代码只是挂起.基于一些好的建议,我决定使用boost ciruclar缓冲区来保存生产者和消费者之间的数据.那里有很多相似的代码,我能够汇集那些想法并自己写点东西.但是,我似乎仍然遇到与以前相同的问题(这是我的程序只是挂起).我以为我没有像以前那样犯同样的错误..

我的代码在下面给出,我已经取出了我之前的代码,我只是我自己的链接列表.

缓冲区头:

#ifndef PCDBUFFER_H
#define PCDBUFFER_H

#include <pcl/io/pcd_io.h>
#include <boost/thread/mutex.hpp>
#include <boost/thread/condition.hpp>
#include <boost/circular_buffer.hpp>

class pcdBuffer
{
    public:
        pcdBuffer(int buffSize);
        void put(int data);
        int get();
        bool isFull();
        bool isEmpty();
        int getSize();
        int getCapacity();
    private:
        boost::mutex bmutex;
        boost::condition_variable buffEmpty;
        boost::condition_variable buffFull;
        boost::circular_buffer<int> buffer;
};


#endif

缓冲源(仅相关部分):

#include "pcdBuffer.h"
#include <iostream>

//boost::mutex io_mutex;

pcdBuffer::pcdBuffer(int buffSize)
{
    buffer.set_capacity(buffSize);
}

void pcdBuffer::put(int data)
{
    {
        boost::mutex::scoped_lock buffLock(bmutex);
        while(buffer.full())
        {
            std::cout << "Buffer is full" << std::endl;
            buffFull.wait(buffLock);
        }
        buffer.push_back(data);
    }
    buffEmpty.notify_one();
}

int pcdBuffer::get()
{
    int data;
    {
        boost::mutex::scoped_lock buffLock(bmutex);
        while(buffer.empty())
        {
            std::cout << "Buffer is empty" << std::endl;
            buffEmpty.wait(buffLock);
        }
        data = buffer.front();
            buffer.pop_front();
    }
    buffFull.notify_one();
    return data;
}

代码的主要驱动程序:

#include <iostream>
#include <boost/thread/mutex.hpp>
#include <boost/thread/thread.hpp>
#include <unistd.h>
#include "pcdBuffer.h"

pcdBuffer buff(100);

void producer()
{
    int i = 10;
    while (true)
    {
        buff.put(i);
        i++;
    }
}

void consumer()
{
    int i;
    while(true)
    {
        i = buff.get();
        std::cout << "Data: " << i << std::endl;
    }
}

int main(int argc, char** argv)
{
    std::cout << "Starting main...." << std::endl;
    std::cout << "Buffer Details: " << std::endl;
    std::cout << "Capacity: " << buff.getCapacity() << ", isEmpty: " << buff.isEmpty() << ", isFull: " << buff.isFull() << std::endl;
    boost::thread cons(consumer);
    sleep(5);
    boost::thread prod(producer);
    prod.join();
    cons.join();
    return 0;
}

我的缓冲区容量被正确初始化为100.消费者线程等待并报告"缓冲区为空"5秒钟,但之后我只是从put方法得到"缓冲区已满"而"数据:10"来自消费者功能在stdout上交替出现.正如你所看到的那样,10是我放入的第一个元素.似乎缓冲区正在填满并且没有通知消费者,但我检查了我的锁,认为它们是正确的.非常感谢任何帮助.

以下是我编写此代码的参考链接:

http://www.boost.org/doc/libs/1_49_0/libs/circular_buffer/doc/circular_buffer.html#classboost_1_1circular__buffer_19ba12c0142a21a7d960877c22fa3ea00

http://www.drdobbs.com/article/print?articleId=184401518&siteSectionName=cpp

线程安全执行循环缓冲区

Answer 1

首先,不要写成自己的列表,你可以只包std::list在pcdQueue写自己的,而不是.这是正确的,原则上不std::list是线程安全的,但无论如何,您在类中提供了必要的同步原语.

您的程序挂起的原因:您保持锁定并填充队列,直到它满了.您对消费者通过的通知notify_one是无用的,因为您的消费者将再次锁定,因为互斥锁已被占用(通过生产者中的锁定).

当您最终通过等待释放锁定(当队列已满)时 condition_variable,您不会唤醒您的消费者,因此您的消费者和生产者都被阻止并且您的程序挂起.

将其更改为:

void pcdQueue::produce()
{
    int i=0;
    while(true)
    {
        {
            boost::mutex::scoped_lock lock(qmutex);
            while( ! qlen < buffSize ) {
                std::cout << "Queue is full" << std::endl;
                full.wait(lock);
            }

            enqueue(i); // or myList.push_back(i) if you switch to std::list
        }

        empty.notify_one();


    }
}

您的方法中存在相同的问题consume().将其更改为:

pcdFrame* pcdQueue::consume()
{
    pcdFrame *frame;

    {
        boost::mutex::scoped_lock lock(qmutex);
        while( qlen == 0 ) {
            std::cout << "Queue is empty" << std::endl;
            empty.wait(lock);
        }

        frame = dequeue();
    }
    full.notify_one();

    return frame;
}

一般情况下,请注意并注意,如果有人在等待,通知只会产生影响.否则,他们就"迷失"了.此外,请注意,在调用时不需要保持互斥锁被锁定notify_one(实际上,它可能会导致额外的上下文切换开销,因为您唤醒另一个线程,然后等待当前(仍然)的互斥锁锁定(由你).首先,释放互斥锁,然后告诉另一个线程继续.

请注意,两个线程都无限运行,因此您的主程序仍将在第一个挂起join()并且永不退出.你可以在你的while-loops中包含一个停止标志,告诉你的线程完成.