В настоящей статье приводится разбор вариантов устранения антипаттерна «Зомби», описанного в первой части: Сказ об опасном std::enable_shared_from_this, или антипаттерн «Зомби».
Введение
Весь код, приведённый в статье, опубликован на гитхабе в ветках. В коде использованы несколько новшеств C++17 — weak_from_this(), if statement with init-statement, может что-то ещё по мелочи.
В примерах из первой части статьи цепочка возникновения проблемного кода выглядит следующим образом:
— задумывается класс асинхронного (неблокирующего) выполнения какого-то длительного процесса;
— для придания процессу асинхронности используется отдельный поток выполнения (std::thread);
— для удобства все нужные для работы процесса данные складываются в поля класса;
— в деструкторе класса предусматривается семафор для информирования потока о необходимости досрочного прекращения процесса в случае уничтожения экземпляра класса;
— поток выполнения нуждается в гарантии валидности экземпляра класса, для чего применяется захват сильной ссылки shared_from_this;
— объект потока выполнения является полем данных класса.
Возникает циклическая ссылка. Уничтожение вышестоящей бизнес-логикой всех сильных ссылок на экземпляр класса не приводит к вызову его деструктора. Неожиданно для разработчика попытка досрочного прекращения процесса в соответствии с RAII не срабатывает. Поток выполнения продолжает работу, что вызывает:
— труднодетектируемую утечку ресурсов (всегда);
— неопределённое поведение (в зависимости от применения);
— сбой высокоуровневой логики (в зависимости от применения).
Вот эту ситуацию и будем развязывать.
Способ очевидный, или weak
Одним из основных элементов антипаттерна «Зомби» является циклическая ссылка. Стандартной техникой её предотвращения является применение std::weak_ptr.
Идея способа: лямбда, запускающаяся в отдельном потоке, должна захватывать weak_from_this(), а не shared_from_this(). Модифицируем только зомби, и обходимся без масштабного рефакторинга.
SimpleZomby
Было:
void Zomby::runOnce(std::shared_ptr<Common::Listener> listener)
{
if (_semaphore) {
throw std::runtime_error("SimpleZomby::Zomby::runOnce() called twice");
}
_listener = listener;
_semaphore = true;
_thread = std::thread([shis = shared_from_this()](){
while (shis && shis->_listener && shis->_semaphore) {
shis->_listener->processData(std::make_shared<Common::Listener::Data>("SimpleZomby is alive!n"));
std::this_thread::sleep_for(std::chrono::seconds(1));
}
});
}
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
============================================================
| Zomby was killed |
============================================================
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
Стало:
void Zomby::runOnce(std::shared_ptr<Common::Listener> listener)
{
if (_semaphore) {
throw std::runtime_error("SimpleZomby::Zomby::runOnce() called twice");
}
_listener = listener;
_semaphore = true;
_thread = std::thread([whis = weak_from_this()](){
while (auto shis = whis.lock()) {
if (shis->_listener && shis->_semaphore) {
shis->_listener->processData(std::make_shared<Common::Listener::Data>("SimpleZomby is alive!n"));
std::this_thread::sleep_for(std::chrono::seconds(1));
}
}
});
}
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
============================================================
| Zomby was killed |
============================================================
N11SimpleZomby5ZombyE::~Zomby
N6Common22WriteToConsoleListenerE::~WriteToConsoleListener
Код вроде бы удалось починить — зомби корректно останавливается.
Кроме того, семафор стал лишним (хотя на самом деле он и до этого не работал) — функционал информирования потока о необходимости остановки удалось возложить на атомарный счётчик ссылок std::shared_ptr. Единственная польза от семафора — защита от повторного вызова функции runOnce, хотя и её можно переложить, например, на проверку _listener.
После удаления семафора:
void Zomby::runOnce(std::shared_ptr<Common::Listener> listener)
{
if (_semaphore) {
throw std::runtime_error("SimpleZomby::Zomby::runOnce() called twice");
}
_listener = listener;
_semaphore = true;
_thread = std::thread([whis = weak_from_this()](){
while (auto shis = whis.lock()) {
if (shis->_listener) {
shis->_listener->processData(std::make_shared<Common::Listener::Data>("SimpleZomby is alive!n"));
std::this_thread::sleep_for(std::chrono::seconds(1));
}
}
});
}
Однако при ближайшем рассмотрении становится понятно, что в теле лямбды сильная ссылка удерживается необоснованно долго — всё время выполнения длительной операции sleep_for.
Устраняется как-то так:
void Zomby::runOnce(std::shared_ptr<Common::Listener> listener)
{
if (_semaphore) {
throw std::runtime_error("SimpleZomby::Zomby::runOnce() called twice");
}
_listener = listener;
_semaphore = true;
_thread = std::thread([whis = weak_from_this()](){
while (auto shis = whis.lock()) {
if (shis->_listener) {
shis->_listener->processData(std::make_shared<Common::Listener::Data>("SimpleZomby is alive!n"));
shis.reset();
std::this_thread::sleep_for(std::chrono::seconds(1));
}
}
});
}
Или вот так:
void Zomby::runOnce(std::shared_ptr<Common::Listener> listener)
{
if (_semaphore) {
throw std::runtime_error("SimpleZomby::Zomby::runOnce() called twice");
}
_listener = listener;
_semaphore = true;
_thread = std::thread([whis = weak_from_this()](){
while (true) {
if (auto shis = whis.lock(); shis && shis->_listener) {
shis->_listener->processData(std::make_shared<Common::Listener::Data>("SimpleZomby is alive!n"));
} else {
break;
}
std::this_thread::sleep_for(std::chrono::seconds(1));
}
});
}
Ну что, переходим к следующему примеру?
Как бы не так!
У нас тут гонка.
Проще всего её продемонстрировать с помощью переноса слипа в чувствительное место:
void Zomby::runOnce(std::shared_ptr<Common::Listener> listener)
{
if (_semaphore) {
throw std::runtime_error("SimpleZomby::Zomby::runOnce() called twice");
}
_listener = listener;
_semaphore = true;
_thread = std::thread([whis = weak_from_this()](){
while (true) {
if (auto shis = whis.lock(); shis && shis->_listener) {
/*!!!*/ std::this_thread::sleep_for(std::chrono::seconds(1));
shis->_listener->processData(std::make_shared<Common::Listener::Data>("SimpleZomby is alive!n"));
} else {
break;
}
}
});
}
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
============================================================
| Zomby was killed |
============================================================
SimpleZomby is alive!
N11SimpleZomby5ZombyE::~Zomby
N6Common22WriteToConsoleListenerE::~WriteToConsoleListener
Вот что произошло:
| main thread | zomby thread |
|---|---|
| вызывает runOnce | |
| захватывает в лямбда-функцию слабую ссылку на экземпляр класса | |
| запускает лямбда-функцию на выполнение в отдельном потоке | |
| пытается взять сильную ссылку на экземпляр класса с проверкой результата; проверка проходит успешно | |
| уничтожает единственную внешнюю сильную ссылку | |
| передаёт сообщение полю данных экземпляра класса — listener'у |
Зомби остановился.
Но всё ещё слишком поздно — от уничтоженного зомби пришло одно сообщение.
В зависимости от применения, такое поведение может как являться приемлемым, так и послужить триггером для какой-либо неожиданной цепочки действий.
Почему приход данных в listener может быть неожиданным?
Зомби вряд ли является единственным игроком на поле.
Данные, прихода которых он ожидает, наверняка предполагают дальнейшую обработку (расшифровку, парсинг, отображение на экране и т.п.). Уничтожение бизнес-логикой зомби может означать, что механизмы обработки данных такого типа также уничтожены. Передача данных на обработку в уничтоженные механизмы грозит неопределённым поведением.
Также сообщение от зомби может предполагать высокоуровневую ответную реакцию. Вряд ли пользователь обрадуется сообщению «Выполнено успешно» после нажатия кнопки «Отмена».
SteppingZomby
Было:
void Zomby::runOnce(std::shared_ptr<Common::Listener> listener)
{
if (_semaphore) {
throw std::runtime_error("SteppingZomby::Zomby::runOnce() called twice");
}
_listener = listener;
_semaphore = true;
_thread = std::thread([shis = shared_from_this()](){
if (shis && shis->_listener && shis->_semaphore) {
shis->resolveDnsName();
}
if (shis && shis->_listener && shis->_semaphore) {
shis->connectTcp();
}
if (shis && shis->_listener && shis->_semaphore) {
shis->establishSsl();
}
if (shis && shis->_listener && shis->_semaphore) {
shis->sendHttpRequest();
}
if (shis && shis->_listener && shis->_semaphore) {
shis->readHttpReply();
}
});
}
N13SteppingZomby5ZombyE::resolveDnsName started
N13SteppingZomby5ZombyE::resolveDnsName finished
N13SteppingZomby5ZombyE::connectTcp started
============================================================
| Zomby was killed |
============================================================
N13SteppingZomby5ZombyE::connectTcp finished
N13SteppingZomby5ZombyE::establishSsl started
N13SteppingZomby5ZombyE::establishSsl finished
N13SteppingZomby5ZombyE::sendHttpRequest started
N13SteppingZomby5ZombyE::sendHttpRequest finished
N13SteppingZomby5ZombyE::readHttpReply started
N13SteppingZomby5ZombyE::readHttpReply finished
N13SteppingZomby5ZombyE::~Zomby
N6Common22WriteToConsoleListenerE::~WriteToConsoleListener
Стало (ненужную проверку семафора сразу убираем):
void Zomby::runOnce(std::shared_ptr<Common::Listener> listener)
{
if (_semaphore) {
throw std::runtime_error("SteppingZomby::Zomby::runOnce() called twice");
}
_listener = listener;
_semaphore = true;
_thread = std::thread([whis = weak_from_this()](){
if (auto shis = whis.lock(); shis && shis->_listener) {
shis->resolveDnsName();
}
if (auto shis = whis.lock(); shis && shis->_listener) {
shis->connectTcp();
}
if (auto shis = whis.lock(); shis && shis->_listener) {
shis->establishSsl();
}
if (auto shis = whis.lock(); shis && shis->_listener) {
shis->sendHttpRequest();
}
if (auto shis = whis.lock(); shis && shis->_listener) {
shis->readHttpReply();
}
});
}
N13SteppingZomby5ZombyE::resolveDnsName started
N13SteppingZomby5ZombyE::resolveDnsName finished
N13SteppingZomby5ZombyE::connectTcp started
============================================================
| Zomby was killed |
============================================================
N13SteppingZomby5ZombyE::connectTcp finished
N13SteppingZomby5ZombyE::~Zomby
N6Common22WriteToConsoleListenerE::~WriteToConsoleListener
Точно так же, как в примере SimpleZomby:
— вроде бы удалось остановить зомби;
— сильная ссылка удерживается слишком долго — функции resolveDnsName, connectTcp, establishSsl, sendHttpRequest, readHttpReply требуют валидного состояния экземпляра класса;
— присутствует состояние гонки.
Начатый шаг всегда отрабатывается до конца и всегда отправляет результат в listener, но цепочка шагов в большинстве случаев может быть прервана вышестоящей бизнес-логикой — кроме случая гонки, из-за которой следующий шаг может всё же начаться после уничтожения внешней сильной ссылки.
Рефакторинг private-методов класса может снизить время удержания сильной ссылки, но не позволит полностью устранить состояние гонки.
BoozdedZomby
Было:
void Zomby::runOnce(std::shared_ptr<Common::Listener> listener)
{
if (_semaphore) {
throw std::runtime_error("BoozdedZomby::Zomby::runOnce() called twice");
}
_listener = listener;
_semaphore = true;
_thread = std::thread([shis = shared_from_this()]() {
while (shis && shis->_semaphore && shis->_listener) {
auto handler = [shis](auto errorCode) {
if (shis && shis->_listener && errorCode == boozd::azzio::io_context::error_code::no_error) {
std::ostringstream buf;
buf << "BoozdedZomby has got a fresh data: ";
for (auto const &elem; : shis->_buffer)
buf << elem << ' ';
buf << std::endl;
shis->_listener->processData(std::make_shared<Common::Listener::Data>(buf.str()));
}
};
shis->_buffer.clear();
shis->_context.async_read(shis->_stream, shis->_buffer, handler);
shis->_context.run();
}
});
}
BoozdedZomby has got a fresh data: 1144108930 101027544 1458777923 1115438165 74243042
BoozdedZomby has got a fresh data: 143542612 1131570933
BoozdedZomby has got a fresh data: 893351816 563613512 704877633
BoozdedZomby has got a fresh data: 1551901393 1399125485 1899894091 937186357 590357944 357571490
============================================================
| Zomby was killed |
============================================================
BoozdedZomby has got a fresh data: 1927702196 130060903 1083454666 2118797801 2035308228 824938981
BoozdedZomby has got a fresh data: 2020739063 1635339425 34075629
BoozdedZomby has got a fresh data: 2146319451 500782188 1269406752 884936716 892053144
BoozdedZomby has got a fresh data: 330111137 1723153177 1070477904
BoozdedZomby has got a fresh data: 343098142 280090412 589673557 889688008 2014119113 388471006
Стало:
void Zomby::runOnce(std::shared_ptr<Common::Listener> listener)
{
if (_semaphore) {
throw std::runtime_error("BoozdedZomby::Zomby::runOnce() called twice");
}
_listener = listener;
_semaphore = true;
_thread = std::thread([whis = weak_from_this()]() {
while (auto shis = whis.lock()) {
if (shis->_semaphore && shis->_listener) {
auto handler = [whis](auto errorCode) {
auto shis = whis.lock();
if (shis && shis->_listener && errorCode == boozd::azzio::io_context::error_code::no_error) {
std::ostringstream buf;
buf << "BoozdedZomby has got a fresh data: ";
for (auto const &elem; : shis->_buffer)
buf << elem << ' ';
buf << std::endl;
shis->_listener->processData(std::make_shared<Common::Listener::Data>(buf.str()));
}
};
shis->_buffer.clear();
shis->_context.async_read(shis->_stream, shis->_buffer, handler);
shis->_context.run();
}
}
});
}
BoozdedZomby has got a fresh data: 1144108930 101027544 1458777923 1115438165 74243042
BoozdedZomby has got a fresh data: 143542612 1131570933
BoozdedZomby has got a fresh data: 893351816 563613512 704877633
BoozdedZomby has got a fresh data: 1551901393 1399125485 1899894091 937186357 590357944 357571490
============================================================
| Zomby was killed |
============================================================
BoozdedZomby has got a fresh data: 1927702196 130060903 1083454666 2118797801 2035308228 824938981
N12BoozdedZomby5ZombyE::~Zomby
N5boozd5azzio10io_contextE::~io_context
N5boozd5azzio13random_streamE::~random_stream
N6Common22WriteToConsoleListenerE::~WriteToConsoleListener
Зомби-поток отдаёт управление функции boozd::azzio::io_context::run(). При этом необходимо обеспечить валидность источника данных, буфера и коллбэк-функции до момента возврата из run(). Коллбэк-функция будет вызвана при валидном экземпляре класса — а значит, она сможет воспользоваться listener'ом.
В коде присутствуют те же самые недостатки, только выражены они ещё сильнее: BoozdedZomby дорабатывает начатую операцию до конца, и внешний код может лишь отменить очередной повтор. В данном случае вполне можно заявить, что способ не сработал. Останавливать надо текущую операцию, а не следующую.
Вывод по weak-способу
Модификация только зомби только в части замены захвата shared_from_this на захват weak_from_this:
— повышает риски обращения к инвалидированному состоянию (примеры приводить не буду, но на практике встречал);
— не способна устранить состояние гонки (взаимный порядок операций в двух потоках влияет на то, будут ли данные переданы в listener после уничтожения зомби бизнес-логикой).
Функционирование зомби требует захвата сильной ссылки хотя бы на короткое время — и в это время уничтожение всех внешних ссылок не приводит к его остановке. В некоторых случаях сильная ссылка удерживается настолько долго, что делает способ неприменимым на практике.
Валидность объекта вряд ли подходит на роль единственного условия продолжения работы.
В то же время, следует отметить, что проблема гонки связана с выбором способа передачи данных от зомби и может растворяться при выборе более подходящего способа передачи или при наличии более терпимого получателя (ещё вернёмся к этому вопросу).
Способ честный, или use_count
Идея способа: по-прежнему захватываем shared_from_this, но при этом проверяем std::shared_ptr::use_count().
Почему честный?
Использование weak_from_this создаёт иллюзию отказа от влияния на собственное время жизни. А в данном варианте — никакого самообмана, экземпляр класса по-прежнему управляет собственным временем жизни. Но при этом детектирует ситуацию отсутствия внешних сильных ссылок.
SimpleZomby
Было:
void Zomby::runOnce(std::shared_ptr<Common::Listener> listener)
{
if (_semaphore) {
throw std::runtime_error("SimpleZomby::Zomby::runOnce() called twice");
}
_listener = listener;
_semaphore = true;
_thread = std::thread([shis = shared_from_this()](){
while (shis && shis->_listener && shis->_semaphore) {
shis->_listener->processData(std::make_shared<Common::Listener::Data>("SimpleZomby is alive!n"));
std::this_thread::sleep_for(std::chrono::seconds(1));
}
});
}
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
============================================================
| Zomby was killed |
============================================================
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
Стало:
void Zomby::runOnce(std::shared_ptr<Common::Listener> listener)
{
if (_semaphore) {
throw std::runtime_error("SimpleZomby::Zomby::runOnce() called twice");
}
_listener = listener;
_semaphore = true;
_thread = std::thread([shis = shared_from_this()](){
while (shis && shis.use_count() > 1 && shis->_listener && shis->_semaphore) {
shis->_listener->processData(std::make_shared<Common::Listener::Data>("SimpleZomby is alive!n"));
std::this_thread::sleep_for(std::chrono::seconds(1));
}
});
}
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
============================================================
| Zomby was killed |
============================================================
N11SimpleZomby5ZombyE::~Zomby
N6Common22WriteToConsoleListenerE::~WriteToConsoleListener
Свойства данной модификации точно совпадают со свойствами weak-способа:
— семафор стал лишним в условии выхода;
— зомби удалось остановить, но присутствует состояние гонки (если проверка use_count() в зомби-потоке прошла успешно — данные будут отправлены даже после уничтожения последней внешней ссылки).
SteppingZomby
Было:
void Zomby::runOnce(std::shared_ptr<Common::Listener> listener)
{
if (_semaphore) {
throw std::runtime_error("SteppingZomby::Zomby::runOnce() called twice");
}
_listener = listener;
_semaphore = true;
_thread = std::thread([shis = shared_from_this()](){
if (shis && shis->_listener && shis->_semaphore) {
shis->resolveDnsName();
}
if (shis && shis->_listener && shis->_semaphore) {
shis->connectTcp();
}
if (shis && shis->_listener && shis->_semaphore) {
shis->establishSsl();
}
if (shis && shis->_listener && shis->_semaphore) {
shis->sendHttpRequest();
}
if (shis && shis->_listener && shis->_semaphore) {
shis->readHttpReply();
}
});
}
N13SteppingZomby5ZombyE::resolveDnsName started
N13SteppingZomby5ZombyE::resolveDnsName finished
N13SteppingZomby5ZombyE::connectTcp started
============================================================
| Zomby was killed |
============================================================
N13SteppingZomby5ZombyE::connectTcp finished
N13SteppingZomby5ZombyE::establishSsl started
N13SteppingZomby5ZombyE::establishSsl finished
N13SteppingZomby5ZombyE::sendHttpRequest started
N13SteppingZomby5ZombyE::sendHttpRequest finished
N13SteppingZomby5ZombyE::readHttpReply started
N13SteppingZomby5ZombyE::readHttpReply finished
N13SteppingZomby5ZombyE::~Zomby
N6Common22WriteToConsoleListenerE::~WriteToConsoleListener
Стало:
void Zomby::runOnce(std::shared_ptr<Common::Listener> listener)
{
if (_semaphore) {
throw std::runtime_error("SteppingZomby::Zomby::runOnce() called twice");
}
_listener = listener;
_semaphore = true;
_thread = std::thread([shis = shared_from_this()](){
if (shis && shis.use_count() > 1 && shis->_listener && shis->_semaphore) {
shis->resolveDnsName();
}
if (shis && shis.use_count() > 1 && shis->_listener && shis->_semaphore) {
shis->connectTcp();
}
if (shis && shis.use_count() > 1 && shis->_listener && shis->_semaphore) {
shis->establishSsl();
}
if (shis && shis.use_count() > 1 && shis->_listener && shis->_semaphore) {
shis->sendHttpRequest();
}
if (shis && shis.use_count() > 1 && shis->_listener && shis->_semaphore) {
shis->readHttpReply();
}
});
}
N13SteppingZomby5ZombyE::resolveDnsName started
N13SteppingZomby5ZombyE::resolveDnsName finished
N13SteppingZomby5ZombyE::connectTcp started
============================================================
| Zomby was killed |
============================================================
N13SteppingZomby5ZombyE::connectTcp finished
N13SteppingZomby5ZombyE::~Zomby
N6Common22WriteToConsoleListenerE::~WriteToConsoleListener
Всё точно как в weak-способе.
BoozdedZomby
Было:
void Zomby::runOnce(std::shared_ptr<Common::Listener> listener)
{
if (_semaphore) {
throw std::runtime_error("BoozdedZomby::Zomby::runOnce() called twice");
}
_listener = listener;
_semaphore = true;
_thread = std::thread([shis = shared_from_this()]() {
while (shis && shis->_semaphore && shis->_listener) {
auto handler = [shis](auto errorCode) {
if (shis && shis->_listener && errorCode == boozd::azzio::io_context::error_code::no_error) {
std::ostringstream buf;
buf << "BoozdedZomby has got a fresh data: ";
for (auto const &elem; : shis->_buffer)
buf << elem << ' ';
buf << std::endl;
shis->_listener->processData(std::make_shared<Common::Listener::Data>(buf.str()));
}
};
shis->_buffer.clear();
shis->_context.async_read(shis->_stream, shis->_buffer, handler);
shis->_context.run();
}
});
}
BoozdedZomby has got a fresh data: 1144108930 101027544 1458777923 1115438165 74243042
BoozdedZomby has got a fresh data: 143542612 1131570933
BoozdedZomby has got a fresh data: 893351816 563613512 704877633
BoozdedZomby has got a fresh data: 1551901393 1399125485 1899894091 937186357 590357944 357571490
============================================================
| Zomby was killed |
============================================================
BoozdedZomby has got a fresh data: 1927702196 130060903 1083454666 2118797801 2035308228 824938981
BoozdedZomby has got a fresh data: 2020739063 1635339425 34075629
BoozdedZomby has got a fresh data: 2146319451 500782188 1269406752 884936716 892053144
BoozdedZomby has got a fresh data: 330111137 1723153177 1070477904
BoozdedZomby has got a fresh data: 343098142 280090412 589673557 889688008 2014119113 388471006
Стало:
void Zomby::runOnce(std::shared_ptr<Common::Listener> listener)
{
if (_semaphore) {
throw std::runtime_error("BoozdedZomby::Zomby::runOnce() called twice");
}
_listener = listener;
_semaphore = true;
_thread = std::thread([shis = shared_from_this()]() {
while (shis && shis.use_count() > 1 && shis->_semaphore && shis->_listener) {
auto handler = [&shis;](auto errorCode) {
if (shis && shis.use_count() > 1 && shis->_listener && errorCode == boozd::azzio::io_context::error_code::no_error) {
std::ostringstream buf;
buf << "BoozdedZomby has got a fresh data: ";
for (auto const &elem; : shis->_buffer)
buf << elem << ' ';
buf << std::endl;
shis->_listener->processData(std::make_shared<Common::Listener::Data>(buf.str()));
}
};
shis->_buffer.clear();
shis->_context.async_read(shis->_stream, shis->_buffer, handler);
shis->_context.run();
}
});
}
BoozdedZomby has got a fresh data: 1144108930 101027544 1458777923 1115438165 74243042
BoozdedZomby has got a fresh data: 143542612 1131570933
BoozdedZomby has got a fresh data: 893351816 563613512 704877633
BoozdedZomby has got a fresh data: 1551901393 1399125485 1899894091 937186357 590357944 357571490
============================================================
| Zomby was killed |
============================================================
N12BoozdedZomby5ZombyE::~Zomby
N5boozd5azzio10io_contextE::~io_context
N5boozd5azzio13random_streamE::~random_stream
N6Common22WriteToConsoleListenerE::~WriteToConsoleListener
Вот тут уже значительно лучше, чем в weak-способе: валидность экземпляра класса не считается достаточным основанием для передачи данных в listener, требуется ещё и наличие хотя бы одной внешней сильной ссылки.
При реализации данного способа надо проявлять повышенную внимательность: очень просто ошибочно создать дополнительные сильные копии внутри зомби (обратите внимание — сейчас shis захватывается в коллбэк-функцию по ссылке, а не по значению). Также следует правильно выбирать места для проверок.
Состояние гонки по-прежнему присутствует.
Вывод по use_count-способу
Добавление в условие выхода проверки use_count при сохранении захвата shared_from_this снижает риск обращения к инвалидированному состоянию по сравнению с weak-способом. При этом по-прежнему сохраняется риск передачи данных в listener после уничтожения зомби вышестоящей бизнес-логикой.
Способы use_count и weak пригодны для применения на практике при соответствующей адаптации способа передачи данных.
Такими адаптациями могут быть:
— предоставление зомби прокси-listener'a, обеспечивающего надёжный разрыв соединения с конечным listener'ом (но не обольщайтесь — этот способ не является ни простым, ни изящным; пример можно посмотреть в репозитории, ветка fixes/weak_with_ProxyListener);
— использование сложных систем типа Qt Signals & Slots (однако тут легко получить иллюзию безопасности вместо безопасности);
— отказ от event-driven в пользу поллинга — в этом случае инициатива всегда принадлежит вышестоящему уровню, и возможность обращения к нижестоящему объекту естественным образом исчезает после уничтожения ссылки на него.
Способ синхронный, или this
Идея способа: отказываемся от наследования от std::shared_from_this. Никаких грязных трюков с управлением собственным временем жизни. Никаких узлов в коде.
Лямбда-функция захватывает this. Время жизни потока синхронизируется со временем жизни экземпляра класса. В деструкторе вместо std::thread::detach() вызывается std::thread::join().
Зомби растворяется.
Вместо него возникает псевдо-асинхронный класс: все его методы являются неблокирующими, кроме деструктора. Уничтожение может длиться неопределённо долго — порядка длительности того процесса, которому мы пытались придать свойство асинхронности. На практике это могут быть единицы и даже десятки секунд, и возможна зависимость от внешних факторов (размера ответа на HTTP-запрос, потери сетевого пакета и т.п.).
Было:
#pragma once
#include <memory>
#include <atomic>
#include <thread>
#include "Common/Manager.h"
namespace Common {
class Listener;
} // namespace Common
namespace SimpleZomby {
class Zomby final : public Common::Manager, public std::enable_shared_from_this<Zomby>
{
public:
static std::shared_ptr<Zomby> create();
~Zomby() override;
void runOnce(std::shared_ptr<Common::Listener> listener) override;
private:
Zomby();
using Semaphore = std::atomic<bool>;
std::shared_ptr<Common::Listener> _listener;
Semaphore _semaphore = false;
std::thread _thread;
};
} // namespace SimpleZomby
#include <sstream>
#include "SimpleZomby.h"
#include "Common/Listener.h"
namespace SimpleZomby {
std::shared_ptr<Zomby> Zomby::create()
{
return std::shared_ptr<Zomby>(new Zomby());
}
Zomby::Zomby() = default;
Zomby::~Zomby()
{
_semaphore = false;
if (_thread.joinable()) {
_thread.detach();
}
if (_listener) {
std::ostringstream buf;
buf << typeid(*this).name() << "::" << __func__ << std::endl;
_listener->processData(std::make_shared<Common::Listener::Data>(buf.str()));
}
}
void Zomby::runOnce(std::shared_ptr<Common::Listener> listener)
{
if (_semaphore) {
throw std::runtime_error("SimpleZomby::Zomby::runOnce() called twice");
}
_listener = listener;
_semaphore = true;
_thread = std::thread([shis = shared_from_this()](){
while (shis && shis->_listener && shis->_semaphore) {
shis->_listener->processData(std::make_shared<Common::Listener::Data>("SimpleZomby is alive!n"));
std::this_thread::sleep_for(std::chrono::seconds(1));
}
});
}
} // namespace SimpleZomby
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
============================================================
| Zomby was killed |
============================================================
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
Стало:
#pragma once
#include <memory>
#include <atomic>
#include <thread>
#include "Common/Manager.h"
namespace Common {
class Listener;
} // namespace Common
namespace SimpleZomby {
class Zomby final : public Common::Manager
{
public:
static std::shared_ptr<Zomby> create();
~Zomby() override;
void runOnce(std::shared_ptr<Common::Listener> listener) override;
private:
Zomby();
using Semaphore = std::atomic<bool>;
std::shared_ptr<Common::Listener> _listener;
Semaphore _semaphore = false;
std::thread _thread;
};
} // namespace SimpleZomby
#include <sstream>
#include "SimpleZomby.h"
#include "Common/Listener.h"
namespace SimpleZomby {
std::shared_ptr<Zomby> Zomby::create()
{
return std::shared_ptr<Zomby>(new Zomby());
}
Zomby::Zomby() = default;
Zomby::~Zomby()
{
_semaphore = false;
if (_thread.joinable()) {
_thread.join();
}
if (_listener) {
std::ostringstream buf;
buf << typeid(*this).name() << "::" << __func__ << std::endl;
_listener->processData(std::make_shared<Common::Listener::Data>(buf.str()));
}
}
void Zomby::runOnce(std::shared_ptr<Common::Listener> listener)
{
if (_semaphore) {
throw std::runtime_error("SimpleZomby::Zomby::runOnce() called twice");
}
_listener = listener;
_semaphore = true;
_thread = std::thread([this](){
while (_listener && _semaphore) {
_listener->processData(std::make_shared<Common::Listener::Data>("SimpleZomby is alive!n"));
std::this_thread::sleep_for(std::chrono::seconds(1));
}
});
}
} // namespace SimpleZomby
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
N11SimpleZomby5ZombyE::~Zomby
============================================================
| Zomby was killed |
============================================================
N6Common22WriteToConsoleListenerE::~WriteToConsoleListener
Отметим:
— это пока единственный пример, в котором деструктор зомби (хотя это уже и не зомби на самом деле) был вызван до вывода «Zomby was killed»;
— семафор наконец-то сыграл ту роль, которая отводилась ему изначально, хотя его можно заменить проверкой валидности _listener;
— требование хранения экземпляра класса исключительно в std::shared_ptr стало излишним, статическую функцию create() можно упразднить в пользу выноса конструктора в public-секцию класса;
— захват this в лямбда-функцию требует постоянного расположения экземпляра класса в памяти — унаследованный от Common::Manager запрет copy- и move-семантики по-прежнему важен.
Модификации SteppingZomby и BoozdedZomby в данном случае полностью аналогичны, приводить их в статье нет смысла. Кому интересно — смотрите в ветке fixes/this репозитория.
Вывод по this-способу
Способ хорош всем, кроме блокирующего деструктора. Блокировка GUI-потока на несколько секунд вполне может заставить операционную систему считать процесс зависшим. В таком виде способ вряд ли пригоден для практического применения.
Способ пригодный, или semaphore_done_right
Во всех рассмотренных до сих пор примерах класс использовался как агрегатор данных, необходимых для работы отдельного потока. Это приводило к необходимости обеспечения времени жизни экземпляра класса, превышающего время жизни запущенного потока.
Да, агрегировать данные в классе удобно.
Но только до тех пор, пока это не приводит к букету сложных проблем.
Идея способа: правильное разграничение ответственностей между классом и потоком.
Обязанности класса:
— запуск потока;
— перевод семафора в запрещающее состояние в деструкторе.
Обязанности потока:
— не продолжать работу при запрещающем состоянии семафора.
Никакой связи по данным (кроме разделяемого семафора), никаких требований по времени жизни (особенно от нижележащего потока в сторону вышележащего класса). Никакого захвата экземпляра класса в лямбду — ни в виде сырого указателя, ни в виде умного.
Для устранения гонки, возникаюшей в способах weak и use_count, придётся использовать синхронизацию с помощью мьютекса. Деструктор захватывает мьютекс на время смены значения семафора, а отдельный поток — на время выполнения операции передачи данных в listener.
Деструктор вызывает std::thread::detach(), не дожидаясь завершения потока (вернёмся к этому вопросу позднее).
Было:
#pragma once
#include <memory>
#include <atomic>
#include <thread>
#include "Common/Manager.h"
namespace Common {
class Listener;
} // namespace Common
namespace SimpleZomby {
class Zomby final : public Common::Manager, public std::enable_shared_from_this<Zomby>
{
public:
static std::shared_ptr<Zomby> create();
~Zomby() override;
void runOnce(std::shared_ptr<Common::Listener> listener) override;
private:
Zomby();
using Semaphore = std::atomic<bool>;
std::shared_ptr<Common::Listener> _listener;
Semaphore _semaphore = false;
std::thread _thread;
};
} // namespace SimpleZomby
#include <sstream>
#include "SimpleZomby.h"
#include "Common/Listener.h"
namespace SimpleZomby {
std::shared_ptr<Zomby> Zomby::create()
{
return std::shared_ptr<Zomby>(new Zomby());
}
Zomby::Zomby() = default;
Zomby::~Zomby()
{
_semaphore = false;
if (_thread.joinable()) {
_thread.detach();
}
if (_listener) {
std::ostringstream buf;
buf << typeid(*this).name() << "::" << __func__ << std::endl;
_listener->processData(std::make_shared<Common::Listener::Data>(buf.str()));
}
}
void Zomby::runOnce(std::shared_ptr<Common::Listener> listener)
{
if (_semaphore) {
throw std::runtime_error("SimpleZomby::Zomby::runOnce() called twice");
}
_listener = listener;
_semaphore = true;
_thread = std::thread([shis = shared_from_this()](){
while (shis && shis->_listener && shis->_semaphore) {
shis->_listener->processData(std::make_shared<Common::Listener::Data>("SimpleZomby is alive!n"));
std::this_thread::sleep_for(std::chrono::seconds(1));
}
});
}
} // namespace SimpleZomby
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
============================================================
| Zomby was killed |
============================================================
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
Стало:
#pragma once
#include <memory>
#include <atomic>
#include <thread>
#include "Common/Manager.h"
namespace Common {
class Listener;
} // namespace Common
namespace SimpleZomby {
class Zomby final : public Common::Manager
{
public:
static std::shared_ptr<Zomby> create();
~Zomby() override;
void runOnce(std::shared_ptr<Common::Listener> listener) override;
private:
Zomby();
using Semaphore = std::pair<std::mutex, bool>;
std::shared_ptr<Common::Listener> _listener;
std::shared_ptr<Semaphore> _semaphore;
std::thread _thread;
};
} // namespace SimpleZomby
#include <sstream>
#include "SimpleZomby.h"
#include "Common/Listener.h"
namespace SimpleZomby {
std::shared_ptr<Zomby> Zomby::create()
{
return std::shared_ptr<Zomby>(new Zomby());
}
Zomby::Zomby() = default;
Zomby::~Zomby()
{
if (_semaphore) {
if (std::this_thread::get_id() != _thread.get_id()) {
auto guard = std::lock_guard(_semaphore->first);
_semaphore->second = false;
} else {
_semaphore->second = false;
}
}
if (_thread.joinable()) {
_thread.detach();
}
if (_listener) {
std::ostringstream buf;
buf << typeid(*this).name() << "::" << __func__ << std::endl;
_listener->processData(std::make_shared<Common::Listener::Data>(buf.str()));
}
}
void Zomby::runOnce(std::shared_ptr<Common::Listener> listener)
{
if (_semaphore) {
throw std::runtime_error("SimpleZomby::Zomby::runOnce() called twice");
}
_listener = listener;
// two separate lines needed because of absence of std::mutex move c-tor
_semaphore = std::make_shared<Semaphore>();
_semaphore->second = true;
_thread = std::thread([listener, semaphore = _semaphore](){
while (listener && semaphore) {
{
auto guard = std::lock_guard(semaphore->first);
if (!semaphore->second) {
break;
}
listener->processData(std::make_shared<Common::Listener::Data>("SimpleZomby is alive!n"));
}
std::this_thread::sleep_for(std::chrono::seconds(1));
}
});
}
} // namespace SimpleZomby
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
N11SimpleZomby5ZombyE::~Zomby
============================================================
| Zomby was killed |
============================================================
N6Common22WriteToConsoleListenerE::~WriteToConsoleListener
Модификации для SteppingZomby и BoozdedZomby аналогичны (см. ветку fixes/semaphore_done_right) — настолько аналогичны, что объявление класса BoozdedZomby совпало с объявлением класса SimpleZomby, а в SteppingZomby присутствуют лишь допонительные статические функции, которые могут быть преобразованы в свободные функции.
Свойства данного способа:
— время выполнения деструктора перестало зависеть от внешних факторов — осталась только конкуренция за мьютекс, защищающий потенциально очень быструю операцию передачи данных в обработку;
— аккуратная работа с мьютексом позволила избавиться от состояния гонки — корректно написанная лямбда-функция гарантированно не отправит данные в listener после завершения деструктора;
— предусмотрена защита от deadlock'а на случай вызова деструктора из обработчика события прихода данных (уничтожение соединения в ответ на получение данных — вполне корректный сценарий);
— требование хранения экземпляра класса исключительно в std::shared_ptr стало излишним, статическую функцию create() можно упразднить в пользу выноса конструктора в public-секцию класса;
— запрет copy- и move-семантики, унаследованный от Common::Manager, стал излишним;
— listener по-прежнему хранится в поле класса — но только для диагностического вывода в деструкторе, он не связывает экземпляр класса с потоком выполнения;
— класс перестал зависеть от специфики данных, необходимых для работы отдельного потока, что открывает дополнительные возможности для обобщения;
— поток может продолжить выполнение в detached-состоянии после уничтожения экземпляра класса.
Последний пункт является потенциально проблемным по следующим причинам:
— маскирует ошибки в условии выхода из функции, выполняющейся в отдельном потоке;
— отдельный поток может дожить до завершения главного потока — тогда он будет аварийно зачищен системой, что приведёт к сбою деинициализации.
Способ финальный, или semaphore_done_right_with_ThreadJoiner
Если detach() потока может создавать проблемы — попробуем что-нибудь с этим сделать (запустим новый поток, который будет асинхронно ждать завершения старого потока).
Идея способа:
— нужна новая сущность (ThreadJoiner), способная взять на себя ответственность за отработавший поток (пожалуй, более подходящее названием для такой сущности — ThreadSanitizer, но оно уже занято);
— ThreadJoiner в отдельном служебном потоке будет ожидать отработки std::thread::join() последовательно для всех потоков, ответственность за которые он принял;
— деструктор ThreadJoiner будет блокировать выполнение до тех пор, пока не завершатся все потоки, ответственность за которые он принял.
Абстрактный интерфейс:
#pragma once
#include <thread>
namespace Common {
class ThreadJoiner
{
public:
ThreadJoiner() = default;
ThreadJoiner(const ThreadJoiner&) = delete;
ThreadJoiner(ThreadJoiner&&) = delete;
ThreadJoiner& operator=(const ThreadJoiner&) = delete;
ThreadJoiner& operator=(ThreadJoiner&&) = delete;
virtual ~ThreadJoiner() = default;
virtual void join(std::thread&&) = 0;
};
} // namespace Common
Конкретная асинхронная реализация:
#pragma once
#include <thread>
#include <queue>
#include <mutex>
#include <optional>
#include <condition_variable>
#include "Common/ThreadJoiner.h"
namespace Common {
class ThreadJoinerAsync final : public ThreadJoiner
{
public:
ThreadJoinerAsync();
ThreadJoinerAsync(const ThreadJoinerAsync&) = delete;
ThreadJoinerAsync(ThreadJoinerAsync&&) = delete;
ThreadJoinerAsync& operator=(const ThreadJoinerAsync&) = delete;
ThreadJoinerAsync& operator=(ThreadJoinerAsync&&) = delete;
~ThreadJoinerAsync() override;
void join(std::thread&&) override;
private:
using Data = std::queue<std::thread>;
using Semaphore = std::atomic<bool>;
std::mutex _mutex;
Data _data;
Semaphore _semaphore;
std::condition_variable _cv;
std::thread _thread;
void threadFn();
};
} // namespace Common
#include "ThreadJoinerAsync.h"
namespace Common {
ThreadJoinerAsync::ThreadJoinerAsync()
: _semaphore(true)
, _thread(&ThreadJoinerAsync::threadFn, this)
{
}
ThreadJoinerAsync::~ThreadJoinerAsync()
{
_semaphore = false;
_cv.notify_one();
if (_thread.joinable()) {
_thread.join();
}
}
void ThreadJoinerAsync::join(std::thread&& thread)
{
{
auto lock = std::lock_guard(_mutex);
_data.push(std::move(thread));
}
_cv.notify_one();
}
void ThreadJoinerAsync::threadFn()
{
while (true) {
auto lock = std::unique_lock(_mutex);
_cv.wait(lock, [this](){ return !_semaphore || !_data.empty(); });
if (!_semaphore && _data.empty()) {
return;
}
auto threadToJoin = std::move(_data.front());
_data.pop();
lock.unlock();
if (threadToJoin.joinable()) {
threadToJoin.join();
}
}
}
} // namespace Common
Применение ThreadJoiner в SimpleZomby:
#pragma once
#include <memory>
#include <atomic>
#include <thread>
#include "Common/Manager.h"
namespace Common {
class Listener;
class ThreadJoiner;
} // namespace Common
namespace SimpleZomby {
class Zomby final : public Common::Manager
{
public:
Zomby(std::shared_ptr<Common::ThreadJoiner>);
~Zomby() override;
void runOnce(std::shared_ptr<Common::Listener> listener) override;
private:
using Semaphore = std::pair<std::mutex, bool>;
std::shared_ptr<Common::Listener> _listener;
std::shared_ptr<Semaphore> _semaphore;
std::thread _thread;
const std::shared_ptr<Common::ThreadJoiner> _joiner;
};
} // namespace SimpleZomby
#include <sstream>
#include "SimpleZomby.h"
#include "Common/Listener.h"
#include "Common/ThreadJoiner.h"
namespace SimpleZomby {
Zomby::Zomby(std::shared_ptr<Common::ThreadJoiner> joiner)
: _joiner(joiner)
{
if (!_joiner) {
throw std::runtime_error("An empty joiner in SimpleZomby::Zomby::Zomby");
}
}
Zomby::~Zomby()
{
if (_semaphore) {
if (std::this_thread::get_id() != _thread.get_id()) {
auto guard = std::lock_guard(_semaphore->first);
_semaphore->second = false;
} else {
_semaphore->second = false;
}
}
if (_thread.joinable()) {
_joiner->join(std::move(_thread));
}
if (_listener) {
std::ostringstream buf;
buf << typeid(*this).name() << "::" << __func__ << std::endl;
_listener->processData(std::make_shared<Common::Listener::Data>(buf.str()));
}
}
void Zomby::runOnce(std::shared_ptr<Common::Listener> listener)
{
if (_semaphore) {
throw std::runtime_error("SimpleZomby::Zomby::runOnce() called twice");
}
_listener = listener;
// two separate lines needed because of absence of std::mutex move c-tor
_semaphore = std::make_shared<Semaphore>();
_semaphore->second = true;
_thread = std::thread([listener, semaphore = _semaphore](){
while (listener && semaphore) {
{
auto guard = std::lock_guard(semaphore->first);
if (!semaphore->second) {
break;
}
listener->processData(std::make_shared<Common::Listener::Data>("SimpleZomby is alive!n"));
}
std::this_thread::sleep_for(std::chrono::seconds(1));
}
});
}
} // namespace SimpleZomby
#include <chrono>
#include <thread>
#include <sstream>
#include "Common/Impl/WriteToConsoleListener.h"
#include "Common/Impl/ThreadJoinerAsync.h"
#include "SimpleZomby/SimpleZomby.h"
int main()
{
auto writeToConsoleListener = Common::WriteToConsoleListener::instance();
auto joiner = std::make_shared<Common::ThreadJoinerAsync>();
{
auto simpleZomby = SimpleZomby::Zomby(joiner);
simpleZomby.runOnce(writeToConsoleListener);
std::this_thread::sleep_for(std::chrono::milliseconds(4500));
} // Zomby should be killed here
{
std::ostringstream buf;
buf << "============================================================n"
<< "| Zomby was killed |n"
<< "============================================================n";
if (writeToConsoleListener) {
writeToConsoleListener->processData(std::make_shared<Common::Listener::Data>(buf.str()));
}
}
std::this_thread::sleep_for(std::chrono::milliseconds(5000));
return 0;
}
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
SimpleZomby is alive!
N11SimpleZomby5ZombyE::~Zomby
============================================================
| Zomby was killed |
============================================================
N6Common22WriteToConsoleListenerE::~WriteToConsoleListener
Предупреждение: код ThreadJoiner тестировался поверхностно; задумаете втащить его в production — я не виноват.
Преимущества способа:
— гарантированное корректное завершение всех потоков, ответственность за которые была возложена на ThreadJoiner;
— гарантированное обнаружение ошибки в условии выхода из отдельного потока, приводящей к бесконечному выполнению — главный поток зависнет при завершении;
— ThreadJoiner может быть дополнен диагностическими сообщениями, упрощающими выявление менее критичных ошибок в условии выхода из отдельного потока — тех, которые приводят к некоторому конечному излишнему времени выполнения потока.
Модификации SteppingZomby и BoozdedZomby — полностью аналогичны (см. ветку fixes/semaphore_done_right_with_ThreadJoiner).
Пара слов об официальных примерах boost
В первой части статьи приведены два примера из официальной документации boost:
Пример HTTP-клиента
Пример Websocket-клиента
Вызов ioc.run() в main() блокирует управление до наступления одного из событий:
— успешная отработка всей цепочки;
— ошибка.
Досрочное уничтожение вышестоящей бизнес-логикой экземпляра класса session в представленном коде невозоможно. Это само по себе гарантирует достаточное время жизни данных, агрегированных в полях класса session. Судя по всему, представленный код будет успешно работать даже при модификации его this-способом. Использование shared_from_this в данном случае избыточно, достаточно захвата сырого указателя this.
Откуда же тогда взялся shared_from_this?
Вероятно, из более реального кода, в котором были предусмотрены сценарии остановки соединения путём его уничтожения. И конечно, отсутствие контроля собственного времени жизни приводило к неопределённому поведению (читай «к крэшам»). Крэши разменяли на зомби, а зомби подарили копипастерам.
Вывод
Решения, приведённые в статье, специфичны. Они учитывают особенности изначального кода — в том числе способ передачи результатов работы асинхронного процесса.
Однако в результате разбора примеров стало понятно, какие архитектурные решения привели к проблемам:
— нарушение принципа single responsibility — на класс были возложены как функция информирования асинхронно выполняющегося процесса о необходимости остановки, так и функция хранения данных для этого процесса;
— отсюда возникло требование от нижележащей сущности к вышележащей — асинхронный процесс нуждался в гарантии валидности экземпляра класса в течение определённого времени;
— отсюда возникла необходимость контроля собственного времени жизни с помощью техники std::enable_shared_from_this.
Наиболее корректный код удалось получить без применения std::enable_shared_from_this.
Чего и Вам желаю.
Автор: Александр Дубовик
