Должен ли я получить блокировку перед вызовом condition_variable.notify_one ()?

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

std::condition_variable cv;
std::mutex cv_m;
int i = 0;
bool done = false;

void waits()
{
std::unique_lock<std::mutex> lk(cv_m);
std::cout << "Waiting... \n";
cv.wait(lk, []{return i == 1;});
std::cout << "...finished waiting. i == 1\n";
done = true;
}

void signals()
{
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "Notifying...\n";
cv.notify_one();

std::unique_lock<std::mutex> lk(cv_m);
i = 1;
while (!done) {
lk.unlock();
std::this_thread::sleep_for(std::chrono::seconds(1));
lk.lock();
std::cerr << "Notifying again...\n";
cv.notify_one();
}
}

int main()
{
std::thread t1(waits), t2(signals);
t1.join(); t2.join();
}

while (!pred())
wait(lock);

if (--f->counter == 0)      // (1)
// we have zeroed this fence's counter, wake up everyone that waits
f->resume.notify_all(); // (2)
else
{
unique_lock<mutex> lock(f->resume_mutex);
f->resume.wait(lock);   // (3)
}

ситуация

Используя vc10 и Boost 1.56, я реализовал параллельную очередь во многом как этот блог предлагает. Автор разблокирует мьютекс для минимизации разногласий, т. Е. notify_one() вызывается с разблокированным мьютексом:

void push(const T& item)
{
std::unique_lock<std::mutex> mlock(mutex_);
queue_.push(item);
mlock.unlock();     // unlock before notificiation to minimize mutex contention
cond_.notify_one(); // notify one waiting thread
}

Разблокировка мьютекса подтверждается примером в Повысить документацию:

void prepare_data_for_processing()
{
retrieve_data();
prepare_data();
{
boost::lock_guard<boost::mutex> lock(mut);
data_ready=true;
}
cond.notify_one();
}

проблема

Тем не менее это привело к следующему ошибочному поведению:

• в то время как notify_one() имеет не был вызван еще cond_.wait() все еще может быть прервана через boost::thread::interrupt()
• один раз notify_one() был вызван впервые cond_.wait() тупики; ожидание не может быть закончено boost::thread::interrupt() или же boost::condition_variable::notify_*() больше.

Решение

Удаление линии mlock.unlock() заставил код работать как ожидалось (уведомления и прерывания заканчивают ожидание). Обратите внимание, что notify_one() вызывается с заблокированным мьютексом, он разблокируется сразу после выхода из области действия:

void push(const T& item)
{
std::lock_guard<std::mutex> mlock(mutex_);
queue_.push(item);
cond_.notify_one(); // notify one waiting thread
}

Это означает, что по крайней мере с моей конкретной реализацией потока мьютекс не должен быть разблокирован перед вызовом boost::condition_variable::notify_one()Хотя оба пути кажутся правильными.

@ Майкл Барр прав. condition_variable::notify_one не требует блокировки переменной. Ничто не мешает вам использовать блокировку в этой ситуации, как показывает пример.

В данном примере блокировка мотивируется одновременным использованием переменной i, Поскольку signals нить модифицирует переменная, она должна гарантировать, что никакой другой поток не получит к ней доступ в течение этого времени.

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

В некоторых случаях, когда cv может быть занят (заблокирован) другими потоками. Вам нужно получить блокировку и снять ее, прежде чем уведомить _ * ().
Если нет, notify _ * () может вообще не выполняться.

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

Чтобы уточнить, вы ДОЛЖНЫ взять блокировку перед вводом wait (lk), потому что wait () разблокирует lk, и было бы неопределенным поведением, если блокировка не была заблокирована. Это не относится к notify_one (), но вам нужно убедиться, что вы не вызовете notify _ * () перед вводом wait () а также наличие этого вызова разблокирует мьютекс; что, очевидно, может быть сделано только путем блокировки этого же мьютекса перед вызовом notify _ * ().

Например, рассмотрим следующий случай:

std::atomic_int count;
std::mutex cancel_mutex;
std::condition_variable cancel_cv;

void stop()
{
if (count.fetch_sub(1) == -999) // Reached -1000 ?
cv.notify_one();
}

bool start()
{
if (count.fetch_add(1) >= 0)
return true;
// Failure.
stop();
return false;
}

void cancel()
{
if (count.fetch_sub(1000) == 0)  // Reached -1000?
return;
// Wait till count reached -1000.
std::unique_lock<std::mutex> lk(cancel_mutex);
cancel_cv.wait(lk);
}

Предупреждение: этот код содержит ошибку.

Идея заключается в следующем: потоки вызывают start () и stop () парами, но только до тех пор, пока start () возвращает true. Например:

if (start())
{
// Do stuff
stop();
}

Один (другой) поток в какой-то момент вызовет метод cancel () и после возврата из метода cancel () уничтожит объекты, необходимые для операции «Делать вещи». Тем не менее, метод cancel () не должен возвращаться при наличии потоков между start () и stop (), а после того, как cancel () выполнил свою первую строку, start () всегда будет возвращать значение false, поэтому новые потоки не войдут в «Do» область вещей.

Работает правильно?

Аргументация следующая:

1) Если какой-либо поток успешно выполнит первую строку start () (и, следовательно, вернет true), ни один из потоков еще не выполнил первую строку cancel () (мы предполагаем, что общее число потоков намного меньше 1000 на путь).

2) Кроме того, хотя поток успешно выполнил первую строку start (), но еще не первую строку stop (), невозможно, чтобы какой-либо поток успешно выполнил первую строку cancel () (обратите внимание, что только один поток) когда-либо вызывает cancel ()): значение, возвращаемое fetch_sub (1000), будет больше 0.

3) Как только поток выполнил первую строку cancel (), первая строка start () всегда будет возвращать false, и поток, вызывающий start (), больше не будет входить в область «Do stuff».

4) Количество вызовов start () и stop () всегда сбалансировано, поэтому после неудачного выполнения первой строки cancel () всегда будет момент, когда (последний) вызов stop () вызывает подсчет достичь -1000 и, следовательно, вызвать notify_one (). Обратите внимание, что это может произойти только тогда, когда первая строка отмены привела к провалу потока.

Помимо проблемы с голоданием, когда так много потоков вызывает функцию start () / stop (), которая никогда не достигает -1000, а метод cancel () никогда не возвращается, что можно принять как «маловероятный и никогда не продолжительный», есть еще одна ошибка:

Возможно, что внутри области «Делать вещи» есть один поток, допустим, он просто вызывает stop (); в этот момент поток выполняет первую строку метода cancel (), считывая значение 1 с fetch_sub (1000) и проваливаясь. Но прежде чем он получит мьютекс и / или сделает вызов wait (lk), первый поток выполняет первую строку stop (), читает -999 и вызывает cv.notify_one ()!

Затем этот вызов notify_one () выполняется ДО того, как мы будем ждать () — с условной переменной! И программа будет бесконечно тупиковой.

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

В этом примере нет условия, однако. Почему я не использовал в качестве условия ‘count == -1000’? Потому что здесь это совсем не интересно: как только будет достигнут -1000, мы уверены, что ни один новый поток не войдет в область «Делать вещи». Более того, потоки могут по-прежнему вызывать start () и увеличивать счетчик (до -999, -998 и т. Д.), Но нас это не волнует. Единственное, что имеет значение, — это то, что было достигнуто -1000, чтобы мы точно знали, что в области «Делать вещи» больше нет потоков. Мы уверены, что это тот случай, когда вызывается notify_one (), но как сделать так, чтобы мы не вызывали notify_one () до отмены () блокировки его мьютекса? Конечно, простое блокирование cancel_mutex незадолго до notify_one () не поможет.

Проблема в том, что, несмотря на то, что мы не ждем условия, там все еще является условие, и нам нужно заблокировать мьютекс

1) до того, как это условие будет достигнуто
2) прежде чем мы позвоним notify_one.

Следовательно, правильный код становится:

void stop()
{
if (count.fetch_sub(1) == -999) // Reached -1000 ?
{
cancel_mutex.lock();
cancel_mutex.unlock();
cv.notify_one();
}
}

[… то же самое начало () …]

void cancel()
{
std::unique_lock<std::mutex> lk(cancel_mutex);
if (count.fetch_sub(1000) == 0)
return;
cancel_cv.wait(lk);
}

Конечно, это только один пример, но другие случаи очень похожи; почти во всех случаях, когда вы используете условную переменную, вы будете необходимость чтобы этот мьютекс был заблокирован (на короткое время) перед вызовом notify_one (), иначе вы можете вызвать его перед вызовом wait ().

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

Этот пример был особенным в том смысле, что строка, которая изменяет условие, выполняется тем же потоком, который вызывает wait ().

Более обычным является случай, когда один поток просто ждет, пока условие станет истинным, а другой поток берет блокировку перед изменением переменных, участвующих в этом условии (что может привести к тому, что оно станет истинным). В этом случае мьютекс является заблокировано непосредственно перед (и после) условием, которое стало истинным — поэтому вполне нормально просто разблокировать мьютекс перед вызовом notify _ * () в этом случае.

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

thread t;

void foo() {
std::mutex m;
std::condition_variable cv;
bool done = false;

t = std::thread([&]() {
{
std::lock_guard<std::mutex> l(m);  // (1)
done = true;  // (2)
}  // (3)
cv.notify_one();  // (4)
});  // (5)

std::unique_lock<std::mutex> lock(m);  // (6)
cv.wait(lock, [&done]() { return done; });  // (7)
}

void main() {
foo();  // (8)
t.join();  // (9)
}

Предположим, что есть переключение контекста на вновь созданный поток t после того, как мы его создали, но прежде чем мы начнем ждать переменную условия (где-то между (5) и (6)). Нить t получает блокировку (1), устанавливает переменную предиката (2) и затем снимает блокировку (3). Предположим, что есть другой переключатель контекста прямо в этой точке, прежде чем notify_one() (4) выполняется. Основной поток получает блокировку (6) и выполняет строку (7), после чего предикат возвращает true и нет причин ждать, поэтому он снимает блокировку и продолжает работу. foo возвращает (8) и переменные в своей области (в том числе cv) уничтожены. Перед тем t может присоединиться к основному потоку (9), он должен завершить свое выполнение, поэтому он продолжает с того места, где остановился, чтобы выполнить cv.notify_one() (4), в какой момент cv уже уничтожен!

Возможное исправление в этом случае — удерживать блокировку при вызове. notify_one (то есть удалить область, заканчивающуюся в строке (3)). Тем самым мы гарантируем, что поток t звонки notify_one до cv.wait может проверить вновь установленную переменную предиката и продолжить, так как это должно было бы получить блокировку, которая t в настоящее время держит, чтобы сделать проверку. Итак, мы гарантируем, что cv не доступен потоком t после foo возвращается.

Подводя итог, можно сказать, что проблема в данном конкретном случае заключается не в потоке, а в времени жизни переменных, захваченных ссылкой. cv захватывается по ссылке через поток tследовательно, вы должны убедиться, cv остается в живых на время выполнения потока. Другие примеры, представленные здесь, не страдают от этой проблемы, потому что condition_variable а также mutex объекты определяются в глобальной области видимости, поэтому они гарантированно будут поддерживаться до выхода из программы.