Há algum tempo, nos deparamos com o problema de limpeza de tuplas nos espaços. . A limpeza precisava ser iniciada não quando o tarantool já estava sem memória, mas com antecedência e em intervalos específicos. Para essa tarefa, o tarantool possui um módulo escrito em Lua, chamado . Após um curto período de uso deste módulo, percebemos que ele não era adequado para nós: durante as limpezas frequentes de grandes volumes de dados, o Lua travava na coleta de lixo (GC). Portanto, pensamos em desenvolver nosso próprio módulo capped expirationd, esperando que o código, escrito em uma linguagem de programação nativa, resolvesse nossas tarefas da melhor forma.
Um bom exemplo para nós foi o módulo tarantool chamado . O approach utilizado nele baseia-se no fato de que, dentro do espaço, é criado um campo separado que indica o tempo de vida da tupla, ou seja, ttl. O módulo, em segundo plano, faz a varredura do espaço, compara o ttl com o tempo atual e decide se deve excluir a tupla ou não. O código do módulo memcached é simples e elegante, mas muito genérico. Primeiramente, ele não considera o tipo de índice pelo qual a varredura e a exclusão estão sendo realizadas. Em segundo lugar, em cada passagem, todas as tuplas são varridas, cuja quantidade pode ser bastante grande. E enquanto, no módulo expirationd, o primeiro problema foi resolvido (o índice de árvore foi destacado em uma classe separada), o segundo ainda não recebeu atenção. Esses três pontos determinaram a escolha de escrever nosso próprio código.
Descrição
A documentação do tarantool contém um muito bom sobre como escrever suas próprias procedures armazenadas na linguagem C. Primeiramente, sugiro que você se familiarize com isso, para entender as inserções com comandos e códigos que aparecerão abaixo. Também vale a pena prestar atenção ao aos objetos que estão disponíveis ao escrever seu próprio módulo capped, nomeadamente , , e .
Vamos começar do zero e ver como fica o módulo capped expirationd por fora:
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 facilitar, iniciamos o tarantool no diretório onde está a nossa biblioteca libcapped-expirationd.so. Duas funções são exportadas da biblioteca: start e kill. Primeiro, precisamos tornar essas funções acessíveis a partir do Lua usando box.schema.func.create e box.schema.user.grant. Em seguida, criamos um espaço, cujas tuplas conterão apenas três campos: o primeiro é um identificador único, o segundo é o e-mail e o terceiro é o tempo de vida da tupla. Construímos um índice tree sobre o primeiro campo e o chamamos de primary. Depois, obtemos um objeto de conexão com a nossa biblioteca nativa.
Após os trabalhos preparatórios, iniciamos a função 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 exemplo funcionará ao escanear exatamente da mesma forma que o módulo expirationd, que foi escrito em Lua. O primeiro argumento para a função start recebe um nome único para a tarefa. O segundo — o identificador do espaço. O terceiro — um índice único pelo qual a remoção de tuplas será efetuada. O quarto — um índice pelo qual a iteração das tuplas será realizada. O quinto — o número do campo da tupla com o tempo de vida (a numeração começa em 1, não em 0!). O sexto e sétimo — as configurações de escaneamento. 1024 é o número máximo de tuplas que são visualizadas em uma única transação. 3600 é o tempo total de escaneamento em segundos.
Observe que para a iteração e remoção no exemplo, o mesmo índice é utilizado. Se este for um índice tree, a iteração ocorre da chave menor para a maior. Se for algum outro, como um índice hash, a iteração geralmente é feita em ordem arbitrária. Todas as tuplas do espaço são visualizadas em uma única varredura.
Vamos inserir no espaço algumas tuplas com um tempo 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 se a inserção foi bem-sucedida:
tarantool> box.space.tester.index.primary:select()
---
- - [0, 'user0@tarantool.io', 1576418976]
- [1, 'user1@tarantool.io', 1576418976]
- [2, 'user2@tarantool.io', 1576418976]
...Repetimos o select após 60+ segundos (contando a partir do início da inserção da primeira tupla) e veremos que o módulo capped expirationd já executou seu trabalho:
tarantool> box.space.tester.index.primary:select()
---
- []
...Pararemos a tarefa:
capped_connection:call('libcapped-expirationd.kill', {'non-indexed'})Vamos considerar um segundo exemplo, onde para a varredura é utilizado um índice separado:
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)Aqui é tudo igual ao primeiro exemplo, com uma pequena exceção. Sobre o terceiro campo, construímos um índice de árvore e o chamamos de exp. Este índice não precisa ser único, ao contrário do índice denominado primary. A varredura será realizada pelo índice exp, enquanto a exclusão será pelo primary. Lembramos que anteriormente tanto uma quanto outra eram feitas apenas utilizando o índice primary.
Após os trabalhos preparatórios, iniciamos a função start com novos 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})Vamos novamente inserir vários tuplos no espaço com um tempo 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}Em 30 segundos, adicionaremos mais alguns tuplos por analogia:
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 se a inserção foi bem-sucedida:
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 o select após 60+ segundos (contando a partir do início da inserção da primeira tupla) e veremos que o módulo capped expirationd já executou seu trabalho:
tarantool> box.space.tester.index.primary:select()
---
- - [3, 'user3@tarantool.io', 1576421287]
- [4, 'user4@tarantool.io', 1576421287]
- [5, 'user5@tarantool.io', 1576421287]
...No espaço, ainda restam tuplos que têm cerca de 30 segundos de vida. Além disso, a varredura parou ao passar do tuplo com o identificador 2 e tempo de vida 1576421257 para o tuplo com o identificador 3 e tempo de vida 1576421287. Os tuplos com tempo de vida 1576421287 e superiores não foram visualizados devido à ordenação das chaves do índice exp. Essa é a economia que desejávamos alcançar desde o início.
Pararemos a tarefa:
capped_connection:call('libcapped-expirationd.kill', {'indexed'})Implementação
O melhor sobre todas as particularidades do projeto sempre será contado pelo seu código-fonte ! Neste artigo, abordaremos apenas os pontos mais importantes, especificamente, os algoritmos de varredura do espaço.
Os argumentos que passamos ao método start são armazenados em uma estrutura chamada 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;
};O atributo name é o nome da tarefa. O atributo space_id é o identificador do espaço. O atributo rm_index_id é o identificador do índice único pelo qual as tuplas serão removidas. O atributo it_index_id é o identificador do índice pelo qual as tuplas serão percorridas. O atributo it_index_type é o tipo de índice pelo qual as tuplas serão percorridas. O atributo field_no é o número do campo da tupla com tempo de vida. O atributo scan_size é o número máximo de tuplas que são visualizadas durante uma transação. O atributo scan_time é o tempo total de escaneamento em segundos.
Não iremos considerar a análise dos argumentos. Essa é uma tarefa meticulosa, mas não difícil, na qual você será auxiliado pela biblioteca . As dificuldades podem surgir apenas com índices que são passados do Lua como uma estrutura de dados complexa do tipo mp_map, e não com tipos simples como mp_bool, mp_double, mp_int, mp_uint e mp_array. Mas não é necessário analisar todo o índice. Basta verificar sua exclusividade, calcular o tipo e extrair o identificador.
Vamos listar os prototypes de todas as funções que são usadas para análise:
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);Agora vamos ao mais importante — a lógica de varredura do espaço e remoção de tuplas. Cada bloco de tuplas, com um tamanho não superior ao scan_size, é analisado e alterado sob uma única transação. Em caso de sucesso, essa transação é confirmada; em caso de erro, é revertida. O último argumento da função expirationd_iterate é um ponteiro para o iterador, a partir do qual a varredura começa ou continua. Este iterador é incrementado até que ocorra um erro, o espaço se esgote ou haja a possibilidade de interromper o processo antecipadamente. A função expirationd_expired verifica a vida útil da tupla, expirationd_delete — remove a tupla, expirationd_breakable — verifica se precisamos continuar.
Código da função 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 da função 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 da função 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 da função 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;
}Aplicativo
Você pode consultar o código fonte em !
Fonte: habr.com
