c++ - boost asio 需要在 m 个工作完成后才发布 n 个工作

Question

我正在寻找一种方法来等待多个作业完成，然后执行另一个完全不同数量的作业。当然是线程。简要说明:我创建了两个工作线程，都在 io_service 上运行。以下代码摘自here .

为了简单起见，我创建了两种类型的作业，CalculateFib 和 CalculateFib2。我希望 CalculateFib2 作业在且仅在 CalculateFib 作业完成后开始。我尝试按照说明使用条件变量 here , 但如果 CalculateFib2 作业不止一个，程序就会挂起。我做错了什么？

谢谢，多多

#include <boost/asio.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/bind.hpp>
#include <iostream>

boost::mutex global_stream_lock;
boost::mutex mx;
boost::condition_variable cv;

void WorkerThread( boost::shared_ptr< boost::asio::io_service > io_service)
{
    global_stream_lock.lock();
    std::cout << "[" << boost::this_thread::get_id()
        << "] Thread Start" << std::endl;
    global_stream_lock.unlock();

    io_service->run();

    global_stream_lock.lock();
    std::cout << "[" << boost::this_thread::get_id()
        << "] Thread Finish" << std::endl;
    global_stream_lock.unlock();
}

size_t fib( size_t n )
{
    if ( n <= 1 )
    {
        return n;
    }
    boost::this_thread::sleep( 
        boost::posix_time::milliseconds( 1000 )
        );
    return fib( n - 1 ) + fib( n - 2);
}

void CalculateFib( size_t n )
{
    global_stream_lock.lock();
    std::cout << "[" << boost::this_thread::get_id()
        << "] Now calculating fib( " << n << " ) " << std::endl;
    global_stream_lock.unlock();

    size_t f = fib( n );

    global_stream_lock.lock();
    std::cout << "[" << boost::this_thread::get_id()
        << "] fib( " << n << " ) = " << f << std::endl;
    global_stream_lock.unlock();

    boost::lock_guard<boost::mutex> lk(mx);
    cv.notify_all();
}

void CalculateFib2( size_t n )
{
    boost::unique_lock<boost::mutex> lk(mx);
    cv.wait(lk);

    global_stream_lock.lock();
    std::cout << "[" << boost::this_thread::get_id()
        << "] Now calculating fib2( " << n << " ) " << std::endl;
    global_stream_lock.unlock();

    size_t f = fib( n );

    global_stream_lock.lock();
    std::cout << "[" << boost::this_thread::get_id()
        << "] fib2( " << n << " ) = " << f << std::endl;
    global_stream_lock.unlock();
}
int main( int argc, char * argv[] )
{
    boost::shared_ptr< boost::asio::io_service > io_service(
        new boost::asio::io_service
        );
    boost::shared_ptr< boost::asio::io_service::work > work(
        new boost::asio::io_service::work( *io_service )
        );

    global_stream_lock.lock();
    std::cout << "[" << boost::this_thread::get_id()
        << "] The program will exit when all work has finished."
        << std::endl;
    global_stream_lock.unlock();

    boost::thread_group worker_threads;
    for( int x = 0; x < 2; ++x )
    {
        worker_threads.create_thread( 
            boost::bind( &WorkerThread, io_service )
            );
    }
    io_service->post( boost::bind( CalculateFib, 5 ) );
    io_service->post( boost::bind( CalculateFib, 4 ) );
    io_service->post( boost::bind( CalculateFib, 3 ) );

    io_service->post( boost::bind( CalculateFib2, 1 ) );
    io_service->post( boost::bind( CalculateFib2, 1 ) );
    work.reset();
    worker_threads.join_all();

    return 0;
}

Answer 1

在 CalculateFib2 中，您要做的第一件事是等待条件 (cv)。此条件仅在CalculateFib 结束时发出信号。因此，按理说，执行永远不会继续，除非条件被触发(通过发布 CalculateFib)作业。

确实，像这样添加任何其他行:

io_service->post( boost::bind( CalculateFib, 5 ) );
io_service->post( boost::bind( CalculateFib, 4 ) );
io_service->post( boost::bind( CalculateFib, 3 ) );

io_service->post( boost::bind( CalculateFib2, 1 ) );
io_service->post( boost::bind( CalculateFib2, 1 ) );

io_service->post( boost::bind( CalculateFib, 5 ) );   // <-- ADDED

使执行运行完成。

为了阐明更多信息:如果您像这样(及时)隔离 Fib2 批处理

io_service->post( boost::bind( CalculateFib, 5 ) );
io_service->post( boost::bind( CalculateFib, 4 ) );
io_service->post( boost::bind( CalculateFib, 3 ) );

boost::this_thread::sleep(boost::posix_time::seconds( 10 ));
io_service->post( boost::bind( CalculateFib2, 1 ) );
io_service->post( boost::bind( CalculateFib2, 1 ) );
io_service->post( boost::bind( CalculateFib2, 1 ) );
io_service->post( boost::bind( CalculateFib2, 1 ) );
io_service->post( boost::bind( CalculateFib2, 1 ) );
io_service->post( boost::bind( CalculateFib2, 1 ) );

所有 Fib2 作业将始终阻塞，无论线程数如何，因为 Fib 作业在发布之前已全部退出。一个简单的

io_service->post( boost::bind( CalculateFib, 1 ) );

将解锁所有等待者(即只有等待线程的数量，即可用线程数 -1，因为 Fib() 作业也占用一个线程。现在有 <7 个线程会死锁，因为甚至没有可用的线程来启动 Fib() 作业(所有线程都被阻塞等待在 Fib2 中)

---

老实说，我不明白您在日程安排方面想要达到的目标。我怀疑您应该监视作业队列并仅在达到所需的项目数量时才明确发布作业(“任务”)。这样您就可以 KISS 并获得一个非常灵活的作业调度界面。

通常，对于线程组(池化)，您希望避免无限期地阻塞线程。这有可能使您的工作安排陷入僵局，否则表现不佳。