Implementación de suscriptores en C++: estamos bailando lejos de la estufa, pero ya estamos muy lejos / Sudo Null IT News
En el artículo anterior, mostramos un notificador completamente razonable para enviar notificaciones a los suscriptores. Es bastante conveniente, no hay nada superfluo: solo 130 líneas de código. Se han pensado los puntos que son importantes para los clientes de este código. Como, por ejemplo, una llamada multiproceso para entregar notificaciones con interbloqueos mínimos y la capacidad de cancelar la suscripción de un suscriptor directamente desde su controlador.
Podríamos detenernos ahí, pero iremos un paso más allá, agregando un poco de magia de plantilla y haciendo que el código sea más “académico”.
Problemas
Si miras la sirena con ojo despejado, notarás que se trata de una plantilla. Incluso el Capitán Obvio se quita el sombrero aquí. Sin embargo, déjame explicarte lo que quiero decir. Las plantillas no son el código final que da como resultado su programa, sino algo a partir de lo cual se generará el código final cuando se sustituyan los argumentos de la plantilla. En inglés: creación de instancias de plantilla. Si tiene una plantilla de código enorme, en la que prácticamente nada depende del parámetro de la plantilla, al compilador no le importa: cuántas referencias a la plantilla con diferentes argumentos existan en el programa, se generarán tantas variantes de código, incluso si es casi igual de variante a variante. En consecuencia, utilizar en un programa de tipo notifier<int> – esta es una copia del código de sirena. notifier<char> – uno más. Completamente, sin reutilización alguna. notifier<int, char> – uno más. Etc.
A decir verdad, no creo que esas 130 líneas conduzcan jamás a un aumento significativo de la sección de código de su programa. Pero en la práctica, hay casos en los que es hora de prestar atención a esto. En el artículo anterior, cuando hablé de la velocidad de búsqueda de un identificador de subproceso, mencioné la localidad del caché de la CPU, es decir, por supuesto, el caché de datos. Pero la CPU también tiene una caché de instrucciones. En consecuencia, hay menos “código activo” que realiza la mayor parte del trabajo útil en el programa; existe una mayor probabilidad de que termine en el caché de instrucciones y se ejecute más rápido. Intentemos resaltar secciones del código de alarma que no cambian debido a cambios en el tipo de argumentos.
Análisis de dependencia de argumentos.
Los argumentos del evento definen el prototipo del funtor que se almacenará en la estructura de suscripción. Y esta estructura se almacena en el contenedor std::list. Y luego, el resto del código trabajando con este contenedor. Todo está conectado. ¿Es posible separar el tipo específico de funtor del método de almacenamiento en la suscripción? Sí, este enfoque se llama borrado de tipo. Podemos intentar almacenar el functor en alguna forma muy general, por ejemplo, como un void* en una suscripción. Al código que no funciona directamente con el funtor no le importará en absoluto lo que esté almacenado allí. Y una copia de este código será suficiente.
Desarrollando esta idea, podemos seleccionar una clase base, digamos, notifier_base (no una plantilla), que contendrá el contenedor std::list y todas las operaciones con él que no afecten directamente al funtor. Y todo lo que dependa de argumentos estará en la clase descendiente de la plantilla. De hecho, recibiremos una copia del código para notifier_base y una cierta cantidad para las clases descendientes (una por combinación de argumentos para el prototipo de entrega de eventos).
Almacenamiento de funtores
Aquí tendremos que rendirnos. std::function<...>porque contiene información inseparable sobre los tipos de argumentos que intentamos descartar. ¡Horneemos nuestros propios pasteles!
En C++ hay muchas entidades invocables. Sin entrar en detalles sobre su taxonometría, sólo señalamos que un invocable es cualquier objeto que puede participar en la expresión obj(args...). Los punteros a funciones nos llegaron del lenguaje 'c', pero en el mundo de C++, cualquier clase que tenga un operador de llamada sobrecargado puede volverse invocable. operator()(...). Esencialmente, una lambda es azúcar sintáctica además de una clase generada automáticamente con un operador de llamada.
¿Cómo guardaría este invocable? Sigamos adelante y creemos una copia en el montón y coloquemos el puntero en la suscripción, digamos, en forma de nulo*. Bueno, no olvides eliminar usando eliminar al final.
¿Qué pasa si este invocable es tan pequeño que ocupa el espacio requerido por uno o más punteros void*? Después de todo, podría ser un buen puntero de función. ¿Por qué entonces trabajar con el montón, introduciendo un nivel adicional de direccionamiento indirecto? Puede construir una copia directamente en el lugar, en lugar de un puntero void*. Creemos una unión para almacenar uno u otro. Sí, es importante no olvidar que a la hora de elegir cómo almacenar un callable, no sólo importa su tamaño, sino también el requisito de alineación. Si es más estricto que lo que se requiere para almacenar un puntero ordinario, por simplicidad nos negaremos a almacenarlo localmente y también haremos una copia en el montón.
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;
...
};
Y definamos una metafunción que le dirá si debe almacenar un objeto invocable “en el lugar” o en el montón:
template <class T>
struct is_inplace {
static constexpr bool value = sizeof(T) <= max_inplace_cb_size
&& alignof(T) <= alignof(void*);
};
¿Cómo eliminar correctamente un funtor? Para hacer esto, necesitamos saber cómo lo colocamos; parece que dicho código debe colocarse en una clase derivada de plantilla. O puede hacerlo de forma más sencilla: generar una función auxiliar para la eliminación correcta y guardar su puntero allí en la suscripción:
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>
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>
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();
}
...
};
Ahora tenemos dos especializaciones repetitivas de operaciones que funcionan con cb_storage: la superior para colocar un invocable en el montón y almacenar solo el puntero resultante, y la inferior para colocar un invocable directamente dentro de cb_storage. Como resultado, copiar el functor y crear una función auxiliar para eliminarlo correctamente se puede hacer con un código simple en la clase descendiente:
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;
}
Aquí, primero se determina la constante is_inp: si almacenaremos el funtor dentro del almacenamiento (verdadero) o en el montón (falso). A continuación, se selecciona la implementación adecuada: ops_t. Y al final, se llama a su función de inicialización de almacenamiento en cb_storage. Que genera internamente la implementación requerida del desinicializador dtor y almacena su puntero en cb_storage para que pueda ser llamado en el destructor.
Suscríbete – bebamos
Anteriormente, el código de suscripción era muy conciso:
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++;
}
Simplemente incluía demasiado conocimiento sobre tipos. ¿Qué estaba haciendo? Debe bloquear el mutex, crear un nuevo elemento de suscripción, colocarlo en m_list y devolver el nuevo ID de suscripción. Casi todo, excepto colocar un invocable, se puede hacer en la clase base. Hagamos esto, y no olvidemos que el mutex debe estar bloqueado no solo en el momento de agregar un elemento a m_list, sino hasta que terminemos de modificar la suscripción. De lo contrario, es posible que alguien vea una suscripción incompleta. Y, como sabes, “a un tonto no le muestran la mitad de su trabajo”.
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;
}
};
¡Toda la cancelación de la suscripción se traslada a la clase base! No existe una sola expresión que dependa de los parámetros de la plantilla.
invocable
Aquí tendrás que jugar. Si observa detenidamente la implementación del envío de notificaciones notify(…), puede encontrar solo una línea que depende de los argumentos de entrada; de hecho, la llamada invocable: it->m_callback(args ...). ¡Ojalá pudiera dejar esta línea en la clase derivada y eliminar las otras docenas de líneas en la clase base que no es de plantilla! Pero, ¿cómo se pueden pasar argumentos de llamada parametrizados al método base sin pasarlos, para no hacerlos dependientes de los argumentos? ¡Hagamos una devolución de llamada desde la clase base a la clase derivada, pasando solo cb_storage allí, y dejemos que la derivada se meta con los argumentos!
La clase derivada, en lugar de los argumentos de notificación reales, debe preparar la dirección de algo local que será llamado desde la clase base. Sea esta una instancia de una clase local ubicada directamente en el método notify(…). Pero el código de la clase base no conoce el tipo de esta clase. Puede hacerlo externo, con una llamada virtual, o simplemente crear una función estática dentro de esta clase y pasarle un puntero. Que haya una segunda opción.
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);
}
Los argumentos de alerta que están en el contexto de llamar a un método de una clase derivada ahora no deben pasarse a la clase base, sino a la local: “interna”. Lo más sencillo es empaquetarlos en std::tuple<...> y moverlo a interior.
Ahora volvamos al objeto invocable almacenado en cb_storage. ¿Cómo llamarlo? Para hacer esto, necesita conocer tanto su tipo como los tipos de todos los argumentos. Pero en el propio cb_storage.m_data esta información se perdió (descartó). Necesitamos otra función de traducción intermedia que pueda, por un lado, tomando la unión “cb_storage.m_data” y, por otro, los tipos descartados, llamar correctamente a un invocable. En el momento de la formación de cb_storage, en el método cb_storage_ops::init (arriba), toda la información sobre los tipos está disponible. Allí mismo puede generar el traductor deseado y guardar un puntero en cb_storage. Por supuesto, es imposible guardar el puntero exacto al prototipo completo del traductor que enumera todos los tipos, pero puedes convertirlo a la dirección de la función void.
() y guárdelo así. El estándar le permite realizar un reinterpret_cast para convertir un tipo de función en otro, siempre que la conversión inversa se realice antes de la llamada real.
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));
}
};
Por cierto, dado que arriba empaquetamos todos los argumentos en std::tuple<...>, hasta la llamada en sí no arrastraremos los argumentos individualmente, sino la tupla completa. C++ tiene una función muy conveniente std::apply(…), que le permite llamar a cualquier invocable con cualquier argumento, pasándolos como una tupla; los descomprimirá él mismo.
Genial, ahora cb_storage tiene un traductor de llamadas que se puede usar en la clase interna local haciendo un reinterpret_cast inverso; allí se conocen todos los tipos de argumentos y no necesitamos el tipo 'T' invocable. El traductor, como puedes ver arriba, es una función que toma cb_storage, std::tuple con argumentos… y listo. No hay más tipos en la declaración.
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);
}
Volvamos al código de llamada:
Ahora todas las líneas aquí deberían quedar claras. Transferimos los argumentos a la variable local args_tuple. Creamos una clase local interna y transferimos este args_tuple allí como un campo de clase. Llamamos a la implementación básica del bucle transversal del suscriptor, que llama a nuestra función estática interna:: llamada para cada invocable. En él, teniendo un puntero a una instancia de esta clase y un enlace a cb_storage, sacamos el puntero al traductor, restauramos su tipo y lo llamamos. Es importante que al llamar al traductor ya no movamos (std::move(…)) los argumentos, sino que los copiemos, porque en el caso general habrá más de un suscriptor y, en consecuencia, más de uno. llamada: todos los suscriptores, excepto el primero, no deberían verse afectados por los argumentos.
¿Realmente funciona? asm volatile("nop")Sorprendentemente, ¡sí! No repetiré el código de prueba del artículo anterior, es el mismo. Al igual que el resultado de la aplicación. Durante el proceso de prueba, me interesó ver cuánto código adicional se generó debido a todo tipo de capas y traductores que tuvieron que introducirse. Es inútil mirar la versión DEBUG del código; por supuesto, hay un desorden allí. Pero en el ensamblaje RELEASE quería separar el código de suscriptor y el código de llamada de notificación(…) de toda la cocina interna. Usé barreras en código como
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");
para poder utilizarlos en el desensamblador y ver dónde comienza tal o cual bloque. En C++ se parece a esto:
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
Agregué argumentos sin sentido a la plantilla de notificador para ver cómo los manejaría el compilador. En el desensamblador, la llamada notify(…) se convierte en estos nops:
...
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>
Todo el código de notificación(…) de la clase derivada con empaquetado de parámetros está integrado directamente en el punto de llamada. La llamada de la máquina se realiza al método de la clase base. Por supuesto, hay una gran cantidad de código de máquina involucrado en recorrer la lista y agregar el ID del hilo al std::vector. Pero esto también estaba en la versión anterior del código; sin esto, toda la lógica de la que tanto se ha hablado no funcionará. Pero, ¿cómo se ve el código inmediatamente después de que se libera el bloqueo dentro del bucle transversal del suscriptor?
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
Aquí call llama a un método estático de la clase interna. ¿Qué sigue en el interior::llamarse a sí mismo?
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
Aquí puedes ver la llamada al traductor en la dirección 0x0000555555557993. Mirando hacia dentro encontraremos esos mismos nops envolventes. Parece que en este caso más simple el compilador ha integrado el código del suscriptor directamente en el traductor: Como resultado, las llamadas automáticas, por supuesto, aumentaron un poco. Nada viene gratis. Pero el compilador también hace un buen trabajo aquí, pasando argumentos a los registros e incrustando código de suscriptor siempre que sea posible. El código de alarma ahora ocupa 260 líneas (incluidos los comentarios) en lugar de 130 en la versión anterior. Y requiere
solicita una introducción a C++ a un nivel más serio.
Y, sin embargo, esta no es una biblioteca de metaplantillas pesada de varios miles de líneas, cuya familiarización requiere tanto esfuerzo como escribir su propia alerta.
Código completo del artículo:
