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

template

void interruptible_wait(std::condition_variable& cv,

std::unique_lock& lk,

Predicate pred) {

interruption_point();

this_thread_interrupt_flag.set_condition_variable(cv);

interrupt_flag::clear_cv_on_destruct guard;

while (!this_thread_interrupt_flag.is_set() && !pred()) {

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

}

interruption_point();

}

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

Ну хорошо, с ожиданием std::condition_variableмы разобрались, а что сказать о std::condition_variable_any? Всё точно так же или можно сделать лучше?

Класс std::condition_variable_anyотличается от std::condition_variableтем, что работает с любым типом блокировки, а не только с std::unique_lock. Как выясняется, это сильно упрощает дело, так что мы сможем добиться более впечатляющих результатов, чем получилось с std::condition_variable. Раз допустим любой тип блокировки, то можно написать и свой собственный класс, который захватывает (освобождает) как внутренний мьютекс set_clear_mutexв классе interrupt_flag, так и блокировку, переданную при вызове wait(). Соответствующий код приведён в листинге ниже.

Листинг 9.12.Реализация interruptible_wait()для std::condition_variable_any

class interrupt_flag {

std::atomic flag;

std::condition_variable* thread_cond;

std::condition_variable_any* thread_cond_any;

std::mutex set_clear_mutex;

public:

interrupt_flag():

thread_cond(0), thread_cond_any(0) {}

void set() {

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

std::lock_guard lk(set_clear_mutex);

if (thread_cond) {

thread_cond->notify_all();

} else if (thread_cond_any) {

thread_cond_any->notify_all();

}

}

template

void wait(std::condition_variable_any& cv, Lockable& lk) {

struct custom_lock {

interrupt_flag* self;

Lockable& lk;

custom_lock(interrupt_flag* self_,

std::condition_variable_any& cond,

Lockable& lk_): self(self_), lk(lk_) {

self->set_clear_mutex.lock(); ← (1)

self->thread_cond_any = &cond; ← (2)

}

void unlock() { ← (3)

lk.unlock();

self->set_clear_mutex.unlock();

}

void lock() {

std::lock(self->set_clear_mutex, lk); ← (4)

}

~custom_lock() {

self->thread_cond_any = 0; ← (5)

self->set_clear_mutex.unlock();

}

};

custom_lock cl(this, cv, lk);

interruption_point();

cv.wait(cl);

interruption_point();

}

// остальное, как и раньше

};

template

void interruptible_wait(std::condition_variable_any& cv,

Lockable& lk) {

this_thread_interrupt_flag.wait(cv, lk);

}

Наш класс блокировки должен захватить внутренний мьютекс set_clear_mutexна этапе конструирования (1)и затем записать в переменную thread_cond_anyуказатель на объект std::condition_variable_any, переданный конструктору (2). Ссылка на объект Lockableсохраняется для последующего использования; он должен быть уже заблокирован. Теперь проверять, был ли поток прерван, можно, не опасаясь гонки. Если в этой точке флаг прерывания установлен, то это было сделано до захвата мьютекса set_clear_mutex. Когда условная переменная вызывает нашу функцию unlock()внутри wait(), мы разблокируем объект Lockable и внутренний мьютекс set_clear_mutex (3). Это позволяет потокам, которые пытаются нас прервать, захватить set_clear_mutexи проверить указатель thread_cond_any, когда мы уже находимся в wait(), но не раньше. Это именно то, чего мы хотели (но не смогли) добиться в случае std::condition_variable. После того как wait()завершит ожидание (из-за получения сигнала или вследствие ложного пробуждения), она вызовет нашу функцию lock(), которая снова захватит внутренний мьютекс set_clear_mutexи заблокирует объект Lockable (4). Теперь можно еще раз проверить, не было ли прерываний, пока мы находились в wait(), и только потом обнулить указатель thread_cond_anyв деструкторе custom_lock (5), где также освобождается set_clear_mutex.

Теперь с прерыванием ожидания условной переменной всё ясно, но как быть с другими блокирующими операциями ожидания: освобождения мьютекса, готовности будущего результата и т.д.? Вообще говоря, приходится прибегать к тому же трюку с таймаутом, который мы использовали для std::condition_variable, потому что, если не влезать в код мьютекса или будущего результата, то нет никакого другого способа прервать ожидание, кроме как обеспечить выполнение ожидаемого условия. Но в отличие от условных переменных, мы точно знаем, чего ждем, поэтому можем организовать цикл внутри функции interruptible_wait().

Вот, например, как выглядит перегрузка interruptible_wait()для std::future<>:

template

void interruptible_wait(std::future& uf) {

while (!this_thread_interrupt_flag.is_set()) {

if (uf.wait_for(lk, std::chrono::milliseconds(1) ==

std::future_status::ready)

break;

}

interruption_point();

}

Здесь мы ждем, пока либо будет установлен флаг прерывания, либо готов будущий результат, но блокирующее ожидание будущего результата продолжается в течение 1 мс на каждой итерации цикла. Это означает, что в среднем запрос на прерывание будет обнаружен с задержкой 0,5 мс, в предположении, что разрешение часов достаточно высокое. Функция wait_forобычно ожидает в течение как минимум одного такта, поэтому если такт системных часов составляет 15 мс, то ждать придётся не одну, а 15 мс. Приемлемо это или нет, зависит от конкретных условий. Таймаут при необходимости можно уменьшить (если часы позволяют), но тогда поток будет чаще просыпаться для проверки флага, что увеличит накладные расходы на переключение задач.

На данный момент мы знаем, как можно обнаружить прерывание с помощью функций interruption_point()и interruptible_wait(), но что с этим потом делать?

С точки зрения прерываемого потока, прерывание — это просто исключение типа thread_interrupted, которое, следовательно, можно обработать, как любое другое исключение. В частности, его можно перехватить в стандартном блоке catch:

try {

do_something();

} catch (thread_interrupted&) {

handle_interruption();

}

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