Създаваме своя capped expirationd модул за tarantool

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

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

Описание

В документацията на 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. Помним, че преди и двете операции се извършваха само с използване на 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:

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

Атрибут name — името на задачата. Атрибут space_id — идентификатор на пространството. Атрибут rm_index_id — идентификатор на уникалния индекс, по който ще се извършва изтриването на кортежи. Атрибут it_index_id — идентификатор на индекса, по който ще се извършва обходът на кортежи. Атрибут it_index_type — тип на индекса, по който ще се извършва обходът на кортежи. Атрибут filed_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