提升线程和 shared_ptrs

Question

我正在和朋友一起做一个附带项目，并正在使用 Boost 线程和 asio 构建一个线程池。我已经启动并运行了一个基本池，但希望能够从池中删除线程。下面的类保留了std::shared_ptr<boost::thread> 的向量。

我遇到的问题是当我运行一个简单的测试时，程序永远不会退出。它似乎到达了终点（见输出），但从未终止。我有一种感觉，这与仍然活着的线程有关，并试图杀死它们但无济于事。我做了一些关于清除共享指针向量的搜索，看起来你真的不应该这样做。同样，thread_group 对join_all() 的调用应该加入并结束所有线程，至少我是这么认为的。

我仍在全面学习 boost 线程、asio 和 shared_ptr，因此非常感谢任何帮助或建议。

编译：`

g++ -std=c++14 main.cpp -lboost_system -lboost_thread -lpthread

类：

  class experimentalPool {
      private:
        boost::asio::io_service ioService;
        boost::asio::io_service::work work;
        boost::thread_group threads;
        std::vector<std::shared_ptr<boost::thread>> workers;
        std::size_t poolsize;
      public:
        experimentalPool(): work( ioService ) {
            poolsize = boost::thread::hardware_concurrency();
            for (size_t i = 0; i < poolsize; i++) {
                std::shared_ptr<boost::thread> t1(new boost::thread(
                    boost::bind(&boost::asio::io_service::run, &ioService)));
                threads.add_thread(t1.get());
                workers.push_back(std::move(t1));
            }
        }

        experimentalPool( std::size_t psize )
          : work( ioService ),
            poolsize( psize )
        {
            for (size_t i = 0; i < poolsize; i++) {
                std::shared_ptr<boost::thread> t1(new boost::thread(
                    boost::bind(&boost::asio::io_service::run, &ioService)));
                threads.add_thread(t1.get());
                workers.push_back(std::move(t1));
            }
        }

        template <typename F, typename... Args>
        void add_work(F f, Args... args){
            ioService.post(boost::bind(f, args...));
        };

        ~experimentalPool(){
          ioService.stop();
          try {
            std::cout << "here" << "\n";
            threads.join_all();
            //threads.interrupt_all();
            std::cout << "here" << "\n";
            //for (size_t i = 0; i < workers.size(); i++) {
              //(*workers[i]).interrupt();
            //}
          }
          catch ( const std::exception& ) {std::cout << "Caught Exception" << "\n";}
        }

        std::size_t size() const { return poolsize;}
    };

主要：

void add1(int& num){
    std::cout << ++num << std::endl;
}

int main(){

    int temp = 5;

    experimentalPool xpool(2);
    std::cout << "Xpool " << pool2.size() << "\n";
    xpool.add_work(add1, temp);

    std::this_thread::sleep_for (std::chrono::seconds(1));
    std::cout << "End" << std::endl;
    return 0;
}

输出：

Xpool 1
6
here
xpool here
xpool here
^C
ap@ubuntu:~/Desktop$ 

Answer 1

你的致命问题在这里：

std::shared_ptr<boost::thread> t1(new boost::thread(
    boost::bind(&boost::asio::io_service::run, &ioService)));
threads.add_thread(t1.get());

从boost::thread_group(http://www.boost.org/doc/libs/1_42_0/doc/html/thread/thread_management.html#thread.thread_management.threadgroup.destructor)的文档来看，其析构函数的效果是

销毁 *this 并删除组中的所有 boost::thread 对象。

因此boost::thread_group 将专门参与其线程的资源管理，您不能将这些线程存储在shared_ptr 中。

要解决这个问题，您需要仔细阅读boost::thread_group 的链接文档，使用create_thread() 并将其存储为原始指针。

Answer 2

说实话，我看了好几分钟的代码，越来越担心了。

为什么线程在两个集合中（thread_group 和 vector）？为什么线程在另一个集合中拥有？为什么工作对象是永久性的？为什么要取消服务而不是让池耗尽？

为什么 poolsize 会被冗余存储？

为什么构造函数99%是重复的？

为什么add1 采用可变引用，但绑定是按值？

简化

Live On Coliru

#include <boost/asio.hpp>
#include <boost/thread.hpp>
#include <iostream>
#include <thread>

namespace Experimental { 
    namespace ba = boost::asio;

    class Pool {
      private:
        ba::io_service ioService;
        boost::thread_group threads;
        boost::optional<ba::io_service::work> work { ioService };

      public:
        Pool(std::size_t psize = boost::thread::hardware_concurrency()) {
            for (size_t i = 0; i < psize; i++) {
                threads.create_thread(boost::bind(&ba::io_service::run, &ioService));
            }
        }

        template <typename F, typename... Args> void add_work(F f, Args... args) {
            ioService.post(boost::bind(f, args...));
        }

        ~Pool() {
            work.reset();
            threads.join_all();
        }

        std::size_t size() const { return threads.size(); }
    };
}

void add1(int &num) { std::cout << ++num << std::endl; }

int main() {
    int temp = 5;

    Experimental::Pool xpool(2);
    std::cout << "Xpool " << xpool.size() << "\n";
    xpool.add_work(add1, std::ref(temp));

    std::this_thread::sleep_for(std::chrono::seconds(1));
    std::cout << "End" << std::endl;
}

打印

Xpool 2
6
End
temp: 6

而且它只占代码的约 50%。