c++ - concurrent_queue内存消耗爆炸，然后程序崩溃

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();

}