Qualche tempo fa ci siamo trovati di fronte al problema della pulizia delle tuple negli spazi . La pulizia doveva essere avviata non quando la memoria di tarantool stava già finendo, ma in anticipo e con una certa periodicità. Per questo compito, in tarantool esiste un modulo scritto in Lua, chiamato . Dopo un breve utilizzo di questo modulo, abbiamo capito che non era adatto a noi: nelle 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 in un linguaggio di programmazione nativo risolvesse i nostri problemi nel migliore dei modi.
Un buon esempio è stato il modulo tarantool chiamato . L'approccio utilizzato si basa sul fatto che in uno spazio viene creato un campo separato, in cui viene indicato il tempo di vita della tupla, in altre parole, ttl. Il modulo scansiona in background lo spazio, confronta il ttl con il tempo attuale e decide se rimuovere 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 la scansione 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 a parte), il secondo non ha ricevuto alcuna attenzione. Questi tre punti hanno predeterminato la scelta di scrivere il nostro codice.
Descrizione
Nella documentazione di tarantool c'è un ottimo su come scrivere le proprie procedure memorizzate in linguaggio C. Innanzitutto, consiglio di familiarizzare con esso, per comprendere quelle inserzioni con comandi e codice che si incontreranno più avanti. Vale anche la pena prestare attenzione al agli oggetti disponibili quando si scrive un proprio modulo capped, in particolare a , , e .
Iniziamo da lontano e vediamo come si presenta il modulo capped expirationd dall'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à, avviamo tarantool nella directory in cui si trova la nostra libreria libcapped-expirationd.so. Dalla libreria vengono esportate due funzioni: start e kill. Inizialmente, è necessario rendere queste funzioni accessibili da Lua utilizzando box.schema.func.create e box.schema.user.grant. Successivamente, creiamo uno spazio, le tuple di cui conterranno solo tre campi: il primo è un identificatore unico, il secondo è l'email, il terzo è il tempo di vita della tupla. Sulla prima colonna costruiamo un indice tree e lo chiamiamo primary. Poi otteniamo un oggetto di connessione alla nostra libreria nativa.
Dopo i lavori preparatori, avviamo 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à durante la scansione esattamente come il modulo expirationd, scritto in Lua. Il primo argomento della funzione start è il nome unico del task. Il secondo è l'identificatore dello spazio. Il terzo è l'indice unico tramite il quale avverrà l'eliminazione delle tuple. Il quarto è l'indice tramite il quale avverrà l'attraversamento delle tuple. Il quinto è il numero del campo della tupla con il tempo di vita (la numerazione parte da 1, non da 0!). Il sesto e il settimo sono le impostazioni per la scansione. 1024 è il numero massimo di tuple che vengono visualizzate all'interno di una singola transazione. 3600 è il tempo totale della scansione in secondi.
Si noti che per l'attraversamento e l'eliminazione nel esempio viene utilizzato lo stesso indice. Se si tratta di un indice tree, l'attraversamento avviene da una chiave minore a una maggiore. Se si tratta di un altro tipo, ad esempio, un indice hash, l'attraversamento avviene, in genere, in ordine casuale. In una singola scansione vengono esaminate tutte le tuple dello spazio.
Facciamo un'inserzione nello spazio di alcune 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}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 select dopo 60+ secondi (contando dall'inizio dell'inserimento della prima tupla) e vedremo che il modulo capped expirationd ha già lavorato:
tarantool> box.space.tester.index.primary:select()
---
- []
...Arrestiamo il task:
capped_connection:call('libcapped-expirationd.kill', {'non-indexed'})Esaminiamo il secondo esempio, quando si utilizza un indice separato per la scansione:
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 è tutto come nel primo esempio, con poche eccezioni. Sulla terza colonna costruiamo l'indice tree e lo chiamiamo exp. Questo indice non deve essere unico, a differenza dell'indice chiamato primary. La scansione avverrà utilizzando l'indice exp, mentre l'eliminazione avverrà tramite primary. Ricordiamo che in precedenza entrambi i processi venivano effettuati solo utilizzando l'indice primary.
Dopo i lavori preparatori, 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 di nuovo 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, aggiungiamo, 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 select dopo 60+ secondi (contando dall'inizio dell'inserimento della prima tupla) e vedremo che il modulo capped expirationd ha già lavorato:
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 tuple che hanno ancora circa 30 secondi di vita. Inoltre, la scansione si è fermata passando dal tuple con identificativo 2 e tempo di vita 1576421257 al tuple con identificativo 3 e tempo di vita 1576421287. I tuple con tempo di vita 1576421287 e superiori non sono stati visualizzati a causa dell'ordinamento delle chiavi dell'indice exp. Questa è l'economia che volevamo ottenere sin dall'inizio.
Arrestiamo il task:
capped_connection:call('libcapped-expirationd.kill', {'indexed'})Implementazione
Le peculiarità del progetto vengono sempre meglio illustrate dal suo codice sorgente. ! Nel contesto di questa pubblicazione ci soffermeremo solo sui punti più importanti, in particolare sugli algoritmi di scansione 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, che verrà utilizzato per eliminare le tuple. L'attributo it_index_id è l'identificatore dell'indice, che verrà utilizzato per scorrere le tuple. L'attributo it_index_type è il tipo di indice, che verrà utilizzato per scorrere le tuple. L'attributo field_no è il numero del campo della tupla con il tempo di vita. L'attributo scan_size è il numero massimo di tuple che vengono esaminate in una singola transazione. L'attributo scan_time è il tempo di scansione completa 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 una struttura di dati complessa di tipo mp_map, e non tramite semplici tipi mp_bool, mp_double, mp_int, mp_uint e mp_array. Ma non è necessario analizzare l'intero indice. È sufficiente controllare la sua unicità, calcolare il tipo ed estrarre l'identificatore.
Elencheremo i prototipi di tutte le funzioni 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);E ora passiamo alla parte più importante: la logica di scansione dello spazio e di eliminazione delle tuple. Ogni blocco di tuple di dimensioni non superiori a scan_size viene esaminato e modificato all'interno di una transazione. In caso di successo, questa transazione viene confermata, in caso di errore viene annullata. L'ultimo argomento nella funzione expirationd_iterate è un puntatore all'iteratore, da cui inizia o prosegue la scansione. Questo iteratore viene incrementato Internamente finché non si verifica un errore, lo spazio non finisce o non si presenta l'opportunità di interrompere anticipatamente il processo. La funzione expirationd_expired verifica la durata di vita della tupla, expirationd_delete elimina la tupla, expirationd_breakable controlla se dobbiamo andare oltre.
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 consultare il codice sorgente su !
Fonte: habr.com
