2

我想做多线程，其中线程 ONE 将数据传递给 4-5 个处理数据的工作线程，所有工作线程都已完成我想继续。我正在使用 boost 来实现，但是我有一个同步问题。意味着程序在某一时刻停止并且不再继续工作。

我以前使用过 OpenMP，效果很好，但我想单独设置线程优先级，但我不知道如何使用 OpenMP 做到这一点，因此我研究了自己的解决方案：

如果有人可以提供提示以找到此代码中的错误或可以帮助我找到解决问题的另一种方法，我将非常高兴。

谢谢你，KmgL

#include <QCoreApplication>

#include <boost/thread.hpp>

#define N_CORE 6
#define N_POINTS 10
#define N_RUNS 100000

class Sema{

public:
    Sema(int _n =0): m_count(_n),m_mut(),m_cond(){}

    void set(int _n)
    {
        boost::unique_lock<boost::mutex> w_lock(m_mut);
        m_count = -_n;
    }

    void wait()
    {
        boost::unique_lock<boost::mutex> lock(m_mut);
        while (m_count < 0)
        {
            m_cond.wait(lock);
        }
        --m_count;
    }
    void post()
    {
        boost::unique_lock<boost::mutex> lock(m_mut);
        ++m_count;
        m_cond.notify_all();
    }


private:
    boost::condition_variable m_cond;
    boost::mutex m_mut;
    int m_count;

};

class Pool
{
private:
    boost::thread m_WorkerThread;
    boost::condition_variable m_startWork;
    bool m_WorkerRun;
    bool m_InnerRun;
    Sema * m_sem;

    std::vector<int> *m_Ep;
    std::vector<int>  m_ret;

    void calc()
    {
        unsigned int    no_pt(m_Ep->size());                
        std::vector<int> c_ret;
        for(unsigned int i=0;i<no_pt;i++)
            c_ret.push_back(100 + m_Ep->at(i));

        m_ret = c_ret;
    }
    void run()
    {
        boost::mutex WaitWorker_MUTEX;
        while(m_WorkerRun)
        {
            boost::unique_lock<boost::mutex> u_lock(WaitWorker_MUTEX);
            m_startWork.wait(u_lock);
            calc();
            m_sem->post();
        }

    }

public:
    Pool():m_WorkerRun(false),m_InnerRun(false){}
    ~Pool(){}
    void start(Sema * _sem){
        m_WorkerRun = true;
        m_sem = _sem;
        m_ret.clear();
        m_WorkerThread = boost::thread(&Pool::run, this);}
    void stop(){m_WorkerRun = false;}
    void join(){m_WorkerThread.join();}

    void newWork(std::vector<int> &Ep)
    {
        m_Ep = &Ep;
        m_startWork.notify_all();
    }
    std::vector<int> getWork(){return m_ret;}



};

int main(int argc, char *argv[])
{
    QCoreApplication a(argc, argv);

    Pool TP[N_CORE];

    Sema _sem(0);
    for(int k=0;k<N_CORE;k++)
        TP[k].start(&_sem);


    boost::this_thread::sleep(boost::posix_time::milliseconds(10));

    std::vector<int> V[N_CORE];

    for(int k=0;k<N_CORE;k++)
        for(int i=0;i<N_POINTS;i++)
        {
            V[k].push_back((k+1)*1000+i);
        }

    for(int j=0;j<N_RUNS;j++)
    {
        _sem.set(N_CORE);
        for(int k=0;k<N_CORE;k++)
        {
            TP[k].newWork(V[k]);
        }

        _sem.wait();

        for(int k=0;k<N_CORE;k++)
        {
            V[k].clear();
            V[k]=TP[k].getWork();
            if(V[k].size()!=N_POINTS)
                std::cout<<"ERROR: "<<"V["<<k<<"].size(): "<<V[k].size()<<std::endl;

        }
       if((j+1)%100==0)
            std::cout<<"LOOP: "<<j+1<<std::endl;
    }
    std::cout<<"FINISHED: "<<std::endl;

    return a.exec();
}
4

2 回答 2

0

这可以通过Boost futures来完成。启动线程，然后等待所有线程完成。无需其他同步。

于 2013-05-15T18:45:00.553 回答
0

Pool::newWork()您在对和的调用之间存在竞争Pool::run()。

您必须记住，信号/广播条件变量不是粘性事件。如果您的线程在发出信号时没有等待条件变量，则信号将丢失。这就是您的程序中可能发生的情况：没有什么可以阻止您的主线程Pool::newWork()在每个 Pool 对象有时间调用wait()您的条件变量之前调用它们。

要解决这个问题，您需要boost::mutex WaitWorker_MUTEX作为类成员移动，而不是作为局部变量。 Pool::newWork()在进行更新之前需要获取该互斥锁：

boost::unique_lock<boost::mutex> u_lock(WaitWorker_MUTEX);
m_Ep = &Ep;
m_startWork.notify(); // no need to use notify_all()

由于您在 中使用条件变量Pool::run()，因此您需要处理虚假唤醒。我建议在构造对象时以及每次完成工作项时将 m_Ep 设置为 NULL：

boost::unique_lock<boost::mutex> u_lock(WaitWorker_MUTEX);
while (1) {
    while (m_Ep == NULL && m_workerRun) {
        m_startWork.wait(u_lock);
    }
    if (!m_workerRun) {
        return;
    }
    calc();
    m_sem->post();
    m_Ep = NULL;
}

stop() 将需要获取互斥锁和 notify()：

boost::unique_lock<boost::mutex> u_lock(WaitWorker_MUTEX);
m_workRun = false;
m_startWork.notify();

这些更改应该使您不必要的 10 毫秒睡眠。你似乎没有打电话Pool::stop()或Pool::join()。您应该更改代码以调用它们。

与最后复制结果相比，m_ret您还可以获得更好的性能。Pool::calc()当你返回作品时，你也在做副本。您可能希望Pool::getWork()将 const ref 返回到m_ret.

我没有运行此代码，因此可能存在其他问题。它应该可以帮助您移动

从您的代码看来，您可能想知道为什么条件变量需要与互斥锁齐头并进（因为您在 中声明了一个本地互斥锁Pool::run()）。我希望我的修复让它更清楚。

于 2013-05-15T22:43:04.563 回答