Jakiś czas temu stanęliśmy przed problemem czyszczenia krotek w przestrzeniach. . Czyszczenie musiało być uruchamiane nie wtedy, gdy w tarantool kończyła się pamięć, ale z wyprzedzeniem i z określoną regularnością. Do tego zadania w tarantool jest moduł napisany w Lua, o nazwie . Po krótkim użytkowaniu tego modułu zdaliśmy sobie sprawę, że nie jest on odpowiedni: przy regularnym czyszczeniu dużych ilości danych Lua zawieszał się w GC. Dlatego postanowiliśmy opracować własny moduł capped expirationd, mając nadzieję, że kod napisany w natywnym języku programowania najlepiej rozwiąże nasze problemy.
Dobrym przykładem był moduł tarantool o nazwie . Stosowane w nim podejście polega na tym, że w przestrzeni tworzona jest osobna kolumna, w której określany jest czas życia krotki, innymi słowy, ttl. Moduł w tle skanuje przestrzeń, porównuje ttl z obecnym czasem i podejmuje decyzję o usunięciu krotki lub nie. Kod modułu memcached jest prosty i elegancki, ale zbyt ogólny. Po pierwsze, nie uwzględnia typu indeksu, według którego odbywa się przeszukiwanie i usuwanie. Po drugie, przy każdym przebiegu skanowane są wszystkie krotki, których liczba może być dość duża. I jeśli w module expirationd pierwszy problem został rozwiązany (indeks tree został wydzielony do osobnej klasy), to drugi nadal pozostaje bez uwagi. Te trzy punkty zadecydowały o wyborze stworzenia własnego kodu.
Opis
W dokumentacji do tarantool znajduje się bardzo dobra na temat tego, jak pisać własne procedury składowane w języku C. Przede wszystkim proponuję zapoznać się z nim, aby zrozumieć te wstawki z poleceniami i kodem, które będą się pojawiać poniżej. Warto również zwrócić uwagę na do obiektów, które są dostępne przy pisaniu własnego modułu capped, a mianowicie na , , i .
Zacznijmy od początku i spójrzmy na to, jak zewnętrznie wygląda moduł 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)Aby uprościć, uruchamiamy tarantoola w katalogu, w którym znajduje się nasza biblioteka libcapped-expirationd.so. Z biblioteki eksportowane są dwie funkcje: start i kill. Najpierw trzeba udostępnić te funkcje w Lua za pomocą box.schema.func.create oraz box.schema.user.grant. Następnie tworzymy spację, w której krotki będą zawierać trzy pola: pierwsze – unikalny identyfikator, drugie – adres e-mail, trzecie – czas życia krotki. Na pierwszym polu budujemy indeks tree i nazywamy go primary. Następnie uzyskujemy obiekt połączenia z naszą natywną biblioteką.
Po pracach przygotowawczych uruchamiamy funkcję 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})Ten przykład będzie działać podczas skanowania dokładnie tak, jak moduł expirationd napisany w Lua. Pierwszym argumentem do funkcji start jest unikalna nazwa zadania. Drugim – identyfikator spacji. Trzecim – unikalny indeks, według którego będą usuwane krotki. Czwartym – indeks, według którego będą przeszukiwane krotki. Piątym – numer pola krotki z czasem życia (numeracja zaczyna się od 1, a nie od 0!). Szóstym i siódmym – ustawienia skanowania. 1024 to maksymalna liczba krotek przeszukiwanych w ramach jednej transakcji. 3600 to czas pełnego skanowania w sekundach.
Zauważ, że do skanowania i usuwania w przykładzie używany jest ten sam indeks. Jeśli to jest indeks tree, skanowanie odbywa się od mniejszego klucza do większego. Jeśli inny, na przykład indeks hash, skanowanie odbywa się zazwyczaj w losowej kolejności. W ramach jednego skanowania przeglądane są wszystkie krotki spacji.
Wstawmy kilka krotek do spacji z czasem życia 60 sekund:
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}Sprawdźmy, czy wstawienie powiodło się:
tarantool> box.space.tester.index.primary:select()
---
- - [0, 'user0@tarantool.io', 1576418976]
- [1, 'user1@tarantool.io', 1576418976]
- [2, 'user2@tarantool.io', 1576418976]
...Powtórzymy select po 60+ sekundach (liczone od rozpoczęcia wstawiania pierwszej krotki) i zobaczymy, że moduł capped expirationd już wykonał swoje zadanie:
tarantool> box.space.tester.index.primary:select()
---
- []
...Zatrzymamy zadanie:
capped_connection:call('libcapped-expirationd.kill', {'non-indexed'})Rozważmy drugi przykład, w którym do przeszukiwania używamy osobnego indeksu:
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)Tutaj wszystko jest takie samo jak w pierwszym przykładzie, z tą małą różnicą, że na trzecim polu budujemy indeks typu tree i nazywamy go exp. Ten indeks nie musi być unikalny, w przeciwieństwie do indeksu o nazwie primary. Przeszukiwanie będzie odbywać się według indeksu exp, a usuwanie według primary. Pamiętamy, że wcześniej wszystko to robiliśmy tylko przy użyciu indeksu primary.
Po pracach przygotowawczych uruchamiamy funkcję start z nowymi argumentami:
capped_connection:call('libcapped-expirationd.start', {'indexed', box.space.tester.id, box.space.tester.index.primary, box.space.tester.index.exp, 3, 1024, 3600})Ponownie wstawimy do space kilka krotek z czasem życia 60 sekund:
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}Po 30 sekundach dodamy jeszcze kilka krotek:
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}Sprawdźmy, czy wstawienie powiodło się:
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]
...Powtórzymy select po 60+ sekundach (liczone od rozpoczęcia wstawiania pierwszej krotki) i zobaczymy, że moduł capped expirationd już wykonał swoje zadanie:
tarantool> box.space.tester.index.primary:select()
---
- - [3, 'user3@tarantool.io', 1576421287]
- [4, 'user4@tarantool.io', 1576421287]
- [5, 'user5@tarantool.io', 1576421287]
...W przestrzeni pozostały krotki, które mają jeszcze około 30 sekund życia. Co więcej, skanowanie zatrzymało się na przejściu od krotki z identyfikatorem 2 i czasem życia 1576421257 do krotki z identyfikatorem 3 i czasem życia 1576421287. Krotki z czasem życia 1576421287 i wyższym nie zostały przeszukane dzięki uporządkowaniu kluczy indeksu exp. To właśnie ta oszczędność, którą chcieliśmy osiągnąć na samym początku.
Zatrzymamy zadanie:
capped_connection:call('libcapped-expirationd.kill', {'indexed'})Realizacja
Najlepiej o wszystkich szczegółach projektu zawsze opowie jego źródło ! W ramach publikacji zatrzymamy się tylko na najważniejszych kwestiach, a mianowicie na algorytmach przeszukiwania przestrzeni.
Argumenty, które przekazujemy do metody start, są przechowywane w strukturze o nazwie 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;
};Atrybut name to nazwa zadania. Atrybut space_id to identyfikator przestrzeni. Atrybut rm_index_id to identyfikator unikalnego indeksu, według którego będą usuwane krotki. Atrybut it_index_id to identyfikator indeksu, według którego będą weryfikowane krotki. Atrybut it_index_type to typ indeksu, według którego będą weryfikowane krotki. Atrybut field_no to numer pola krotki z czasem życia. Atrybut scan_size to maksymalna liczba krotek, które są przeglądane w ramach jednej transakcji. Atrybut scan_time to czas pełnego skanowania w sekundach.
Nie będziemy rozważać parsowania argumentów. To żmudna, ale nie skomplikowana praca, w której pomoże ci biblioteka . Problemy mogą pojawić się jedynie z indeksami, które są przekazywane z Lua jako złożona struktura danych o typie mp_map, a nie za pomocą prostych typów mp_bool, mp_double, mp_int, mp_uint i mp_array. Jednak nie trzeba całego indeksu parsować. Wystarczy sprawdzić jego unikalność, obliczyć typ i wyodrębnić identyfikator.
Wymieńmy prototypy wszystkich funkcji, które są używane do parsowania:
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);A teraz przejdźmy do najważniejszego — logiki przeszukiwania przestrzeni i usuwania krotek. Każdy blok krotek o rozmiarze nieprzekraczającym scan_size jest przeglądany i modyfikowany w ramach jednej transakcji. W przypadku powodzenia ta transakcja jest zatwierdzana, w przypadku błędu — wycofywana. Ostatnim argumentem w funkcji expirationd_iterate jest wskaźnik na iterator, od którego zaczyna się lub kontynuuje skanowanie. Ten iterator jest inkrementowany, aż wystąpi błąd, skończy się przestrzeń, lub będzie możliwość wcześniejszego zatrzymania procesu. Funkcja expirationd_expired sprawdza czas życia krotki, expirationd_delete — usuwa krotkę, expirationd_breakable — sprawdza, czy powinniśmy przejść dalej.
Kod funkcji 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;
}Kod funkcji 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;
}Kod funkcji 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);
}Kod funkcji 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;
}Aplikacja
Można zapoznać się z kodem źródłowym na !
Źródło: habr.com
