Néhány idővel ezelőtt felmerült a jegyzékek tisztításának problémája a spászokban. . A tisztítást nem akkor kellett elindítani, amikor a tarantool memóriája már kifogyott, hanem előre és meghatározott időközönként. Ehhez a feladathoz a tarantoolban van egy Lua nyelven írt modul, amelynek neve . Rövid ideig tartó használat után rájöttünk, hogy nem felel meg nekünk: a nagy mennyiségű adatok állandó tisztításánál a Lua a GC-n akadt el. Ezért úgy döntöttünk, hogy kifejlesztünk egy saját capped expirationd modult, remélve, hogy a natív programozási nyelven írt kód a lehető legjobban megoldja a feladatainkat.
Jó példa számunkra a tarantoolban található modul, amelynek neve . Az abban használt megközelítés lényege, hogy a spászban létrehozunk egy külön mezőt, amelyben meg van adva a jegyzék élettartama, más szóval, a ttl. A modul a háttérben átvizsgálja a spászt, összehasonlítja a ttl-t a jelenlegi idővel, és döntést hoz arról, hogy törölje-e a jegyzéket vagy sem. A memcached modul kódja egyszerű és elegáns, de túl általános. Először is, nem veszi figyelembe az index típusát, amelyen a bejárás és törlés történik. Másodszor, minden egyes átvizsgáláskor az összes jegyzék bejárásra kerül, amelyek mennyisége elég nagy lehet. Míg az expirationd modulban az első problémát megoldották (a fa-index külön osztályra lett kiemelve), a másodikra továbbra sincs figyelem fordítva. Ezek a három pont döntő szerepet játszottak abban, hogy a saját kódunk írása mellett döntsünk.
Leírás
A tarantool dokumentációjában található egy nagyon jó a saját tárolt eljárások C nyelven történő írásáról. Mindenekelőtt javaslom, hogy ismerkedjen meg vele, hogy megérthesse az alábbiakban előforduló parancsokkal és kódrészletekkel kapcsolatos betéteket. Érdemes figyelni a azokhoz az objektumokhoz, amelyek elérhetőek a saját capped modul írása során, mégpedig a , , és .
Kezdjük távolról, és nézzük meg, hogyan néz ki a capped expirationd modul kívülről:
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)A könnyű indítás érdekében a tarantoolt abban a könyvtárban indítjuk, ahol a libcapped-expirationd.so könyvtárunk található. A könyvtárból két funkciót exportálunk: a start-ot és a kill-t. Először is biztosítani kell, hogy ezek a funkciók elérhetőek legyenek Lua-ból a box.schema.func.create és a box.schema.user.grant segítségével. Ezután létre kell hozni egy space-t, amelynek tuple-jei mindössze három mezőt tartalmaznak: az első egyedi azonosító, a második elektronikus levél, a harmadik a tuple élettartama. Az első mező fölött tree-indexet építünk, és primary-nak nevezzük. Ezt követően megszerezzük a kapcsolat objektumot a natív könyvtárunkhoz.
Az előkészítő munkálatok után indítsuk el a start funkciót:
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})Ez a példa pontosan úgy fog működni, mint ahogy a Lua-ban írt expirationd modul. Az első argumentum, amelyet a start függvénybe adunk, egyedi feladatnév. A második - a space azonosítója. A harmadik - az egyedi index, amely alapján a tuple-k törlése történik. A negyedik - az index, amely alapján a tuple-k bejárása történik. Az ötödik - a tuple élettartammal rendelkező mező száma (a számozás 1-től kezdődik, nem 0-tól!). A hatodik és hetedik - a bejárás beállításai. 1024 - a maximális tuple-k száma, amelyeket egy tranzakció keretében áttekintünk. 3600 - a teljes bejárás ideje másodpercekben.
Vegye figyelembe, hogy a bejárás és törlés érdekében a példában ugyanazt az indexet használjuk. Ha ez egy tree-index, a bejárás a kisebb kulcsról a nagyobbra történik. Ha valamilyen más, például hash-index, akkor a bejárás általában véletlenszerű sorrendben történik. Egy bejárás alatt az összes tuple-t áttekintjük a space-ben.
Most hajtsunk végre néhány tuple beszúrást a space-ba, 60 másodperces élettartammal:
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}Ellenőrizzük, hogy a beszúrás sikeres volt:
tarantool> box.space.tester.index.primary:select()
---
- - [0, 'user0@tarantool.io', 1576418976]
- [1, 'user1@tarantool.io', 1576418976]
- [2, 'user2@tarantool.io', 1576418976]
...Ismételjük meg a select-t 60+ másodperc múlva (az első tuple beszúrásának kezdetétől számítva), és látni fogjuk, hogy a capped expirationd modul már végrehajtotta a feladatát:
tarantool> box.space.tester.index.primary:select()
---
- []
...Állítsuk le a feladatot:
capped_connection:call('libcapped-expirationd.kill', {'non-indexed'})Nézzük meg a második példát, amikor speciális indexet használunk az iterációhoz:
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)Itt minden ugyanaz, mint az első példában, egy apró kivétellel. A harmadik mező fölé tree-indexet építünk, és exp néven hívjuk. Ez az index nem kötelezően egyedi, ellentétben a primary nevű indexszel. Az iteráció az exp indexen keresztül történik, míg a törlés a primary alapján. Emlékszünk, hogy korábban mindkettőt csak a primary index használatával hajtottuk végre.
A előkészítő munkák után elindítjuk a start funkciót új paraméterekkel:
capped_connection:call('libcapped-expirationd.start', {'indexed', box.space.tester.id, box.space.tester.index.primary, box.space.tester.index.exp, 3, 1024, 3600})Ismételten beillesztünk a space-be több tuple-t, amelyek életideje 60 másodperc:
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}30 másodperc múlva hasonlóan további tuple-kat adunk hozzá:
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}Ellenőrizzük, hogy a beszúrás sikeres volt:
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]
...Ismételjük meg a select-t 60+ másodperc múlva (az első tuple beszúrásának kezdetétől számítva), és látni fogjuk, hogy a capped expirationd modul már végrehajtotta a feladatát:
tarantool> box.space.tester.index.primary:select()
---
- - [3, 'user3@tarantool.io', 1576421287]
- [4, 'user4@tarantool.io', 1576421287]
- [5, 'user5@tarantool.io', 1576421287]
...A space-ben még 30 másodpercig élő tuple-k maradtak. Továbbá, a szkennelés megállt 2-es számú tuple-ről, 1576421257-es élettartammal 3-as számú tuple-hez ugrálva, 1576421287-es élettartammal. Az 1576421287-es vagy annál nagyobb életidejű tuple-k nem lettek megnézve, a key exp index sorrendessége miatt. Ez az az megtakarítás, amit kezdetben el akartunk érni.
Állítsuk le a feladatot:
capped_connection:call('libcapped-expirationd.kill', {'indexed'})Megvalósítás
A projekt minden jellemzőjét legjobban az alapforrás fogja bemutatni. ! A publikáció keretében csak a legfontosabb szempontokra fogunk összpontosítani, nevezetesen a space iterációs algoritmusaira.
A start metódusnak átadott argumentumok egy expirationd_task nevű struktúrában kerülnek tárolásra:
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;
};A name attribútum a feladat neve. A space_id attribútum a tér azonosítója. A rm_index_id attribútum a törléshez használt egyedi index azonosítója. Az it_index_id attribútum az a index azonosító, amellyel a tuple-ok bejárásra kerülnek. Az it_index_type attribútum az a index típusa, amellyel a tuple-ok bejárásra kerülnek. A filed_no attribútum a tuple életciklusát reprezentáló mező száma. A scan_size attribútum a max. tuple-ok száma, amelyeket egy tranzakció keretén belül vizsgálunk. A scan_time attribútum a teljes szkennelés ideje másodpercben.
Az argumentumok elemzésével nem foglalkozunk. Ez egy aprólékos, de nem bonyolult munka, amelyhez segíteni fog a . Nehézségek csak az indexekkel adódhatnak, amelyek Lua-ban komplex adattípus alakjában, mp_map típusként kerülnek bemutatásra, nem pedig egyszerű típusok, mint mp_bool, mp_double, mp_int, mp_uint és mp_array segítségével. De az index teljes elemzése nem szükséges. Elég csak az egyediségét ellenőrizni, a típust kiszámítani és az azonosítót kinyerni.
Soroljunk fel minden funkció prototípusát, amelyet az elemzéshez használnak:
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);Mostra, hogy áthelyezzük a helyet és eltávolítjuk a tuple-öket. Minden tuple blokk, amely nem nagyobb, mint a scan_size, egy tranzakció alatt van megvizsgálva és módosítva. Siker esetén ez a tranzakció elköteleződik, hiba esetén pedig visszagördül. Az expirationd_iterate funkció utolsó argumentumaként az iterátor mutatóját adjuk meg, ahonnan a szkennelés kezdődik vagy folytatódik. Ez az iterátor inkrementálódik belül, amíg hiba nem történik, a hely nem fogy el, vagy nem adódik lehetőség a folyamat előzetes leállítására. Az expirationd_expired funkció ellenőrzi a tuple életciklusát, az expirationd_delete - eltávolítja a tuple-t, az expirationd_breakable - ellenőrzi, hogy kell-e tovább mennünk.
Az expirationd_iterate funkció kódja:
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;
}Az expirationd_expired funkció kódja:
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;
}Az expirationd_delete funkció kódja:
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);
}Az expirationd_breakable funkció kódja:
static bool
expirationd_breakable(struct expirationd_task *task)
{
return task->it_index_id != task->rm_index_id && task->it_index_type == ITER_GT;
}Az alkalmazás
Az eredeti kód megtekinthető itt: !
Forrás: habr.com
