Синхронизация потоков и безопасная работа с общими данными в многопоточной среде на C++ и Qt

Представьте, что вы с другом пишете реферат и у вас один учебник на двоих. Вы сели писать реферат, написали несколько страниц, оставили учебник открытым с намерением дописать его позже и ушли по своим делам. В это время ваш друг, который живет с вами в одной комнате, вернулся домой, увидел, что у вас что-то написано на листах, а рядом лежит учебник и решил, что вы закончили свой реферат. Он начинает листать учебник, выкидывая все ваши закладки, находит нужный ему материал и пишет свой реферат. Спустя несколько часов вы возвращаетесь домой, ваш друг уже спит, и вы решаете дописать реферат. Садитесь за письменный стол и «о, ужас!» все ваши закладки сбиты и вы не можете найти то место, где остановились. Неприятная ситуация? А ведь такая же может возникнуть в многопоточной среде, когда одни и те же внешние данные используются несколькими потоками. О том, как с этим жить, я расскажу в этой статье.

Содержание статьи

1. Проблема
2. Мьютексы (mutex)
3. Семафоры (semaphore)
4. Переменные условия ­(Condition variables или Wait conditions)
5. Другие способы
6. Выводы

Описанная выше задача известная как race conditions – или просто “гонки”, когда несколько параллельных задач конкурируют за общие ресурсы. Решается она разграничением доступа к данным во времени. Для этих целей используются следующие сущности:

QtC++
QMutex
QSemaphore
QWaitCondition
std::mutex
std::semaphore
std::condition_variable

Применение этих сущностей я покажу на примерах и расскажу как они устроены и как их применять. Приведённый далее код будет написан на Qt, т. к. С++ код практически идентичен, есть лишь некоторые отличия между чистым С++ и Qt, и в таких случаях я приведу оба примера кода.

1. Проблема

Давайте перейдем к примерам и поставим задачу. Пусть имеется 2 банковских счета с 1000 рублей на каждом. Два пользователя одновременно инициируют операции перевода между ними: со счета №1 переводится 500 рублей на счет №2, а со счета №2 на счет №1 переводится 1200 рублей. Чтобы переводы выполнялись «одновременно», создадим объект выполняющий операцию перевода и его выполнение запустим в двух разных потоках.

#include <QCoreApplication>
#include <QThread>
#include <QDebug>

struct Account {
    int balance{1000};
};

class Transfer : public QThread {
    Account& from;
    Account& to;
    int amount;
    void run() override {
        if(from.balance >= amount) {
            from.balance -= amount;
            QThread::usleep(1);
            to.balance += amount;
        }
    }
public:
     Transfer(Account& from, 
     Account& to,
     int amount, 
     QObject* parent = nullptr) : 
     QThread(parent),
     from(from),
     to(to),
     amount(amount)
    {

    }
};

int main(int argc, char *argv[])
{
    QCoreApplication a(argc, argv);

    Account acc1;
    Account acc2;

    Transfer t1(acc1, acc2, 500);
    Transfer t2(acc2, acc1, 1200);

    t1.start();
    t2.start();

    t1.wait();
    t2.wait();

    qDebug() << "Acc1.balance=" << acc1.balance;
    qDebug() << "Acc2.balance=" << acc2.balance;


    return a.exec();
}

Запускаем программу и получаем следующий результат:

Acc1.balance = 500
Acc2.balance = 1500

Ну что, вас ничего не смущает? Как мы видим, выполнилась лишь первая операция — перевод 500 рублей с первого счета на второй, а перевести 1200 рублей со второго на первый не удалось, т. к. в момент начала операции на втором счете ещё не было нужной суммы. Давайте решать эту проблему.

2. Мьютексы (mutex)

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

QMutex m;
...
m.lock();
// Осуществляем монопольный доступ
m.unlock();

Будьте осторожны с мьютексом. Его надо обязательно отпустить, иначе его нельзя будет захватить. Поскольку это частая ошибка и в больших проектах выявить её очень сложно, то были придуманы сущности для работы с мьютексами с использованием идиомы RAII (Resource Acquisition Is Initialization).

QtC++
QMutexLockerstd::lock_guard

RAII — идиома, смысл которой заключается в том, что получение ресурса совмещено с инициализацией объекта, а освобождение с уничтожением объекта. Таким образом, вместе с уничтожением объекта QMutexLocker автоматически вызывается unlock мьютекса, если требуется. Давайте применим эти сущности для исправления нашего кода с помощью мьютексов.

#include <QCoreApplication>
#include <QThread>
#include <QDebug>
#include <QMutex>
#include <QMutexLocker>

QMutex m;

struct Account {
    int balance{1000};
};

class Transfer : public QThread {
    Account& from;
    Account& to;
    int amount;
    void run() override {
        QMutexLocker lk(&m);
        if(from.balance >= amount) {
            from.balance -= amount;
            QThread::usleep(1);
            to.balance += amount;
        }
    }
public:
    Transfer(Account& from, Account& to, int amount, QObject* parent = nullptr) :
     QThread(parent),
     from(from),
     to(to),
     amount(amount)
    {

    }
};

int main(int argc, char *argv[])
{
    QCoreApplication a(argc, argv);

    Account acc1;
    Account acc2;

    Transfer t1(acc1, acc2, 500);
    Transfer t2(acc2, acc1, 1200);

    t1.start();
    t2.start();

    t1.wait();
    t2.wait();

    qDebug() << "Acc1.balance=" << acc1.balance;
    qDebug() << "Acc2.balance=" << acc2.balance;


    return a.exec();
}

Получаем результат:

Acc1.balance = 1700
Acc2.balance = 300

Другое дело!

3. Семафоры (semaphore)

Семафоры, по сути своей, есть те же мьютексты, но с одним отличием. Семафор является счетным объектом, т. к. его можно выставить несколько раз и потом столько же раз его можно отпустить. Т.е. мьютекс является как бы бинарным — выставлен или нет, а семафор является счетчиком.

Для дальнейших примером давайте рассматривать следующую задачу. Первый поток — это работодатель, который каждую секунду в течение 12 секунд перечисляет в кошелек отцу 400 рублей. У отца есть сын Вася, который просит у отца на мороженное стоимостью 500 рублей. Если у отца достаточно денег, он выдает ему, если денег недостаточно, то и мороженного нет. Баланс кошелька мы можем хранить в семафоре. Тогда получим следующее решение:

#include <QCoreApplication>
#include <QThread>
#include <QDebug>
#include <QSemaphore>

struct Account {
    QSemaphore balance;
};

class BuyIceCream : public QThread {
    Account& from;
    static const int cost = 500;
    QString name;
    void run() override {
        while (1) {
            from.balance.acquire(cost);
            qDebug() << name << "! Enough money! Wallet = " << from.balance.available()+cost << "-" << cost << "=" << from.balance.available();
        }


    }
public:
    BuyIceCream(QStringView name, Account& from, QObject* parent = nullptr) :
    QThread(parent),
     from(from),
     name(name.toString())
{}
};

class Transfer : public QThread {
    Account& to;
    int amount;
    void run() override {
        for(int i = 0; i < 12; ++i) {
            QThread::msleep(1000);

            qDebug() << i+1 << ") Transfer from WORK: " << amount << ". Wallet = " << to.balance.available()+amount;
            to.balance.release(amount);

        }
    }
public:
    Transfer(Account& to, int amount, QObject* parent = nullptr) :
    QThread(parent),
    to(to),
    amount(amount)
    {

    }
};

int main(int argc, char *argv[])
{
    QCoreApplication a(argc, argv);

    Account fathersWallet;

    Transfer t1(fathersWallet, 400);
    BuyIceCream son1(u"Vasya", fathersWallet);

    t1.start();
    son1.start();

    t1.wait();

    son1.quit();


    return a.exec();
}

и выводимый результат:

1 ) Transfer from WORK:  400 . Wallet =  400
2 ) Transfer from WORK:  400 . Wallet =  800
"Vasya" ! Enough money! Wallet =  800 - 500 = 300
3 ) Transfer from WORK:  400 . Wallet =  700
"Vasya" ! Enough money! Wallet =  700 - 500 = 200
4 ) Transfer from WORK:  400 . Wallet =  600
"Vasya" ! Enough money! Wallet =  600 - 500 = 100
5 ) Transfer from WORK:  400 . Wallet =  500
"Vasya" ! Enough money! Wallet =  500 - 500 = 0
6 ) Transfer from WORK:  400 . Wallet =  400
7 ) Transfer from WORK:  400 . Wallet =  800
"Vasya" ! Enough money! Wallet =  800 - 500 = 300
8 ) Transfer from WORK:  400 . Wallet =  700
"Vasya" ! Enough money! Wallet =  700 - 500 = 200
9 ) Transfer from WORK:  400 . Wallet =  600
"Vasya" ! Enough money! Wallet =  600 - 500 = 100
10 ) Transfer from WORK:  400 . Wallet =  500
"Vasya" ! Enough money! Wallet =  500 - 500 = 0
11 ) Transfer from WORK:  400 . Wallet =  400
12 ) Transfer from WORK:  400 . Wallet =  800
"Vasya" ! Enough money! Wallet =  800 - 500 = 300

4. Переменные условия­ (Condition variables или Wait conditions)

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

QtC++
QWaitConditionstd::condition_vairable

Давайте также разберем на примере задачи про отца и мороженное, только теперь у отца будет два сына, содержимое кошелька будет снова определяться переменной типа int, а не семафором.

#include <QCoreApplication>
#include <QThread>
#include <QDebug>
#include <QWaitCondition>
#include <QMutex>

QMutex m;

struct Account {
    int balance{0};
    QWaitCondition enoughMoney;
};

class BuyIceCream : public QThread {
    Account& from;
    static const int cost = 500;
    QString name;
    void run() override {
        while (1) {
            QMutexLocker lk(&m);
            from.enoughMoney.wait(&m);

            if(from.balance >= cost) {
                qDebug() << name << "! Enough money! Wallet = " << from.balance << "-" << cost << "=" << from.balance-cost;
                from.balance -= cost;
            }
        }


    }
public:
    BuyIceCream(QStringView name, Account& from, QObject* parent = nullptr) :
     QThread(parent),
    from(from),
     name(name.toString())
 {}
};

class Transfer : public QThread {
    Account& from;
    Account& to;
    int amount;
    void run() override {
        for(int i = 0; i < 12; ++i) {
            QThread::msleep(1000);

            QMutexLocker lk(&m);
            if(from.balance >= amount) {
                from.balance -= amount;
                to.balance += amount;
                qDebug() << i+1 << ") Transfer from WORK: " << amount << ". Wallet = " << to.balance;
            }

            to.enoughMoney.notify_one();
        }
    }
public:
    Transfer(Account& from, Account& to, int amount, QObject* parent = nullptr) :
     QThread(parent),
     from(from), to(to),
     amount(amount)
    {

    }
};

int main(int argc, char *argv[])
{
    QCoreApplication a(argc, argv);

    Account work;
    work.balance = 10'0000'000;
    Account fathersWallet;

    Transfer t1(work, fathersWallet, 900);
    BuyIceCream son1(u"Vasya", fathersWallet);
    BuyIceCream son2(u"Petya", fathersWallet);

    t1.start();
    son1.start();
    son2.start();

    t1.wait();

    son1.quit();
    son2.quit();


    return a.exec();
}

Обратите внимание, что в строке 54 мы вызываем метод notify_one(), который пробуждает один поток, поэтому лишь один сын сможет купить мороженное и отец может накапливать деньги.

Получаемый результат:

1 ) Transfer from WORK:  1400 . Wallet =  1400
"Vasya" ! Enough money! Wallet =  1400 - 500 = 900
2 ) Transfer from WORK:  1400 . Wallet =  2300
"Petya" ! Enough money! Wallet =  2300 - 500 = 1800
3 ) Transfer from WORK:  1400 . Wallet =  3200
"Vasya" ! Enough money! Wallet =  3200 - 500 = 2700
4 ) Transfer from WORK:  1400 . Wallet =  4100
"Petya" ! Enough money! Wallet =  4100 - 500 = 3600
5 ) Transfer from WORK:  1400 . Wallet =  5000
"Vasya" ! Enough money! Wallet =  5000 - 500 = 4500
6 ) Transfer from WORK:  1400 . Wallet =  5900
"Petya" ! Enough money! Wallet =  5900 - 500 = 5400
7 ) Transfer from WORK:  1400 . Wallet =  6800
"Vasya" ! Enough money! Wallet =  6800 - 500 = 6300
8 ) Transfer from WORK:  1400 . Wallet =  7700
"Petya" ! Enough money! Wallet =  7700 - 500 = 7200
9 ) Transfer from WORK:  1400 . Wallet =  8600
"Vasya" ! Enough money! Wallet =  8600 - 500 = 8100
10 ) Transfer from WORK:  1400 . Wallet =  9500
"Petya" ! Enough money! Wallet =  9500 - 500 = 9000
11 ) Transfer from WORK:  1400 . Wallet =  10400
"Vasya" ! Enough money! Wallet =  10400 - 500 = 9900
12 ) Transfer from WORK:  1400 . Wallet =  11300
"Petya" ! Enough money! Wallet =  11300 - 500 = 10800

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

Если же мы захотим порадовать мороженным обоих сыновей, при наличие такой возможности, то используем метод notify_all(). Для наглядного решения уменьшим зарплату отца до 900 рублей. Получим следующий результат:

1 ) Transfer from WORK:  900 . Wallet =  900
"Vasya" ! Enough money! Wallet =  900 - 500 = 400
2 ) Transfer from WORK:  900 . Wallet =  1300
"Vasya" ! Enough money! Wallet =  1300 - 500 = 800
"Petya" ! Enough money! Wallet =  800 - 500 = 300
3 ) Transfer from WORK:  900 . Wallet =  1200
"Vasya" ! Enough money! Wallet =  1200 - 500 = 700
"Petya" ! Enough money! Wallet =  700 - 500 = 200
4 ) Transfer from WORK:  900 . Wallet =  1100
"Vasya" ! Enough money! Wallet =  1100 - 500 = 600
"Petya" ! Enough money! Wallet =  600 - 500 = 100
5 ) Transfer from WORK:  900 . Wallet =  1000
"Vasya" ! Enough money! Wallet =  1000 - 500 = 500
"Petya" ! Enough money! Wallet =  500 - 500 = 0
6 ) Transfer from WORK:  900 . Wallet =  900
"Vasya" ! Enough money! Wallet =  900 - 500 = 400
7 ) Transfer from WORK:  900 . Wallet =  1300
"Vasya" ! Enough money! Wallet =  1300 - 500 = 800
"Petya" ! Enough money! Wallet =  800 - 500 = 300
8 ) Transfer from WORK:  900 . Wallet =  1200
"Vasya" ! Enough money! Wallet =  1200 - 500 = 700
"Petya" ! Enough money! Wallet =  700 - 500 = 200
9 ) Transfer from WORK:  900 . Wallet =  1100
"Vasya" ! Enough money! Wallet =  1100 - 500 = 600
"Petya" ! Enough money! Wallet =  600 - 500 = 100
10 ) Transfer from WORK:  900 . Wallet =  1000
"Vasya" ! Enough money! Wallet =  1000 - 500 = 500
"Petya" ! Enough money! Wallet =  500 - 500 = 0
11 ) Transfer from WORK:  900 . Wallet =  900
"Vasya" ! Enough money! Wallet =  900 - 500 = 400
12 ) Transfer from WORK:  900 . Wallet =  1300
"Petya" ! Enough money! Wallet =  1300 - 500 = 800
"Vasya" ! Enough money! Wallet =  800 - 500 = 300

Теперь в январе, июне и ноябре мороженное ест лишь один сын (к счастью для Васи и к несчастью для Пети).

В С++ CV предоставляют лаконичный интерфейс, позволяя совместить проверку условия и ожидание CV:

std::condition_variable cw
std::mutex m;
…
int balance;
std::unique_lock<std::mutex> lk(m);
cw.wait(lk, []{
    return (balance > cost);
});

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

5. Другие способы

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

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

QtC++
QReadWriteLockstd::shared_mutex
QReadLocker
QWriteLocker
std::shared_lock
std::lock_guard

Пример кода:

#include <QCoreApplication>
#include <QThread>
#include <QDebug>
#include <QStringView>
#include <QReadWriteLock>
#include <QStringList>
#include <QWriteLocker>
#include <QReadLocker>

QReadWriteLock lock;

QString msg { "Hello "};

class Transfer : public QThread {
    void run() override {
        QStringList list {"Hi", "Privet"};
        for(int i = 0; i < list.size(); ++i) {
            QThread::sleep(1);
            QWriteLocker locker(&lock);
            msg = list.at(i);
        }
        QThread::sleep(1);
    }
public:
    Transfer(QObject* parent = nullptr) :
        QThread(parent)
    {

    }
};

class Receive : public QThread {
    QString name;
    void run() override {
        while(1) {
            QReadLocker locker(&lock);
            qDebug() << msg << name;
            QThread::msleep(500);
        }
    }
public:
    Receive(QStringView name, QObject* parent = nullptr) :
        QThread(parent),
        name(name.toString())
    {

    }
};

int main(int argc, char *argv[])
{
    QCoreApplication a(argc, argv);

    Receive t1(u"Vasya");
    Receive t2(u"Petya");

    Transfer w;

    t1.start();
    t2.start();
    w.start();

    QObject::connect(&w, &QThread::finished, &t1, &QThread::terminate);
    QObject::connect(&w, &QThread::finished, &t2, &QThread::terminate);

    w.wait();


    return a.exec();
}

и результат:

"Hello " "Vasya"
"Hello " "Petya"
"Hello " "Vasya"
"Hello " "Petya"
"Hello " "Vasya"
"Hi" "Vasya"
"Hi" "Petya"
"Hi" "Vasya"
"Hi" "Petya"
"Hi" "Petya"
"Hi" "Vasya"
"Privet" "Petya"
"Privet" "Vasya"
"Privet" "Vasya"
"Privet" "Petya"

6. Выводы

В целом нет жестких рекомендаций о том, какие сущности где и как применять. Здесь вы вольны сами выбирать подходящий вам инструмент. Однако, не забывайте про то, что за всё приходится платить. Всегда задавайте себе вопрос, а как часто будет пробуждаться поток, как часто он будет работать, а не гоняю ли я его зря? А может есть более дешевый способ выполнить эту работу? Всё это так или иначе приходит с опытом. Пробуйте, практикуйтесь и исследуйте! Всего доброго!