В прошлой статье был показан вполне себе разумный оповещатель (notifier) для рассылки уведомлений подписчикам (subscribers). Он достаточно удобен, там нет ничего лишнего — всего 130 строчек кода. Моменты, важные для клиентов этого кода, продуманы. Как, например, многопоточный вызов доставки оповещений с минимальными взаимными блокировками и возможность отписывать подписчика прямо из его обработчика.
На этом бы и остановиться, но мы шагнём дальше, добавив немного шаблонной магии и сделав код «академичнее».
Проблема
Если посмотреть на оповещатель незамутнённым взглядом, то можно заметить, что это шаблон! Здесь даже Капитан Очевидность снимает шляпу. Однако, позвольте пояснять, к чему это я. Шаблоны — это не конечный код, получающийся в вашей программе, а нечто, из чего конечный код будет генерироваться при подстановке аргументов шаблона в него. На английском — template instantiation. Если у вас есть огромный шаблон кода, в котором от шаблонного параметра практически ничего не зависит, компилятору плевать на это — сколько упоминаний шаблона с разными аргументами существует в программе, столько и будет генерироваться вариантов кода, пусть даже он почти не отличается от варианта к варианту. Соответственно, использование в программе типа notifier<int> — это один экземпляр кода оповещателя. notifier<char> — ещё один. Полностью, без какого либо переиспользования. notifier<int, char> — ещё один. И т.п.
Сказать по правде, я не думаю, что эти 130 строк приведут когда-либо к существенному разбуханию секции с кодом в вашей программе. Но на практике бывают случаи, когда на это уже пора обратить внимание. В прошлой статье при обсуждении быстродействия поиска идентификатора потока я упоминал о локальности кэша CPU, имея ввиду, конечно же, кэш данных. Но в CPU есть и кэш инструкций. Соответственно, меньше «горячего кода», выполняющего большинство полезной работы в программе — больше вероятность, что он окажется в кэше инструкций и будет выполняться быстрее. Давайте попробуем выделить участки кода оповещателя, которые не меняются из-за изменения типа аргументов.
Анализ зависимостей от аргументов
Аргументы для события определяют прототип функтора, который будет храниться в структуре подписки — subscription. А эта структура хранится в контейнере std::list. А далее — остальной код, работающий с этим контейнером. Всё связано. А можно ли отделить конкретный вид функтора от способа хранения в subscription? Да, этот подход называется type erasure. Мы можем попробовать сохранить функтор в каком-то самом общем виде, например, в виде void* в subscription. Тому коду, который непосредственно с функтором не работает, вообще будет не важно, что там хранится. И достаточно будет одного экземпляра этого кода.
Развивая эту мысль, мы можем выделить базовый класс, скажем, notifier_base (не шаблонный), в котором будет контейнер std::list и все операции работы с ним, которые не затрагивают непосредственно функтора. А всё аргументо-зависимое будет в шаблонном классе-наследнике. По факту мы получим один экземпляр кода для notifier_base и какое-то количество для классов-наследников — по одному на комбинацию аргументов для прототипа доставки события.
Хранение функтора
Здесь нам придётся отказаться от std::function<...>, потому что она содержит в себе неотделимую информацию о типах аргументов, от которой мы стремимся отказаться. Будем печь пироги самостоятельно!
В c++ есть множество вызываемых сущностей — callable. Не вдаваясь детально в их таксонометрию, отметим для себя только, что callable — это любой объект, который может участвовать в выражении obj(args...). Из языка ‘c’ к нам пришли указатели на функции, но в мире c++ вызываемым может стать любой класс, имеющий перегруженный оператор вызова operator()(...). По сути лямбда — это синтаксический сахар поверх такого автоматически генерируемого класса с оператором вызова.
Как бы сохранить этот callable? Да просто возьмём и создадим копию в куче (heap), а указатель положим в subscription, скажем, в виде void*. Ну и не забудем удалить через delete в конце.
А вдруг этот callable настолько маленький, что сам по себе занимает место, требуемое для одного или нескольких указателей void*? В конце концов это может быть старый добрый указатель на функцию. Зачем тогда вообще работать с кучей, вводить дополнительный уровень косвенности? Можно сконструировать копию прямо по месту (in place) — вместо указателя void*. Заведём union чтобы хранить либо одно, либо другое. Да, важно не забыть, что при выборе, как хранить callable, имеет значение не только его размер, но и требование по выравниванию (alignment). Если он строже того, который требуется для хранения обыкновенного указателя, для простоты откажемся от хранения его по месту и также будем делать копию в куче.
constexpr std::size_t max_inplace_cb_size = sizeof(void*) * 2; struct cb_storage { union s { void* m_ptr; char m_obj[max_inplace_cb_size]; } m_data; ... }; И определим мета-функцию, которая подскажет, использовать ли хранение callable объекта «по-месту» или в куче:
template <class T> struct is_inplace { static constexpr bool value = sizeof(T) <= max_inplace_cb_size && alignof(T) <= alignof(void*); }; Как правильно удалить функтор? Для этого мы должны знать, как мы его расположили — кажется, что подобный код необходимо размещать в шаблонном классе-наследнике. А можно сделать проще — сгенерировать вспомогательную функцию правильного удаления и сохранить её указатель там же в subscription:
struct cb_storage { ... // as above void (*m_dtor)(cb_storage&) = nullptr; ~cb_storage() { if (m_dtor) (*m_dtor)(*this); } }; // operations with callback storage 'cb_storage' template <bool> struct cb_storage_ops { template <class T, class ... Args> static void init(cb_storage& s, T&& t) { // get real callable type from the deduced reference type T typedef std::decay_t<T> T1; s.m_data.m_ptr = new T1(std::forward<T>(t)); s.m_dtor = &dtor<T1>; ... } template <class T1> static void dtor(cb_storage& s) { delete static_cast<T1*>(s.m_data.m_ptr); } ... }; template <> struct cb_storage_ops<true> { template <class T, class ... Args> static void init(cb_storage& s, T&& t) { typedef std::decay_t<T> T1; new (s.m_data.m_obj) T1(std::forward<T>(t)); s.m_dtor = &dtor<T1>; ... } template <class T1> static void dtor(cb_storage& s) { std::launder(reinterpret_cast<T1*>(s.m_data.m_obj)) ->T::~T(); } ... }; Теперь у нас есть две шаблонных специализации операций, работающих с cb_storage — верхняя для размещения callable в куче и сохранения только полученного указателя, а нижняя — для размещения callable прямо внутри cb_storage. По итогу копирование функтора и создание вспомогательной функции для правильного его удаления делаются простым кодом в классе-наследнике:
template <class F> sub_id_t subscribe(F&& cb) { typedef std::decay_t<F> F1; const bool is_inp = details::is_inplace<F1>::value; typedef details::cb_storage_ops<is_inp> ops_t; cb_storage s = ... // see below ops_t::template init<F, Args...>(s.m_cb_storage, std::forward<F>(cb)); return id; } Здесь сначала определяется константа is_inp — будем ли мы хранить функтор внутри хранилища (true) или в куче (false). Далее выбирается соответствующая реализация — ops_t. И в конце вызывается её функция инициализации хранилища в cb_storage. Которая внутри генерирует требуемую реализацию деинициализатора dtor и сохраняет его указатель в cb_storage, чтобы в деструкторе можно было его вызвать.
Подписка — давайте попилим
Раньше код подписки был очень лаконичным:
sub_id_t subscribe(std::function<void(Args ...)> callback) { std::lock_guard l{m_list_mtx}; m_list.emplace_back(std::move(callback), m_next_id); return m_next_id++; } Вот только включал в себя слишком много знаний о типах. Что он делал? Нужно заблокировать mutex, создать новый элемент subscription и положить его в m_list, и вернуть новый идентификатор подписки. Почти всё кроме размещения callable можно сделать в базовом классе. Давайте так и сделаем, и не забудем, что mutex должен быть заблокирован не только в момент добавления элемента в m_list, а пока мы не закончим с модификацией subscription. Иначе кто-то может увидеть недоделанный subscription. А, как известно, «дураку полработы не показывают».
class notifier_base { ... std::tuple<subscription&, std::unique_lock<std::mutex>, sub_id_t> subscribe() { std::unique_lock l{m_list_mtx}; auto r = std::make_tuple( std::ref(m_list.emplace_back(m_next_id)), std::move(l), m_next_id); ++m_next_id; return r; } ... }; template <class ... Args> class notifier : public details::notifier_base { ... template <class F> sub_id_t subscribe(F&& cb) { ... // as above auto [s, l, id] = notifier_base::subscribe(); ops_t::template init<F, Args...>(s.m_cb_storage, std::forward<F>(cb)); return id; } }; Отписка вообще вся переезжает в базовый класс! Там нет ни одного выражения, которое бы зависело от шаблонных параметров.
Вызов callable
Здесь придётся повозиться. Если посмотреть внимательно на реализацию рассылки оповещений notify(…), можно найти всего одну строчку, которая зависит от входных аргументов — собственно, вызов callable: it->m_callback(args ...). Вот бы эту строчку оставить в производном классе, а все остальные дюжины строк убрать в базовый не шаблонный класс! Но как передать параметризованные аргументы вызова в базовый метод, и при этом не передавать их, чтобы не делать зависимым от аргументов? Сделаем обратный вызов из базового класса в производный, передавая туда только cb_storage, и пусть производный мутит с аргументами!
Производный класс вместо реальных аргументов оповещения должен подготовить адрес чего-то локального, что будет вызвано обратно из базового класса. Пусть это будет экземпляр локального класса, расположенного прямо в методе notify(…). Но тип этого класса не известен коду базового класса. Можно либо делать его внешним, с виртуальным вызовом, либо просто сделать внутри этого класса статическую функцию и передать указатель на неё. Пусть будет второй вариант.
template <class ... ArgsI> void notify(ArgsI&& ... args) { ... struct inner { ... static void call(void* ctx, details::cb_storage& s) { auto ths = static_cast<inner*>(ctx); ... } } i; // call base class implementation which will call back inner::call notifier_base::notify(&inner::call, &i); } Аргументы оповещения, которые находятся в контексте вызова метода производного класса, теперь нужно передать не в базовый, а локальный — «inner». Самое простое — запаковать их в std::tuple<…>, а тот переместить в inner.
Теперь вернёмся к объекту callable, сохранённому в cb_storage. Как его вызывать? Для этого нужно знать и его тип, и типы всех аргументов. Но в самом cb_storage.m_data эта информация была утеряна (отброшена). Нам нужна ещё одна промежуточная функция-транслятор, которая сможет, с одной стороны взяв union «cb_storage.m_data», а с другой — отброшенные типы, правильно вызвать callable. На момент формирования cb_storage в методе cb_storage_ops::init (выше), вся информация о типах есть. Прямо там можно сгенерировать нужный транслятор и сохранить указатель на него в cb_storage. Точный указатель на полный прототип транслятора с перечислением всех типов конечно сохранить нельзя, но можно преобразовать его к адресу на функцию void (*)(), и сохранить в таком виде. Стандарт позволяет выполнить reinterpret_cast для преобразования одного типа функции в другой, если потом перед реальным вызовом будет сделано обратное преобразование.
Кстати, раз уж выше мы запаковали все аргументы в std::tuple<…>, то и до самого вызова будем тянуть не аргументы по отдельности, а весь tuple. В c++ есть очень удобная функция std::apply(…), которая позволяет вызвать любой callable с любыми аргументами, передав их в виде tuple — она сама их распакует.
template <bool> struct cb_storage_ops { template <class T, class ... Args> static void init(cb_storage& s, T&& t) { ... // as above s.m_translator = reinterpret_cast<void (*)()>(&translator<T1, Args ...>); } ... // as above template <class T, class ... Args> static void translator(cb_storage& s, std::tuple<Args ...> args) { std::apply(*static_cast<T*>(s.m_data.m_ptr), std::move(args)); } }; Отлично, теперь в cb_storage есть транслятор вызова, который можно использовать в локальном классе inner, сделав обратный reinterpret_cast — там то все типы аргументов известны, а тип callable ‘T’ нам и не нужен. Транслятор, как видно выше, это функция, принимающая cb_storage, std::tuple с аргументами… и всё. Никаких больше типов в декларации нет.
Вернёмся к коду вызова:
template <class ... ArgsI> void notify(ArgsI&& ... args) { auto args_tuple = std::make_tuple(std::forward<ArgsI>(args) ...); struct inner { decltype(args_tuple) m_args; inner(decltype(args_tuple)&& args) : m_args(std::move(args)) {} static void call(void* ctx, details::cb_storage& s) { auto ths = static_cast<inner*>(ctx); auto tr = reinterpret_cast< void(*)(details::cb_storage&, std::tuple<Args...>)>(s.m_translator); (*tr)(s, ths->m_args); } } i{std::move(args_tuple)}; notifier_base::notify(&inner::call, &i); } Теперь все строчки здесь должны быть понятны. Переносим аргументы в локальную переменную args_tuple. Создаём локальный класс inner и переносим эту args_tuple туда в качестве поля класса. Вызываем базовую реализацию цикла обхода подписчиков, которая для каждого callable вызывает нашу статическую функцию inner::call. В ней, имея указатель на экземпляр этого класса и ссылку на cb_storage, достаём указатель на транслятор, восстанавливаем его тип и вызываем. Важно, что, вызывая транслятор, мы уже не переносим (std::move(…)) аргументы, а копируем, потому что в общем случае будет больше одного подписчика и, соответственно, больше одного вызова — все подписчики кроме первого не должны получить огрызок от аргументов.
Оно, правда, работает?
Удивительно, но да! Не буду повторять код теста из прошлой статьи, он такое же. Как и вывод приложения. В процессе тестирования мне было интересно посмотреть, сколько же лишнего кода генерируется из-за всяких там прослоек и трансляторов, которые пришлось ввести. Смотреть DEBUG вариант кода бесполезно — там, разумеется, нагромождение. А вот в RELEASE сборке хотелось отделить код подписчика и код вызова notify(…) от всей внутренней кухни. Я использовал барьеры в коде вида asm volatile("nop"), чтобы по ним в дизассемблере посмотреть, где начинается тот или иной блок. В c++ выглядит примерно так:
notifier<int, char, const test_move_only&> s; auto id1 = s.subscribe([](int, char, auto&){ asm volatile("nop"); g_sync_logger() << "subscriber 1 executed"; asm volatile("nop"); }); asm volatile("nop"); s.notify(1, 'a', test_move_only{}); asm volatile("nop"); Я добавил бессмысленных аргументов в шаблон оповещателя, чтобы посмотреть, как компилятор с ними справится. В дизассемблере вызов notify(…) оборачивается этими nop’ами:
0x00005555555566ee <+254>:nop 0x00005555555566ef <+255>:lea 0x126a(%rip),%r14 # 0x555555557960 <notifier<int, char, test_move_only const&>::notify<int, char, test_move_only>(int&&, char&&, test_move_only&&)::inner::call(void*, details::cb_storage&)> 0x00005555555566f6 <+262>:lea 0x20(%rsp),%rdx 0x00005555555566fb <+267>:mov %rbp,%rdi 0x00005555555566fe <+270>:movabs $0x100000061,%rax 0x0000555555556708 <+280>:mov %r14,%rsi 0x000055555555670b <+283>:mov %rax,0x20(%rsp) 0x0000555555556710 <+288>:call 0x555555558000 <details::notifier_base::notify(void (*)(void*, details::cb_storage&), void*)> 0x0000555555556715 <+293>:nop Весь код notify(…) производного класса с паковкой параметров встроился прямо в точку вызова. Машинный call делается уже к методу базового класса. Там, конечно, много машинного кода, связанного с прохождением списка и добавлением идентификатора потока в std::vector. Но это было и в прошлой версии кода, без этого вся логика, о которой столько говорилось, работать не будет. Но вот как выглядит код сразу после снятия блокировки внутри цикла обхода подписчиков?
... l.unlock(); try { (*ptr)(ctx, it->m_cb_storage); } catch (...) {} l.lock(); ... 0x00005555555580f9 <+249>:call 0x555555556340 <pthread_mutex_unlock@plt> 0x00005555555580fe <+254>:movb $0x0,0x38(%rsp) 0x0000555555558103 <+259>:mov 0x8(%rsp),%rdi 0x0000555555558108 <+264>:lea 0x10(%rbx),%rsi 0x000055555555810c <+268>:call *%r14 0x000055555555810f <+271>:mov %r12,%rdi 0x0000555555558112 <+274>:call 0x555555556430 <pthread_mutex_lock@plt> Здесь call вызывает статический метод класса inner. А что там дальше, в самом inner::call?
0x0000555555557960 <+0>:endbr64 0x0000555555557964 <+4>:sub $0x28,%rsp 0x0000555555557968 <+8>:mov %rsi,%rax 0x000055555555796b <+11>:mov %fs:0x28,%rdx 0x0000555555557974 <+20>:mov %rdx,0x18(%rsp) 0x0000555555557979 <+25>:xor %edx,%edx 0x000055555555797b <+27>:movzbl (%rdi),%edx 0x000055555555797e <+30>:mov %rdi,(%rsp) 0x0000555555557982 <+34>:mov %rsp,%rsi 0x0000555555557985 <+37>:mov %dl,0x8(%rsp) 0x0000555555557989 <+41>:mov 0x4(%rdi),%edx 0x000055555555798c <+44>:mov %rax,%rdi 0x000055555555798f <+47>:mov %edx,0xc(%rsp) 0x0000555555557993 <+51>:call *0x18(%rax) 0x0000555555557996 <+54>:mov 0x18(%rsp),%rax 0x000055555555799b <+59>:sub %fs:0x28,%rax 0x00005555555579a4 <+68>:jne 0x5555555579ab <notifier<int, char, test_move_only const&>::notify<int, char, test_move_only>(int&&, char&&, test_move_only&&)::inner::call(void*, details::cb_storage&)+75> 0x00005555555579a6 <+70>:add $0x28,%rsp 0x00005555555579aa <+74>:ret Здесь виден вызов транслятора по адресу 0x0000555555557993. Посмотрев внутрь мы обнаружим те самые оборачивающие nop’ы. Похоже, в этом простейшем случае компилятор встроил код подписчика прямо в транслятор:
0x0000555555557020 <+0>:endbr64 0x0000555555557024 <+4>:push %rbp 0x0000555555557025 <+5>:push %rbx 0x0000555555557026 <+6>:sub $0x38,%rsp 0x000055555555702a <+10>:mov %fs:0x28,%rax 0x0000555555557033 <+19>:mov %rax,0x28(%rsp) 0x0000555555557038 <+24>:xor %eax,%eax 0x000055555555703a <+26>:nop ... // subscriber implementation here 0x0000555555557079 <+89>:nop 0x000055555555707a <+90>:mov 0x28(%rsp),%rax 0x000055555555707f <+95>:sub %fs:0x28,%rax 0x0000555555557088 <+104>:jne 0x55555555710b <details::cb_storage_ops<true>::translator<main()::<lambda(int, char, auto:3&)>, int, char, const test_move_only&>(details::cb_storage &, std::tuple<int, char, test_move_only const&>)+235> 0x000055555555708e <+110>:add $0x38,%rsp 0x0000555555557092 <+114>:pop %rbx 0x0000555555557093 <+115>:pop %rbp 0x0000555555557094 <+116>:ret По итогу, машинных вызовов, разумеется, стало несколько больше. Ничто не бывает бесплатно. Но компилятор и тут справляется неплохо, перекидывая аргументы в регистры и встраивая код подписчика, где это возможно. Код оповещателя стал занимать теперь 260 строчек (вместе с комментариями) вместо 130 в прошлой версии. И требует просит знакомства с c++ на более серьёзном уровне.
И всё-таки, это не тяжеловесная мета-шаблонная библиотека на несколько тысяч строчек, знакомство с которой требует столько же усилий, как написать свой оповещатель.
«Keep it simple!»
Полный код для статьи: https://github.com/Corosan/subscribers/blob/main/src/notifs-3.cpp
ссылка на оригинал статьи https://habr.com/ru/articles/843530/
Добавить комментарий