For some time now, we have faced the problem of cleaning tuples in spaces. . The cleaning needed to be initiated not when tarantool was already running out of memory, but in advance and at specific intervals. For this task, tarantool has a module written in Lua called . After using this module for a short period, we realized that it wasn't suitable for us: during the constant cleanup of large volumes of data, Lua hung in GC. Therefore, we considered developing our own capped expirationd module, hoping that the code written in a native programming language would address our tasks more effectively.
A good example for us was the tarantool module called . The approach used in it is based on the fact that a separate field is created in the space, specifying the lifetime of the tuple, in other words, ttl. The module scans the space in the background, compares the ttl with the current time, and decides whether to delete the tuple or not. The code of the memcached module is simple and elegant, but too general. Firstly, it does not take into account the type of index used for traversal and deletion. Secondly, on each pass, all tuples are scanned, the number of which can be quite large. While the expirationd module addressed the first issue (the tree index is allocated to a separate class), the second still received no attention. These three points predetermined the choice in favor of writing our own code.
Description
The documentation for tarantool has a very good on how to write your own stored procedures in C. First of all, I suggest familiarizing yourself with it to understand the inserts with commands and code that will be encountered below. It is also worth paying attention to the to objects that are available when writing your own capped module, namely to , , og .
Let's start from afar and see what the capped expirationd module looks like from the outside:
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)For simplicity, we will run tarantool in the directory where our library libcapped-expirationd.so is located. Two functions are exported from the library: start and kill. First, we need to make these functions accessible from Lua using box.schema.func.create and box.schema.user.grant. Then, we create a space whose tuples will contain just three fields: the first is a unique identifier, the second is the email address, and the third is the tuple's lifetime. We build a tree index on the first field and call it primary. Next, we obtain a connection object to our native library.
After the preparatory work, we launch the start function:
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})This example will work exactly like the expirationd module written in Lua when scanning. The first argument passed to the start function is a unique task name. The second is the space identifier. The third is a unique index that will be used for deleting tuples. The fourth is the index that will be used for traversing tuples. The fifth is the tuple field number with the lifetime (numbering starts at 1, not 0!). The sixth and seventh are the scanning settings. 1024 is the maximum number of tuples that are viewed in one transaction. 3600 is the total scanning time in seconds.
Note that the same index is used for traversing and deleting in the example. If it is a tree index, then the traversal occurs from the smaller key to the larger one. If it is some other type, like a hash index, the traversal generally occurs in random order. All tuples in the space are scanned in one operation.
Let's insert several tuples with a lifetime of 60 seconds into the space:
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}Let's check that the insertion was successful:
tarantool> box.space.tester.index.primary:select()
---
- - [0, 'user0@tarantool.io', 1576418976]
- [1, 'user1@tarantool.io', 1576418976]
- [2, 'user2@tarantool.io', 1576418976]
...Let's repeat the select after 60+ seconds (counting from the insertion of the first tuple) and see that the capped expirationd module has already worked:
tarantool> box.space.tester.index.primary:select()
---
- []
...We will stop the task:
capped_connection:call('libcapped-expirationd.kill', {'non-indexed'})La oss se pÄ et annet eksempel der en egen indeks brukes for traversering:
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)Her er alt det samme som i det fĂžrste eksemplet, med unntak av Ă©n liten detalj. Over det tredje feltet bygger vi en tree-indeks og kaller den exp. Denne indeksen trenger ikke vĂŠre unik, i motsetning til indeksen kalt primary. Traverseringen vil skje via exp-indeksen, mens slettingen vil skje via primary. Vi husker at tidligere ble begge deler gjort kun ved hjelp av primary-indeksen.
Etter forberedelsene kjĂžrer vi funksjonen start med nye argumenter:
capped_connection:call('libcapped-expirationd.start', {'indexed', box.space.tester.id, box.space.tester.index.primary, box.space.tester.index.exp, 3, 1024, 3600})Vi vil igjen sette inn flere tuples i spacet med en levetid pÄ 60 sekunder:
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}Etter 30 sekunder legger vi til flere tuples pÄ lignende mÄte:
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}Let's check that the insertion was successful:
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]
...Let's repeat the select after 60+ seconds (counting from the insertion of the first tuple) and see that the capped expirationd module has already worked:
tarantool> box.space.tester.index.primary:select()
---
- - [3, 'user3@tarantool.io', 1576421287]
- [4, 'user4@tarantool.io', 1576421287]
- [5, 'user5@tarantool.io', 1576421287]
...Det er fortsatt tuples i spacet som vil leve i omtrent 30 sekunder til. Videre stoppet skanningen ved overgangen fra tuplen med ID 2 og levetid 1576421257 til tuplen med ID 3 og levetid 1576421287. Tuplene med levetid 1576421287 og hÞyere har ikke blitt sett pÄ grunn av ordningen i nÞklene til exp-indeksen. Dette er den besparelsen vi Þnsket Ä oppnÄ i utgangspunktet.
We will stop the task:
capped_connection:call('libcapped-expirationd.kill', {'indexed'})Implementering
Det beste er alltid Ä fortelle om alle detaljene i prosjektet ved Ä se pÄ koden I denne publikasjonen vil vi kun fokusere pÄ de viktigste punktene, spesielt algoritmene for traversering av spacet.
Argumentene vi sender til metoden start lagres i en struktur kalt 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;
};Attributet name er navnet pÄ oppgaven. Attributet space_id er identifikatoren for rommet. Attributet rm_index_id er identifikatoren for den unike indeksen som tuple-fjerning vil bli utfÞrt pÄ. Attributet it_index_id er identifikatoren for indeksen som tuple-gjennomgang vil bli utfÞrt pÄ. Attributet it_index_type er typen av indeksen som tuple-gjennomgang vil bli utfÞrt pÄ. Attributet field_no er nummeret pÄ tuple-feltet som angir levetid. Attributet scan_size er det maksimale antallet tupler som blir inspisert i lÞpet av én transaksjon. Attributet scan_time er tiden for full skanning i sekunder.
Vi vil ikke ta for oss argumentparsing. Dette er et omstendelig, men ikke vanskelig arbeid, der biblioteket . kan vÊre til hjelp. Vanskeligheter kan oppstÄ med indekser som sendes fra Lua i form av komplekse datatyper av typen mp_map, snarere enn enkle typer som mp_bool, mp_double, mp_int, mp_uint og mp_array. Men det er ikke nÞdvendig Ä parse hele indeksen. Det er tilstrekkelig Ä sjekke dens unikhet, beregne typen og hente identifikatoren.
La oss liste opp prototype for alle funksjoner som brukes til 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);Og nĂ„ gĂ„r vi videre til det viktigste â logikken for Ă„ navigere i plassen og slette tupler. Hver blokk av tupler, som er maks scan_size, blir gjennomgĂ„tt og endret under Ă©n transaksjon. Ved suksess blir denne transaksjonen bekreftet, ved feil blir den rullet tilbake. Den siste argumentet i funksjonen expirationd_iterate er en peker til iteratoren, som angir hvor skanningen starter eller fortsetter. Denne iteratoren inkrementeres inntil det oppstĂ„r en feil, plassen gĂ„r tom, eller det blir gitt mulighet til Ă„ stoppe prosessen pĂ„ forhĂ„nd. Funksjonen expirationd_expired sjekker levetiden til tuplen, expirationd_delete â sletter tuplen, expirationd_breakable â sjekker om vi skal fortsette.
Koden til funksjonen 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;
}Koden til funksjonen 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;
}Koden til funksjonen 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);
}Koden til funksjonen 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;
}Applikasjon
Du kan se kildekoden pÄ !
Kilde: habr.com
Russian
Italian
German
Polish
Estonian
Albanian
Bulgarian
Croatian
Czech
Danish
Dutch
English
French
Hungarian
Latvian
Norwegian
Portuguese
Romanian
Slovak
Slovenian
Spanish
Swedish