C++
Многопоточность

Синтаксис

• нить()
• thread (thread && other)
• явный поток (Function && func, Args && ... args)

параметры

замечания

Режимы работы

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

Часто, в какой-то момент, вам нужно (возможно, поток уже может быть выполнен) ждать завершения потока, потому что вы хотите использовать результат, например.

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

Вы также можете detach резьбу, разрешив ее выполнить:

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

Передача ссылки на поток

Вы не можете передать ссылку (или ссылку на const ) непосредственно в поток, потому что std::thread будет копировать / перемещать их. Вместо этого используйте 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

Создание std :: thread

В C ++ потоки создаются с использованием класса std :: thread. Нить - отдельный поток исполнения; это аналогично тому, как помощник выполняет одну задачу, одновременно выполняя другую. Когда весь код в потоке выполняется, он завершается . При создании потока вам нужно передать что-то, что будет выполнено на нем. Несколько вещей, которые вы можете передать в поток:

• Бесплатные функции
• Функции участников
• Объекты Functor
• Лямбда-выражения

Пример бесплатной функции - выполняет функцию в отдельном потоке ( Live Example ):

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

Пример функции-члена - выполняет функцию-член в отдельном потоке ( Live Example ):

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

Пример объекта functor (пример Live ):

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

Пример выражения Lambda (пример Live ):

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

Операции над текущей нитью

std::this_thread - это namespace которое имеет функции делать интересные вещи в текущем потоке из функции, из которой она вызывается.

Получение текущего идентификатора потоков с помощью 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

Спящий в течение 3 секунд с использованием 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";

Спящий до 3 часов в будущем с использованием 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";

Разрешить другим потокам приоритет, используя 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();

Использование std :: async вместо std :: thread

std::async также может создавать потоки. По сравнению с std::thread он считается менее мощным, но более простым в использовании, когда вы просто хотите запустить функцию асинхронно.

Асинхронное вызов функции

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

Общие проблемы

• std::async возвращает std::future которое содержит возвращаемое значение, которое будет вычисляться функцией. Когда это future будет уничтожено, он ждет, пока поток не завершится, делая ваш код эффективно одиночным. Это легко упустить из виду, когда вам не требуется возвращаемое значение:

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

• std::async работает без политики запуска, поэтому std::async(square, 5); компилирует. Когда вы это сделаете, система решит, хочет ли он создать поток или нет. Идея заключалась в том, что система выбирает поток, если в нем уже запущено больше потоков, чем он может работать эффективно. К сожалению, в реализациях обычно просто не нужно создавать поток в этой ситуации, поэтому вам нужно переопределить это поведение с помощью std::launch::async который заставляет систему создавать поток.

• Остерегайтесь условий гонки.

Подробнее о асинхронных фьючерсах и обещаниях

Обеспечение непрерывности резьбы

Когда деструктор std::thread вызывается, вызов либо join() или detach() , должно быть сделано. Если поток не был соединен или отсоединен, то по умолчанию будет вызываться std::terminate . Используя RAII , это, как правило, достаточно просто для выполнения:

class thread_joiner
{
public:

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

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

private:

    std::thread t_;
}

Затем это используется так:

 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

Это также обеспечивает безопасность исключений; если бы мы создали наш поток нормально, и работа, выполненная в t() выполняющая другие вычисления, вызвала исключение, join() никогда бы не вызывался в нашем потоке, и наш процесс был бы прекращен.

Переназначение объектов потоков

Мы можем создавать пустые объекты потока и назначать им работу позже.

Если мы назначим объект потока другому активному, joinable потоку, std::terminate будет вызываться автоматически до замены потока.

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

Базовая синхронизация

Синхронизация потоков может быть выполнена с использованием мьютексов, среди других примитивов синхронизации. Существует несколько типов мьютексов, предоставляемых стандартной библиотекой, но самым простым является std::mutex . Чтобы заблокировать мьютекс, вы создаете на нем блокировку. Самый простой тип блокировки - 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

С std::lock_guard мьютекс заблокирован на весь срок службы объекта блокировки. В случаях, когда вам необходимо вручную управлять регионами для блокировки, вместо этого используйте 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
}

Дополнительные структуры синхронизации потоков

Использование переменных условий

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

Один из них ожидает std::condition_variable с std::unique_lock<std::mutex> . Это позволяет коду безопасно исследовать состояние общего доступа, прежде чем принимать решение о продолжении приобретения.

Ниже представлен эскиз производителя-потребителя, который использует std::thread , std::condition_variable , std::mutex и несколько других, чтобы сделать интересными вещи.

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

Создайте простой пул потоков

C ++ 11-примитивы с потоками по-прежнему относительно низки. Они могут использоваться для написания конструкции более высокого уровня, например пула потоков:

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; } ) возвращает std::future<std::string> , который, когда объект задач приближается к запуску, заполняется hello world .

Вы создаете потоки, запустив tasks.start(10) (который запускает 10 потоков).

Использование packaged_task<void()> заключается только в том, что нет эквивалента std::function типу, который хранит типы только для перемещения. Написание пользовательского одного из них, вероятно, будет быстрее, чем использование packaged_task<void()> .

Живой пример .

Подробнее о мьютексах .

Потоковое локальное хранилище

thread_local локальное хранилище может быть создано с использованием thread_local слова thread_local . thread_local переменная, объявленная с thread_local спецификатора thread_local имеет длительность хранения потоков.

• Каждый поток в программе имеет свою собственную копию каждой локальной переменной.
• Локальная переменная потока с функциональной (локальной) областью будет инициализирована, первый элемент управления проходит через его определение. Такая переменная неявно статична, если не объявлен extern .
• Локальная переменная потока с пространством имен или классом (нелокальной) будет инициализирована как часть запуска потока.
• Потоковые локальные переменные уничтожаются при завершении потока.
• Член класса может быть только поточно-локальным, если он статичен. Следовательно, будет одна копия этой переменной для потока, а не одна копия в (поток, экземпляр).

Пример:

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