itmo_conspects

Программирование на C++ с элементами многопоточности

Лекция 1. Идиомы C++

Идиома RAII

С++ - язык с ручным управлением ресурсами, в нем нужно разработчику управлять выделением и освобождение памяти

Поэтому появилась идиома RAII (Resource Acquisition is Initialization) - “Получение ресурса есть инициализация”

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

Принципу RAII соответствуют умные указатели std::unique_ptr и мьютексы std::lock_guard. В случае мьютексов код, не следующии RAII, выглядит так:

{
    mutex1.lock()

    // критическая секция
    // здесь может быть return или throw

    mutex1.unlock()
}

Здесь нужно быть внимательным, чтобы mutex1.unlock() однозначно вызвался. std::lock_guard избавляет от этого:

{
    std::lock_guard lck(mutex1);
    // конструктор делает mutex1.lock() 

    // критическая секция

    // деструктор делает mutex1.unlock()
}

Деструктор вызывается в любом случае при выходе из скоупа, поэтому мьютекс будет однозначно разблокирован

Уникальный указатель std::unique_ptr

Умные указатели std::unique_ptr и пара std::shared_ptr-std::weak_ptr также следуют идиоме RAII и используются для управления памяти

Уникальный указатель std::unique_ptr владеет объектом единолично. Копирование такое указателя запрещено, но разрешено перемещение

std::unique_ptr создается с помощью std::make_unique, а не через конструктор. Рассмотрим такой пример:

Foo(new Bar(), std::unique_ptr(new Boo()));

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

  1. Создается new Bar()
  2. Создается new Boo(), выделив память, конструктор которого вызывает исключение
  3. Создается std::unique_ptr(new Boo())

Исключение перехватывается на уровне выше по стеку вызовов, но память, выделившаяся в конструкторе new Bar(), не освободиться, поэтому возникнет утечка памяти. Вместо этого лучше использовать std::make_unique:

Foo(std::make_unique<Bar>(), std::make_unique<Boo>());

Разделяемый указатель std::shared_ptr и слабый указатель std::weak_ptr

Разделяемый указатель std::shared_ptr используется, когда объект должен иметь несколько владельцев. Каждый разделяемый указатель увеличивает счетчик ссылок при создании и копировании, уменьшает при уничтожении, а когда счетчик становится равен нулю - объект уничтожается

Рекомендуется создавать std::shared_ptr с помощью std::make_shared по тем же причинам

Вместе с std::shared_ptr в комплекте идет тип слабого указателя. Слабый указатель std::weak_ptr создан для того, чтобы избавиться от кольцевых зависимостей

struct B;

struct A {
    std::shared_ptr<B> b;
};

struct B {
    std::shared_ptr<A> a;
};

Если A и B ссылаются друг на друга через std::shared_ptr, счетчик ссылок никогда не станет нулем, и возникнет утечка памяти

Слабый указатель weak_ptr используется вместе с shared_ptr, но сам по себе ничего не хранит. Вместо этого можно вызвать метод .lock(), который вернет shared_ptr, если объект существует внутри, или nullptr, если он был уничтожен (то есть больше нет живых std::shared_ptr):

std::shared_ptr<Foo> shptr = std::make_shared<Foo>();

// ...

std::weak_ptr<Foo> wptr = shptr;

if (auto sptr = wptr.lock()) {
    // объект еще существует
} else {
    // объект уже уничтожен
}

std::shared_ptr при создании через std::make_shared выделяет один непрерывный блок памяти c самим объектом, счетчиком разделяемых указателей и счетчиком слабых указателей. Это эффективнее по памяти и быстрее по времени

Если счетчик std::shared_ptr становится равен нулю, объект уничтожается, но управляющий блок (в котором хранятся счетчики) остается в памяти, пока существуют std::weak_ptr

Если подразумевается, что слабых указателей во время жизни программы будет много, то не рекомендуется использовать std::make_shared

Также копирование и уничтожение разделяемых указателей потокобезопасно, так как инкремент и декремент - атомарные операции

Функтор

Также для создания программ на C++ полезны функторы

Функтор (functor) — это объект, который можно вызывать как функцию. Он реализует оператор operator()

Функторы реализуют паттерн «Команда» — объект, хранящий в себе функцию и ее состояние.

Пример:

struct Add {
    int x;
    Add(int x) : x(x) {}

    int operator()(int y) const {
        return x + y;
    }
};

int main() {
    Add number(10);

    std::cout << number(5); // 15
}

Потоки

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

Лекция 2. Примитивы синхронизации в C++

В качестве объектов для синхронизации потоков используются классы:

std::mutex m;
int shared_counter = 0;

void increment() {
    std::lock_guard<std::mutex> lock(m); // lock() вызывается здесь
    ++shared_counter;
} // unlock() вызывается автоматически при разрушении lock

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

Также существуют модификация мьютекса – разделенный мьютекс. Разделенный мьютекс (или RW-mutex) std::shared_mutex работает аналогично обычному мьютексу, но допускает несколько одновременных блокировок для чтения и только одну для записи. Это полезно, когда операции чтения значительно преобладают над записью:

Операция Обёртка Метод мьютекса
Запись (единичный доступ) std::unique_lock lock()
Чтение (совместный доступ) std::shared_lock lock_shared()

Пример:

std::shared_mutex rw_mutex;
std::map<int, std::string> data;

// Читать могут несколько потоков одновременно
std::string read(int key) {
    std::shared_lock lock(rw_mutex);
    return data.at(key);
}

// Писать может только один поток
void write(int key, std::string value) {
    std::unique_lock lock(rw_mutex);
    data[key] = std::move(value);
}

Тип std::atomic<T>

Рассмотрим простой счётчик, к которому обращаются два потока:

int counter = 0;

void increment() {
    ++counter; // НЕ атомарная операция!
}

На уровне машинных инструкций ++counter — это три отдельных шага:

lw      a5,-20(s0)    # прочитать значение counter из памяти в регистр
addi    a5,a5,1       # прибавить 1 к регистру  
sw      a5,-20(s0)    # записать результат обратно в память

Если два потока выполняют эти шаги одновременно, возникает гонка обновлений:

  1. Поток A: lw a5,-20(s0) - читает 0
  2. Поток B: lw a5,-20(s0) - читает 0
  3. Поток A: addi a5,a5,1, sw a5,-20(s0) - пишет 1
  4. Поток B: addi a5,a5,1, sw a5,-20(s0) - пишет 1, но должен быть 2

Тип std::atomic<T> позволяет выполнять такие операции над переменной атомарно без явного использования мьютекса:

std::atomic<int> counter{0};

// Безопасно из нескольких потоков без мьютекса
void increment() {
    counter.fetch_add(1); // или просто ++counter
}

std::atomic<T> гарантирует, что все операции над переменной выполняются неделимо, без возможности вмешательства другого потока. Под капотом процессор использует специальные инструкции (например, LOCK XADD, CMPXCHG на x86-архитектуре) или шины памяти, которые делают операцию неделимой на аппаратном уровне - без блокировок и переключения контекста

Основные методы:

Каждый метод atomic принимает опциональный параметр std::memory_order, который управляет тем, как компилятор и процессор могут переупорядочивать инструкции вокруг атомарной операции, например, x.store(1, std::memory_order_relaxed);

Memory Order Смысл
relaxed Никаких гарантий порядка - только атомарность самой операции, поэтому максимальная производительность
acquire Все последующие операции с памятью в этом потоке выполнятся после этой загрузки, используется при load
release Все предшествующие операции с памятью в этом потоке выполнятся до этой записи, используется при store
acq_rel Комбинация acquire и release, используется для exchange, fetch_add
seq_cst Полная последовательная согласованность. Самый безопасный, поэтому значение по умолчанию, но самый медленный

Типичный паттерн acquire/release — передача данных между потоками без мьютекса:

std::atomic<bool> ready{false};
int data = 0;

// Поток A (производитель)
void producer() {
    data = 42;                               // (1) запись данных
    ready.store(true, std::memory_order_release); // (2) публикуем флаг
    // release гарантирует, что (1) виден до (2)
}

// Поток B (потребитель)
void consumer() {
    while (!ready.load(std::memory_order_acquire)); // (3) ждём флага
    // acquire гарантирует, что после (3) мы видим (1)
    assert(data == 42); // всегда верно
}

Без acquire и release компилятор или процессор мог бы переставить инструкции так, что data читался бы до того, как производитель его записал

Если нет уверенности, какой порядок использовать, то лучше оставить значение по умолчанию seq_cst, так как оптимизировать такое стоит только тогда, когда есть реальная проблема производительности

Тип std::recursive_mutex

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

Типичный случай - это методы класса, которые вызывают друг друга, при этом каждый берёт блокировку:

class SafeCollection {
    std::recursive_mutex m;
    std::vector<int> data;

public:
    void add(int x) {
        std::lock_guard lock(m);
        data.push_back(x);
    }

    void add_twice(int x) {
        std::lock_guard lock(m); // первый захват
        add(x);                  // второй захват того же мьютекса
        add(x);
    }
};

С обычным std::mutex вызов add() внутри add_twice() привёл бы к взаимной блокировке, так как поток попытался бы заблокировать уже захваченный им мьютекс

Тип std::scoped_lock

std::scoped_lock - аналог std::lock_guard, но позволяет захватить несколько мьютексов одновременно, избегая взаимной блокировки:

std::mutex m1, m2;

void transfer(Account& from, Account& to, int amount) {
    // Захватываем оба мьютекса атомарно
    std::scoped_lock lock(m1, m2);
    from.balance -= amount;
    to.balance   += amount;
}

Если бы два потока захватывали m1 и m2 в разном порядке по отдельности - была бы классическая взаимная блокировка, а тип scoped_lock этого не допускает

Spinlock

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

#include <atomic>

class Spinlock {
    std::atomic_flag flag = ATOMIC_FLAG_INIT;
public:
    void lock()   { while (flag.test_and_set(std::memory_order_acquire)); }
    void unlock() { flag.clear(std::memory_order_release); }
};

Петля событий

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

Создаём поток; если есть разделяемый ресурс (очередь задач) — создаём мьютекс для доступа к нему. Вот как выглядит типичный event loop:

std::queue<std::function<void()>> tasks;
std::mutex m;

Task t;
while (true) {
    {
        std::lock_guard l(m);
        if (!tasks.empty()) {
            t = std::move(tasks.front());
            tasks.pop();
        }
    } // мьютекс освобождается здесь

    if (t) {
        t();
        t = nullptr;
    }

    std::this_thread::sleep_for(std::chrono::nanoseconds(100));
}

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

std::condition_variable

std::condition_variable позволяет потоку заснуть до наступления условия и быть разбуженным другим потоком

std::mutex m;
std::queue<std::function<void()>> tasks;
std::condition_variable cv;
bool done = false;

// Поток-потребитель
void worker() {
    while (true) {
        std::unique_lock lock(m); // condition_variable требует unique_lock

        // Засыпаем, пока очередь пуста (и не завершаем работу)
        cv.wait(lock, [] { return !tasks.empty() || done; });

        if (done && tasks.empty()) break;

        auto task = std::move(tasks.front());
        tasks.pop();
        lock.unlock(); // освобождаем мьютекс перед выполнением задачи

        task();
    }
}

// Поток-производитель
void enqueue(std::function<void()> task) {
    {
        std::lock_guard lock(m);
        tasks.push(std::move(task));
    }
    cv.notify_one(); // будим одного спящего потребителя
}

Ключевые методы:

Метод Описание
cv.wait(lock, predicate) Засыпает, пока предикат не вернёт true. Атомарно освобождает мьютекс при засыпании и снова захватывает при пробуждении.
cv.notify_one() Будит один ожидающий поток.
cv.notify_all() Будит все ожидающие потоки.

condition_variable::wait должен временно освободить мьютекс, пока поток спит - lock_guard этого не умеет, а unique_lock поддерживает ручное lock() и unlock()

Также важно заметить, что поток может проснуться без вызова notify. Именно поэтому wait принимает предикат - без него нужно писать цикл while (!predicate()) cv.wait(lock); вручную

This site is open source. Improve this page.