Vor einiger Zeit standen wir vor dem Problem, die Tupel in den Spaces zu bereinigen. . Die Bereinigung musste nicht dann gestartet werden, wenn der Speicher bei tarantool bereits zu Ende ging, sondern im Voraus und in bestimmten Intervallen. Für diese Aufgabe gibt es in tarantool ein Modul, das in Lua geschrieben ist, mit dem Namen . Nach kurzer Nutzung dieses Moduls stellten wir fest, dass es nicht zu uns passt: Bei den kontinuierlichen Bereinigungen großer Datenmengen hing Lua im Garbage Collector. Daher dachten wir über die Entwicklung eines eigenen Capped Expirationd-Moduls nach, in der Hoffnung, dass der in einer nativen Programmiersprache geschriebene Code unsere Aufgaben am besten lösen würde.
Ein gutes Beispiel war das tarantool-Modul namens . Der in ihm verwendete Ansatz basiert darauf, dass in dem Space ein separates Feld angelegt wird, in dem die Lebensdauer des Tupels angegeben wird, anders ausgedrückt, ttl. Das Modul scannt im Hintergrund den Space, vergleicht ttl mit der aktuellen Zeit und entscheidet, ob das Tupel gelöscht werden soll oder nicht. Der Code des Moduls memcached ist einfach und elegant, aber zu allgemein. Erstens berücksichtigt er nicht den Typ des Indexes, nach dem die Traversierung und Löschung erfolgt. Zweitens werden bei jedem Durchlauf alle Tupel gescannt, deren Anzahl ziemlich groß sein kann. Und während im Modul expirationd das erste Problem gelöst wurde (Tree-Index wurde in eine separate Klasse ausgegliedert), wurde das zweite immer noch nicht ausreichend beachtet. Diese drei Punkte bestimmten die Wahl zu Gunsten der Erstellung unseres eigenen Codes.
Beschreibung
In der Dokumentation zu tarantool gibt es ein sehr gutes über das Schreiben eigener gespeicherter Prozeduren in C. Zunächst empfehle ich, sich damit vertraut zu machen, um die Einfügungen mit Befehlen und Code zu verstehen, die später vorkommen werden. Auch sollte auf den zu den Objekten geachtet werden, die beim Schreiben eines eigenen Capped Moduls zur Verfügung stehen, nämlich auf , , und .
Lassen Sie uns von hinten anfangen und uns ansehen, wie das Capped Expirationd-Modul von außen aussieht:
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)Zur Vereinfachung starten wir Tarantool im Verzeichnis, in dem sich unsere Bibliothek libcapped-expirationd.so befindet. Aus der Bibliothek werden zwei Funktionen exportiert: start und kill. Zunächst müssen diese Funktionen mit box.schema.func.create und box.schema.user.grant für Lua verfügbar gemacht werden. Dann erstellen wir einen Space, dessen Tupel insgesamt drei Felder enthalten: das erste ist eine eindeutige ID, das zweite eine E-Mail-Adresse, das dritte die Lebensdauer des Tupels. Über das erste Feld bauen wir einen Tree-Index und nennen ihn primary. Danach erhalten wir ein Verbindungsobjekt zu unserer nativen Bibliothek.
Nach den Vorarbeiten starten wir die Funktion 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})Dieses Beispiel funktioniert beim Scannen genau wie das Modul expirationd, das in Lua geschrieben wurde. Das erste Argument in der Funktion start ist der eindeutige Name der Aufgabe. Das zweite ist die ID des Spaces. Das dritte ist der eindeutige Index, anhand dessen die Tupel gelöscht werden. Das vierte ist der Index, mit dem die Tupel durchlaufen werden. Das fünfte ist die Nummer des Tupelfeldes mit der Lebensdauer (die Nummerierung beginnt bei 1 und nicht bei 0!). Das sechste und siebte sind die Einstellungen für das Scannen. 1024 ist die maximale Anzahl von Tupeln, die in einer Transaktion angesehen werden. 3600 ist die Zeit für das vollständige Scannen in Sekunden.
Bitte beachten Sie, dass im Beispiel derselbe Index für das Durchlaufen und Löschen verwendet wird. Wenn dies ein Tree-Index ist, erfolgt das Durchlaufen von dem kleineren Schlüssel zum größeren. Wenn es ein anderer ist, beispielsweise ein Hash-Index, erfolgt das Durchlaufen normalerweise in zufälliger Reihenfolge. Bei einem Scan werden alle Tupel des Spaces angesehen.
Lassen Sie uns einige Tupel mit einer Lebensdauer von 60 Sekunden in den Space einfügen:
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}Überprüfen wir, ob der Einfügevorgang erfolgreich war:
tarantool> box.space.tester.index.primary:select()
---
- - [0, 'user0@tarantool.io', 1576418976]
- [1, 'user1@tarantool.io', 1576418976]
- [2, 'user2@tarantool.io', 1576418976]
...Wiederholen wir die Auswahl nach 60+ Sekunden (gerechnet ab dem Zeitpunkt der Einfügung des ersten Tupels) und sehen wir, dass das Modul capped expirationd bereits gearbeitet hat:
tarantool> box.space.tester.index.primary:select()
---
- []
...Stoppen wir die Aufgabe:
capped_connection:call('libcapped-expirationd.kill', {'non-indexed'})Lassen Sie uns das zweite Beispiel betrachten, in dem ein separater Index für die Durchquerung verwendet wird:
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)Hier ist alles dasselbe wie im ersten Beispiel, mit Ausnahme von Kleinigkeiten. Über das dritte Feld bauen wir einen tree-Index und nennen ihn exp. Dieser Index muss im Gegensatz zum Index mit dem Namen primary nicht einzigartig sein. Die Durchquerung erfolgt über den exp-Index, das Löschen über den primary. Wir erinnern uns, dass zuvor beides nur unter Verwendung des primary-Indexes durchgeführt wurde.
Nach den Vorbereitungsarbeiten starten wir die Funktion start mit neuen Argumenten:
capped_connection:call('libcapped-expirationd.start', {'indexed', box.space.tester.id, box.space.tester.index.primary, box.space.tester.index.exp, 3, 1024, 3600})Wieder werden wir mehrere Tupel mit einer Lebensdauer von 60 Sekunden in den Space einfügen:
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}Nach 30 Sekunden fügen wir analog noch einige Tupel hinzu:
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}Überprüfen wir, ob der Einfügevorgang erfolgreich war:
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]
...Wiederholen wir die Auswahl nach 60+ Sekunden (gerechnet ab dem Zeitpunkt der Einfügung des ersten Tupels) und sehen wir, dass das Modul capped expirationd bereits gearbeitet hat:
tarantool> box.space.tester.index.primary:select()
---
- - [3, 'user3@tarantool.io', 1576421287]
- [4, 'user4@tarantool.io', 1576421287]
- [5, 'user5@tarantool.io', 1576421287]
...Im Space sind noch Tupel verblieben, die noch etwa 30 Sekunden leben. Darüber hinaus stoppte das Scannen beim Übergang vom Tupel mit der ID 2 und der Lebensdauer 1576421257 zum Tupel mit der ID 3 und der Lebensdauer 1576421287. Tupel mit einer Lebensdauer von 1576421287 oder mehr wurden aufgrund der Ordnung der Schlüssel im exp-Index nicht angezeigt. Dies ist die Einsparung, die wir zu Beginn erreichen wollten.
Stoppen wir die Aufgabe:
capped_connection:call('libcapped-expirationd.kill', {'indexed'})Implementierung
Am besten erzählt man immer von allen Besonderheiten des Projekts durch dessen Ausgangscode. ! Im Rahmen dieser Veröffentlichung werden wir uns nur auf die wichtigsten Punkte konzentrieren, insbesondere auf die Algorithmen zur Durchquerung des Spaces.
Die Argumente, die wir an die Methode start übergeben, werden in einer Struktur namens expirationd_task gespeichert:
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;
};Das Attribut name ist der Name der Aufgabe. Das Attribut space_id ist die Identifikation des Spaces. Das Attribut rm_index_id ist die Identifikation des einzigartigen Index, über den Tupel gelöscht werden. Das Attribut it_index_id ist die Identifikation des Index, über den Tupel durchlaufen werden. Das Attribut it_index_type ist der Typ des Index, über den Tupel durchlaufen werden. Das Attribut filed_no ist die Nummer des Tupelfelds mit der Lebensdauer. Das Attribut scan_size ist die maximale Anzahl an Tupeln, die im Rahmen einer Transaktion durchsucht werden. Das Attribut scan_time ist die Zeit für einen vollständigen Scan in Sekunden.
Das Parsen von Argumenten werden wir nicht behandeln. Es ist eine mühsame, aber nicht komplizierte Aufgabe, bei der Ihnen die Bibliothek . Schwierigkeiten können nur mit Indizes auftreten, die aus Lua in Form einer komplexen Datenstruktur vom Typ mp_map übergeben werden, und nicht mit einfachen Typen wie mp_bool, mp_double, mp_int, mp_uint und mp_array. Aber es ist nicht notwendig, den gesamten Index zu parsen. Es genügt, seine Eindeutigkeit zu überprüfen, den Typ zu bestimmen und die Identifikation zu extrahieren.
Hier sind die Prototypen aller Funktionen aufgelistet, die für das Parsen verwendet werden:
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);Und jetzt kommen wir zum Wichtigsten – zur Logik des Scannens des Raums und der Löschung von Tupeln. Jedes Block mit Tupeln, das nicht größer als scan_size ist, wird unter einer Transaktion durchsucht und geändert. Im Erfolgsfall wird diese Transaktion festgeschrieben, im Fehlerfall zurückgesetzt. Der letzte Parameter der Funktion expirationd_iterate ist ein Zeiger auf den Iterator, von dem aus das Scannen beginnt oder fortgesetzt wird. Dieser Iterator wird innerhalb der Schleife inkrementiert, bis ein Fehler auftritt, der Raum endet oder die Möglichkeit besteht, den Prozess vorzeitig zu stoppen. Die Funktion expirationd_expired prüft die Lebensdauer des Tupels, expirationd_delete löscht das Tupel, expirationd_breakable prüft, ob wir weitermachen müssen.
Code der Funktion 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;
}Code der Funktion 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;
}Code der Funktion 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);
}Code der Funktion 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;
}Anwendung
Den Quellcode finden Sie unter !
Quelle: habr.com
