Contention
#опытным

Thread Contention (соревнование потоков) — это ситуация в многопоточном программировании, когда несколько потоков одновременно пытаются получить доступ к одному и тому же разделяемому ресурсу, но только один поток может использовать его в данный момент времени.

Это нормальная ситуация, на любом мьютексе потоки соревнуются. Но иногда это выходит за грани нормальности.

Многопоточное программирование же у нас должно повышать эффективность вычислений за счет разделения потоков обработки данных на независимые части и помещать их на свои потоки исполнения. Однако рано или поздно наступает приход в точку синхронизации: потоки конкурируют между собой за доступ к разделяемым данным.

И вот тут может появиться проблема. Один ресурс, а желающих завладеть им слишком много. Только один в итоге овладевает, а все остальные отправляются спать. И это конечно приводит к простою потоков и замедление общего прогресса.

template <Key, Value>
class ThreadSafeMap {
    mutable std::mutex mtx;
    std::map<Key, Value> map;

public:
    void Insert(const Key &key, const Value &value) {
        std::lock_guard lg{mtx};
        map.insert(key, value);
    }
    Value &Get(const Key &key) const {
        std::lock_guard lg{mtx};
        return map.at(key);
    }
};

Если к такой мапе одновременно будет получать доступ куча потоков, то все кроме одного будут простаивать. А если таких потоков 10 или 20? Неприятненько.

Как можно снизить Contention?

👉🏿 Read-Write Lock. Если у вас много читателе и мало писателей, то можно разрешить нескольким читателям одновременно получать доступ к данным с помощью std::shared_mutex:

template <Key, Value>
class ThreadSafeMap {
    mutable std::shared_mutex mtx;
    std::map<Key, Value> map;

public:
    void Insert(const Key &key, const Value &value) {
        std::unique_lock ul{mtx};
        map.insert(key, value);
    }
    Value &Get(const Key &key) const {
        std::shared_lock sl{mtx};
        return map.at(key);
    }
};

👉🏿 Thread-Local Storage. Потоки пишут данные в свои локальные буферы, которые централизованно синхронизируют данные друг с другом, чтобы как можно меньше блокировать потоки.

👉🏿 Можно организовать свою структуру данных так, чтобы у нее была ячеистая структура и к каждой ячейке был отдельный замок. Теперь потребители данных распределятся по разным ячейкам и не будут толкаться.

template <Key, Value>
class FineGrainedMap {
    struct Node {
        std::mutex mtx;
        std::map<Key, Value> data;
    };
    std::vector<Node> buckets{16};  // Много мелких блокировок
    
public:
    Value& Get(const Key& key) const {
        auto& bucket = buckets[std::hash<Key>{}(key) % buckets.size()];
        std::lock_guard lock(bucket.mtx);
        return bucket.data.at(key);
    }
};

👉🏿 Используйте lock-free структуры данных. Ну как бы тут логично: нет мьютексов, нет и сontention. Не в каждой задаче это реально применить, но иногда все же можно.

Compete and win. Stay cool.

#concurrency

​​Starvation
#опытным

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

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

Эта сцена наглядно демонстрирует еще одну проблему многопоточного мира - starvation или голодание.

Голодовка в многопоточной передаче происходит, когда один или несколько потоков постоянно блокируются при доступе к ресурсам, в результате чего у них редко бывает возможность выполниться(потенциально никогда). В то время как дедлок замораживает все вовлеченные треды, голодание затрагивает только те невезучие потоки, которые остаются «ожидать в очереди», в то время как другие занимают все ресурсы.

Какие предпосылки появления голодания?

👉🏿 Приоритеты потоков. Хоть в стандарте С++ нельзя выставить приоритет потоков, это можно сделать, например, в pthreads. Потоки с большим приоритетом могут забирать всю работу у низкоприоритетных.

👉🏿 Короткий доступ к мьютексу. Есть два вида замков: справедливые и несправедливые. Поток, только что освободивший unfair мьютекс, имеет преимущество по его захвату, потому что мьютекс все еще может быть в кэше этого потока и у него еще не закончилось время на работу. И это может приводить к простую других потоков. Справедливая реализация учитывает порядок запроса блокировки мьютекса, например с помощью очереди.

👉🏿 Все хотят доступ к одному ресурсу. Когда много потоков пытаются получить доступ к ресурсу, охраняемому всего одним мьютексом, то полезную работу делает только один из них, а все остальные ждут.

👉🏿 Длинные задачи под мьютексом. В дополнение к предыдущему пункту. Мало того, что потоки просто долго ждут очереди, чтобы занять замок, так еще и каждый из них вечность делает свою задачу.

Простой пример:

std::mutex mtx;
int counter = 0;

void worker(int id) {
    for (int i = 0; i < 100; ++i) {
        std::lock_guard lg{mtx};
        ++counter;
        std::cout << "Thread " << id
                  << " entered critical section, counter = " << counter
                  << std::endl;
        // do work
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
}

int main() {
    std::jthread t1(worker, 1);
    std::jthread t2(worker, 2);
}

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

Но! Если вы добавите слип после релиза мьютекса, то картина становится более справедливой.

Как избавиться от голодания?

✅ Справедливый шедулинг и замки. В стандартных плюсах на это мы не можем повлиять, но в системном апи или самописных реализациях можем.

✅ Минимальный размер критической секции. Она должна менеждить хранение задачи, но не быть ответственной за выполненеие задачи. Это позволит ограничивать простой других потоков.

✅ Грамотно проектируйте разделяемые данные. Если у вас 100 потоков пинают одну несчастную потокобезопасную мапу, то есть высока вероятность пересмотреть архитектуру и межпоточное взаимодействие.

✅ Давайте возможность другим войти в критическую секцию. Учитывая второй пункт, поток, который постоянно стучится в критическую секцию, скорее всего выполняет в ней лишний код. Разгрузите секцию, займите поток чем-нибудь в перерывах между критическими секциями и будет вам счастье.

Remember that you have the highest priority. Stay cool.

#concurrency

Голодание. Приоритетные очереди
#опытным

Голодание бывает не только у потоков, но и у других сущностей с приоритетами.

Допустим у вас есть система задач с 3-мя приоритетами: High, Medium, Low. Продюсеры кладут каждую задачу в очередь, соответствующую ее приоритету. А консюмеры всегда должны потреблять задачи с самым высоким возможным приоритетом.

То есть, пока High очередь не опустеет, никто не будет брать Middle задачи. И никто не возьмет в обработку Low задачи, пока High и Middle очереди не пусты.

Может возникнуть такая ситуация, при которой задачи High будут постоянно приходить так, что обработчики редко будут брать задачи Middle и никогда не дойдут до Low очереди. Таким образом, эти очереди будут голодать от недостатка обработки.

class Scheduler {
private:
    std::vector<ThreadSafeQueue<std::string>> queues;
    std::vector<std::string> priority_names;

public:
    Scheduler() : queues(3), priority_names{"HIGH", "MEDIUM", "LOW"} {}

    std::string Get() {
        while(true) {
            for(int i = 0; i < queues.size(); ++i) {
                auto task = queues[i].take();
                if (!task)
                    continue;
                std::cout << "Get task " << priority_names[i] << ": " << task << std::endl;
                return task;
            }
            // some kind of waiting mechanism in case of every queue is full
            }
    }

    void AddTask(int priority, const std::string& task) {
        queues[priority].push(task);
        std::cout << "Add task " << priority_names[priority] << ": " << task << std::endl;
    }
};

Допустим, что эта проблема возникает не всегда, а только периодически. Если она постоянная, то проблема здесь в количестве обработчиков и/или их вычислительной мощности, либо вообще ваши задачи нужно обрабатывать как-то по-другому.

Кстати сам алгоритм называется Fixed-priority pre-emptive scheduling. В каждый момент времени выполняется задача с самым высоким приоритетом.

Решение проблемы - сменить алгоритм взятия задач из очередей.

Например, можно установить правило, что вы обрабатываете не более f(priority) элементов в любой данной очереди, прежде чем рассматривать элементы из очереди с более низким приоритетом.

Функция f может быть:

👉🏿 Линейной: f(p) = p. Обрабатывается не более 4 элементов с приоритетом 4 (высший), затем не более 3 с приоритетом 3,..., 1 с приоритетом 1.

👉🏿 Экспоненциальной: f(p) = 2^(p-1). Обрабатывается не более 8 элементов с приоритетом 4 (высший), затем не более 4 с приоритетом 3, затем не более 2 с приоритетом 2,..., 1 с приоритетом 1.

Конкретная функция выбирается из ожидаемой частоты появления задач

Возьмем экспоненциальный случай и предположим, что в каждой очереди много ожидающих задач. Мы планируем: 8 высших, 4 высоких, 2 средних, 1 низкий, 8 высших и т.д... Каждый цикл содержит 8 + 4 + 2 + 1 = 15 задач, поэтому задачи высшего приоритета занимают 8/15 времени потребителя, следующие — 4/15, следующие — 2/15, следующие — 1/15.

Сравниваем эти частоты с ожидаемыми и корректируем коэффициенты или используем другую функцию.

You are the highest priority. Stay cool.

#concurrency

​​Тулзы для поиска проблем многопоточности
#опытным

Мы уже с вами убедились, что в мире многопоточности куча проблем. И шанс на них наткнуться, мягко говоря, немаленький. А на самом деле почти любой мало-мальски полезный конкурентный код, написаный с нуля, будет содержать как минимум одну такую проблему.

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

✅ Юнит тесты. Код без тестов - деньги на ветер. Это я перефразировал известную поговорку, но она и в данном контексте хорошо отражает суть. Если вы не тестируете код, то проблема может проявиться в самый неподходящий момент и это может стоить вам кучу зеленых фантиков.

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

✅ Cppcheck. Пользуйтесь инструментами статического анализа, например Cppcheck. Определять проблемы синхронизации по коду программы - занятие конечно увлекательное и вряд ли вы много багов так найдете, но собственно почему бы и нет.

Надо лишь установить сам cppcheck, а запускается он просто:

cppcheck --enable=all --inconclusive thread_app.cpp

✅ Thread San. Без динамического анализа в многопоточке никуда. ThreadSanitizer - это детектор гонок данных для C/C++. Санитайзер определяет гонку ровно как в стандарте: если у вас много потоков получают доступ к ячейке памяти и хотя бы один из них - несинхронизированная запись. И это же и является принципом детектирования гонок.

Работает на GCC и Clang. Достаточно лишь при сборке указать нужные флаги и ждать прилета сообщений о багах:

clang++ -fsanitize=thread -g -O2 -o my_app main.cpp

g++ -fsanitize=thread -g -O2 -o my_app main.cpp

✅ Helgrind. Это одна из тулзов Valgrind'а, работающая конкретно с багами многопоточности. Достаточно при запуске валгринда указать --tool=helgrind и ждите писем счастья. Главное, чтобы ваши примитивы синхронизации использовали под капотом pthread.

Helgrind детектирует такие проблемы, как:
- разблокировка невалидного мьютекса
- разблокировка не заблокированного мьютекса
- разблокировка мьютекса, удерживаемого другим потоком
- уничтожение невалидного или заблокированного мьютекса
- рекурсивная блокировка нерекурсивного мьютекса
- освобождение памяти, содержащей заблокированный мьютекс
и еще кучу всего.

✅ Vtune. Не все проблемы конкурентности связаны с некорректным использованием инструментов. С точки зрения стандартов, программа может корректно работать, но в ней будут лайв локи или голодовки. Тогда нужен хороший профилировщик, способный отследить, например, влияние lock contention на общую производительность, неэффективную синхронизацию или неравномерную нагрузку между потоками.

vtune -collect threading -result-dir my_analysis ./my_application

VTune в принципе очень мощный профилировщик даже не касательно многопоточности. Если есть возможность заморочится с ним, то это стоит сделать.

Test your system. Stay cool.

#concurrency #tools

​​Двойной unlock
#опытным

Если не пользоваться RAII, то можно наткнуться на массу проблем. Все знают про double free. Но менее известна проблема double unlock.

Все просто, вы используете ручной lock-unlock мьютекса и возможно попадаете в ситуацию двойного освобождения:

void unsafe_function(int value) {
    mtx.lock();
    
    if (value < 0) {
        std::cout << "Error: negative value\n";
        mtx.unlock();
        // forget to return!
    }
    
    shared_data = value;
    std::cout << "Data has updated: " << shared_data << std::endl;
    
    mtx.unlock();  // second unlock
}

Практически всегда двойной unlock происходит из-за некорректного кода в той или иной степени. Забыть вызвать return кажется детской проблемой, но если вы например не написали тесты на эту ветку, то возможно вы наткнетесь на проблемы только в проде.

А проблемы могут быть примерно любыми. Потому что двойной unlock мьютекса - UB по стандарту. Соответственно, можете получить много непрятностей, от сегфолта до бесконечного ожидания.

Поэтому просто используйте RAII и спина болеть не будет:

void safe_function(int value) {
    std::lock_guard lg{mtx};
    
    if (value < 0) {
        std::cout << "Error: negative value\n";
        return;
    }
    
    shared_data = value;
    std::cout << "Data has updated: " << shared_data << std::endl;
}

Use safe technics. Stay cool.

#concurrency #cpp11