使用boost :: thread的Actor计算模型

Question

我正在尝试使用boost :: thread在C++上的线程上实现Actor计算模型.但程序在执行期间抛出奇怪的异常.异常不稳定,有时程序以正确的方式工作.

我的代码:

actor.hpp

class Actor {

  public:
    typedef boost::function<int()> Job;

  private:
    std::queue<Job>             d_jobQueue;
    boost::mutex                d_jobQueueMutex;
    boost::condition_variable   d_hasJob;
    boost::atomic<bool>         d_keepWorkerRunning;
    boost::thread               d_worker;

    void workerThread();

  public:
    Actor();
    virtual ~Actor();

    void execJobAsync(const Job& job);

    int execJobSync(const Job& job);
};

actor.cpp

namespace {

int executeJobSync(std::string          *error,
                   boost::promise<int> *promise,
                   const Actor::Job     *job)
{
    int rc = (*job)();

    promise->set_value(rc);
    return 0;
}

}

void Actor::workerThread()
{
    while (d_keepWorkerRunning) try {
        Job job;
        {
            boost::unique_lock<boost::mutex> g(d_jobQueueMutex);

            while (d_jobQueue.empty()) {
                d_hasJob.wait(g);
            }

            job = d_jobQueue.front();
            d_jobQueue.pop();
        }

        job();
    }
    catch (...) {
        // Log error
    }
}

void Actor::execJobAsync(const Job& job)
{
    boost::mutex::scoped_lock g(d_jobQueueMutex);
    d_jobQueue.push(job);
    d_hasJob.notify_one();
}

int Actor::execJobSync(const Job& job)
{
    std::string error;
    boost::promise<int> promise;
    boost::unique_future<int> future = promise.get_future();

    {
        boost::mutex::scoped_lock g(d_jobQueueMutex);
        d_jobQueue.push(boost::bind(executeJobSync, &error, &promise, &job));
        d_hasJob.notify_one();
    }

    int rc = future.get();

    if (rc) {
        ErrorUtil::setLastError(rc, error.c_str());
    }

    return rc;
}

Actor::Actor()
: d_keepWorkerRunning(true)
, d_worker(&Actor::workerThread, this)
{
}

Actor::~Actor()
{
    d_keepWorkerRunning = false;
    {
        boost::mutex::scoped_lock g(d_jobQueueMutex);
        d_hasJob.notify_one();
    }
    d_worker.join();
}

实际上抛出的异常是boost :: thread_interrupted int rc = future.get();.但形式提升文档我不能推理这个例外.文件说

抛出: - 如果与*this关联的结果在调用点没有准备好,并且当前线程被中断,则boost :: thread_interrupted.

但我的工作线程不能处于中断状态.

当我使用gdb并设置"catch throw"时,我看到后面的跟踪看起来像

抛出thread_interrupted

提高::详细:: interruption_checker :: check_for_interruption

提高::详细:: interruption_checker :: interruption_checker

提高:: condition_variable ::等待

提高::详细:: future_object_base :: wait_internal

提高::详细:: future_object_base ::等待

提高::详细:: future_object ::得到

提高:: unique_future ::得到

我查看了boost源但无法理解为什么interruption_checker决定工作线程被中断.

所以有人C++大师,请帮助我.我需要做些什么来获得正确的代码？我正在使用:

提升1_53

Linux版本2.6.18-194.32.1.el5 Red Hat 4.1.2-48

gcc 4.7

编辑

固定它!感谢Evgeny Panasyuk和Lazin.问题出在TLS管理中.boost :: thread和boost :: thread_specific_ptr正在使用相同的TLS存储.在我的情况下,当他们都试图在创建时更改此存储时出现问题(不幸的是我没有明白为什么会发生这种情况).所以TLS被破坏了.

我用__thread指定的变量替换了我的代码中的boost :: thread_specific_ptr.

Offtop:在调试过程中我发现外部库中的内存损坏并修复了它=)

.

编辑2 我遇到了确切的问题......这是GCC中的一个错误=)_GLIBCXX_DEBUG编译标志打破了ABI.你可以看看关于boost bugtracker的讨论:https: //svn.boost.org/trac/boost/ticket/7666

Answer 1

我发现了几个错误:

Actor::workerThread功能可以双重解锁d_jobQueueMutex.首先解锁是手动d_jobQueueMutex.unlock();,第二个是析构函数boost::unique_lock<boost::mutex>.

您应该阻止解锁之一,例如:和之间的释放关联:unique_lockmutex

g.release(); // <------------ PATCH
d_jobQueueMutex.unlock();

或者添加其他代码块+ default-construct Job.

有可能workerThread永远不会留下以下循环:

while (d_jobQueue.empty()) {
    d_hasJob.wait(g);
}

想象一下以下情况:d_jobQueue是空的,Actor::~Actor()被调用,它设置标志并通知工作线程:

d_keepWorkerRunning = false;
d_hasJob.notify_one();

workerThread 在while循环中唤醒,看到该队列为空并再次睡眠.

通常的做法是发送特殊的最终作业来停止工作线程:

~Actor()
{
    execJobSync([this]()->int
    {
        d_keepWorkerRunning = false;
        return 0;
    });
    d_worker.join();
}

在这种情况下,d_keepWorkerRunning不需要是原子的.

Coliru上的现场演示

编辑:

我在您的示例中添加了事件队列代码.

您在这两个具有并发队列EventQueueImpl和Actor,但对于不同的类型.可以将公共部分提取到concurrent_queue<T>适用于任何类型的单独实体中.在一个地方调试和测试队列比捕获分散在不同类上的错误要容易得多.

所以,你可以尝试使用它concurrent_queue<T>(在Coliru上)