Un po' di tempo fa, ci siamo trovati davanti al problema della pulizia delle tuple negli spazi. . La pulizia doveva essere avviata non quando la memoria di tarantool stava per esaurirsi, ma in modo anticipato e con una certa periodicità. Per questo compito, in tarantool esiste un modulo scritto in Lua, chiamato . Dopo un breve utilizzo di questo modulo, ci siamo resi conto che non era adatto: durante le pulizie costanti di grandi volumi di dati, Lua si bloccava nel GC. Pertanto, abbiamo pensato di sviluppare il nostro modulo capped expirationd, sperando che il codice scritto nel linguaggio di programmazione nativo risolvesse al meglio le nostre esigenze.
Un buon esempio per noi è stato il modulo tarantool chiamato . L'approccio utilizzato si basa sul fatto che in un database si crea un campo separato, in cui viene specificato il tempo di vita della tupla, in altre parole, ttl. Un modulo in background scansisce il database, confronta il ttl con l'orario attuale e decide se eliminare la tupla o meno. Il codice del modulo memcached è semplice ed elegante, ma troppo generico. In primo luogo, non tiene conto del tipo di indice utilizzato per l'iterazione e la rimozione. In secondo luogo, ad ogni passaggio vengono scansionate tutte le tuple, il cui numero può essere piuttosto elevato. E se nel modulo expirationd il primo problema è stato risolto (l'indice ad albero è stato separato in una classe distinta), il secondo continua a non ricevere alcuna attenzione. Questi tre punti hanno determinato la scelta di scrivere il proprio codice.
Descrizione
Nella documentazione di tarantool c'è un'ottima guida su come scrivere le proprie procedure memorizzate in linguaggio C. Prima di tutto, consiglio di familiarizzare con essa, per comprendere le istruzioni e il codice che appariranno qui di seguito. È anche importante prestare attenzione al agli oggetti disponibili durante la scrittura del proprio modulo capped, in particolare a , , e .
Iniziamo con calma e vediamo come appare il modulo capped expirationd all'esterno:
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)Per semplicità, eseguiamo Tarantool nella directory in cui si trova la nostra libreria libcapped-expirationd.so. Due funzioni vengono esportate dalla libreria: start e kill. Innanzitutto, è necessario rendere queste funzioni accessibili da Lua utilizzando box.schema.func.create e box.schema.user.grant. Quindi creeremo uno spazio, i cui tuple conterranno solo tre campi: il primo è un identificatore unico, il secondo è un'email, il terzo è il tempo di vita del tuple. Costruiamo un indice tree sul primo campo e lo chiamiamo primary. Successivamente, otteniamo un oggetto di connessione alla nostra libreria nativa.
Dopo i preparativi, eseguiamo la funzione 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})Questo esempio funzionerà esattamente come il modulo expirationd, scritto in Lua. Il primo argomento della funzione start è il nome univoco del task. Il secondo è l'identificatore dello spazio. Il terzo è l'indice unico, utilizzato per la rimozione delle tuple. Il quarto è l'indice per l'iterazione delle tuple. Il quinto è il numero del campo della tupla relativo alla vita utile (la numerazione inizia da 1, non da 0!). Il sesto e il settimo sono le impostazioni per la scansione. 1024 è il numero massimo di tuple esaminate in una singola transazione. 3600 è il tempo totale di scansione in secondi.
Si noti che per l'iterazione e la rimozione nell'esempio viene utilizzato lo stesso indice. Se si tratta di un indice tree, l'iterazione avviene dal valore della chiave più piccolo a quello più grande. Se è un altro tipo, come un indice hash, l'iterazione avviene di solito in ordine casuale. Durante una scansione, tutte le tuple dello spazio vengono esaminate.
Facciamo un'inserzione nello spazio di alcune tuple con una vita utile di 60 secondi:
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}Verifichiamo che l'inserimento sia andato a buon fine:
tarantool> box.space.tester.index.primary:select()
---
- - [0, 'user0@tarantool.io', 1576418976]
- [1, 'user1@tarantool.io', 1576418976]
- [2, 'user2@tarantool.io', 1576418976]
...Ripetiamo la selezione dopo 60+ secondi (calcolando dall'inizio dell'inserimento del primo tuple) e vedremo che il modulo capped expirationd ha già eseguito:
tarantool> box.space.tester.index.primary:select()
---
- []
...Fermiamo il task:
capped_connection:call('libcapped-expirationd.kill', {'non-indexed'})Esaminiamo un secondo esempio, dove viene utilizzato un indice separato per l'attraversamento:
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)Qui troviamo tutto ciò che si trova nel primo esempio, con una piccola eccezione. Costruiamo un indice ad albero sopra il terzo campo e lo chiamiamo exp. Questo indice non deve essere unico, a differenza dell'indice chiamato primary. La scansione avverrà attraverso l'indice exp, mentre la cancellazione utilizzerà il primary. Ricordiamo che prima tutto ciò è stato fatto utilizzando solo l'indice primary.
Dopo i preparativi, avviamo la funzione start con nuovi argomenti:
capped_connection:call('libcapped-expirationd.start', {'indexed', box.space.tester.id, box.space.tester.index.primary, box.space.tester.index.exp, 3, 1024, 3600})Inseriamo nuovamente nello spazio alcuni tuple con un tempo di vita di 60 secondi:
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}Dopo 30 secondi, aggiungeremo analogamente altri tuple:
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}Verifichiamo che l'inserimento sia andato a buon fine:
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]
...Ripetiamo la selezione dopo 60+ secondi (calcolando dall'inizio dell'inserimento del primo tuple) e vedremo che il modulo capped expirationd ha già eseguito:
tarantool> box.space.tester.index.primary:select()
---
- - [3, 'user3@tarantool.io', 1576421287]
- [4, 'user4@tarantool.io', 1576421287]
- [5, 'user5@tarantool.io', 1576421287]
...Nello spazio sono rimasti dei tuple, che vivranno ancora per circa 30 secondi. Inoltre, la scansione si è bloccata passando dal tuple con identificativo 2 e tempo di vita 1576421257 al tuple con identificativo 3 e tempo di vita 1576421287. I tuple con un tempo di vita di 1576421287 o superiore non sono stati visualizzati a causa dell'ordinamento delle chiavi dell'indice exp. Questa è l'economia che volevamo ottenere fin dall'inizio.
Fermiamo il task:
capped_connection:call('libcapped-expirationd.kill', {'indexed'})Implementazione
Il progetto è sempre meglio descritto dal suo codice sorgente ! In questa pubblicazione ci concentreremo solo sugli aspetti più importanti, ovvero sugli algoritmi di traversamento dello spazio.
Gli argomenti che passiamo al metodo start vengono memorizzati in una struttura chiamata 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;
};L'attributo name è il nome del task. L'attributo space_id è l'identificatore dello spazio. L'attributo rm_index_id è l'identificatore dell'indice unico, secondo cui verranno rimossi le tuple. L'attributo it_index_id è l'identificatore dell'indice, secondo cui verranno attraversate le tuple. L'attributo it_index_type è il tipo di indice, secondo cui verranno attraversate le tuple. L'attributo filed_no è il numero del campo della tupla con la durata. L'attributo scan_size è il numero massimo di tuple che vengono esaminate all'interno di una singola transazione. L'attributo scan_time è il tempo totale di scansione in secondi.
Non considereremo il parsing degli argomenti. È un lavoro meticoloso, ma non difficile, in cui ti aiuterà la libreria . Le difficoltà possono sorgere solo con gli indici che vengono passati da Lua come struttura dati complessa di tipo mp_map, e non con semplici tipi mp_bool, mp_double, mp_int, mp_uint e mp_array. Ma non è necessario effettuare il parsing dell'intero indice. È sufficiente controllarne l'unicità, calcolare il tipo e estrarre l'identificatore.
Elencare i prototipi di tutte le funzioni che vengono utilizzate per il parsing:
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);Ora passiamo al punto principale: la logica di scansione dello spazio e rimozione delle tuple. Ogni blocco di tuple, di dimensione non superiore a scan_size, viene esaminato e modificato all'interno di una singola transazione. In caso di successo, questa transazione viene impegnata, in caso di errore viene annullata. L'ultimo argomento della funzione expirationd_iterate è un puntatore all'iteratore da cui inizia o prosegue la scansione. Questo iteratore viene incrementato finché non si verifica un errore, non si esaurisce lo spazio, o non si presenta l'opportunità di interrompere il processo in anticipo. La funzione expirationd_expired verifica il tempo di vita della tupla, expirationd_delete rimuove la tupla, expirationd_breakable verifica se dobbiamo procedere.
Codice della funzione 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;
}Codice della funzione 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;
}Codice della funzione 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);
}Codice della funzione 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;
}Applicazione
Puoi trovare il codice sorgente su !
Fonte: habr.com
