Пишем своя capped expirationd модул за tarantool

Преди известно време се сблъскахме с проблема за почистването на кортежи в спейсовете. tarantool. Процесът на почистване трябваше да се стартира не само когато паметта на tarantool вече свършва, а предварително и с определена периодичност. За тази задача в tarantool има модул, написан на Lua, наречен expirationd. След краткото използване на този модул осъзнахме, че не е подходящ за нас: при постоянно почистване на големи обеми данни, Lua се забавяше поради GC. Затова решихме да разработим собствен модул capped expirationd, като се надявахме, че кодът, написан на нативен програминг език, ще реши задачите ни по най-добрия начин.

Добър пример за нас е модулът tarantool, наречен memcached. Използваният в него подход се основава на това, че в спейса се създава отделно поле, в което се посочва времето на живот на кортежа, тоест ttl. Модулът на заден план сканира спейса, сравнява ttl с текущото време и взема решение дали да изтрие кортежа или не. Кодът на модула memcached е прост и елегантен, но твърде общ. Първо, той не взема предвид типа индекс, по който се извършва обходът и изтритето. Второ, при всеки обход се сканират всички кортежи, чийто брой може да бъде доста голям. И ако в модула expirationd първият проблем е решен (tree-индексът е отделен в свой клас), вторият все още не е получил никакво внимание. Тези три точки предопределиха избора в полза на написването на собствен код.

Описание

В документацията на tarantool има много добър урок относно това как да пишете свои храними процедури на C. Веднага предлагам да се запознаете с него, за да разберете настъпленията с команди и код, които ще срещнете по-долу. Също така е важно да обърнете внимание на референса к обектите, налични при написването на собствен capped модул, а именно на box, fiber, index и txn.

Нека започнем отдалеч и видим как изглежда модулът capped expirationd отвън:

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)

За по-лесно стартиране, стартираме tarantool в директорията, в която се намира нашата библиотека libcapped-expirationd.so. От библиотеката се експортират две функции: start и kill. В самото начало трябва да направим тези функции достъпни от Lua с помощта на box.schema.func.create и box.schema.user.grant. След това създаваме спейс, кортежите на който ще съдържат само три полета: първото - уникален идентификатор, второто - електронна поща, третото - време на живот на кортежа. Върху първото поле изграждаме tree-индекс и го наричаме primary. След това получаваме обект за свързване с нашата местна библиотека.

След подготовката стартираме функцията 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})

Този пример ще работи по абсолютно същия начин, както модулът expirationd, написан на Lua. Първият аргумент на функцията start е уникалното име на задачата. Вторият - идентификатор на спейса. Третият - уникален индекс, по който ще става премахването на кортежите. Четвъртият - индекс, по който ще става обходът на кортежите. Петият - номер на полето на кортежа с времето на живот (нумерацията започва от 1, а не от 0!). Шестият и седмият - настройки за сканиране. 1024 - това е максималният брой кортежи, които се преглеждат в рамките на една транзакция. 3600 - времето за пълно сканиране в секунди.

Обърнете внимание, че за обход и премахване в примера се използва един и същ индекс. Ако това е tree-индекс, обходът се извършва от по-малкия ключ до по-голямия. Ако е някакъв друг, например hash-индекс, обходът обикновено се извършва в произволен ред. При едно сканиране се преглеждат всички кортежи на спейса.

Нека направим вмъкване в спейса на няколко кортежа с време на живот 60 секунди:

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}

Да проверим, че вмъкването е успешно:

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

Да повторим select след 60+ секунди (изчисляваме от началото на вмъкването на първия кортеж) и ще видим, че модулът capped expirationd вече е сработил:

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)

Тук всичко е същото, както в първия пример, с малки изключения. Върху третото поле изграждаме tree индекс и го наричаме exp. Този индекс не е задължен да бъде уникален, за разлика от индекса с името primary. Обходът ще се извършва по индекса exp, а изтриването по primary. Помним, че преди и двете действия се извършваха само с помощта на основния индекс.

След подготовката стартираме функцията start с нови аргументи:

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, '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 секунди, по аналогия, добавяме още няколко кортежа:

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}

Да проверим, че вмъкването е успешно:

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

Да повторим select след 60+ секунди (изчисляваме от началото на вмъкването на първия кортеж) и ще видим, че модулът capped expirationd вече е сработил:

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

В спейса останаха кортежи, които ще са в сила още около 30 секунди. Освен това, сканирането спря при преминаване от кортежа с идентификатор 2 и времетраене 1576421257 към кортежа с идентификатор 3 и времетраене 1576421287. Кортежите с времетраене 1576421287 и повече не бяха прегледани благодарение на подредбата на ключовете на индекса exp. Това е икономията, която искахме да постигнем в началото.

Да спрем задачата:

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

Реализация

Най-добре за всички особености на проекта винаги ще разкаже неговият източник. код! В рамките на публикацията можем да се спрем само на най-важните моменти, а именно, на алгоритмите за обход на спейса.

Аргументите, които предаваме на метода start, се съхраняват в структура, наречена expirationd_task:

структура 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;
};

Атрибут name — това е името на задачата. Атрибут space_id — идентификатор на пространството. Атрибут rm_index_id — идентификатор на уникалния индекс, по който ще се извършва изтриването на записи. Атрибут it_index_id — идентификатор на индекса, по който ще се извършва обхода на записите. Атрибут it_index_type — тип на индекса, по който ще се извършва обхода на записите. Атрибут field_no — номер на полето на записа с времето на живот. Атрибут scan_size — максималното количество записи, което ще се прегледа в рамките на една транзакция. Атрибут scan_time — времето за пълно сканиране в секунди.

Няма да разглеждаме парсирането на аргументите. Това е трудоемка, но не сложна работа, в която библиотеката msgpuck. Може да възникнат трудности единствено с индексите, които се предават от Lua под формата на сложни структури от данни с тип mp_map, а не с помощта на прости типове mp_bool, mp_double, mp_int, mp_uint и mp_array. Но не е нужно да се парсира целият индекс. Достатъчно е само да се провери уникалността му, да се изчисли типът и да се извлече идентификаторът.

Нека изброим прототипите на всички функции, които се използват за парсиране:

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

А сега да преминем към най-важното — логиката за обход на спейса и изтриване на кортежи. Всеки блок кортежи с размер не по-голям от scan_size се преглежда и модифицира в рамките на една транзакция. При успех тази транзакция се комитва, а при грешка — се откатва. Последният аргумент в функцията expirationd_iterate е указателят към итератора, с който започва или продължава сканирането. Този итератор се инкрементира, докато не възникне грешка, не свърши спейсът или не се появи възможност да се прекрати процесът предварително. Функцията expirationd_expired проверява времето на живот на кортежа, expirationd_delete — изтрива кортежа, expirationd_breakable — проверява дали трябва да продължим напред.

Код на функцията 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;
}

Код на функцията 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;
}

Код на функцията 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);
}

Код на функцията 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;
}

Приложение

Можете да се запознаете с изходния код на тук!

Източник: habr.com