C++
threading

Sintassi

• filo()
• thread (thread e altro)
• thread esplicito (Function && func, Args && ... args)

Parametri

Osservazioni

Operazioni di thread

Quando si avvia un thread, verrà eseguito fino al completamento.

Spesso, ad un certo punto, è necessario (probabilmente - il thread potrebbe già essere fatto) attendere il termine del thread, perché si desidera utilizzare il risultato, ad esempio.

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

Puoi anche detach il thread, lasciandolo eseguire liberamente:

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

Passando un riferimento a un thread

Non è possibile passare un riferimento (o riferimento const ) direttamente a un thread perché std::thread lo copia / sposta. Invece, usa 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

Creare uno std :: thread

In C ++, i thread vengono creati utilizzando la classe std :: thread. Una discussione è un flusso separato di esecuzione; è analogo al fatto che un aiutante esegua un compito mentre contemporaneamente ne esegue un altro. Quando viene eseguito tutto il codice nel thread, termina . Quando si crea un thread, è necessario passare qualcosa da eseguire su di esso. Alcune cose che puoi passare ad un thread:

• Funzioni libere
• Funzioni membro
• Oggetti Functor
• Espressioni Lambda

Esempio di funzione libera - esegue una funzione su un thread separato ( esempio dal 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;
}

Esempio di funzione membro - esegue una funzione membro su un thread separato ( esempio dal 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;
}

Esempio di oggetto Functor (esempio dal 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;
}

Esempio di espressione Lambda (esempio dal 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;
}

Operazioni sul thread corrente

std::this_thread è uno namespace che ha funzioni per fare cose interessanti sul thread corrente dalla funzione da cui è chiamato.

Ottenere l'id corrente dei thread 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

Dormire per 3 secondi 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";

Dormire fino a 3 ore in 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";

Lasciando che altri thread std::this_thread::yield priorità usando 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 invece di std :: thread

std::async è anche in grado di creare thread. Rispetto a std::thread è considerato meno potente ma più facile da usare quando si desidera eseguire una funzione in modo asincrono.

Chiamata in modo asincrono di una funzione

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

Insidie ​​comuni

• std::async restituisce uno std::future che contiene il valore restituito che verrà calcolato dalla funzione. Quando quel future viene distrutto, attende il completamento del thread, rendendo il tuo codice effettivamente thread singolo. Questo è facilmente trascurato quando non hai bisogno del valore di ritorno:

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

• std::async funziona senza una politica di avvio, quindi std::async(square, 5); compila. Quando lo fai, il sistema decide se vuole creare o meno una discussione. L'idea era che il sistema scelga di creare un thread a meno che non stia già eseguendo più thread di quanto possa funzionare in modo efficiente. Sfortunatamente le implementazioni comunemente scelgono di non creare un thread in quella situazione, mai, quindi è necessario sovrascrivere quel comportamento con std::launch::async che forza il sistema a creare un thread.

• Attenzione alle condizioni di gara.

Altro su async su Futures and Promises

Assicurandosi che un thread sia sempre unito

Quando viene chiamato il distruttore per std::thread , deve essere stata effettuata una chiamata a join() o a detach() . Se un thread non è stato unito o rimosso, per default verrà chiamato std::terminate . Usando RAII , questo è generalmente abbastanza semplice da realizzare:

class thread_joiner
{
public:

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

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

private:

    std::thread t_;
}

Questo viene quindi utilizzato in questo modo:

 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

Questo fornisce anche la sicurezza delle eccezioni; se avessimo creato normalmente il nostro thread e il lavoro svolto in t() eseguendo altri calcoli avesse generato un'eccezione, join() non sarebbe mai stato chiamato sul nostro thread e il nostro processo sarebbe stato terminato.

Riassegnazione degli oggetti thread

Possiamo creare oggetti thread vuoti e assegnare loro il lavoro in un secondo momento.

Se assegniamo un oggetto thread a un altro thread joinable attivo, std::terminate verrà chiamato automaticamente prima che il thread venga sostituito.

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

Sincronizzazione di base

La sincronizzazione dei thread può essere eseguita utilizzando i mutex, tra le altre primitive di sincronizzazione. Esistono diversi tipi di mutex forniti dalla libreria standard, ma il più semplice è std::mutex . Per bloccare un mutex, costruisci un blocco su di esso. Il tipo di blocco più semplice è 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 il mutex è bloccato per l'intera vita dell'oggetto lock. Nei casi in cui è necessario controllare manualmente le regioni per il blocco, utilizzare invece std::unique_lock :

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
}

Altre strutture di sincronizzazione dei thread

Uso delle variabili di condizione

Una variabile di condizione è una primitiva usata in congiunzione con un mutex per orchestrare la comunicazione tra i thread. Anche se non è né il modo esclusivo né il modo più efficace per farlo, può essere tra i più semplici per coloro che hanno familiarità con il modello.

Si attende una std::unique_lock<std::mutex> std::condition_variable con std::unique_lock<std::mutex> . Ciò consente al codice di esaminare in sicurezza lo stato condiviso prima di decidere se procedere o meno con l'acquisizione.

Di seguito è riportato uno schizzo produttore-consumatore che utilizza std::thread , std::condition_variable , std::mutex e pochi altri per rendere le cose interessanti.

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

Creare un pool di thread semplice

I primitivi di filettatura del C ++ 11 sono ancora relativamente bassi. Possono essere usati per scrivere un costrutto di livello superiore, come un pool di thread:

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; } ) restituisce uno std::future<std::string> , che quando l'oggetto delle attività ottiene l'esecuzione, viene popolato con hello world .

Puoi creare thread eseguendo tasks.start(10) (che avvia 10 thread).

L'uso di packaged_task<void()> è semplicemente perché non esiste un equivalente std::function cancellato dal tipo che memorizza i tipi di spostamento. Scrivendo uno di quelli personalizzati sarebbe probabilmente più veloce di usare packaged_task<void()> .

Esempio dal vivo

Altro su Mutex .

Memorizzazione locale del thread

L'archiviazione locale del thread può essere creata usando la parola chiave thread_local . Si dice che una variabile dichiarata con l' thread_local abbia durata di memorizzazione del thread.

• Ogni thread in un programma ha la propria copia di ogni variabile locale del thread.
• Una variabile locale del thread con ambito di funzione (locale) verrà inizializzata al primo passaggio del controllo attraverso la sua definizione. Tale variabile è implicitamente statica, a meno che non sia dichiarata extern .
• Una variabile locale del thread con spazio dei nomi o classe (non locale) verrà inizializzata come parte dell'avvio del thread.
• Le variabili locali del thread vengono distrutte al termine della terminazione del thread.
• Un membro di una classe può essere thread-local solo se è statico. Ci sarà quindi una copia di quella variabile per thread, piuttosto che una copia per coppia (thread, istanza).

Esempio:

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