Энтони Уильямс - Параллельное программирование на С++ в действии. Практика разработки многопоточных программ

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

void interruption_point() {

if (this_thread_interrupt_flag.is_set()) {

throw thread_interrupted();

}

}

Обращаться к этой функции можно там, где нам удобно:

void fоо() {

while (!done) {

interruption_point();

process_next_item();

}

}

Такое решение работает, но оно не идеально. Лучше всего прерывать поток тогда, когда он блокирован в ожидании чего-либо, но именно в этот момент поток как раз и не работает, а, значит, не может вызвать interruption_point()! Нам требуется какой-то механизм прерываемого ожидания.

Итак, мы можем обнаруживать прерывание в подходящих местах программы с помощью обращений к функции interruption_point(), но это ничем не помогает в случае, когда поток блокирован в ожидании какого-то события, например сигнала условной переменной. Нам необходима еще одна функция, interruptible_wait(), которую можно будет перегрузить для различных ожидаемых событий, и нужно придумать, как вообще прерывать ожидание. Я уже говорил, что среди прочего ожидать можно сигнала условной переменной, поэтому с нее и начнем. Что необходимо для того, чтобы можно было прервать поток, ожидающий условную переменную? Проще всего было бы известить условную переменную в момент установки флага и поставить точку прерывания сразу после ожидания. Но в этом случае придётся разбудить все потоки, ждущие эту условную переменную, хотя заинтересован в этом только прерываемый поток. Впрочем, потоки, ждущие условную переменную, в любом случае должны обрабатывать ложные пробуждения, а отличить посланный нами сигнал от любого другого они не могут, так что ничего страшного не случится. В структуре interrupt_flagнужно будет сохранить указатель на условную переменную, чтобы при вызове set()ей можно было послать сигнал. В следующем листинге показана возможная реализация функции interruptible_wait()для условных переменных.

Листинг 9.10.Неправильная реализация interruptible_wait()для std::condition_variable

void interruptible_wait(std::condition_variable& cv,

std::unique_lock& lk) {

interruption_point();← (1)

this_thread_interrupt_flag.set_condition_variable(cv);

cv.wait(lk); ← (2)

this_thread_interrupt_flag.clear_condition_variable();← (3)

interruption_point();

}

В предположении, что существуют функции, которые устанавливают и разрывают ассоциацию условной переменной с флагом прерывания, этот код выглядит просто и понятно. Он проверяет, не было ли прерывания, ассоциирует условную переменную с флагом interrupt_flagдля текущего потока (1), ждет условную переменную (2), разрывает ассоциацию с условной переменной (3)и снова проверяет, не было ли прерывания. Если поток прерывается во время ожидания условной переменной, то прерывающий поток пошлёт этой переменной сигнал, что пробудит нас и даст возможность проверить факт. К сожалению, этот код не работает, в нем есть две проблемы. Первая довольно очевидна: функция std::condition_variable::wait()может возбуждать исключения, поэтому из interruptible_wait()возможен выход без разрыва ассоциации флага прерывания с условной переменной. Это легко исправляется с помощью структуры, которая разрывает ассоциацию в ее деструкторе.

Вторая, не столь очевидная, проблема связана с гонкой. Если поток прерывается после первого обращения к interruption_point(), но до обращения к wait(), то не имеет значения, ассоциирована условная переменная с флагом прерывания или нет, потому что поток еще ничего не ждет и, следовательно, не может быть разбужен сигналом, посланным условной переменной . Мы должны гарантировать, что потоку не может быть послан сигнал между последней проверкой прерывания и обращением к wait(). Если не залезать в код класса std::condition_variable, то сделать это можно только одним способом: использовать для защиты мьютекс, хранящийся в lk, который, следовательно, нужно передавать функции set_condition_variable(). К сожалению, при этом возникают новые проблемы: мы передаём ссылку на мьютекс, о времени жизни которого ничего не знаем, другому потоку (тому, который выполняет прерывание), чтобы тот его захватил (внутри interrupt()). Но может случиться, что этот поток уже удерживает данный мьютекс, и тогда возникнет взаимоблокировка. К тому же, появляется возможность доступа к уже уничтоженному мьютексу. В общем, это решение не годится. Но если мы не можем надежно прерывать ожидание условной переменной, то нечего было и затевать это дело — почти того же самого можно было бы добиться и без специальной функции interruptible_wait(). Так какие еще есть варианты? Можно, к примеру, задать таймаут ожидания; использовать вместо wait()функцию wait_for()с очень коротким таймаутом (скажем, 1 мс). Это ограничивает сверху время до момента, когда поток обнаружит прерывание (с учетом промежутка между тактами часов). Если поступить так, что ожидающий поток будет видеть больше ложных пробуждений из-за срабатывания таймера, но тут уж ничего не попишешь. Такая реализация показана в листинге ниже вместе с соответствующей реализацией interrupt_flag.

Листинг 9.11.Реализация interruptible_wait()для std::condition_variableс таймаутом

class interrupt_flag {

std::atomic flag;

std::condition_variable* thread_cond;

std::mutex set_clear_mutex;

public:

interrupt_flag(): thread_cond(0) {}

void set() {

flag.store(true, std::memory_order_relaxed);

std::lock_guard lk(set_clear_mutex);

if (thread_cond) {

thread_cond->notify_all();

}

}

bool is_set() const {

return flag.load(std::memory_order_relaxed);

}

void set_condition_variable(std::condition_variable& cv) {

std::lock_guard lk(set_clear_mutex);

thread_cond = &cv;

}

void clear_condition_variable() {

std::lock_guard lk(set_clear_mutex);

thread_cond = 0;

}

struct clear_cv_on_destruct {

~clear_cv_on_destruct() {

this_thread_interrupt_flag.clear_condition_variable();

}

};

};

void interruptible_wait(std::condition_variable& cv,

std::unique_lock& lk) {

interruption_point();

this_thread_interrupt_flag.set_condition_variable(cv);

interrupt_flag::clear_cv_on_destruct guard;

interruption_point();

cv.wait_for(lk, std::chrono::milliseconds(1));

interruption_point();

}

Если мы ждем какой-то предикат, то таймаут продолжительностью 1 мс можно полностью скрыть внутри цикла проверки предиката: