Agrupación de subprocesos en C ++ 11

Preguntas relevantes :

Sobre C ++ 11:

• C ++ 11: std :: hilo agrupado?
• ¿Async (launch :: async) en C ++ 11 hará que los grupos de subprocesos sean obsoletos para evitar la creación de subprocesos costosos?

Sobre Boost:

• C ++ impulsar el hilo reutilizando hilos
• boost :: thread y creando un grupo de ellos!

¿Cómo consigo un grupo de subprocesos para enviar tareas , sin crearlas y eliminarlas una y otra vez? Esto significa hilos persistentes para resincronizar sin unirse.

Tengo un código que se ve así:

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;
}

En lugar de crear y unir subprocesos en cada iteración, preferiría enviar tareas a mis subprocesos de trabajo en cada iteración y solo crearlas una vez.

Puede usar la biblioteca de grupos de subprocesos de C ++, https://github.com/vit-vit/ctpl .

Luego, el código que escribió puede reemplazarse con el siguiente

#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);
    }
}

Obtendrá el número deseado de hilos y no los creará y eliminará una y otra vez en las iteraciones.

Esto se copia de mi respuesta a otra publicación muy similar, espero que pueda ayudar:

1) Comience con el número máximo de subprocesos que un sistema puede admitir:

int Num_Threads =  thread::hardware_concurrency();

2) Para una implementación eficiente de conjunto de subprocesos, una vez que los subprocesos se crean de acuerdo con Num_Threads, es mejor no crear nuevos o destruir los antiguos (uniéndose). Habrá una penalización de rendimiento, incluso podría hacer que su aplicación vaya más lenta que la versión en serie.

Cada subproceso de C ++ 11 debe ejecutarse en su función con un bucle infinito, esperando constantemente que se lleven a cabo nuevas tareas.

Aquí se explica cómo adjuntar dicha función al grupo de subprocesos:

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) La función de bucle infinito

Este es un ciclo "while (verdadero)" que espera la cola de tareas

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) Haga una función para agregar trabajo a su cola

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

5) Vincula una función arbitraria a tu Cola

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

Una vez que integre estos ingredientes, tiene su propio grupo dinámico de subprocesos. Estos subprocesos siempre se ejecutan, esperando el trabajo que hacer.

Pido disculpas si hay algunos errores de sintaxis, escribí este código y tengo mala memoria. Lamento no poder proporcionarle el código completo del grupo de subprocesos, eso violaría la integridad de mi trabajo.

Editar: para terminar el grupo, llame al método 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() 
}

Un grupo de subprocesos significa que todos sus subprocesos se están ejecutando, todo el tiempo; en otras palabras, la función de subproceso nunca regresa. Para dar a los hilos algo significativo que hacer, debe diseñar un sistema de comunicación entre hilos, tanto con el fin de decirle al hilo que hay algo que hacer, como para comunicar los datos de trabajo reales.

Por lo general, esto implicará algún tipo de estructura de datos concurrentes, y cada hilo probablemente duerma en algún tipo de variable de condición, que se notificará cuando haya trabajo por hacer. Al recibir la notificación, uno o varios subprocesos se activan, recuperan una tarea de la estructura de datos concurrente, la procesan y almacenan el resultado de manera análoga.

Luego, el hilo continuará para verificar si hay aún más trabajo por hacer y, si no, volver a dormir.

El resultado es que tiene que diseñar todo esto usted mismo, ya que no existe una noción natural de "trabajo" que sea de aplicación universal. Es bastante trabajo, y hay algunos problemas sutiles que debes resolver. (Puede programar en Go si desea un sistema que se encargue de la gestión de subprocesos detrás de escena).

Un conjunto de subprocesos es, en esencia, un conjunto de subprocesos todos vinculados a una función que funciona como un bucle de eventos. Estos subprocesos esperarán sin cesar hasta que se ejecute una tarea o su propia terminación.

El trabajo de grupo de subprocesos es proporcionar una interfaz para enviar trabajos, definir (y quizás modificar) la política de ejecución de estos trabajos (reglas de programación, creación de instancias de subprocesos, tamaño del grupo) y supervisar el estado de los subprocesos y recursos relacionados.

Entonces, para un grupo versátil, uno debe comenzar definiendo qué es una tarea, cómo se inicia, se interrumpe, cuál es el resultado (vea la noción de promesa y futuro para esa pregunta), qué tipo de eventos tendrán que responder los hilos. a, cómo los manejarán, cómo estos eventos serán discriminados de los manejados por las tareas. Como puede ver, esto puede volverse bastante complicado e imponer restricciones sobre cómo funcionarán los subprocesos, a medida que la solución se involucra cada vez más.

La herramienta actual para manejar eventos es bastante básica (*): primitivas como mutexes, variables de condición y algunas abstracciones además de eso (bloqueos, barreras). Pero en algunos casos, estas abstracciones pueden resultar no aptas (ver esta pregunta relacionada ), y uno debe volver a usar las primitivas.

Otros problemas también deben ser manejados:

• señal
• i / o
• hardware (afinidad del procesador, configuración heterogénea)

¿Cómo se desarrollarían en tu entorno?

Esta respuesta a una pregunta similar apunta a una implementación existente destinada a boost y stl.

Me ofrecieron una aplicación muy cruda de un conjunto de subprocesos para otra pregunta, que no aborda muchos de los problemas descritos anteriormente. Es posible que desee construir sobre él. También es posible que desee echar un vistazo a los marcos existentes en otros idiomas, para encontrar inspiración.

_{(*) No lo veo como un problema, sino todo lo contrario. Creo que es el espíritu de C ++ heredado de 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();
    }
}

Algo como esto podría ayudar (tomado de una aplicación que funciona).

#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;
};

Puedes usarlo así:

thread_pool pool(2);

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

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

Tenga en cuenta que reinventar un mecanismo de cola asíncrono eficiente no es trivial.

Boost :: asio :: io_service es una implementación muy eficiente, o en realidad es una colección de contenedores específicos de la plataforma (por ejemplo, envuelve los puertos de finalización de E / S en Windows).

Editar: Esto ahora requiere C ++ 17 y conceptos. (A partir del 12/09/16, solo g ++ 6.0+ es suficiente).

Sin embargo, la deducción de plantillas es mucho más precisa debido a ello, por lo que vale la pena el esfuerzo de obtener un compilador más nuevo. Todavía no he encontrado una función que requiera argumentos de plantilla explícitos.

Ahora también toma cualquier objeto invocable apropiado (¡ y todavía es estáticamente seguro! ).

Ahora también incluye un grupo de subprocesos de prioridad de subprocesamiento verde opcional con la misma API. Sin embargo, esta clase es solo POSIX. Utiliza la ucontext_tAPI para el cambio de tareas del espacio de usuario.

Creé una biblioteca simple para esto. Un ejemplo de uso se da a continuación. (Estoy respondiendo esto porque fue una de las cosas que encontré antes de decidir que era necesario escribirlo yo mismo).

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;
  }  
}

Puede pasar asynccualquier función con cualquier valor de retorno (o nulo) y cualquier argumento (o no) y devolverá un correspondiente std::future. Para obtener el resultado (o simplemente esperar hasta que una tarea se haya completado), llama get()al futuro.

Aquí está el github: https://github.com/Tyler-Hardin/thread_pool .

Esta es otra implementación de grupo de subprocesos que es muy simple, fácil de entender y usar, usa solo la biblioteca estándar de C ++ 11 y se puede ver o modificar para sus usos, debería ser un buen comienzo si desea comenzar a usar el subproceso quinielas:

https://github.com/progschj/ThreadPool

Puede usar thread_pool de la biblioteca 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, []() {
      ...
    });
}

También puede usar threadpool de la comunidad de código abierto:

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);
}

Un conjunto de subprocesos sin dependencias fuera de STL es completamente posible. Recientemente escribí una pequeña biblioteca de subprocesos de solo encabezado para abordar exactamente el mismo problema. Admite el cambio de tamaño dinámico del grupo (cambio del número de trabajadores en tiempo de ejecución), espera, detención, pausa, reanudación, etc. Espero que le sea útil.