c++ - 如何使用 Boost 在 C++ 中通过上下文切换执行两个任务

Question

这是我的第一个问题，所以我希望我能在规则之后描述我的问题，并希望你能理解。

我尝试在 MySQL 中编写一些附加功能。为此，我需要同步查询任务。

开始时，两个任务都将等待执行。一个任务是生产者(主)，另一任务只是消费者(从)。

当我想切换任务时，每个查询都会进入方法execQuery。关于QueueElement e，他们彼此了解。

如果它有一个从属对象，它本身就是主对象，如果它有一个主对象，它本身就是从属对象，在其他情况下，从属对象或主对象是0。 因此，我通过第三个线程启动主任务。当它完成第一次计算时，它会进入 execQuery 方法并开始执行从站并在那里等待。从机启动并读取主机的数据(不在代码中)，并进入 execQuery 方法并再次启动主机并等待，依此类推。如果所有数据都计算并读取完毕，则结束。 通常它是这样工作的。但有时我会遇到以下问题:

主节点进入execQuery方法并执行slave->cond.notify_all();因此从节点任务启动并执行所有内容并进入execQuery 方法并在主任务执行下一行之前执行 master->cond.notify_all();:e->cond.wait(lock); > 所以从机没有唤醒任何东西，然后两个线程都在休眠。因为这里的slave任务太快了。

我需要的是一个解决方案，如何安排该任务唤醒另一个任务并在另一个任务开始执行之前开始等待。有解决办法吗？

void QueryQueue::execQuery(JOIN* toSearch) {
if (QueryExport::isSystemQuery(toSearch)) {
    return;
}
QueueElement* e = getQueryElement(toSearch);
BOOST_LOG_TRIVIAL(info) << "Enter execMultiQuery with id: " << e->join->thd->query_id;
if (e) {
    if (e->slave) {
        QueueElement* slave = e->slave;
        BOOST_LOG_TRIVIAL(info) << "Wait for Query " << e->join->thd->query_id << " and start " << slave->join->thd->query_id << " (Slave) ";



        boost::unique_lock<boost::mutex> lock(e->mut);

        slave->cond.notify_all();
        e->cond.wait(lock);
        BOOST_LOG_TRIVIAL(info) << "Wake up Query " << e->join->thd->query_id << " (Master) ";

    }
    if (e->master) {
        QueueElement* master = e->master;
        BOOST_LOG_TRIVIAL(info) << "Wait for Query " << e->join->thd->query_id << " and start " << master->join->thd->query_id << " (Master) ";


        boost::unique_lock<boost::mutex> lock(e->mut);

        master->cond.notify_all();            
        e->cond.wait(lock);
        BOOST_LOG_TRIVIAL(info) << "Wake up Query " << e->join->thd->query_id << " (Slave) ";
    }
}

}

Answer 1

我找到了解决问题的方法。所以我想把解决方案分享给大家。可能它帮助了其他人。

我需要一个新变量来保存状态，并且工作线程(从属线程)需要在工作时保存主线程的互斥体。所以在从属部分中必须有一个新的锁部分。

我发现一个论坛也有类似的问题:http://boost.2283326.n4.nabble.com/Deadlock-issue-with-boost-condition-td2554763.html

从我的形式来看，这个问题更容易理解。但这是同样的问题。

void QueryQueue::execQuery(JOIN* toSearch) {
if (QueryExport::isSystemQuery(toSearch)) {
    return;
}
QueueElement* e = getQueryElement(toSearch);
BOOST_LOG_TRIVIAL(info) << "Enter execMultiQuery with id: " << e->join->thd->query_id;
if (e) {
    if (e->slave) {
        QueueElement* slave = e->slave;
        BOOST_LOG_TRIVIAL(info) << "Wait for Query " << e->join->thd->query_id << " and start " << slave->join->thd->query_id << " (Slave) ";

        slave->cond.notify_all();


        boost::unique_lock<boost::mutex> lock(e->mut);


        e->cond.wait(lock);
        while(!e->worker_thread_finished ) e->cond.wait(lock);
        e->worker_thread_finished = false;   
        BOOST_LOG_TRIVIAL(info) << "Wake up Query " << e->join->thd->query_id << " (Master) ";

    }
    if (e->master) {
        QueueElement* master = e->master;
        BOOST_LOG_TRIVIAL(info) << "Wait for Query " << e->join->thd->query_id << " and start " << master->join->thd->query_id << " (Master) ";

        {
            boost::unique_lock<boost::mutex> lock(e->master->mut,boost::try_to_lock);
            if (!lock.owns_lock()) {
                BOOST_LOG_TRIVIAL(warning) << "Can't lock the Query with ID " << e->master->join->thd->query_id << " (Master) ";
            }
            master->worker_thread_finished = true;
            master->cond.notify_one();
        }

        boost::unique_lock<boost::mutex> lock(e->mut);

        e->cond.wait(lock);
        BOOST_LOG_TRIVIAL(info) << "Wake up Query " << e->join->thd->query_id << " (Slave) ";
    }
}

}