БЛОГ НА HUSL


std::mutex и как его все таки нужно использовать для разных участков

Автор вопроса: Optimus1

Я все таки окончательно запутался в мьютексах и как их нужно правильно применять к синхронизации разных участков кода:

Вот два простых примера: различаются только доп. наличием переменной std::my_mutex1

1)

#include <iostream>           
    #include <thread>             
    #include <mutex>              
    #include <condition_variable> 
    #include <chrono> 
    
    std::mutex my_mutex;
    
    std::condition_variable my_cond_var;
    std::list<int>my_list;
    int global_int;
    
    
    
    void waiting_threads(int id)
    {
    repeat:
    
        std::cout << "waiting_threads_start" << std::endl;
    
        //----------------------------------------------------------------
        {
            std::unique_lock<std::mutex> my_lock(my_mutex);
    
    
            while (global_int == 0)
            {
                my_cond_var.wait(my_lock);
            }
            std::cout << "thread_id:" << std::this_thread::get_id() << ":" << id << std::endl; 
    
        }
        //----------------------------------------------------------------
    
        std::this_thread::sleep_for(std::chrono::seconds(3));
    
        my_mutex.lock();
        global_int = 0;
        my_mutex.unlock();
    
        goto repeat;
    }
    
    
    
    
    
    int main()
    {
        global_int = 0;
    
    
        //------------------------------------------------------------------------Создадим пул потоков:Начало-----------------------------
        std::vector<std::thread> my_thread_vec;
        my_thread_vec.resize(4);
    
        for (int i = 0; i < 4; i++)
        {
            my_thread_vec[i] = std::thread(waiting_threads, i);     
        }
        //------------------------------------------------------------------------Создадим пул потоков:Конец--------------------
    
    
        std::this_thread::sleep_for(std::chrono::seconds(1));
    
    
        my_mutex.lock();
        std::cout << "main_start" << std::endl;
        
        global_int = 1;
        my_cond_var.notify_one();    
        global_int = 1;
        my_cond_var.notify_one();   
        global_int = 1;
        my_cond_var.notify_one(); 
        global_int = 1;
        my_cond_var.notify_one();
        
        std::cout << "sleep_main_start" << std::endl;
        std::this_thread::sleep_for(std::chrono::seconds(10));
        std::cout << "sleep_main_end" << std::endl;
    
        global_int = 1;
        my_cond_var.notify_one();
        std::cout << "main_mutex_end" << std::endl;
        my_mutex.unlock();
    
    
    
        //------------------------------------------------Отпускаем потоки, так как у них бесконечный цикл---------------------------------------
        for (int i = 0; i < 4; i++)
        {
            my_thread_vec[i].join();
        }
        //---------------------------------------------------------------------------------------------------------------------------------------
    
    }

#include <iostream>           
#include <thread>             
#include <mutex>              
#include <condition_variable> 
#include <chrono> 

std::mutex my_mutex;
std::mutex my_mutex1;

std::condition_variable my_cond_var;
std::list<int>my_list;
int global_int;



void waiting_threads(int id)
{
repeat:

    std::cout << "waiting_threads_start" << std::endl;

    //----------------------------------------------------------------
    {
        std::unique_lock<std::mutex> my_lock(my_mutex);


        while (global_int == 0)
        {
            my_cond_var.wait(my_lock);
        }
        std::cout << "thread_id:" << std::this_thread::get_id() << ":" << id << std::endl; 

    }
    //----------------------------------------------------------------

    std::this_thread::sleep_for(std::chrono::seconds(3));

    my_mutex.lock();
    global_int = 0;
    my_mutex.unlock();

    goto repeat;
}





int main()
{
    global_int = 0;


    //------------------------------------------------------------------------Создадим пул потоков:Начало-----------------------------
    std::vector<std::thread> my_thread_vec;
    my_thread_vec.resize(4);

    for (int i = 0; i < 4; i++)
    {
        my_thread_vec[i] = std::thread(waiting_threads, i);     
    }
    //------------------------------------------------------------------------Создадим пул потоков:Конец--------------------


    std::this_thread::sleep_for(std::chrono::seconds(1));


    my_mutex1.lock();
    std::cout << "main_start" << std::endl;
    
    global_int = 1;
    my_cond_var.notify_one();    
    global_int = 1;
    my_cond_var.notify_one();   
    global_int = 1;
    my_cond_var.notify_one(); 
    global_int = 1;
    my_cond_var.notify_one();
    
    std::cout << "sleep_main_start" << std::endl;
    std::this_thread::sleep_for(std::chrono::seconds(10));
    std::cout << "sleep_main_end" << std::endl;

    global_int = 1;
    my_cond_var.notify_one();
    std::cout << "main_mutex_end" << std::endl;
    my_mutex1.unlock();



    //------------------------------------------------Отпускаем потоки, так как у них бесконечный цикл---------------------------------------
    for (int i = 0; i < 4; i++)
    {
        my_thread_vec[i].join();
    }
    //---------------------------------------------------------------------------------------------------------------------------------------

}

В первом случае изменение значения global_int заблокированы my_mutex, во втором my_mutex1. Из за этого у код разное поведение.

-В первом случае: потоки пробуждаются только после main_mutex_end - пробуждаются каждый по одному разу и почему то 5ый раз пробуждение не срабатывает.

-Во втором случае: потоки пробуждаются сразу после вызова notify_one() и во втором случае пробуждается один и з потоков и 5ый раз.

Не могу сообразить, что происходит:

-В первом случае создаются 4 потока и в них запускается функция waiting_threads, поток сразу же захватывает my_mutex и так как переменная global_int равна 0, то поток засыпает блокируя остальные потоки. Потом выполнение кода доходит до строки с захватам того же мьютекса в main() и вот тут не понятно, что происходит.

-В втором случае создаются 4 потока и в них запускается функция waiting_threads, поток сразу же захватывает my_mutex и так как переменная global_int равна 0, то поток засыпает блокируя остальные потоки. Потом выполнение кода доходит до строки с захватам my_mutex1 в main()(другого нежели в ф-ии waiting_threads) и вот тут тоже, что то не понятно, что происходит. Вроде бы участок с кода в котором присутствует обращение к переменной global_int уже под мьютексом my_mutex, а тут захватывается еще один мьютекс my_mutex1 и в участке кода тоже есть обращение к переменной global_int.

Я в конец запутался.


Источник

Ответы (1 шт):

Автор решения: Slava

Вы не понимаете принцип работы std::condition_variable. Давайте начнем с понимания, что делает метод std::condition_variable::wait(). А делает он две вещи:

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

  • по наступлении события (другой поток вызвал std::condition_variable::notify_one() или std::condition_variable::notify_all() поток переводится из режима ожидания в режим выполнения, и он пытается заблокировть мьютекс и затем продолжить выполнение. Поток может быть также разбужен системными событиями (spurious wakeup).

То есть, во время ожидания события мьютекс разблокирован, но чтобы вернутся из ожидания и продолжить выполнение кода, поток должен быть как разбужен событием (или может быть разбужен например сигналом (spurious wakeup), поэтому условие в ожидании проверяется в цикле) но и опять захватить мьютекс.

Далее, поток, который должен сообщить одному или более ожидающим потокам, что событие наступило делает следующие вещи:

  1. блокирует мьютекс
  2. устанавливает состояние, которое ожидают
  3. посылает сообщение notify_one() или notify_all()
  4. разблокирует мьютекс

Причем шаг 3 и шаг 4 можно менять местами.

Что же делает вызов метода std::condition_variable::notify_one()? если в данный момент на данной переменной один из потоков ожидает события, то он выводится из режима ожидания. Если таких потоков нет то вызов std::condition_variable::notify_one() игнорируется и не запоминается

При этом разница между std::condition_variable::notify_one() и std::condition_variable::notify_all() только в том, что первый метод переведет из режима ожидания только один поток, даже если ожидают многие, а второй их все. Но и в том и в другом случае поток уже в данный момент должен находится в режиме ожидания, чтобы этот сигнал получить.

Если вы поймете принцип работы, то вам станет понятен ответ на ваш вопрос

-В первом случае: потоки пробуждаются только после main_mutex_end - пробуждаются каждый по одному разу и почему то 5ый раз пробуждение не срабатывает.

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

Кроме того, ваша функция ожидания некорректна, вы не должны сбрасывать переменную в ноль, да еще и после отдельного захвата мьютекса, должна она выглядеть примерно вот так:

void waiting_threads(int id)
{
    std::cout << "waiting_threads_start" << std::endl;

    while( true ) {
        {
            //-----------------------------------
            std::unique_lock<std::mutex> my_lock(my_mutex);

            while (global_int == 0)
            {
                my_cond_var.wait(my_lock);
            }
            --global_int;
            std::cout << "thread_id:" <<  std::this_thread::get_id() << ":" << id << std::endl; 
           //-------------------------------------------------------         
        }   
        std::this_thread::sleep_for(std::chrono::seconds(3));
    }
}

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

Далее, если вы посмотрите на участок кода, между комментарием //----- то заметите, что выполняется он под захваченным мьютексом. Это значит, что выполняется в один момент только один поток. То есть, когда вы находитесь здесь, главный поток либо уже установил переменную global_int в ненулевое значение и цикл ожидания не будет выполнятся, либо еше нет, но тогда вы не пропустите сигнала на std::condition_variable поскольку послать сигнал основной поток сможет только тогда, когды вы отпустите мьютекс и находитесь в режиме ожидания.

Следует заметить, что событие, которое вы ожидаете, это не сигнал на std::condition_vazriable, а состояние, когда global_int не равно нулю. С данной логикой событие вы не пропустите, вне зависимости от того, получили вы сигнал от std::condition_variable или нет.

Кроме того, основная функция не должна устанавливать global_int в значение 1, а должна увеличивать ее на 1. Вы должны понять, что событие завязано на переменную global_int и она в данном случае определяет, какое количество работы стоит в очереди. std::condition_variable здесь вспомогательное средство, и вы не должны строить свою логику на количестве вызовов notify_one или notify_all - это лишь средство сообщить ожидающим потокам, что состояние изменилось.

→ Ссылка

licensed under cc by-sa 3.0 with attribution.