如何使 boost::thread_group 在其线程中具有 boost::asio::io_service::run 更小？

Question

我看到的一般是common way to create thread pools via "io_service + thread_group"。它非常适合 const 大小的线程池。或者只能变大的水池。但我想知道如何在不停止所有 io_service 的情况下使这样的池更小？

所以我们有shown

// class variables
asio::io_service io_service;
boost::thread_group threads;
asio::io_service::work *work;

// some pool init function
work = new asio::io_service::work(io_service);
int cores_number = boost::thread::hardware_concurrency(); 
  for (std::size_t i = 0; i < cores_number; ++i)
    threads.create_thread(boost::bind(&asio::io_service::run, &io_service));

// and now we can simply post tasks
io_service.post(boost::bind(&class_name::an_expensive_calculation, this, 42));
io_service.post(boost::bind(&class_name::a_long_running_task, this, 123));

// and it is really eazy to make pool biger - just call (mutexes may be required)
threads.create_thread(boost::bind(&asio::io_service::run, &io_service));

但是如果我们想从我们的线程池中删除线程呢？我们不能只是简单地调用threads.remove_thread(thread* thrd);，因为它不会停止在其中运行&asio::io_service::run（恕我直言）所以我想知道 - 是否有可能以及如何真正从这样的池中删除线程？ （不仅仅是打断 tham，而是等到当前线程任务超出范围）？

更新：

这里有一些简单的可编译代码：线程池，线程所需的生命周期。

#include <stdio.h>
#include <iostream>
#include <fstream>

//Boost
#include <boost/asio.hpp>
#include <boost/thread.hpp>
#include <boost/thread/locks.hpp>

boost::asio::io_service io_service;
boost::asio::io_service::work *work;
boost::thread_group threads;
boost::mutex threads_creation;
int time_limit;

int calculate_the_answer_to_life_the_universe_and_everything(int i)
{
    boost::this_thread::sleep(boost::posix_time::milliseconds(i));
    std::cout << i << std::endl;
    return i;
}

void run(boost::shared_ptr<boost::thread> thread_ptr)
{
    try
    {
        io_service.run();
    }
    catch(std::exception &e)
    {
        std::cout << "exeption: " << e.what() << std::endl;
        boost::mutex::scoped_lock lock(threads_creation);
        threads.remove_thread(thread_ptr.get());
        lock.unlock();
        std::cout << "thread removed from group" << std::endl;
        return;
    }

}

void pool_item( int i)
{
    boost::packaged_task<int> pt(boost::bind(calculate_the_answer_to_life_the_universe_and_everything, i));
    boost::unique_future<int> fi=pt.get_future();

    boost::thread *task = new boost::thread(std::move(pt)); // launch task on a thread

    if(fi.timed_wait(boost::posix_time::milliseconds(time_limit)))
    {
        std::cout << "sucsess function returned: " << fi.get() << std::endl;
    }
    else
    {
        std::cout << "request took way 2 long!" << std::endl;

        std::cout << "current group size:" << threads.size() << std::endl;

        boost::shared_ptr<boost::thread> thread;
        boost::packaged_task<void> pt(boost::bind(run, thread));
        thread = boost::shared_ptr<boost::thread>( new boost::thread(std::move(pt)));

        boost::mutex::scoped_lock lock(threads_creation);
        threads.add_thread(thread.get());
        lock.unlock();

        task->join();

        throw std::runtime_error("killed joined thread");

    }
}

int main()
{
    time_limit = 500;

    work = new boost::asio::io_service::work(io_service);
    int cores_number = boost::thread::hardware_concurrency(); 
    for (std::size_t i = 0; i < cores_number; ++i)
    {

        boost::shared_ptr<boost::thread> thread;
        boost::packaged_task<void> pt(boost::bind(run, thread));
        thread = boost::shared_ptr<boost::thread>( new boost::thread(std::move(pt)));
        threads.add_thread(thread.get());
    }

    int i = 800;
    io_service.post(boost::bind(pool_item, i));

    boost::this_thread::sleep(boost::posix_time::milliseconds(i*2));
    std::cout << "thread should be removed by now." << std::endl
        << "group size:" << threads.size() << std::endl;

    std::cin.get();
    return 0;
}

正如您所见，即使在.remove_thread(ptr); 调用之后，线程也不会从线程池中删除。=（为什么？

更新 #2：

好吧，无论如何我都以服装线程组结束...

#include <stdio.h>
#include <iostream>
#include <fstream>
#include <set>

//Boost
#include <boost/asio.hpp>
#include <boost/thread.hpp>
#include <boost/thread/locks.hpp>

//cf service interface
//#include <service.hpp>

//cf-server
//#include <server.h>

#include <boost/foreach.hpp>

class thread_group
{
public:
    void add( boost::shared_ptr<boost::thread> to_add)
    {
        boost::mutex::scoped_lock lock(m);
        ds_.insert(to_add);
    }
    void remove( boost::shared_ptr<boost::thread> to_remove)
    {
        boost::mutex::scoped_lock lock(m);
        ds_.erase(to_remove);
    }

    int size()
    {
        boost::mutex::scoped_lock lock(m);
        return ds_.size();
    }

    void join_all(boost::posix_time::milliseconds interuption_time=boost::posix_time::milliseconds(1000))
    {
        boost::mutex::scoped_lock lock(m);
        BOOST_FOREACH(boost::shared_ptr<boost::thread> t, ds_)
        {
            boost::thread interrupter(boost::bind(&thread_group::interupt_thread, this, t, interuption_time));
        }
    }

private:
    std::set< boost::shared_ptr<boost::thread> > ds_;
    boost::mutex m;
    void interupt_thread(boost::shared_ptr<boost::thread> t, boost::posix_time::milliseconds interuption_time)
    {
        try
        {
            if(!t->timed_join(interuption_time))
                t->interrupt();

        }
        catch(std::exception &e)
        {
        }
    }
};

boost::asio::io_service io_service;
boost::asio::io_service::work *work;
thread_group threads;
int time_limit;



int calculate_the_answer_to_life_the_universe_and_everything(int i)
{
    boost::this_thread::sleep(boost::posix_time::milliseconds(i));
    std::cout << i << std::endl;
    return i;
}

void run(boost::shared_ptr<boost::thread> thread_ptr)
{
    try
    {
        io_service.run();
    }
    catch(std::exception &e)
    {
        std::cout << "exeption: " << e.what() << std::endl;
        threads.remove(thread_ptr);
        std::cout << "thread removed from group" << std::endl;
        return;
    }

}

void pool_item( int i)
{
    boost::packaged_task<int> pt(boost::bind(calculate_the_answer_to_life_the_universe_and_everything, i));
    boost::unique_future<int> fi=pt.get_future();

    boost::thread *task = new boost::thread(std::move(pt)); // launch task on a thread

    if(fi.timed_wait(boost::posix_time::milliseconds(time_limit)))
    {
        std::cout << "sucsess function returned: " << fi.get() << std::endl;
    }
    else
    {
        std::cout << "request took way 2 long!" << std::endl;

        std::cout << "current group size:" << threads.size() << std::endl;
        std::cout << "we want to add thread!" << std::endl;
        boost::shared_ptr<boost::thread> thread;
        boost::packaged_task<void> pt(boost::bind(run, thread));
        threads.add(thread);
        std::cout << "thread added" << std::endl
            << "current group size:" << threads.size() << std::endl;
        task->join();

        throw std::runtime_error("killed joined thread");

    }
}

int main()
{
    time_limit = 500;

    work = new boost::asio::io_service::work(io_service);
    int cores_number = boost::thread::hardware_concurrency(); 
    for (std::size_t i = 0; i < cores_number; ++i)
    {

        boost::shared_ptr<boost::thread> thread;
        boost::packaged_task<void> pt(boost::bind(run, thread));
        thread = boost::shared_ptr<boost::thread>( new boost::thread(std::move(pt)));
        threads.add(thread);
    }

    int i = 800;
    io_service.post(boost::bind(pool_item, i));

    boost::this_thread::sleep(boost::posix_time::milliseconds(i*2));
    std::cout << "thread should be removed by now." << std::endl
        << "group size:" << threads.size() << std::endl;

    std::cin.get();
    return 0;
}

Answer 1

过去我已经能够通过利用run() 如果回调抛出异常将退出这一事实来实现这一点。我没有直接在线程中启动run()，而是调用一个实用函数，如果抛出适当的异常则退出线程：

void RunIOService()
{
   try
   {
       io_service.run();
   }
   catch(std::exception ex)
   {
   }
}

那么你所要做的就是安排一个会抛出异常的回调：

static void KillThreadCallback() 
{
    // throw some exception that you catch above
}

io_service.post(&KillThreadCallback);

这将导致执行此回调的线程退出，实质上将线程池计数大小减少 1。使用它，您可以非常轻松地扩展和收缩 io_service 线程池。

Answer 2

可用于彻底关闭 I/O 服务的一种模式是（使用 C++0x lambda）：

void ThreadLoop()
{
    while(m_keepRunning) {
        try {
            io_service.run_one();
        } catch(const std::exception& e) {
            // error handling
        }
    }
}

void Stop()
{
   // Using C++0x lambdas
   io_service.post([=]{ m_keepRunning = false; });
   // or
   io_service.post(boost::bind(&ThisClass::StopCallback, this));
}

void StopCallback()
{
   m_keepRunning = false;
}

其中 m_keepRunning 是一个成员变量。应该只在 I/O 服务线程中被触及。