Пул потоков в C ++ 11

Актуальные вопросы :

О C ++ 11:

• С ++ 11: пул std :: thread?
• Будет ли async (launch :: async) в C ++ 11 делать пулы потоков устаревшими, чтобы избежать создания дорогостоящих потоков?

О Boost:

• C ++ boost thread повторное использование потоков
• boost :: thread и создание их пула!

Как получить пул потоков для отправки задач , не создавая и не удаляя их снова и снова? Это означает, что постоянные потоки будут повторно синхронизироваться без присоединения.

У меня есть код, который выглядит так:

namespace {
  std::vector<std::thread> workers;

  int total = 4;
  int arr[4] = {0};

  void each_thread_does(int i) {
    arr[i] += 2;
  }
}

int main(int argc, char *argv[]) {
  for (int i = 0; i < 8; ++i) { // for 8 iterations,
    for (int j = 0; j < 4; ++j) {
      workers.push_back(std::thread(each_thread_does, j));
    }
    for (std::thread &t: workers) {
      if (t.joinable()) {
        t.join();
      }
    }
    arr[4] = std::min_element(arr, arr+4);
  }
  return 0;
}

Вместо того, чтобы создавать и объединять потоки на каждой итерации, я бы предпочел отправлять задачи своим рабочим потокам на каждой итерации и создавать их только один раз.

Вы можете использовать библиотеку пула потоков C ++, https://github.com/vit-vit/ctpl .

Затем код, который вы написали, можно заменить следующим

#include <ctpl.h>  // or <ctpl_stl.h> if ou do not have Boost library

int main (int argc, char *argv[]) {
    ctpl::thread_pool p(2 /* two threads in the pool */);
    int arr[4] = {0};
    std::vector<std::future<void>> results(4);
    for (int i = 0; i < 8; ++i) { // for 8 iterations,
        for (int j = 0; j < 4; ++j) {
            results[j] = p.push([&arr, j](int){ arr[j] +=2; });
        }
        for (int j = 0; j < 4; ++j) {
            results[j].get();
        }
        arr[4] = std::min_element(arr, arr + 4);
    }
}

Вы получите желаемое количество потоков и не будете создавать и удалять их снова и снова на итерациях.

Это скопировано из моего ответа на другой очень похожий пост, надеюсь, это поможет:

1) Начните с максимального количества потоков, которое может поддерживать система:

int Num_Threads =  thread::hardware_concurrency();

2) Для эффективной реализации пула потоков после создания потоков в соответствии с Num_Threads лучше не создавать новые и не уничтожать старые (путем присоединения). Это приведет к снижению производительности, может даже сделать ваше приложение медленнее, чем последовательная версия.

Каждый поток C ++ 11 должен выполняться в своей функции с бесконечным циклом, постоянно ожидая захвата и запуска новых задач.

Вот как прикрепить такую ​​функцию к пулу потоков:

int Num_Threads = thread::hardware_concurrency();
vector<thread> Pool;
for(int ii = 0; ii < Num_Threads; ii++)
{  Pool.push_back(thread(Infinite_loop_function));}

3) Функция Infinite_loop_function

Это цикл while (true), ожидающий очереди задач

void The_Pool:: Infinite_loop_function()
{
    while(true)
    {
        {
            unique_lock<mutex> lock(Queue_Mutex);

            condition.wait(lock, []{return !Queue.empty() || terminate_pool});
            Job = Queue.front();
            Queue.pop();
        }
        Job(); // function<void()> type
    }
};

4) Сделайте функцию для добавления работы в вашу очередь

void The_Pool:: Add_Job(function<void()> New_Job)
{
    {
        unique_lock<mutex> lock(Queue_Mutex);
        Queue.push(New_Job);
    }
    condition.notify_one();
}

5) Привяжите произвольную функцию к вашей очереди

Pool_Obj.Add_Job(std::bind(&Some_Class::Some_Method, &Some_object));

После интеграции этих компонентов у вас будет собственный динамический пул потоков. Эти потоки всегда выполняются, ожидая выполнения задания.

Прошу прощения, если есть какие-то синтаксические ошибки, я набрал этот код и у меня плохая память. Извините, что я не могу предоставить вам полный код пула потоков, это нарушит мою честность работы.

Изменить: чтобы завершить работу пула, вызовите метод shutdown ():

XXXX::shutdown(){
{
    unique_lock<mutex> lock(threadpool_mutex);
    terminate_pool = true;} // use this flag in condition.wait

    condition.notify_all(); // wake up all threads.

    // Join all threads.
    for(std::thread &every_thread : thread_vector)
    {   every_thread.join();}

    thread_vector.clear();  
    stopped = true; // use this flag in destructor, if not set, call shutdown() 
}

Пул потоков означает, что все ваши потоки работают постоянно - другими словами, функция потока никогда не возвращается. Чтобы дать потокам что-то значимое, вы должны разработать систему межпотокового взаимодействия, как с целью сообщить потоку, что нужно что-то сделать, так и для передачи фактических рабочих данных.

Обычно это будет связано с какой-то параллельной структурой данных, и каждый поток предположительно будет спать на какой-то переменной условия, которая будет уведомлена, когда есть работа. После получения уведомления один или несколько потоков просыпаются, восстанавливают задачу из параллельной структуры данных, обрабатывают ее и аналогичным образом сохраняют результат.

Затем поток продолжит проверять, нужно ли еще поработать, и если не вернется в режим сна.

В результате вам придется спроектировать все это самостоятельно, поскольку не существует универсального универсального понятия «работа». Это довольно большая работа, и есть некоторые тонкие проблемы, которые вам нужно решить. (Вы можете программировать на Go, если вам нравится система, которая негласно позаботится об управлении потоками.)

Пул потоков - это, по сути, набор потоков, связанных с функцией, работающей как цикл событий. Эти потоки будут бесконечно ждать выполнения задачи или своего собственного завершения.

Задача пула потоков заключается в предоставлении интерфейса для отправки заданий, определения (и, возможно, изменения) политики выполнения этих заданий (правила планирования, создание экземпляров потоков, размер пула) и мониторинг состояния потоков и связанных ресурсов.

Итак, для универсального пула нужно начать с определения, что такое задача, как она запускается, прерывается, каков результат (см. Понятие обещания и будущего для этого вопроса), на какие события потоки должны будут реагировать. как они будут их обрабатывать, как эти события должны отличаться от тех, которые обрабатываются задачами. Как видите, это может стать довольно сложным и накладывать ограничения на работу потоков, поскольку решение становится все более сложным.

Текущие инструменты для обработки событий довольно просты (*): примитивы, такие как мьютексы, условные переменные, и несколько абстракций в дополнение к этому (блокировки, барьеры). Но в некоторых случаях эти абстракции могут оказаться непригодными (см. Связанный с этим вопрос ), и нужно вернуться к использованию примитивов.

Также необходимо решить другие проблемы:

• сигнал
• I / O
• аппаратное обеспечение (сходство с процессором, неоднородная настройка)

Как это отразится в вашем окружении?

Этот ответ на аналогичный вопрос указывает на существующую реализацию, предназначенную для boost и stl.

Я предложил очень грубую реализацию пула потоков для другого вопроса, который не решает многие проблемы, описанные выше. Возможно, вы захотите развить это. Вы также можете взглянуть на существующие фреймворки на других языках, чтобы найти вдохновение.

_{(*) Я не вижу в этом проблемы, скорее, наоборот. Я думаю, что это сам дух C ++, унаследованный от C.}

Follwoing [PhD EcE](https://stackoverflow.com/users/3818417/phd-ece) suggestion, I implemented the thread pool:

function_pool.h

#pragma once
#include <queue>
#include <functional>
#include <mutex>
#include <condition_variable>
#include <atomic>
#include <cassert>

class Function_pool
{

private:
    std::queue<std::function<void()>> m_function_queue;
    std::mutex m_lock;
    std::condition_variable m_data_condition;
    std::atomic<bool> m_accept_functions;

public:

    Function_pool();
    ~Function_pool();
    void push(std::function<void()> func);
    void done();
    void infinite_loop_func();
};

function_pool.cpp

#include "function_pool.h"

Function_pool::Function_pool() : m_function_queue(), m_lock(), m_data_condition(), m_accept_functions(true)
{
}

Function_pool::~Function_pool()
{
}

void Function_pool::push(std::function<void()> func)
{
    std::unique_lock<std::mutex> lock(m_lock);
    m_function_queue.push(func);
    // when we send the notification immediately, the consumer will try to get the lock , so unlock asap
    lock.unlock();
    m_data_condition.notify_one();
}

void Function_pool::done()
{
    std::unique_lock<std::mutex> lock(m_lock);
    m_accept_functions = false;
    lock.unlock();
    // when we send the notification immediately, the consumer will try to get the lock , so unlock asap
    m_data_condition.notify_all();
    //notify all waiting threads.
}

void Function_pool::infinite_loop_func()
{
    std::function<void()> func;
    while (true)
    {
        {
            std::unique_lock<std::mutex> lock(m_lock);
            m_data_condition.wait(lock, [this]() {return !m_function_queue.empty() || !m_accept_functions; });
            if (!m_accept_functions && m_function_queue.empty())
            {
                //lock will be release automatically.
                //finish the thread loop and let it join in the main thread.
                return;
            }
            func = m_function_queue.front();
            m_function_queue.pop();
            //release the lock
        }
        func();
    }
}

main.cpp

#include "function_pool.h"
#include <string>
#include <iostream>
#include <mutex>
#include <functional>
#include <thread>
#include <vector>

Function_pool func_pool;

class quit_worker_exception : public std::exception {};

void example_function()
{
    std::cout << "bla" << std::endl;
}

int main()
{
    std::cout << "stating operation" << std::endl;
    int num_threads = std::thread::hardware_concurrency();
    std::cout << "number of threads = " << num_threads << std::endl;
    std::vector<std::thread> thread_pool;
    for (int i = 0; i < num_threads; i++)
    {
        thread_pool.push_back(std::thread(&Function_pool::infinite_loop_func, &func_pool));
    }

    //here we should send our functions
    for (int i = 0; i < 50; i++)
    {
        func_pool.push(example_function);
    }
    func_pool.done();
    for (unsigned int i = 0; i < thread_pool.size(); i++)
    {
        thread_pool.at(i).join();
    }
}

Что-то вроде этого может помочь (взято из рабочего приложения).

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

struct thread_pool {
  typedef std::unique_ptr<boost::asio::io_service::work> asio_worker;

  thread_pool(int threads) :service(), service_worker(new asio_worker::element_type(service)) {
    for (int i = 0; i < threads; ++i) {
      auto worker = [this] { return service.run(); };
      grp.add_thread(new boost::thread(worker));
    }
  }

  template<class F>
  void enqueue(F f) {
    service.post(f);
  }

  ~thread_pool() {
    service_worker.reset();
    grp.join_all();
    service.stop();
  }

private:
  boost::asio::io_service service;
  asio_worker service_worker;
  boost::thread_group grp;
};

Вы можете использовать это так:

thread_pool pool(2);

pool.enqueue([] {
  std::cout << "Hello from Task 1\n";
});

pool.enqueue([] {
  std::cout << "Hello from Task 2\n";
});

Имейте в виду, что изобретение эффективного механизма асинхронной организации очередей - нетривиальная задача.

Boost :: asio :: io_service - очень эффективная реализация или фактически представляет собой набор специфичных для платформы оболочек (например, он обертывает порты завершения ввода-вывода в Windows).

Изменить: теперь для этого требуется C ++ 17 и концепции. (По состоянию на 9/12/16 достаточно только g ++ 6.0+.)

Тем не менее, вывод шаблонов из-за этого намного точнее, поэтому стоит потратить усилия на получение более нового компилятора. Я еще не нашел функцию, которая требует явных аргументов шаблона.

Теперь он также принимает любой подходящий вызываемый объект ( и все еще статически типизирован !!! ).

Он также теперь включает дополнительный пул потоков с приоритетом зеленого потока, использующий тот же API. Однако этот класс - только POSIX. Он использует ucontext_tAPI для переключения задач в пользовательском пространстве.

Я создал для этого простую библиотеку. Пример использования приведен ниже. (Я отвечаю на это, потому что это была одна из вещей, которые я нашел до того, как решил, что необходимо написать это сам.)

bool is_prime(int n){
  // Determine if n is prime.
}

int main(){
  thread_pool pool(8); // 8 threads

  list<future<bool>> results;
  for(int n = 2;n < 10000;n++){
    // Submit a job to the pool.
    results.emplace_back(pool.async(is_prime, n));
  }

  int n = 2;
  for(auto i = results.begin();i != results.end();i++, n++){
    // i is an iterator pointing to a future representing the result of is_prime(n)
    cout << n << " ";
    bool prime = i->get(); // Wait for the task is_prime(n) to finish and get the result.
    if(prime)
      cout << "is prime";
    else
      cout << "is not prime";
    cout << endl;
  }  
}

Вы можете передать asyncлюбую функцию с любым (или недействительным) возвращаемым значением и любыми (или без) аргументами, и она вернет соответствующий std::future. Чтобы получить результат (или просто дождаться выполнения задачи), вы обращаетесь get()к будущему.

Вот гитхаб: https://github.com/Tyler-Hardin/thread_pool .

Это еще одна реализация пула потоков, которая очень проста, легка в понимании и использовании, использует только стандартную библиотеку C ++ 11 и может быть просмотрена или изменена для ваших целей, должна быть хорошим стартером, если вы хотите начать использовать поток бассейны:

https://github.com/progschj/ThreadPool

Вы можете использовать thread_pool из библиотеки boost:

void my_task(){...}

int main(){
    int threadNumbers = thread::hardware_concurrency();
    boost::asio::thread_pool pool(threadNumbers);

    // Submit a function to the pool.
    boost::asio::post(pool, my_task);

    // Submit a lambda object to the pool.
    boost::asio::post(pool, []() {
      ...
    });
}

Вы также можете использовать threadpool из сообщества с открытым исходным кодом:

void first_task() {...}    
void second_task() {...}

int main(){
    int threadNumbers = thread::hardware_concurrency();
    pool tp(threadNumbers);

    // Add some tasks to the pool.
    tp.schedule(&first_task);
    tp.schedule(&second_task);
}

Пул потоков без зависимостей вне STL вполне возможен. Недавно я написал небольшую библиотеку пула потоков только для заголовков, чтобы решить ту же проблему. Он поддерживает динамическое изменение размера пула (изменение количества рабочих во время выполнения), ожидание, остановку, приостановку, возобновление и так далее. Надеюсь, вы сочтете это полезным.