构建多个线程以获得最佳性能的正确方法是什么？

Question

我想知道在使用 lambda 表达式来定义线程时，它是否有助于提高一些性能增益。就我而言，我必须运行多个线程。这是基于实时的应用程序。因此，如果有人向我建议创建多个线程的最佳方式是什么。创建线程发生在实际代码库的每次迭代中。这是高级迭代中发生的事情的一个示例。因此，这是一种需要优化的昂贵操作。

  #include <iostream>
  #include <thread>
  #include <vector>
  #include <algorithm>

  class Task
  {
    public:
    void execute(std::string command)
    {
      //TODO actual logic
      for(int i = 0; i < 5; i++)
      {
        std::cout<<command<<std::endl;
      }
    }
  };

  int main()
  {          
      Task* taskPtr = new Task();
      std::vector<std::thread> workers_older;
      for (int i = 0; i < 2; i++) {
          workers_older.push_back(std::thread(&Task::execute, taskPtr, "Task: without lambda expression"+ std::to_string(i)));
      }
      std::for_each(workers_older.begin(), workers_older.end(), [](std::thread &t) 
      {
          t.join();
      });

      std::vector<std::thread> workers;
      for (int i = 0; i < 2; i++) {
          workers.push_back(std::thread([&]() 
          {
              taskPtr->execute("Task: "+ std::to_string(i));
          }));
      }
      std::for_each(workers.begin(), workers.end(), [](std::thread &t) 
      {
          t.join();
      });
      return 0;
  }

编辑： 在关于要做什么的宝贵意见之后，我已经按照他们的建议提供了答案

Answer 1

非常感谢大家提出的所有非常宝贵的想法。我决定为该任务使用线程池。对不起，关于没有解释整个逻辑很长，我认为这不是必需的。

这是我提出的解决方案。我从here 获得了初始代码，并按照我想要的方式进行了修改。

    #include <iostream>
    #include <unistd.h>
    #include <iostream>
    #include <thread>
    #include <vector>
    #include <algorithm>
    #include <boost/shared_ptr.hpp>
    #include <boost/make_shared.hpp>

    #include <boost/thread.hpp>
    #include <boost/bind.hpp>
    #include <boost/asio.hpp>
    #include <boost/move/move.hpp>
    #include <boost/make_unique.hpp>

    namespace asio = boost::asio; 

    typedef boost::packaged_task<int> task_t;
    typedef boost::shared_ptr<task_t> ptask_t;

    class Task
    {
    public:
    int execute(std::string command)
    {
      //TODO actual logic
      std::cout<< "\nThread:" << command << std::endl;
      int sum = 0;
      for(int i = 0; i < 5; i++)
      {
        sum+=i;
      }
      return sum;
    }
  };


    void push_job(Task* worker, std::string seconds, boost::asio::io_service& io_service
                , std::vector<boost::shared_future<int> >& pending_data) {
      ptask_t task = boost::make_shared<task_t>(boost::bind(&Task::execute, worker, seconds));
      boost::shared_future<int> fut(task->get_future());
      pending_data.push_back(fut);
      io_service.post(boost::bind(&task_t::operator(), task));
    }

    int main()
    {
        Task* taskPtr = new Task();

        boost::asio::io_service io_service;
        boost::thread_group threads;
        std::unique_ptr<boost::asio::io_service::work> service_work;
        service_work = boost::make_unique<boost::asio::io_service::work>(io_service);
        for (int i = 0; i < boost::thread::hardware_concurrency() ; ++i)
        {
          threads.create_thread(boost::bind(&boost::asio::io_service::run,
            &io_service));
        }
        std::vector<boost::shared_future<int> > pending_data; // vector of futures

        push_job(taskPtr, "4", io_service, pending_data);
        push_job(taskPtr, "5", io_service, pending_data);
        push_job(taskPtr, "6", io_service, pending_data);
        push_job(taskPtr, "7", io_service, pending_data);

        boost::wait_for_all(pending_data.begin(), pending_data.end());
        int total_sum = 0;
        for(auto result : pending_data){
           total_sum += result.get();
        }
        std::cout<< "Total sum: "<< total_sum << std::endl;
        return 0;
    }

Answer 2

使用线程时最大的开销来自启动线程、调度、上下文切换和缓存利用。与此相比，额外间接函数指针的开销可以忽略不计。

为了获得最佳性能，请记住以下几点：

• 保留一个包含 N 个线程的池，其中 N = std::thread::hardware_concurrency()（系统中的逻辑处理器数）
• 将 N-1 个作业提交到池中，并在调用线程中运行第 N 个作业。不将第 N 个作业提交到池中可以节省大量资金
• 避免虚假共享。不同线程写入的数据应该在不同的缓存行中
• 更多活动线程通常意味着更大的工作集。因此 D-cache 利用率可能会降低，从而影响性能

这是我的工作示例：

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

struct thread_pool {
    thread_pool(int threads = std::thread::hardware_concurrency()) : size(threads) {
        grp.reserve(threads);
        for (int i = 0; i < threads; ++i)
            grp.emplace_back([this] { return service.run(); });
    }

    template<typename F, typename ...Args>
    auto enqueue(F& f, Args... args) -> std::future<decltype(f(args...))> {
        return boost::asio::post(service,
            std::packaged_task<decltype(f(args...))()>([&f, args...]{ return f(args...); })
        );
    }

    ~thread_pool() {
        service_work.reset();
        for (auto &t : grp)
            if (t.joinable())
                t.join();
        service.stop();
    }

    const int size;
private:
    boost::asio::io_service service;
    std::unique_ptr<boost::asio::io_service::work> service_work {new boost::asio::io_service::work(service)};
    std::vector<std::thread> grp;
};

int main() {
    thread_pool pool;
    std::vector<std::future<int>> results;
    auto task = [](int i) { return i + 1; };
    for (int i = 0; i < pool.size - 1; i++) {
        results.emplace_back(pool.enqueue(task, i));
    }
    int sum = task(pool.size - 1); // last task run synchronously
    for (auto& res : results) {
        sum += res.get();
    }
    std::cout << sum << std::endl;
}

Answer 3

将成员函数的地址和一组参数传递给 std::thread 构造函数与传递带有适当捕获的 lambda 函数之间的开销差别很小。

std::thread 构造函数中的大开销实际上是启动线程本身。

如果您知道要在程序中的多个位置使用相同数量的工作线程，那么将它们作为具有任务队列的长时间运行线程保留可能是值得的。