Сегодня я хочу поделиться реальным примером, который работает в продакшене. Речь пойдет о написании сопрограмм, которые можно прервать извне. Изначально мне это понадобилось для реализации механизма deadline или timeout (кому как больше нравится). Согласитесь, довольно часто возникающая задача. На этом примере я продемонстрирую использование еще нескольких возможностей, предоставляемых С++. Речь пойдет об await_transform и конструкторе объекта promise_type.
В этот раз не обойдется без boost::asio. Но он будет использован исключительно в демонстрационных целях — в детали его работы я углубляться сильно не буду. А начну по традиции с конца. Давайте напишем простое приложение, которое может быть завершено нажатием CTRL+C. Как раз в нем получилось больше boost, чем во всём остальном коде.
task coroutine() {
// вся логика приложения вынесена в эту функцию. По понятным причинам
// main() не может быть корутиной - программа просто завершилась бы
// при первом же вызове co_await. Но это только теория, компилятор не
// позволит вам проверить это на практике
// ...
}
int main(int argc, char *argv[]) {
// эта переменная - сердце boost::asio обеспечивающее связь вашего
// приложения и операций ввода/вывода операционной системы
boost::asio::io_context context;
// SIGINT - это тот сигнал, который получит ваше приложение, когда вы нажмете CTRL+C
boost::asio::signal_set signals(context, SIGINT);
// запускаем наш "настоящий main"
auto job = coroutine();
// регистрируем в boost::asio наш обработчик сигнала SIGINT
signals.async_wait([job = std::move(job)] (auto code) {
// при получении сигнала прерываем наше приложение
job.terminate();
});
// context.run() будет выполняться до тех пор, пока не будут обработаны
// все зарегистрированные в этом context события. Так что, мы получим
// красивый так называемый graceful shutdown
// на самом деле, я так не делаю. Дело в том, что такой подход лишает
// ваше приложение возможности завершиться по собственному желанию. В
// context будет висеть и ждать своего часа обработчик сигнала. Эта
// проблема может быть решена различными способами, но для
// упрощения примера мы не будем рассматривать данную ситуацию
context.run();
}
Что же дают нам для реализации подобного механизма корутины С++? И всё, и ничего. В комментариях к моей статье генераторы на корутинах С++ мне задавали вопрос: что случится, если генератор (корутину) не вызовут после очередного co_yield? Ответ был прост — всё будет хорошо, если все ресурсы внутри генератора освобождаются при помощи RAII. Завершаясь, корутина уничтожит все локальные переменные, вызывая для них деструкторы. Поэтому, используя идиому RAII, вы как бы объясняете компилятору, какие ресурсы принадлежат корутине. А использованные вами операторы co_await и co_yield будут как раз теми точками, где вы будете проверять флаг — требование завершить корутину. На самом деле всё еще круче! Вместо того, чтобы думать о том, как завершить корутину, вам надо просто её не возобновлять. То есть, не делать ничего. Компилятор всё сделает за вас. Казалось бы — успех! Ставим точку, дальше и писать не о чем
В действительности это хорошо сработает только при использовании co_yield.
template <typename Duration>
auto sleep(Duration duration) {
struct [[nodisacrd]] awaitable {
// для асинхронного sleep я использовал таймер из boost
boost::asio::steady_timer timer;
// резервируем в нашем awaitable объекте место под результат
boost::system::error_code error{};
bool await_ready() const { return false; }
void await_suspend(std::coroutine_handle<> coro) {
// регистрируем событие в io_context, ссылку на который мы
// неявно получили через executor при создании steady_timer
// смотри ниже
timer.async_wait([coro] (auto ec) mutable {
if (!error) { error = ec; }
// при получении события возобновляем корутину
coro.resume();
});
}
void on_terminate(boost::system::error_code ec) {
error = ec;
// cancel прервет текущую асинхронную операцию, что приведет к вызову
// callback (нашей лямбды) с ec = boost::asio::error:peration_aborted
timer.cancel();
}
// нашему awaitable нечего возврящать, но в случае
// возникновения ошибки он выбрасывает исключение
void await_resume() {
if (error) { throw boost::system::system_error(error); }
}
};
// на самом деле, не очень честно заводить таймер здесь, потому что
// отсчет времени начнется с момента создания awaitable объекта, а не
// с момента вызова co_await. В некоторых случаях это может быть
// существенно
// еще вы можете спросить: что такое executor и откуда он взялся?
// это переменная типа boost::asio::any_io_executor, обычно я передаю
// её в подобные функции параметром, но всё больше склоняюсь к мысли
// использовать для этого глобальные thread_local переменные
return awaitable{.timer = boost::asio::steady_timer{executor, std::forward<Duration>(duration)}};
};
В этом нам как раз и поможет метод promise_type::await_transform(). Если он определен, то перед каждом вызовом co_await в него передается awaitable объект. Метод, в свою очередь, должен вернуть тоже awaitable объект (может вернуть тот же, может другой). И именно этот, полученный от await_transform объект, будет передан в co_await.
class task {
public:
struct promise_type {
struct state_t {
// функтор для вызова on_terminate из текущего awaitable объекта
std::function<void(boost::system::error_code)> callback;
};
std::shared_ptr<state_t> state = std::make_shared<state_t>();
// никаких лишних остановок корутины ни до
std::suspend_never initial_suspend() const noexcept { return {}; }
// ни после
std::suspend_never final_suspend() const noexcept { return {}; }
// нашему task надо иметь ссылку на состояние, чтобы вызвать on_terminate
auto get_return_object() { return task{state}; }
// корутина ничего не возвращает
void return_void() { }
// просто игнорируем все непойманные исключения,
// они будут приводить к корректному завершению корутин
void unhandled_exception() { }
template <typename Awaitable>
auto await_transform(Awaitable &&awaitable) {
// методы wrapper вызывают соответствующие методы из базового awaitable
struct [[nodiscard]] wrapper {
Awaitable base;
std::shared_ptr<state_t> m_state;
auto await_ready() { return base.await_ready(); }
auto await_suspend(std::experimental::coroutine_handle<> coro) {
// сохраняем ссылку на awaitable через замыкание
m_state->callback = [this] (boost::system::error_code ec) {
base.on_terminate(ec);
});
return base.await_suspend(coro);
}
auto await_resume() {
return base.await_resume();
}
};
return wrapper{std::forward<Awaitable>(awaitable), state};
}
};
// как только корутина завершится, promise_type будет уничтожен
// наш weak_ptr будет указывать на null. В остальных случаях,
// при однопоточной работе, state будет указывать на текущий
// awaitable объект, который прервал выполнение корутины
void terminate(boost::system::error_code ec = boost::asio::error::interrupted) {
if (auto ptr = m_state.lock(); ptr != nullptr && ptr->callback != nullptr) {
ptr->callback(ec);
// защищаемся от повторного вызова terminate()
ptr->callback = nullptr;
}
}
private:
// сохрани мы ссылку на promise_type - и перед нами встал бы вопрос:
// как узнать, завершилась ли корутина. Данный подход избавляет нас
// от этой диллемы
std::weak_ptr<promise_type::state_t> m_state;
explicit task(std::weak_ptr<promise_type::state_t> state)
: m_state{std::move(state)} {}
};
Мы получили искомое поведение. Правда, наш terminate не завершает корутину немедленно, а лишь приводит к возникновению ошибки (exception). Как по мне, это более прямой и гибкий путь, но незначительно поменяв логику await_suspend, можно сделать и немедленное завершение.
Полный код получившегося приложения
Например, если вы уже выполняете операцию очистки при завершении корутины. В таких случаях RAII спасает не всегда. Иногда хочется выполнить какие-нибудь асинхронные вызовы, а деструктор не может быть корутиной. Или же для операции требуется большое количество локальных переменных. Что можно сделать в этой ситуации?
struct promise_type {
struct state_t {
bool locked = false;
std::function<void(boost::system::error_code)> callback;
void lock() { locked = true; }
void unlock { locked = false; }
using ptr = std::shared_ptr<state_t>;
};
ptr state;
// на случай, если наша корутина вообще не имеет параметров
promise_type() : state{std::make_shared<state_t>()} { }
// этот конструктор подойдет под любой набор параметров
template <typename ...Args>
explicit promise_type(Args & ...args) : promise_type{} {
// используем fold expression: для каждого параметра будет создана
// лямбда, затем все их результаты сложены при помощи логического "или"
//
// конструкция выглядит страшно, но фактически она скомпилируется
// в одну операцию присваивания для первого подходящего по типу
// параметра. Остальное будет безжалостно выкинуто в процессе
// оптимизации. Кто не верит, вот ссылка на compiler explorer
// https://godbolt.org/z/hsrPErqfn
([this] (auto ¶m) {
if constexpr (std::is_same_v<Args, state_t:tr>) {
// мы ЗАПИСЫВАЕМ новое значение в параметр корутины
// и делаем это фактически до её запуска
param = state;
return true;
}
return false;
}(args) || ...);
}
// ...
};
Это небольшое улучшение позволяет нам написать следующий код:
task coroutine(task::state_t lock = {}) {
using std::chrono_literals;
try {
std::cout << "we are doing our work here" << std::endl;
co_await sleep(10s);
std::cout << "work is complete" << std::endl;
} catch (...) { }
std::cout << "final cleanup" << std::endl;
// я назвал методы state - lock и unlock, это дает
// мне возможность использовать std::lock_guard
const std::lock_guard g(*lock);
// следующий sleep уже не будет прерван
co_await sleep(5s);
}
Но при аккуратном подходе, в первую очередь разработчиками библиотек и фреймворков, конструкторы promise_type и await_transform могут позволить изящно решить массу задач асинхронного программирования.
habr.com
Корутины можно рассматривать как легковесные потоки. Во многом подходы к принудительному завершению потоков и корутин схожи.
В этот раз не обойдется без boost::asio. Но он будет использован исключительно в демонстрационных целях — в детали его работы я углубляться сильно не буду. А начну по традиции с конца. Давайте напишем простое приложение, которое может быть завершено нажатием CTRL+C. Как раз в нем получилось больше boost, чем во всём остальном коде.
task coroutine() {
// вся логика приложения вынесена в эту функцию. По понятным причинам
// main() не может быть корутиной - программа просто завершилась бы
// при первом же вызове co_await. Но это только теория, компилятор не
// позволит вам проверить это на практике
// ...
}
int main(int argc, char *argv[]) {
// эта переменная - сердце boost::asio обеспечивающее связь вашего
// приложения и операций ввода/вывода операционной системы
boost::asio::io_context context;
// SIGINT - это тот сигнал, который получит ваше приложение, когда вы нажмете CTRL+C
boost::asio::signal_set signals(context, SIGINT);
// запускаем наш "настоящий main"
auto job = coroutine();
// регистрируем в boost::asio наш обработчик сигнала SIGINT
signals.async_wait([job = std::move(job)] (auto code) {
// при получении сигнала прерываем наше приложение
job.terminate();
});
// context.run() будет выполняться до тех пор, пока не будут обработаны
// все зарегистрированные в этом context события. Так что, мы получим
// красивый так называемый graceful shutdown
// на самом деле, я так не делаю. Дело в том, что такой подход лишает
// ваше приложение возможности завершиться по собственному желанию. В
// context будет висеть и ждать своего часа обработчик сигнала. Эта
// проблема может быть решена различными способами, но для
// упрощения примера мы не будем рассматривать данную ситуацию
context.run();
}
Что же дают нам для реализации подобного механизма корутины С++? И всё, и ничего. В комментариях к моей статье генераторы на корутинах С++ мне задавали вопрос: что случится, если генератор (корутину) не вызовут после очередного co_yield? Ответ был прост — всё будет хорошо, если все ресурсы внутри генератора освобождаются при помощи RAII. Завершаясь, корутина уничтожит все локальные переменные, вызывая для них деструкторы. Поэтому, используя идиому RAII, вы как бы объясняете компилятору, какие ресурсы принадлежат корутине. А использованные вами операторы co_await и co_yield будут как раз теми точками, где вы будете проверять флаг — требование завершить корутину. На самом деле всё еще круче! Вместо того, чтобы думать о том, как завершить корутину, вам надо просто её не возобновлять. То есть, не делать ничего. Компилятор всё сделает за вас. Казалось бы — успех! Ставим точку, дальше и писать не о чем
В действительности это хорошо сработает только при использовании co_yield.
Не всё так просто или что делать, если вам надоело ждать
В чем же проблема co_await ? Я уже касался этой проблемы в другой моей статье про каналы на С++. Дело в том, что co_await не просто ждёт возобновления корутины, он ждет возникновения некого события (например чтения при помощи boost::asio::async_read). И это событие неплохо было бы отменить. Во-первых — чтобы не загружать приложение ненужной работой. А во вторых awaitable объект может использовать ссылки на локальные переменные корутины. Когда awaitable дождется события и начнет сохранять полученные данные в несуществующий уже буфер, в лучшем случае вы получите аварийное завершение приложения.Для решения этой проблемы я решил сделать свои awaitable объекты прерываемыми. И для этого добавил в них еще один метод — on_terminate. Давайте рассмотрим пример реализации операции sleep. Она может выглядеть следующим образом:
template <typename Duration>
auto sleep(Duration duration) {
struct [[nodisacrd]] awaitable {
// для асинхронного sleep я использовал таймер из boost
boost::asio::steady_timer timer;
// резервируем в нашем awaitable объекте место под результат
boost::system::error_code error{};
bool await_ready() const { return false; }
void await_suspend(std::coroutine_handle<> coro) {
// регистрируем событие в io_context, ссылку на который мы
// неявно получили через executor при создании steady_timer
// смотри ниже
timer.async_wait([coro] (auto ec) mutable {
if (!error) { error = ec; }
// при получении события возобновляем корутину
coro.resume();
});
}
void on_terminate(boost::system::error_code ec) {
error = ec;
// cancel прервет текущую асинхронную операцию, что приведет к вызову
// callback (нашей лямбды) с ec = boost::asio::error:
peration_abortedtimer.cancel();
}
// нашему awaitable нечего возврящать, но в случае
// возникновения ошибки он выбрасывает исключение
void await_resume() {
if (error) { throw boost::system::system_error(error); }
}
};
// на самом деле, не очень честно заводить таймер здесь, потому что
// отсчет времени начнется с момента создания awaitable объекта, а не
// с момента вызова co_await. В некоторых случаях это может быть
// существенно
// еще вы можете спросить: что такое executor и откуда он взялся?
// это переменная типа boost::asio::any_io_executor, обычно я передаю
// её в подобные функции параметром, но всё больше склоняюсь к мысли
// использовать для этого глобальные thread_local переменные
return awaitable{.timer = boost::asio::steady_timer{executor, std::forward<Duration>(duration)}};
};
Для того, чтобы понять как этот метод нам поможет, давайте теперь напишем наш task. Идея в следующем: при каждом вызове co_await мы будем запоминать ссылку на awaitable объект и в случае вызова terminate — вызывать метод on_terminate.
В этом нам как раз и поможет метод promise_type::await_transform(). Если он определен, то перед каждом вызовом co_await в него передается awaitable объект. Метод, в свою очередь, должен вернуть тоже awaitable объект (может вернуть тот же, может другой). И именно этот, полученный от await_transform объект, будет передан в co_await.
class task {
public:
struct promise_type {
struct state_t {
// функтор для вызова on_terminate из текущего awaitable объекта
std::function<void(boost::system::error_code)> callback;
};
std::shared_ptr<state_t> state = std::make_shared<state_t>();
// никаких лишних остановок корутины ни до
std::suspend_never initial_suspend() const noexcept { return {}; }
// ни после
std::suspend_never final_suspend() const noexcept { return {}; }
// нашему task надо иметь ссылку на состояние, чтобы вызвать on_terminate
auto get_return_object() { return task{state}; }
// корутина ничего не возвращает
void return_void() { }
// просто игнорируем все непойманные исключения,
// они будут приводить к корректному завершению корутин
void unhandled_exception() { }
template <typename Awaitable>
auto await_transform(Awaitable &&awaitable) {
// методы wrapper вызывают соответствующие методы из базового awaitable
struct [[nodiscard]] wrapper {
Awaitable base;
std::shared_ptr<state_t> m_state;
auto await_ready() { return base.await_ready(); }
auto await_suspend(std::experimental::coroutine_handle<> coro) {
// сохраняем ссылку на awaitable через замыкание
m_state->callback = [this] (boost::system::error_code ec) {
base.on_terminate(ec);
});
return base.await_suspend(coro);
}
auto await_resume() {
return base.await_resume();
}
};
return wrapper{std::forward<Awaitable>(awaitable), state};
}
};
// как только корутина завершится, promise_type будет уничтожен
// наш weak_ptr будет указывать на null. В остальных случаях,
// при однопоточной работе, state будет указывать на текущий
// awaitable объект, который прервал выполнение корутины
void terminate(boost::system::error_code ec = boost::asio::error::interrupted) {
if (auto ptr = m_state.lock(); ptr != nullptr && ptr->callback != nullptr) {
ptr->callback(ec);
// защищаемся от повторного вызова terminate()
ptr->callback = nullptr;
}
}
private:
// сохрани мы ссылку на promise_type - и перед нами встал бы вопрос:
// как узнать, завершилась ли корутина. Данный подход избавляет нас
// от этой диллемы
std::weak_ptr<promise_type::state_t> m_state;
explicit task(std::weak_ptr<promise_type::state_t> state)
: m_state{std::move(state)} {}
};
Мы получили искомое поведение. Правда, наш terminate не завершает корутину немедленно, а лишь приводит к возникновению ошибки (exception). Как по мне, это более прямой и гибкий путь, но незначительно поменяв логику await_suspend, можно сделать и немедленное завершение.
Полный код получившегося приложения
Не прерывайте прерывание
Но иногда хочется сказать: «Постойте! Вот сейчас меня прерывать не надо»Например, если вы уже выполняете операцию очистки при завершении корутины. В таких случаях RAII спасает не всегда. Иногда хочется выполнить какие-нибудь асинхронные вызовы, а деструктор не может быть корутиной. Или же для операции требуется большое количество локальных переменных. Что можно сделать в этой ситуации?
- запустить еще одну корутину в конце текущей или из деструктора и выполнить требуемые действия в ней;
- можно выполнить какой-нибудь специальный co_await и обработать его в том же await_transform;
- можно написать шаблон-декоратор, который скроет от await_transform метод on_terminate, и научить наш task работать с такими awaitable объектами.
Таким образом, мы получаем в конструкторе promise_type доступ к параметрам корутины. Причем мы можем не только читать их, но и менять!
struct promise_type {
struct state_t {
bool locked = false;
std::function<void(boost::system::error_code)> callback;
void lock() { locked = true; }
void unlock { locked = false; }
using ptr = std::shared_ptr<state_t>;
};
ptr state;
// на случай, если наша корутина вообще не имеет параметров
promise_type() : state{std::make_shared<state_t>()} { }
// этот конструктор подойдет под любой набор параметров
template <typename ...Args>
explicit promise_type(Args & ...args) : promise_type{} {
// используем fold expression: для каждого параметра будет создана
// лямбда, затем все их результаты сложены при помощи логического "или"
//
// конструкция выглядит страшно, но фактически она скомпилируется
// в одну операцию присваивания для первого подходящего по типу
// параметра. Остальное будет безжалостно выкинуто в процессе
// оптимизации. Кто не верит, вот ссылка на compiler explorer
// https://godbolt.org/z/hsrPErqfn
([this] (auto ¶m) {
if constexpr (std::is_same_v<Args, state_t:
tr>) {// мы ЗАПИСЫВАЕМ новое значение в параметр корутины
// и делаем это фактически до её запуска
param = state;
return true;
}
return false;
}(args) || ...);
}
// ...
};
Это небольшое улучшение позволяет нам написать следующий код:
task coroutine(task::state_t lock = {}) {
using std::chrono_literals;
try {
std::cout << "we are doing our work here" << std::endl;
co_await sleep(10s);
std::cout << "work is complete" << std::endl;
} catch (...) { }
std::cout << "final cleanup" << std::endl;
// я назвал методы state - lock и unlock, это дает
// мне возможность использовать std::lock_guard
const std::lock_guard g(*lock);
// следующий sleep уже не будет прерван
co_await sleep(5s);
}
З.Ы.
Механизмы, рассмотренные сегодня, позволяют реализовывать поистине фантастические вещи. А их сочетание может дать просто волшебные возможности и сорвать крышу. Только подумайте, оператор внутри функции может получить неявный доступ к её параметрам. Не уверен, что могу сходу придумать удачный пример использования подобной связки.Но при аккуратном подходе, в первую очередь разработчиками библиотек и фреймворков, конструкторы promise_type и await_transform могут позволить изящно решить массу задач асинхронного программирования.
Прерывая корутины С++
В своих предыдущих постах я описывал задачи, которые были решены скорее в академических целях. Сегодня я хочу поделиться реальным примером, который работает в продакшене. Речь пойдет о написании...
habr.com