Писање сопственог модула са ограниченим роком трајања за тарантоол

Пре неког времена суочили смо се са проблемом чишћења тупле у просторима тарантоол. Чишћење је требало започети не када је тарантоол-у већ понестало меморије, већ унапред и на одређеној фреквенцији. За овај задатак, тарантоол има модул написан у Луа језику који се зове истицање. Након кратког коришћења овог модула, схватили смо да није прикладан за нас: због сталног чишћења велике количине података, Луа је висила у ГЦ-у. Стога смо размишљали о развоју сопственог модула са ограниченим роком трајања, надајући се да ће код написан на матерњем програмском језику решити наше проблеме на најбољи могући начин.

Добар пример за нас је био тарантоол модул под називом мемцацхед. Приступ који се у њему користи заснива се на чињеници да се у простору креира посебно поље које указује на животни век тупле, другим речима, ттл. Модул у позадини скенира простор, упоређује ТТЛ са тренутним временом и одлучује да ли да обрише тупле или не. Код модула мемцацхед је једноставан и елегантан, али превише генерички. Прво, не узима у обзир тип индекса који се пописује и брише. Друго, на сваком пролазу се скенирају све торке, чији број може бити прилично велик. И ако је у модулу истекао први проблем решен (индекс стабла је издвојен у посебну класу), онда се на други и даље није обраћала пажња. Ове три тачке су предодредиле избор у корист писања сопственог кода.

Опис

Документација за тарантоол има веома добру Приручник о томе како да напишете своје ускладиштене процедуре у Ц. Пре свега, предлажем да се упознате са тим како бисте разумели оне уметке са командама и кодом који ће се појавити испод. Такође вреди обратити пажњу референца на објекте који су доступни приликом писања вашег сопственог ограниченог модула, наиме кутија, влакно, индекс и ткн.

Почнимо издалека и погледајмо како споља изгледа модул са ограниченим роком трајања:

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)

Ради једноставности, покрећемо тарантоол у ​​директоријуму где се налази наша библиотека либцаппед-екпиратионд.со. Две функције се извозе из библиотеке: старт и килл. Први корак је да ове функције учините доступним из Луа користећи бок.сцхема.фунц.цреате и бок.сцхема.усер.грант. Затим направите простор чије ће торке садржати само три поља: прво је јединствени идентификатор, друго је е-пошта, а треће је животни век торке. Ми градимо индекс стабла на врху првог поља и називамо га примарним. Затим добијамо објекат везе са нашом матичном библиотеком.

Након припремних радова, покрените функцију старт:

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

Овај пример ће функционисати током скенирања потпуно исто као и истекао модул, који је написан у Луа. Први аргумент функције старт је јединствено име задатка. Други је идентификатор простора. Трећи је јединствени индекс којим ће се тупле брисати. Четврти је индекс којим ће се прећи торке. Пети је број поља тупле са животним веком (нумерисање почиње од 1, а не од 0!). Шеста и седма су подешавања скенирања. 1024 је максималан број торки које се могу видети у једној трансакцији. 3600 — пуно време скенирања у секундама.

Имајте на уму да пример користи исти индекс за пописивање и брисање. Ако је ово индекс стабла, онда се прелазак врши од мањег кључа до већег. Ако постоји неки други, на пример, хеш индекс, онда се обилазак врши, по правилу, случајним редоследом. Сви просторни токови се скенирају у једном скенирању.

Хајде да убацимо неколико тупле у простор са животним веком од 60 секунди:

box.space.tester:insert{0, '[email protected]', math.floor(fiber.time()) + 60}
box.space.tester:insert{1, '[email protected]', math.floor(fiber.time()) + 60}
box.space.tester:insert{2, '[email protected]', math.floor(fiber.time()) + 60}

Хајде да проверимо да ли је уметање било успешно:

tarantool> box.space.tester.index.primary:select()
---
- - [0, '[email protected]', 1576418976]
  - [1, '[email protected]', 1576418976]
  - [2, '[email protected]', 1576418976]
...

Поновимо избор након 60+ секунди (рачунајући од почетка уметања првог тупле) и видимо да је ограничени модул истека већ обрађен:

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

Зауставимо задатак:

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

Хајде да погледамо други пример где се за пописивање користи посебан индекс:

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)

Овде је све исто као у првом примеру, са неколико изузетака. Ми градимо индекс стабла на врху трећег поља и зовемо га екп. Овај индекс не мора да буде јединствен, за разлику од индекса који се зове примарни. Прелазак ће се вршити по индексу екп, а брисање по примарном. Сетимо се да су претходно оба рађена само помоћу примарног индекса.

Након припремних радова, покрећемо функцију старт са новим аргументима:

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

Хајде да поново убацимо неколико тупле у простор са животним веком од 60 секунди:

box.space.tester:insert{0, '[email protected]', math.floor(fiber.time()) + 60}
box.space.tester:insert{1, '[email protected]', math.floor(fiber.time()) + 60}
box.space.tester:insert{2, '[email protected]', math.floor(fiber.time()) + 60}

После 30 секунди, по аналогији, додаћемо још неколико тупле-а:

box.space.tester:insert{3, '[email protected]', math.floor(fiber.time()) + 60}
box.space.tester:insert{4, '[email protected]', math.floor(fiber.time()) + 60}
box.space.tester:insert{5, '[email protected]', math.floor(fiber.time()) + 60}

Хајде да проверимо да ли је уметање било успешно:

tarantool> box.space.tester.index.primary:select()
---
- - [0, '[email protected]', 1576421257]
  - [1, '[email protected]', 1576421257]
  - [2, '[email protected]', 1576421257]
  - [3, '[email protected]', 1576421287]
  - [4, '[email protected]', 1576421287]
  - [5, '[email protected]', 1576421287]
...

Поновимо избор након 60+ секунди (рачунајући од почетка уметања првог тупле) и видимо да је ограничени модул истека већ обрађен:

tarantool> box.space.tester.index.primary:select()
---
- - [3, '[email protected]', 1576421287]
  - [4, '[email protected]', 1576421287]
  - [5, '[email protected]', 1576421287]
...

Остало је још неколико тупле у простору који ће имати још око 30 секунди живота. Штавише, скенирање је заустављено када се пређе са торке са ИД-ом 2 и животним веком од 1576421257 у торку са ИД-ом 3 и животним веком од 1576421287. Торке са животним веком од 1576421287 или више нису скениране због редоследа кључеви индекса екп. То је уштеда коју смо желели да постигнемо на самом почетку.

Зауставимо задатак:

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

Имплементација

Најбољи начин да се исприча о свим карактеристикама пројекта је његов изворни извор. код! У оквиру публикације, фокусираћемо се само на најважније тачке, а то су алгоритми за обилазак простора.

Аргументи које прослеђујемо у старт метод се чувају у структури која се зове екпиратионд_таск:

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

Атрибут наме је назив задатка. Атрибут спаце_ид је идентификатор простора. Атрибут рм_индек_ид је идентификатор јединственог индекса којим ће торке бити обрисане. Атрибут ит_индек_ид је идентификатор индекса помоћу којег ће се торке прелазити. Атрибут ит_индек_типе је тип индекса помоћу којег ће се торке прелазити. Атрибут филед_но је број поља тупле са животним веком. Атрибут сцан_сизе је максималан број торки које се скенирају у једној трансакцији. Атрибут сцан_тиме је пуно време скенирања у секундама.

Нећемо разматрати рашчлањивање аргумената. Ово је мукотрпан, али једноставан посао, у чему ће вам библиотека помоћи мсгпуцк. Потешкоће могу настати само са индексима који се прослеђују из Луа као сложена структура података са типом мп_мап, а не коришћењем једноставних типова мп_боол, мп_доубле, мп_инт, мп_уинт и мп_арраи. Али нема потребе да се анализира цео индекс. Само треба да проверите његову јединственост, израчунате тип и издвојите идентификатор.

Наводимо прототипове свих функција које се користе за рашчлањивање:

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

Сада пређимо на оно најважније - логику заобилажења простора и брисања тупле. Сваки блок торки не већи од сцан_сизе се скенира и модификује у оквиру једне трансакције. Ако је успешна, ова трансакција је урезана; ако дође до грешке, она се враћа назад. Последњи аргумент функције екпиратионд_итерате је показивач на итератор из којег скенирање почиње или се наставља. Овај итератор се интерно повећава све док не дође до грешке, док не понестане простора или није могуће зауставити процес унапред. Функција екпиратионд_екпиред проверава животни век тупле, екпиратионд_делете брише тупле, екпиратионд_бреакабле проверава да ли треба да идемо даље.

Екпиратионд_итерате код функције:

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 < task->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;
}

Код функције екпиратионд_екпиред:

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

Екпиратионд_делете код функције:

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

Код функције екпиратион_бреакабле:

static bool
expirationd_breakable(struct expirationd_task *task)
{
  return task->it_index_id != task->rm_index_id && task->it_index_type == ITER_GT;
}

апликација

Изворни код можете погледати на овде!

Извор: ввв.хабр.цом