Skriver vores capped expirationd modul til tarantool

For nogle tid siden stod vi over for problemet med at rense tuples i spaces. tarantool. Rensningen skulle ikke startes, når tarantool allerede var løbet tør for hukommelse, men i stedet på forhånd og med bestemte intervaller. Til denne opgave findes der i tarantool et modul, skrevet i Lua, kaldet expirationd. Efter kort tids brug af dette modul indså vi, at det ikke var egnet til vores behov: ved udførelse af regelmæssige rensninger af store datamængder, hang Lua i GC. Derfor overvejede vi at udvikle vores eget capped expirationd modul, i håbet om at kode skrevet i et nativesprog ville løse vores opgaver på bedste vis.

Et godt eksempel for os var modulen fra tarantool kaldet memcached. Den anvendte tilgang bygger på, at der i spacet oprettes et separat felt, hvor levetiden for tuplet angives, med andre ord, ttl. Modulet scanner i baggrunden spacet, sammenligner ttl med den aktuelle tid og træffer beslutning om, hvorvidt tuplet skal slettes eller ej. Koden til memcached-modulet er enkel og elegant, men for generelt. For det første tager det ikke højde for typen af indeks, som bruges til gennemgang og sletning. For det andet bliver alle tuples scannet ved hver gennemgang, hvor antallet kan være ret stort. Og mens det i expirationd-modulet blev løst, at det første problem (tree-indekset er udskilt i en separat klasse), blev der stadig ikke givet opmærksomhed til det andet. Disse tre punkter udgjorde grundlaget for beslutningen om at skrive vores egen kode.

Description

I dokumentationen til tarantool findes der en meget god tutorial om hvordan man skriver sine egne lagrede procedurer i sprog C. Jeg anbefaler først at sætte sig ind i det, så man forstår de indstik med kommandoer og kode, der vil blive præsenteret nedenfor. Det er også værd at lægge mærke til referencen til objekterne, der er tilgængelige ved skrivning af sin egen capped modul, nemlig til box, fiber, index og txn.

Lad os starte fra starten og se på, hvordan capped expirationd-modulet ser ud udenfor:

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 launch Tarantool in the directory where our library libcapped-expirationd.so is located. The library exports two functions: 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, the tuples of which will contain only three fields: the first is a unique identifier, the second is an email address, and the third is the lifetime of the tuple. We build a tree index on the first field and name it primary. Next, we obtain the 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 when scanning just like the expirationd module written in Lua. The first argument passed to the start function is a unique task name. The second is the identifier of the space. The third is the unique index by which the tuples will be removed. The fourth is the index for iterating over the tuples. The fifth is the number of the field of the tuple with the lifetime (numbering starts from 1, not 0!). The sixth and seventh are scanning settings. 1024 is the maximum number of tuples that are viewed within one transaction. 3600 is the total scanning time in seconds.

Note that in the example, the same index is used for iteration and deletion. If it is a tree index, iteration is performed from the smaller key to the larger one. If it is some other, for example, a hash index, then iteration usually occurs in arbitrary order. During one scan, all tuples of the space are examined.

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 start of the insertion of the first tuple) and we will 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'})

Lad os tage et kig på det andet eksempel, hvor der bruges et separat indeks til traversal:

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 eksempel, med undtagelse af en lille detalje. Vi opretter et tree-indeks over det tredje felt og kalder det exp. Dette indeks behøver ikke at være unikt, i modsætning til indekset kaldet primary. Traverseringen vil finde sted via exp-indekset, mens sletningen vil ske via primary. Vi husker, at tidligere blev begge dele kun udført med brug af primary-indekset.

Efter forberedelserne starter vi funktionen 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})

Lad os igen indsætte flere tupler i rummet med en livstid 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}

Efter 30 sekunder tilføjer vi analogt endnu et par tupler:

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 start of the insertion of the first tuple) and we will 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]
...

I rummet er der stadig tupler, der har cirka 30 sekunder tilbage at leve. Desuden stoppede scanningen, da den gik fra tuplet med ID 2 og livstid 1576421257 til tuplet med ID 3 og livstid 1576421287. Tupler med livstid 1576421287 og derover blev ikke set på grund af nøglernes orden i exp-indekset. Dette er den besparelse, vi ønskede at opnå fra starten.

We will stop the task:

capped_connection:call('libcapped-expirationd.kill', {'indexed'})

Implementering

Den bedste forklaring af alle projektets funktioner findes altid i sin kilde. koden! I denne publikation vil vi kun fokusere på de vigtigste aspekter, nemlig algoritmerne for traversal i rummet.

De argumenter, vi sender til metoden start, gemmes i en struktur kaldet 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;
};

Attributten name er navnet på opgaven. Attributten space_id er identifikatoren for rummet. Attributten rm_index_id er identifikatoren for den unikke indeks, som tuple- sletning vil blive udført på. Attributten it_index_id er identifikatoren for den indeks, som traversal af tuples vil blive udført på. Attributten it_index_type er typen af indeks, som traversal af tuples vil blive udført på. Attributten field_no er tuple-feltet med levetid. Attributten scan_size er det maksimale antal tuples, der gennemgås under en enkelt transaktion. Attributten scan_time er tiden for en fuld scanning i sekunder.

Vi vil ikke overveje parsing af argumenter. Det er et omhyggeligt, men ikke vanskeligt arbejde, hvor du vil få hjælp af biblioteket msgpuck. Vanskeligheder kan kun opstå med indekserne, som overføres fra Lua i form af en kompleks datstruktur af typen mp_map, og ikke ved hjælp af simple typer som mp_bool, mp_double, mp_int, mp_uint og mp_array. Men det er ikke nødvendigt at parse hele indekset. Det er tilstrækkeligt kun at kontrollere dets unikhed, beregne typen og udtrække identifikatoren.

Lad os opremse prototyperne for alle de funktioner, der bruges 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);

Lad os nu gå videre til det vigtigste — logikken bag scanning af plads og sletning af tuples. Hver blok af tuples, som ikke overstiger scan_size, gennemgås og ændres under én transaktion. Ved succes commit'en denne transaktion, ved fejl bliver den rullet tilbage. Den sidste argument i funktionen expirationd_iterate modtager en peger til iteratoren, hvor scanning enten starter eller fortsætter fra. Denne iterator inkrementeres indeni, indtil der opstår en fejl, pladsen er udtømt, eller der gives mulighed for at stoppe processen på forhånd. Funktionen expirationd_expired tjekker tuple's levetid, expirationd_delete — sletter tuple, expirationd_breakable — tjekker, om vi skal fortsætte videre.

Koden for funktionen 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 for funktionen 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 for funktionen 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 for funktionen 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;
}

Applikation

Du kan finde kildekoden på her!

Kilde: habr.com