Hace algún tiempo, nos encontramos con el problema de limpiar las tuplas en los espacios. . La limpieza debía iniciarse no cuando tarantool se estaba quedando sin memoria, sino de antemano y con una periodicidad determinada. Para esta tarea, tarantool cuenta con un módulo, escrito en Lua, llamado . Después de un breve uso de este módulo, nos dimos cuenta de que no era adecuado para nosotros: durante limpiezas constantes de grandes volúmenes de datos, Lua se estancaba en el GC. Por lo tanto, pensamos en desarrollar nuestro propio módulo capped expirationd, esperando que el código, escrito en un lenguaje de programación nativo, resolvería nuestras necesidades de la mejor manera.
Un buen ejemplo para nosotros fue el módulo tarantool llamado . El enfoque utilizado se basa en que en el espacio se introduce un campo separado, en el que se especifica el tiempo de vida de la tupla, en otras palabras, ttl. El módulo escanea en segundo plano el espacio, compara el ttl con la hora actual y decide si eliminar la tupla o no. El código del módulo memcached es simple y elegante, pero demasiado general. En primer lugar, no toma en cuenta el tipo de índice por el cual se lleva a cabo la iteración y eliminación. En segundo lugar, en cada iteración se escanean todas las tuplas, cuya cantidad puede ser bastante grande. Y si en el módulo expirationd se resolvió el primer problema (el índice tree se выделó en una clase separada), el segundo no recibió ninguna atención. Estos tres puntos definieron la elección de escribir nuestro propio código.
Descripción
En la documentación de tarantool hay un muy buen sobre cómo escribir sus propios procedimientos almacenados en C. En primer lugar, recomiendo familiarizarse con ello para entender las inserciones de comandos y código que aparecerán a continuación. También vale la pena prestar atención a a los objetos que están disponibles al escribir su propio módulo capped, específicamente a , , y .
Comencemos desde el principio y veamos cómo se ve el módulo capped expirationd desde afuera:
fiber = require('fiber')
net_box = require('net.box')
box.cfg{listen = 3300}
box.schema.func.create('libcapped-expirationd.start', {language = 'C'})
box.schema.user.grant('guest', 'execute', 'function', 'libcapped-expirationd.start')
box.schema.func.create('libcapped-expirationd.kill', {language = 'C'})
box.schema.user.grant('guest', 'execute', 'function', 'libcapped-expirationd.kill')
box.schema.space.create('tester')
box.space.tester:create_index('primary', {unique = true, parts = {1, 'unsigned'}})
capped_connection = net_box:new(3300)Para simplificar, ejecutamos tarantool en el directorio donde se encuentra nuestra biblioteca libcapped-expirationd.so. De la biblioteca se exportan dos funciones: start y kill. Primero, debemos hacer que estas funciones sean accesibles desde Lua utilizando box.schema.func.create y box.schema.user.grant. Luego, creamos un espacio cuyos tuplas contendrán solo tres campos: el primero es un identificador único, el segundo un correo electrónico, y el tercero el tiempo de vida de la tupla. Sobre el primer campo construimos un índice tipo tree y lo llamamos primary. A continuación, obtenemos un objeto de conexión a nuestra biblioteca nativa.
Después de los trabajos preparatorios, ejecutamos la función start:
capped_connection:call('libcapped-expirationd.start', {'non-indexed', box.space.tester.id, box.space.tester.index.primary, box.space.tester.index.primary, 3, 1024, 3600})Este ejemplo funcionará al escanear de la misma manera que el módulo expirationd, que está escrito en Lua. El primer argumento de la función start es un nombre único para la tarea. El segundo es el identificador del espacio. El tercero es el índice único con el que se eliminarán las tuplas. El cuarto es el índice por el cual se recorrerán las tuplas. El quinto es el número del campo de la tupla con el tiempo de vida (la numeración comienza desde 1, no desde 0). El sexto y séptimo son las configuraciones de escaneo. 1024 es la cantidad máxima de tuplas que se revisan en una sola transacción. 3600 es el tiempo total de escaneo en segundos.
Tenga en cuenta que para el recorrido y la eliminación en el ejemplo se utiliza el mismo índice. Si se trata de un índice tipo tree, el recorrido se realiza de menor a mayor clave. Si se usa otro tipo, como un índice hash, el recorrido generalmente se realiza en un orden arbitrario. En una sola exploración, se revisan todas las tuplas del espacio.
Vamos a insertar en el espacio varias tuplas con un tiempo de vida de 60 segundos:
box.space.tester:insert{0, 'user0@tarantool.io', math.floor(fiber.time()) + 60}
box.space.tester:insert{1, 'user1@tarantool.io', math.floor(fiber.time()) + 60}
box.space.tester:insert{2, 'user2@tarantool.io', math.floor(fiber.time()) + 60}Verificamos que la inserción se haya realizado con éxito:
tarantool> box.space.tester.index.primary:select()
---
- - [0, 'user0@tarantool.io', 1576418976]
- [1, 'user1@tarantool.io', 1576418976]
- [2, 'user2@tarantool.io', 1576418976]
...Repetimos el select después de 60+ segundos (contando desde el inicio de la inserción de la primera tupla) y veremos que el módulo capped expirationd ya ha hecho su trabajo:
tarantool> box.space.tester.index.primary:select()
---
- []
...Detenemos la tarea:
capped_connection:call('libcapped-expirationd.kill', {'non-indexed'})Veamos un segundo ejemplo, en el que se utiliza un índice separado para el recorrido:
fiber = require('fiber')
net_box = require('net.box')
box.cfg{listen = 3300}
box.schema.func.create('libcapped-expirationd.start', {language = 'C'})
box.schema.user.grant('guest', 'execute', 'function', 'libcapped-expirationd.start')
box.schema.func.create('libcapped-expirationd.kill', {language = 'C'})
box.schema.user.grant('guest', 'execute', 'function', 'libcapped-expirationd.kill')
box.schema.space.create('tester')
box.space.tester:create_index('primary', {unique = true, parts = {1, 'unsigned'}})
box.space.tester:create_index('exp', {unique = false, parts = {3, 'unsigned'}})
capped_connection = net_box:new(3300)Aquí todo es lo mismo que en el primer ejemplo, con la pequeña excepción de que construimos un índice de árbol sobre el tercer campo y lo llamamos exp. Este índice no necesita ser único, a diferencia del índice llamado primary. El recorrido se realizará utilizando el índice exp, mientras que la eliminación será por el primary. Recordamos que antes, ambas operaciones se realizaban solo con el índice primary.
Después de los trabajos preparatorios, iniciamos la función start con nuevos argumentos:
capped_connection:call('libcapped-expirationd.start', {'indexed', box.space.tester.id, box.space.tester.index.primary, box.space.tester.index.exp, 3, 1024, 3600})Nuevamente, haremos inserciones en el espacio con varios tuplas con un tiempo de vida de 60 segundos:
box.space.tester:insert{0, 'user0@tarantool.io', math.floor(fiber.time()) + 60}
box.space.tester:insert{1, 'user1@tarantool.io', math.floor(fiber.time()) + 60}
box.space.tester:insert{2, 'user2@tarantool.io', math.floor(fiber.time()) + 60}Después de 30 segundos, añadiremos por analogía algunas tuplas más:
box.space.tester:insert{3, 'user3@tarantool.io', math.floor(fiber.time()) + 60}
box.space.tester:insert{4, 'user4@tarantool.io', math.floor(fiber.time()) + 60}
box.space.tester:insert{5, 'user5@tarantool.io', math.floor(fiber.time()) + 60}Verificamos que la inserción se haya realizado con éxito:
tarantool> box.space.tester.index.primary:select()
---
- - [0, 'user0@tarantool.io', 1576421257]
- [1, 'user1@tarantool.io', 1576421257]
- [2, 'user2@tarantool.io', 1576421257]
- [3, 'user3@tarantool.io', 1576421287]
- [4, 'user4@tarantool.io', 1576421287]
- [5, 'user5@tarantool.io', 1576421287]
...Repetimos el select después de 60+ segundos (contando desde el inicio de la inserción de la primera tupla) y veremos que el módulo capped expirationd ya ha hecho su trabajo:
tarantool> box.space.tester.index.primary:select()
---
- - [3, 'user3@tarantool.io', 1576421287]
- [4, 'user4@tarantool.io', 1576421287]
- [5, 'user5@tarantool.io', 1576421287]
...Quedan tuplas en el espacio que tienen aproximadamente 30 segundos de vida. Más aún, el escaneo se detuvo al pasar de la tupla con el identificador 2 y tiempo de vida 1576421257 a la tupla con el identificador 3 y tiempo de vida 1576421287. Las tuplas con tiempo de vida 1576421287 o más no fueron visualizadas debido al ordenamiento de las claves del índice exp. Esta es la economía que queríamos lograr desde el principio.
Detenemos la tarea:
capped_connection:call('libcapped-expirationd.kill', {'indexed'})Implementación
Siempre será mejor que el propio código del proyecto explique todas las características del mismo. ! En esta publicación nos detendremos sólo en los aspectos más importantes, específicamente, en los algoritmos de recorrido del espacio.
Los argumentos que pasamos al método start se guardan en una estructura llamada expirationd_task:
struct expirationd_task
{
char name[256];
uint32_t space_id;
uint32_t rm_index_id;
uint32_t it_index_id;
uint32_t it_index_type;
uint32_t field_no;
uint32_t scan_size;
uint32_t scan_time;
};El atributo name es el nombre de la tarea. El atributo space_id es el identificador del espacio. El atributo rm_index_id es el identificador del índice único que se utilizará para eliminar tuplas. El atributo it_index_id es el identificador del índice que se utilizará para recorrer tuplas. El atributo it_index_type es el tipo de índice que se utilizará para recorrer tuplas. El atributo field_no es el número del campo de la tupla que contiene el tiempo de vida. El atributo scan_size es el número máximo de tuplas que se revisan en una transacción. El atributo scan_time es el tiempo de escaneo completo en segundos.
No consideraremos el análisis de argumentos. Es un trabajo meticuloso pero no complicado, en el que te ayudará la biblioteca . Las dificultades pueden surgir solo con los índices que se pasan desde Lua en forma de una estructura de datos compleja con tipo mp_map, y no con tipos simples como mp_bool, mp_double, mp_int, mp_uint y mp_array. Pero no es necesario analizar todo el índice. Es suficiente con verificar su unicidad, calcular el tipo y extraer el identificador.
Enumeraremos los prototipos de todas las funciones que se utilizan para el análisis:
bool expirationd_parse_name(struct expirationd_task *task, const char **pos);
bool expirationd_parse_space_id(struct expirationd_task *task, const char **pos);
bool expirationd_parse_rm_index_id(struct expirationd_task *task, const char **pos);
bool expirationd_parse_rm_index_unique(struct expirationd_task *task, const char **pos);
bool expirationd_parse_rm_index(struct expirationd_task *task, const char **pos);
bool expirationd_parse_it_index_id(struct expirationd_task *task, const char **pos);
bool expirationd_parse_it_index_type(struct expirationd_task *task, const char **pos);
bool expirationd_parse_it_index(struct expirationd_task *task, const char **pos);
bool expirationd_parse_field_no(struct expirationd_task *task, const char **pos);
bool expirationd_parse_scan_size(struct expirationd_task *task, const char **pos);
bool expirationd_parse_scan_time(struct expirationd_task *task, const char **pos);Ahora pasemos a lo más importante: la lógica de escaneo y eliminación de tuplas. Cada bloque de tuplas, de tamaño no mayor que scan_size, se revisa y se modifica en una sola transacción. Si tiene éxito, esta transacción se confirma; en caso de error, se revierte. El último argumento de la función expirationd_iterate es un puntero al iterador desde el cual comienza o continúa el escaneo. Este iterador se incrementa hasta que ocurra un error, se termine el espacio o se presente la oportunidad de detener el proceso anticipadamente. La función expirationd_expired verifica el tiempo de vida de la tupla, expirationd_delete elimina la tupla, y expirationd_breakable verifica si debemos continuar.
Código de la función expirationd_iterate:
static bool
expirationd_iterate(struct expirationd_task *task, box_iterator_t **iterp)
{
box_iterator_t *iter = *iterp;
box_txn_begin();
for (uint32_t i = 0; i scan_size; ++i) {
box_tuple_t *tuple = NULL;
if (box_iterator_next(iter, &tuple) < 0) {
box_iterator_free(iter);
*iterp = NULL;
box_txn_rollback();
return false;
}
if (!tuple) {
box_iterator_free(iter);
*iterp = NULL;
box_txn_commit();
return true;
}
if (expirationd_expired(task, tuple))
expirationd_delete(task, tuple);
else if (expirationd_breakable(task))
break;
}
box_txn_commit();
return true;
}Código de la función expirationd_expired:
static bool
expirationd_expired(struct expirationd_task *task, box_tuple_t *tuple)
{
const char *buf = box_tuple_field(tuple, task->field_no - 1);
if (!buf || mp_typeof(*buf) != MP_UINT)
return false;
uint64_t val = mp_decode_uint(&buf);
if (val > fiber_time64() / 1000000)
return false;
return true;
}Código de la función expirationd_delete:
static void
expirationd_delete(struct expirationd_task *task, box_tuple_t *tuple)
{
uint32_t len;
const char *str = box_tuple_extract_key(tuple, task->space_id, task->rm_index_id, &len);
box_delete(task->space_id, task->rm_index_id, str, str + len, NULL);
}Código de la función expirationd_breakable:
static bool
expirationd_breakable(struct expirationd_task *task)
{
return task->it_index_id != task->rm_index_id && task->it_index_type == ITER_GT;
}Aplicación
Puede consultar el código fuente en !
Fuente: habr.com
