🥇Vi skriver vår egen capped expirationd-modul för tarantool

För en tid sedan ställdes vi inför problemet att rensa tuppluster i utrymmen. tarantool. Rensningen behövde inte startas när tarantool redan hade slut på minne, utan i förväg och med viss periodicitet. För detta ändamål finns en modul i tarantool, skriven i Lua, som kallas expirationd. Efter en kort tids användning av denna modul insåg vi att den inte passade oss: vid permanenta rensningar av stora datamängder hängde Lua i GC. Därför började vi tänka på att utveckla vår egen capped expirationd-modul, i förhoppningen att koden, skriven på ett inbyggt programmeringsspråk, skulle lösa våra problem på bästa sätt.

Ett bra exempel för oss var modulen tarantool som heter memcached. Den använda metoden baseras på att en separat fält skapas i utrymmet, där livslängden för tupplusten anges, med andra ord ttl. Modulen skannar i bakgrunden utrymmet, jämför ttl med den aktuella tiden och fattar beslut om att ta bort tupplusten eller inte. Koden för modulen memcached är enkel och elegant, men för generell. För det första tar den inte hänsyn till indexets typ för genomgång och borttagning. För det andra skannas alla tuppluster vid varje genomgång, vars antal kan vara ganska stort. Och i expirationd-modulen löstes det första problemet (tree-indexet har avsatts i en separat klass), men det andra har fortfarande inte uppmärksammats. Dessa tre punkter avgjorde valet av att skriva vår egen kod.

Beskrivning

I dokumentationen till tarantool finns det en mycket bra handledning om hur man skriver sina egna lagrade procedurer på C-språket. Jag rekommenderar först att bekanta sig med den för att förstå de insatser med kommandon och kod som kommer att dyka upp nedan. Det är också värt att notera referensen till objekt som är tillgängliga när man skriver sin egen capped modul, nämligen till box, fiber, index och txn.

Låt oss börja från början och se hur den capped expirationd-modulen ser ut utifrån:

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)

För enkelhetens skull kör vi tarantool i katalogen där vårt bibliotek libcapped-expirationd.so ligger. Från biblioteket exporteras två funktioner: start och kill. Först måste vi göra dessa funktioner tillgängliga från Lua med hjälp av box.schema.func.create och box.schema.user.grant. Sedan skapar vi ett space vars tuples kommer att innehålla endast tre fält: det första — ett unikt identifierare, det andra — en e-postadress, det tredje — livslängden för tuple. Över det första fältet bygger vi ett tree-index och kallar det primary. Vi får även en anslutningsobjekt till vårt inbyggda bibliotek.

Efter förberedelserna kör vi funktionen 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})

Detta exempel kommer att fungera på samma sätt som modulen expirationd, som är skriven i Lua. Det första argumentet till funktionen start överför ett unikt namn för uppgiften. Det andra — identifieraren för spacet. Det tredje — det unika indexet som används för att ta bort tuples. Det fjärde — indexet som används för att gå igenom tuples. Det femte — numret på tuple-fältet med livslängd (numreringen börjar vid 1, inte 0!). Det sjätte och sjunde — inställningar för skanning. 1024 — det maximala antalet tuples som granskas inom en transaktion. 3600 — tiden för en fullständig skanning i sekunder.

Observera att samma index används för både genomgång och borttagning i exemplet. Om det är ett tree-index, sker genomgången från lägre till högre nyckel. Om det är något annat, till exempel ett hash-index, så sker genomgången ofta i slumpmässig ordning. I en skanning granskas alla tuples i spacet.

Låt oss göra en insättning i spacet av flera tuples med livslängd 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}

Låt oss kontrollera att insättningen lyckades:

tarantool> box.space.tester.index.primary:select()
---
- - [0, 'user0@tarantool.io', 1576418976]
  - [1, 'user1@tarantool.io', 1576418976]
  - [2, 'user2@tarantool.io', 1576418976]
...

Vi upprepar select efter 60+ sekunder (räknat från starten av insättningen av den första tuplen) och ser att modulen capped expirationd redan har arbetat:

tarantool> box.space.tester.index.primary:select()
---
  - []
...

Vi stoppar uppgiften:

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

Låt oss titta på det andra exemplet, där en separat index används för genomgång:

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)

Här är allt detsamma som i det första exemplet, med en liten skillnad. Ovanpå det tredje fältet bygger vi en tree-index och kallar den exp. Denna index behöver inte vara unik, till skillnad från indexet som kallas primary. Genomgången kommer att ske via exp-indexet, medan borttagningen sker via primary. Vi minns att tidigare både detta och det andra endast gjordes med hjälp av primary-indexet.

Efter förberedelserna startar vi funktionen start med nya argument:

capped_connection:call('libcapped-expirationd.start', {'indexed', box.space.tester.id, box.space.tester.index.primary, box.space.tester.index.exp, 3, 1024, 3600})

Låt oss återigen göra en insättning i space med flera tuples med en livslängd 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}

Om 30 sekunder lägger vi till ytterligare några tuples på samma sätt:

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}

Låt oss kontrollera att insättningen lyckades:

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]
...

Vi upprepar select efter 60+ sekunder (räknat från starten av insättningen av den första tuplen) och ser att modulen capped expirationd redan har arbetat:

tarantool> box.space.tester.index.primary:select()
---
- - [3, 'user3@tarantool.io', 1576421287]
  - [4, 'user4@tarantool.io', 1576421287]
  - [5, 'user5@tarantool.io', 1576421287]
...

I space finns fortfarande tuples som har en livslängd på cirka 30 sekunder kvar. Dessutom stannade skanningen vid övergången från tuple med identifieraren 2 och livslängden 1576421257 till tuple med identifieraren 3 och livslängden 1576421287. Tuples med livslängden 1576421287 och mer har inte granskats på grund av ordningen på nycklarna i exp-indexet. Detta är just den ekonomin vi ville uppnå från början.

Vi stoppar uppgiften:

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

Implementering

Bästa sättet att berätta om alla särdrag i projektet är alltid dess ursprungliga kod! I den här publikation kommer vi bara att fokusera på de mest viktiga punkterna, nämligen algoritmerna för att genomgå space.

Argumenten som vi skickar till metoden start sparas i en struktur som kallas 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 är namnet på uppgiften. Attributet space_id är identifieraren för utrymmet. Attributet rm_index_id är identifieraren för den unika indexen som används för att ta bort tupler. Attributet it_index_id är identifieraren för indexet som används för att traversera tupler. Attributet it_index_type är typen av index som används för att traversera tupler. Attributet filed_no är numret på fältet av tuplen med livslängd. Attributet scan_size är det maximala antalet tupler som granskas i en transaktion. Attributet scan_time är tiden för en fullständig skanning i sekunder.

Vi kommer inte att diskutera parsning av argument. Det är ett arbetet som kräver noggrannhet, men det är inte svårt, och biblioteket msgpuckkan vara till hjälp. Svårigheter kan endast uppstå med index som skickas från Lua i form av en komplex dataskstruktur av typen mp_map, och inte med hjälp av enkla typer som mp_bool, mp_double, mp_int, mp_uint och mp_array. Men det är inte nödvändigt att parsa hela indexet. Det räcker att kontrollera dess unika identitet, beräkna typen och extrahera identifieraren.

Låt oss lista prototyperna för alla funktioner som används för parsning:

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);

Nu går vi vidare till det viktigaste — logiken för att traversera utrymmet och radera tupler. Varje block av tupler som är högst scan_size granskas och ändras under en och samma transaktion. Vid framgång committas denna transaktion, vid fel rullas den tillbaka. Den sista argumentet i funktionen expirationd_iterate är en pekare till iteratorn där skanningen börjar eller fortsätter. Denna iterator inkrementeras inuti tills ett fel inträffar, utrymmet tar slut, eller det finns möjlighet att avbryta processen i förväg. Funktionen expirationd_expired kontrollerar livslängden för tuplen, expirationd_delete tar bort tuplen, expirationd_breakable kontrollerar om vi behöver gå vidare.

Koden för 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 för 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 för 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 för 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 bekanta dig med källkoden på ). Ett typiskt exempel på användning:!

Källa: habr.com

Vi skriver vår capped expirationd modul för tarantool

Beskrivning

Implementering

Applikation