c++ - 并发队列内存消耗爆炸，然后程序崩溃

Question

这是一个使用VS 2010并发队列的典型生产者/消费者模式，问题是当我运行程序时，内存消耗超过1GB然后程序崩溃，有人可以指出这段代码中的问题吗？

#include <iostream>
#include <fstream>
#include <string>
#include <cstdlib> 
#include <ctime> 

#include <boost\shared_ptr.hpp>
#include <boost\thread.hpp>
#include <concurrent_queue.h>



void wait2(int milliseconds)
{
    boost::this_thread::sleep(boost::posix_time::milliseconds(milliseconds)); 
}

class CQueue
{
    Concurrency::concurrent_queue<int>  Q;

    boost::mutex                m;
    boost::condition_variable   cv;

public:

    CQueue():QValue(-1)
    {
    }

    int QRead()
    {
        while(Q.empty())
        {
            boost::unique_lock<boost::mutex> lk(m);
            cv.wait(lk);
        }

        int res;
        if(Q.try_pop(res))
        {
            QValue = res;
            return true;
        }
        return false;
    }

    void QWrite(int i)
    {
        Q.push(i);
        cv.notify_one();
    }

    int QValue;
};

CQueue myqueue;

void write()
{
    int i = 0;
    while(true)
    {
        myqueue.QWrite(++i);
    }
}


void read()
{
    while(true)
    {
        if( myqueue.QRead())
            std::cout << myqueue.QValue << std::endl;
        else
            std::cout << "failed to read" << std::endl;
    }
}
void main ()
{

    boost::thread w(write);
    boost::thread r(read);

    w.join();
    r.join();

}

Answer 1

我在简单的双核上使用 VS'13 和 Boost 1.52 构建并测试了您的代码。

正如已经说过的，由于您的生产者-消费者设计没有定义当库存(并发队列)达到给定水平时阻止生产者的阈值，因此生产者在队列中推送了太多数据，因此内存不足增加，窗口开始交换，卡住，如果超过最大提交大小，进程可能会崩溃，等等......

请注意，提交大小限制取决于多个因素，编译器、编译器选项、程序运行的操作系统，...

因此，在下面我添加了一种方法，如果队列大小达到阈值，则阻止生产者，如果队列大小低于阈值，则消费者唤醒生产者。

通过这些更改，我们添加了一些同步，这可以限制一点并行性，但使用中的内存受到控制。

#include <iostream>
#include <fstream>
#include <string>
#include <cstdlib> 
#include <ctime> 

#include "..\..\..\boost\boost\shared_ptr.hpp"
#include "..\..\..\boost\boost\thread.hpp"

#include <concurrent_queue.h>

#define STOCK_THRESHOLD 1000

void wait2(int milliseconds)
{
    boost::this_thread::sleep(boost::posix_time::milliseconds(milliseconds)); 
}

class CQueue
{
    Concurrency::concurrent_queue<int>  Q;

    boost::mutex                consumerMutex;
    boost::condition_variable   consumerCV;

    boost::mutex                producerMutex;
    boost::condition_variable   producerCV;

public:

    CQueue():QValue(-1)
    {
    }

    int QRead()
    {
        while(Q.empty())
        {
            boost::unique_lock<boost::mutex> lk(consumerMutex);
            consumerCV.wait(lk);
        }

        int res;
        if(Q.try_pop(res))
        {
            QValue = res;
            if(Q.unsafe_size() <= STOCK_THRESHOLD)
            {
                producerCV.notify_one();
            }
            return true;
        }
        return false;
    }

    void QWrite(int i)
    {
        while(Q.unsafe_size() > STOCK_THRESHOLD){
            boost::unique_lock<boost::mutex> lk(producerMutex);
            producerCV.wait_for(lk, boost::chrono::milliseconds(10));
        }
        Q.push(i);
        consumerCV.notify_one();
    }

    int QValue;
};

CQueue myqueue;

void write()
{
    int i = 0;
    while(true)
    {
        myqueue.QWrite(++i);

    }
}


void read()
{
    while(true)
    {
        if( myqueue.QRead())
            std::cout << myqueue.QValue << std::endl;
        else
            std::cout << "failed to read" << std::endl;
    }
}

void main ()
{

    boost::thread w(write);
    boost::thread r(read);

    w.join();
    r.join();

}