C++
Enhebrado

Sintaxis

• hilo()
• hilo (hilo y & otros)
• Hilo explícito (Function && func, Args && ... args)

Parámetros

Observaciones

Operaciones de hilo

Cuando comienzas un hilo, se ejecutará hasta que termine.

A menudo, en algún momento, debe (posiblemente, el subproceso ya esté terminado) esperar a que el subproceso finalice, porque desea utilizar el resultado, por ejemplo.

int n;
std::thread thread{ calculateSomething, std::ref(n) };

//Doing some other stuff

//We need 'n' now!
//Wait for the thread to finish - if it is not already done
thread.join();

//Now 'n' has the result of the calculation done in the seperate thread
std::cout << n << '\n';

También puede detach el hilo, permitiendo que se ejecute libremente:

std::thread thread{ doSomething };

//Detaching the thread, we don't need it anymore (for whatever reason)
thread.detach();

//The thread will terminate when it is done, or when the main thread returns

Pasando una referencia a un hilo

No puede pasar una referencia (o una referencia const ) directamente a un hilo porque std::thread los copiará / moverá. En su lugar, use std::reference_wrapper :

void foo(int& b)
{
    b = 10;
}

int a = 1;
std::thread thread{ foo, std::ref(a) }; //'a' is now really passed as reference

thread.join();
std::cout << a << '\n'; //Outputs 10

void bar(const ComplexObject& co)
{
    co.doCalculations();
}

ComplexObject object;
std::thread thread{ bar, std::cref(object) }; //'object' is passed as const&

thread.join();
std::cout << object.getResult() << '\n'; //Outputs the result

Creando un std :: thread

En C ++, los subprocesos se crean utilizando la clase std :: thread. Un hilo es un flujo de ejecución separado; es análogo a que un ayudante realice una tarea mientras usted realiza otra. Cuando se ejecuta todo el código en el hilo, termina . Cuando creas un hilo, necesitas pasar algo para ser ejecutado en él. Algunas cosas que puedes pasar a un hilo:

• Funciones libres
• Funciones miembro
• Objetos funcionales
• Expresiones lambda

Ejemplo de función libre: ejecuta una función en un subproceso separado ( ejemplo en vivo ):

#include <iostream>
#include <thread>
 
void foo(int a)
{
    std::cout << a << '\n';
}
 
int main()
{
    // Create and execute the thread
    std::thread thread(foo, 10); // foo is the function to execute, 10 is the
                                 // argument to pass to it
 
    // Keep going; the thread is executed separately
 
    // Wait for the thread to finish; we stay here until it is done
    thread.join();
 
    return 0;
}

Ejemplo de función miembro: ejecuta una función miembro en un subproceso independiente ( ejemplo en vivo ):

#include <iostream>
#include <thread>
 
class Bar
{
public:
    void foo(int a)
    {
        std::cout << a << '\n';
    }
};
 
int main()
{
    Bar bar;
    
    // Create and execute the thread
    std::thread thread(&Bar::foo, &bar, 10); // Pass 10 to member function
 
    // The member function will be executed in a separate thread
 
    // Wait for the thread to finish, this is a blocking operation
    thread.join();
 
    return 0;
}

Ejemplo de objeto funcional (ejemplo en vivo ):

#include <iostream>
#include <thread>
 
class Bar
{
public:
    void operator()(int a)
    {
        std::cout << a << '\n';
    }
};
 
int main()
{
    Bar bar;
    
    // Create and execute the thread
    std::thread thread(bar, 10); // Pass 10 to functor object
 
    // The functor object will be executed in a separate thread
 
    // Wait for the thread to finish, this is a blocking operation
    thread.join();
 
    return 0;
}

Ejemplo de expresión lambda (ejemplo vivo ):

#include <iostream>
#include <thread>
 
int main()
{
    auto lambda = [](int a) { std::cout << a << '\n'; };

    // Create and execute the thread
    std::thread thread(lambda, 10); // Pass 10 to the lambda expression
 
    // The lambda expression will be executed in a separate thread
 
    // Wait for the thread to finish, this is a blocking operation
    thread.join();
 
    return 0;
}

Operaciones en el hilo actual

std::this_thread es un namespace que tiene funciones para hacer cosas interesantes en el hilo actual desde la función desde la que se llama.

Obteniendo el id de los hilos actuales usando std::this_thread::get_id :

void foo()
{
    //Print this threads id
    std::cout << std::this_thread::get_id() << '\n';
}

std::thread thread{ foo };
thread.join(); //'threads' id has now been printed, should be something like 12556

foo(); //The id of the main thread is printed, should be something like 2420

Durmiendo por 3 segundos usando std::this_thread::sleep_for :

void foo()
{
    std::this_thread::sleep_for(std::chrono::seconds(3));
}

std::thread thread{ foo };
foo.join();

std::cout << "Waited for 3 seconds!\n";

Dormir hasta 3 horas en el futuro usando std::this_thread::sleep_until :

void foo()
{
    std::this_thread::sleep_until(std::chrono::system_clock::now() + std::chrono::hours(3));
}

std::thread thread{ foo };
thread.join();

std::cout << "We are now located 3 hours after the thread has been called\n";

Permitir que otros subprocesos tengan prioridad utilizando std::this_thread::yield :

void foo(int a)
{
    for (int i = 0; i < al ++i)
        std::this_thread::yield(); //Now other threads take priority, because this thread
                                   //isn't doing anything important

    std::cout << "Hello World!\n";
}

std::thread thread{ foo, 10 };
thread.join();

Usando std :: async en lugar de std :: thread

std::async también es capaz de hacer hilos. En comparación con std::thread , se considera menos potente pero más fácil de usar cuando solo desea ejecutar una función de forma asíncrona.

Asincrónicamente llamando a una función

#include <future>
#include <iostream>

unsigned int square(unsigned int i){
    return i*i;
}

int main() {
    auto f = std::async(std::launch::async, square, 8);
    std::cout << "square currently running\n"; //do something while square is running
    std::cout << "result is " << f.get() << '\n'; //getting the result from square
}

Errores comunes

• std::async devuelve un std::future que contiene el valor de retorno que la función calculará. Cuando ese future se destruye, espera hasta que se complete el subproceso, haciendo que su código sea efectivamente único. Esto se pasa por alto fácilmente cuando no necesita el valor de retorno:

  std::async(std::launch::async, square, 5);
  //thread already completed at this point, because the returning future got destroyed
  

• std::async funciona sin una política de lanzamiento, así que std::async(square, 5); compila Cuando haces eso, el sistema decide si quiere crear un hilo o no. La idea era que el sistema elige crear un subproceso a menos que ya esté ejecutando más subprocesos de los que puede ejecutar de manera eficiente. Desafortunadamente, las implementaciones comúnmente solo eligen no crear un subproceso en esa situación, por lo que es necesario anular ese comportamiento con std::launch::async que obliga al sistema a crear un subproceso.

• Cuidado con las condiciones de la carrera.

Más sobre asíncrono sobre futuros y promesas.

Asegurando un hilo siempre está unido

Cuando se invoca el destructor para std::thread se debe haber realizado una llamada a join() o detach() . Si un subproceso no se ha unido o separado, se llamará de forma predeterminada a std::terminate . Usando RAII , esto es generalmente bastante simple de lograr:

class thread_joiner
{
public:

    thread_joiner(std::thread t)
        : t_(std::move(t))
    { }

    ~thread_joiner()
    {
        if(t_.joinable()) {
            t_.join();
        }
    }

private:

    std::thread t_;
}

Esto se usa entonces como tal:

 void perform_work()
 {
     // Perform some work
 }

 void t()
 {
     thread_joiner j{std::thread(perform_work)};
     // Do some other calculations while thread is running
 } // Thread is automatically joined here

Esto también proporciona una excepción de seguridad; si hubiéramos creado nuestro hilo normalmente y el trabajo realizado en t() realizando otros cálculos hubiera generado una excepción, nunca se habría llamado join() en nuestro hilo y nuestro proceso habría terminado.

Reasignando objetos de hilo

Podemos crear objetos de hilo vacíos y asignarles trabajo más tarde.

Si asignamos un objeto de subproceso a otro subproceso activo, que se puede joinable , se llamará automáticamente a std::terminate antes de que se reemplace el subproceso.

#include <thread>

void foo()
{
    std::this_thread::sleep_for(std::chrono::seconds(3));
}
//create 100 thread objects that do nothing
std::thread executors[100];

// Some code

// I want to create some threads now

for (int i = 0;i < 100;i++)
{
    // If this object doesn't have a thread assigned
    if (!executors[i].joinable())
         executors[i] = std::thread(foo);
}

Sincronizacion basica

La sincronización de subprocesos se puede realizar utilizando mutexes, entre otras primitivas de sincronización. La biblioteca estándar proporciona varios tipos de exclusión mutua, pero el más simple es std::mutex . Para bloquear un mutex, construyes un bloqueo en él. El tipo de bloqueo más simple es std::lock_guard :

std::mutex m;
void worker() {
    std::lock_guard<std::mutex> guard(m); // Acquires a lock on the mutex
    // Synchronized code here
} // the mutex is automatically released when guard goes out of scope

Con std::lock_guard el mutex está bloqueado durante toda la vida útil del objeto de bloqueo. En los casos en que necesite controlar manualmente las regiones para el bloqueo, use std::unique_lock lugar:

std::mutex m;
void worker() {
    // by default, constructing a unique_lock from a mutex will lock the mutex
    // by passing the std::defer_lock as a second argument, we
    // can construct the guard in an unlocked state instead and
    // manually lock later.
    std::unique_lock<std::mutex> guard(m, std::defer_lock);
    // the mutex is not locked yet!
    guard.lock();
    // critical section
    guard.unlock();
    // mutex is again released
}

Más estructuras de sincronización de hilos

Uso de variables de condición

Una variable de condición es una primitiva que se usa junto con un mutex para orquestar la comunicación entre hilos. Si bien no es la forma exclusiva ni más eficiente de lograr esto, puede ser una de las más sencillas para aquellos familiarizados con el patrón.

Uno espera en un std::condition_variable con un std::unique_lock<std::mutex> . Esto permite que el código examine de forma segura el estado compartido antes de decidir si proceder o no con la adquisición.

A continuación se muestra un boceto productor-consumidor que usa std::thread , std::condition_variable , std::mutex , y algunos otros para hacer las cosas interesantes.

#include <condition_variable>
#include <cstddef>
#include <iostream>
#include <mutex>
#include <queue>
#include <random>
#include <thread>


int main()
{
    std::condition_variable cond;
    std::mutex mtx;
    std::queue<int> intq;
    bool stopped = false;

    std::thread producer{[&]()
    {
        // Prepare a random number generator.
        // Our producer will simply push random numbers to intq.
        //
        std::default_random_engine gen{};
        std::uniform_int_distribution<int> dist{};

        std::size_t count = 4006;    
        while(count--)
        {    
            // Always lock before changing
            // state guarded by a mutex and
            // condition_variable (a.k.a. "condvar").
            std::lock_guard<std::mutex> L{mtx};

            // Push a random int into the queue
            intq.push(dist(gen));

            // Tell the consumer it has an int
            cond.notify_one();
        }

        // All done.
        // Acquire the lock, set the stopped flag,
        // then inform the consumer.
        std::lock_guard<std::mutex> L{mtx};

        std::cout << "Producer is done!" << std::endl;

        stopped = true;
        cond.notify_one();
    }};

    std::thread consumer{[&]()
    {
        do{
            std::unique_lock<std::mutex> L{mtx};
            cond.wait(L,[&]()
            {
                // Acquire the lock only if
                // we've stopped or the queue
                // isn't empty
                return stopped || ! intq.empty();
            });

            // We own the mutex here; pop the queue
            // until it empties out.

            while( ! intq.empty())
            {
                const auto val = intq.front();
                intq.pop();

                std::cout << "Consumer popped: " << val << std::endl;
            }

            if(stopped){
                // producer has signaled a stop
                std::cout << "Consumer is done!" << std::endl;
                break;
            }

        }while(true);
    }};

    consumer.join();
    producer.join();
    
    std::cout << "Example Completed!" << std::endl;

    return 0;
}

Crear un grupo de subprocesos simple

Las primitivas de subprocesos de C ++ 11 son todavía un nivel relativamente bajo. Se pueden usar para escribir una construcción de nivel superior, como un grupo de hilos:

struct tasks {
  // the mutex, condition variable and deque form a single
  // thread-safe triggered queue of tasks:
  std::mutex m;
  std::condition_variable v;
  // note that a packaged_task<void> can store a packaged_task<R>:
  std::deque<std::packaged_task<void()>> work;

  // this holds futures representing the worker threads being done:
  std::vector<std::future<void>> finished;

  // queue( lambda ) will enqueue the lambda into the tasks for the threads
  // to use.  A future of the type the lambda returns is given to let you get
  // the result out.
  template<class F, class R=std::result_of_t<F&()>>
  std::future<R> queue(F&& f) {
    // wrap the function object into a packaged task, splitting
    // execution from the return value:
    std::packaged_task<R()> p(std::forward<F>(f));

    auto r=p.get_future(); // get the return value before we hand off the task
    {
      std::unique_lock<std::mutex> l(m);
      work.emplace_back(std::move(p)); // store the task<R()> as a task<void()>
    }
    v.notify_one(); // wake a thread to work on the task

    return r; // return the future result of the task
  }

  // start N threads in the thread pool.
  void start(std::size_t N=1){
    for (std::size_t i = 0; i < N; ++i)
    {
      // each thread is a std::async running this->thread_task():
      finished.push_back(
        std::async(
          std::launch::async,
          [this]{ thread_task(); }
        )
      );
    }
  }
  // abort() cancels all non-started tasks, and tells every working thread
  // stop running, and waits for them to finish up.
  void abort() {
    cancel_pending();
    finish();
  }
  // cancel_pending() merely cancels all non-started tasks:
  void cancel_pending() {
    std::unique_lock<std::mutex> l(m);
    work.clear();
  }
  // finish enques a "stop the thread" message for every thread, then waits for them:
  void finish() {
    {
      std::unique_lock<std::mutex> l(m);
      for(auto&&unused:finished){
        work.push_back({});
      }
    }
    v.notify_all();
    finished.clear();
  }
  ~tasks() {
    finish();
  }
private:
  // the work that a worker thread does:
  void thread_task() {
    while(true){
      // pop a task off the queue:
      std::packaged_task<void()> f;
      {
        // usual thread-safe queue code:
        std::unique_lock<std::mutex> l(m);
        if (work.empty()){
          v.wait(l,[&]{return !work.empty();});
        }
        f = std::move(work.front());
        work.pop_front();
      }
      // if the task is invalid, it means we are asked to abort:
      if (!f.valid()) return;
      // otherwise, run the task:
      f();
    }
  }
};

tasks.queue( []{ return "hello world"s; } ) devuelve std::future<std::string> , que cuando el objeto de tareas se ejecuta, se rellena con hello world .

Puede crear subprocesos ejecutando tasks.start(10) (que inicia 10 subprocesos).

El uso de packaged_task<void()> se debe simplemente a que no hay una std::function borrado de tipo que almacene tipos de solo movimiento. Escribir uno personalizado de esos probablemente sería más rápido que usar packaged_task<void()> .

Ejemplo vivo .

Más sobre Mutexes .

Almacenamiento de hilo local

El almacenamiento local de subprocesos se puede crear utilizando la palabra clave thread_local . Una variable declarada con el especificador thread_local se dice que tiene duración de almacenamiento de subprocesos.

• Cada subproceso en un programa tiene su propia copia de cada variable de subproceso local.
• Una variable de subproceso local con función (local) se inicializará la primera vez que el control pase por su definición. Dicha variable es implícitamente estática, a menos que se declare extern .
• Una variable de subproceso local con espacio de nombres o ámbito de clase (no local) se inicializará como parte del inicio de subproceso.
• Las variables locales del hilo se destruyen al terminar el hilo.
• Un miembro de una clase solo puede ser subproceso local si es estático. Por lo tanto, habrá una copia de esa variable por subproceso, en lugar de una copia por par (hilo, instancia).

Ejemplo:

void debug_counter() {
    thread_local int count = 0;
    Logger::log("This function has been called %d times by this thread", ++count);
}