🥇Ние пишем наш собствен модул с ограничен срок на валидност за tarantool

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

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

описание

Документацията за tarantool е много добра урок за това как да напишете вашите съхранени процедури в C. Преди всичко ви предлагам да се запознаете с него, за да разберете онези вмъквания с команди и код, които ще се появят по-долу. Също така си струва да се обърне внимание справка към обекти, които са налични, когато пишете свой собствен ограничен модул, а именно кутия, влакно, индекс и txn.

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

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. Две функции се експортират от библиотеката: стартиране и убиване. Първата стъпка е да направите тези функции достъпни от Lua с помощта на box.schema.func.create и box.schema.user.grant. След това създайте пространство, чиито кортежи ще съдържат само три полета: първото е уникален идентификатор, второто е имейл, а третото е продължителността на живота на кортежа. Изграждаме дървовиден индекс върху първото поле и го наричаме първично. След това получаваме обекта за връзка към нашата родна библиотека.

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

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 — време за пълно сканиране в секунди.

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

Нека вмъкнем няколко кортежа в пространството с живот от 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+ секунди (броейки от началото на вмъкването на първия кортеж) и видим, че ограниченият модул 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)

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

След подготвителната работа стартираме функцията 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, '[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]
...

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 или повече не бяха сканирани поради подреждането на ключовете за индекс на 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 < 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;
}

Функционален код 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;
}

App

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

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

Пишем наш собствен модул с ограничен срок на валидност за tarantool

описание

Изпълнение

App

Добавяне на нов коментар

Пишем наш собствен модул с ограничен срок на валидност за tarantool

описание

Изпълнение

App

Добавяне на нов коментар Отказване на отговора

Добавяне на нов коментар