Nějakou dobu zpět jsme čelili problému s údržbou položek v prostorech . Údržbu bylo potřeba spouštět nikoli až tehdy, když měl tarantool již nedostatek paměti, ale předem a s určitou periodicitou. Pro tento úkol existuje v tarantoolu modul napsaný v Lua, nazvaný . Po krátkém používání tohoto modulu jsme zjistili, že nám nevyhovuje: při pravidelné údržbě velkých objemů dat Lua zaviselo v GC. Proto nás napadla myšlenka vyvinout náš vlastní modul capped expirationd, v naději, že kód napsaný v nativním programovacím jazyce vyřeší naše úkoly nejlepším způsobem.
Dobrým příkladem nám posloužil modul tarantool nazvaný . Používaný přístup v něm spočívá v tom, že se v prostoru vytvoří zvláštní pole, ve kterém je uvedena doba života položky, jinými slovy, ttl. Modul na pozadí skenuje prostor, porovnává ttl s aktuálním časem a rozhoduje, zda položku smazat nebo ne. Kód modulu memcached je jednoduchý a elegantní, ale příliš obecný. Za prvé, nezohledňuje typ indexu, podle kterého se provádí procházení a mazání. Za druhé, při každém průchodu se skenují všechny položky, jejichž počet může být poměrně velký. A pokud v modulu expirationd první problém byl vyřešen (tree-index byl vyčleněn do samostatné třídy), druhý stále nezískal žádnou pozornost. Tyto tři body určili volbu ve prospěch napsání vlastního kódu.
Popis
V dokumentaci k tarantoolu je velmi dobrý o tom, jak psát své uložené procedury v jazyce C. V první řadě doporučuji se s ním seznámit, abyste rozuměli těm vložením se příkazy a kódem, které se níže objeví. Také si dejte pozor na k objektům, které jsou dostupné při psaní vlastního capped modulu, a to konkrétně na , , a .
Pojďme začít z dálky a podívat se, jak vypadá capped expirationd modul z vnějšku:
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)Pro zjednodušení spouštíme tarantool v adresáři, ve kterém se nachází naše knihovna libcapped-expirationd.so. Z knihovny jsou exportovány dvě funkce: start a kill. Nejprve je nutné zpřístupnit tyto funkce z Lua pomocí box.schema.func.create a box.schema.user.grant. Poté vytvoříme space, jehož tuple budou obsahovat pouze tři pole: první — unikátní identifikátor, druhé — e-mail, třetí — časový limit tuple. Na první pole vytvoříme tree-index a nazveme ho primary. Dále získáme objekt připojení k naší nativní knihovně.
Po přípravných pracích spustíme funkci 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})Tento příklad bude fungovat při skenování stejně jako modul expirationd napsaný v Lua. Prvním argumentem předáváme funkci start unikátní název úkolu. Druhým — identifikátor space. Třetím — unikátní index, podle kterého se budou odstraňovat tuple. Čtvrtým — index, podle kterého se bude procházet tuple. Pátým — číslo pole tuple s časovým limitem (číslení začíná od 1, nikoliv od 0!). Šestým a sedmým — nastavujeme parametry skenování. 1024 — to je maximální počet tuple, které se prohlížejí v rámci jedné transakce. 3600 — doba úplného skenování v sekundách.
Všimněte si, že pro procházení a odstraňování se v příkladu používá stejný index. Pokud je to tree-index, pak se prochází od menšího klíče k většímu. Pokud je to nějaký jiný, například hash-index, pak se prochází obvykle v libovolném pořadí. Při jednom skenování se prohlíží všechny tuple space.
Pojďme vložit do space několik tuple s časovým limitem 60 sekund:
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}Zkontrolujeme, že vložení proběhlo úspěšně:
tarantool> box.space.tester.index.primary:select()
---
- - [0, 'user0@tarantool.io', 1576418976]
- [1, 'user1@tarantool.io', 1576418976]
- [2, 'user2@tarantool.io', 1576418976]
...Opakujeme výběr po 60+ sekundách (počítáme od začátku vložení prvního tuple) a uvidíme, že modul capped expirationd již zafungoval:
tarantool> box.space.tester.index.primary:select()
---
- []
...Zastavíme úkol:
capped_connection:call('libcapped-expirationd.kill', {'non-indexed'})Podívejme se na druhý příklad, kdy je pro procházení použit samostatný index:
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)Zde je všechno stejné jako v prvním příkladu, s malým rozdílem. Na třetím poli vytváříme tree-index a nazýváme ho exp. Tento index nemusí být unikátní, na rozdíl od indexu nazvaného primary. Procházení bude probíhat pomocí exp indexu, zatímco mazání pomocí primary. Pamatujeme si, že dříve bylo obojí provedeno pouze s použitím primary indexu.
Po přípravných pracích spouštíme funkci start s novými argumenty:
capped_connection:call('libcapped-expirationd.start', {'indexed', box.space.tester.id, box.space.tester.index.primary, box.space.tester.index.exp, 3, 1024, 3600})Znovu provedeme vložení několika kódu s životností 60 sekund:
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}Po 30 sekundách přidáme analogicky další kódy:
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}Zkontrolujeme, že vložení proběhlo úspěšně:
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]
...Opakujeme výběr po 60+ sekundách (počítáme od začátku vložení prvního tuple) a uvidíme, že modul capped expirationd již zafungoval:
tarantool> box.space.tester.index.primary:select()
---
- - [3, 'user3@tarantool.io', 1576421287]
- [4, 'user4@tarantool.io', 1576421287]
- [5, 'user5@tarantool.io', 1576421287]
...Ve space zůstaly kódy, které mají ještě přibližně 30 sekund životnosti. Navíc skenování se zastavilo při přechodu od kódu s identifikátorem 2 a životností 1576421257 ke kódu s identifikátorem 3 a životností 1576421287. Kódy s životností 1576421287 a více nebyly prohlíženy díky seřazení klíčů indexu exp. To je ta úspora, které jsme chtěli na začátku dosáhnout.
Zastavíme úkol:
capped_connection:call('libcapped-expirationd.kill', {'indexed'})Implementace
O všech vlastnostech projektu nejlépe vypráví jeho zdrojový ! V rámci publikace se zastavíme pouze u nejdůležitějších momentů, a to u algoritmů procházení space.
Argumenty, které předáváme metodě start, se ukládají do struktury nazvané 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;
};Atribut name je jméno úkolu. Atribut space_id je identifikátor prostoru. Atribut rm_index_id je identifikátor jedinečného indexu, podle kterého bude probíhat odstranění křivek. Atribut it_index_id je identifikátor indexu, podle kterého se budou procházet křivky. Atribut it_index_type je typ indexu, podle kterého se budou procházet křivky. Atribut filed_no je číslo pole křivky s časovým limitem. Atribut scan_size je maximální počet křivek, které se prohlédnou v rámci jedné transakce. Atribut scan_time je doba úplného skenování v sekundách.
Oparsing argumentů se nebudeme bavit. Je to pracná, ale jednoduchá práce, při které vám pomůže knihovna . Problémy mohou nastat pouze s indexy, které se předávají z Lua ve formě komplexní datové struktury typu mp_map, a ne pomocí jednoduchých typů mp_bool, mp_double, mp_int, mp_uint a mp_array. Ale není nutné parsovat celý index. Stačí jen zkontrolovat jeho jedinečnost, vypočítat typ a extrahovat identifikátor.
Uveďme prototypy všech funkcí, které se používají pro 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);A teď se dostáváme k nejdůležitějšímu — logice procházení prostoru a odstraňování n-tic. Každý blok n-tic s velikostí maximálně scan_size se prohlíží a mění v rámci jedné transakce. V případě úspěchu se tato transakce uloží, v případě chyby se vrátí zpět. Posledním argumentem funkce expirationd_iterate je ukazatel na iterátor, od kterého začne nebo se bude pokračovat ve skenování. Tento iterátor se inkrementuje uvnitř, dokud nedojde k chybě, nebo dokud neskončí prostor, nebo nebude možnost proces předčasně zastavit. Funkce expirationd_expired kontroluje životnost n-tice, expirationd_delete — odstraňuje n-tici, expirationd_breakable — zkontroluje, jestli máme pokračovat dál.
Kód funkce 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;
}Kód funkce 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;
}Kód funkce 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);
}Kód funkce 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;
}Aplikace
Zdrojový kód je k dispozici na !
Zdroj: habr.com
