Vor einiger Zeit standen wir vor der Herausforderung, Tupel in Spaces zu reinigen. . Die Reinigung musste nicht dann gestartet werden, wenn der Speicher von tarantool bereits zur Neige ging, sondern im Voraus und in bestimmten Intervallen. Für diese Aufgabe gibt es in tarantool ein Modul, das in Lua geschrieben ist, namens . Nach kurzer Nutzung dieses Moduls stellten wir fest, dass es für uns nicht geeignet war: Bei permanenten Bereinigungen großer Datenmengen hing Lua im Garbage Collector. Daher dachten wir darüber nach, unser eigenes Capped-Expirationd-Modul zu entwickeln, in der Hoffnung, dass der in einer nativen Programmiersprache geschriebene Code unsere Anforderungen optimal erfüllen würde.
Ein gutes Beispiel für uns war das tarantool-Modul mit dem Namen . Der verwendete Ansatz basiert darauf, dass im Space ein separates Feld erstellt wird, in dem die Lebensdauer des Tupels, mit anderen Worten, ttl, angegeben ist. Das Modul scannt im Hintergrund den Space, vergleicht ttl mit der aktuellen Zeit und trifft die Entscheidung, ob das Tupel gelöscht werden soll oder nicht. Der Code des Memcached-Moduls ist einfach und elegant, jedoch zu allgemein. Erstens berücksichtigt er nicht den Typ des Indexes, auf dem die Durchsuchung und Löschung erfolgt. Zweitens werden bei jedem Durchlauf alle Tupel gescannt, deren Anzahl recht groß sein kann. Und während im Modul expirationd das erste Problem gelöst wurde (der Baum-Index wurde in eine separate Klasse ausgegliedert), wird dem zweiten weiterhin keinerlei Aufmerksamkeit geschenkt. Diese drei Punkte haben die Entscheidung für die Erstellung eines eigenen Codes bestimmt.
Beschreibung
In der Dokumentation zu Tarantool gibt es sehr gute hinweise, wie man eigene Stored Procedures in C schreibt. Zunächst empfehle ich, sich damit vertraut zu machen, um die folgenden Einfügungen mit Befehlen und Code zu verstehen, die später auftauchen werden. Es lohnt sich auch, auf die zu den Objekten, die beim Schreiben eines eigenen Capped-Moduls verfügbar sind, insbesondere auf , , und .
Lassen Sie uns von der Basis aus beginnen 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 dieser Bibliothek werden zwei Funktionen exportiert: start und kill. Zuerst müssen diese Funktionen über box.schema.func.create und box.schema.user.grant in Lua verfügbar gemacht werden. Dann erstellen wir einen Space, dessen Tupel nur drei Felder enthalten: das erste ist eine eindeutige Kennung, das zweite die E-Mail-Adresse, das dritte die Lebensdauer des Tupels. Über das erste Feld bauen wir einen Baum-Index und nennen ihn primary. Anschließend erhalten wir ein Verbindungsobjekt zu unserer nativen Bibliothek.
Nach den Vorbereitungsarbeiten rufen wir die Funktion start auf:
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 expirationd-Modul, das in Lua geschrieben ist. Das erste Argument der Funktion start ist der eindeutige Name der Aufgabe. Das zweite ist die Identifikationsnummer des Spaces. Das dritte ist der eindeutige Index, nach dem Tupel entfernt werden. Das vierte ist der Index, nach dem die Tupel durchlaufen werden. Das fünfte ist die Feldnummer des Tupels mit der Lebensdauer (die Zählung beginnt bei 1, nicht bei 0!). Das sechste und siebte sind die Scaneinstellungen. 1024 ist die maximale Anzahl von Tupeln, die innerhalb einer Transaktion durchsucht werden. 3600 ist die gesamte Scanzeit in Sekunden.
Bitte beachten Sie, dass im Beispiel derselbe Index für das Durchlaufen und Entfernen verwendet wird. Wenn es sich um einen Baum-Index handelt, erfolgt die Durchquerung von kleineren zu größeren Schlüsseln. Bei einem anderen, beispielsweise einem Hash-Index, erfolgt die Durchquerung in der Regel in zufälliger Reihenfolge. Bei einem Scan werden alle Tupel des Spaces betrachtet.
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}Lass uns überprüfen, ob der Eintrag erfolgreich war:
tarantool> box.space.tester.index.primary:select()
---
- - [0, 'user0@tarantool.io', 1576418976]
- [1, 'user1@tarantool.io', 1576418976]
- [2, 'user2@tarantool.io', 1576418976]
...Lass uns die Auswahl nach mehr als 60 Sekunden wiederholen (gerechnet vom Beginn des ersten Eintrags) und sehen, dass das Modul capped expirationd bereits gearbeitet hat:
tarantool> box.space.tester.index.primary:select()
---
- []
...Lass uns die Aufgabe beenden:
capped_connection:call('libcapped-expirationd.kill', {'non-indexed'})Betrachten wir das zweite Beispiel, bei dem ein separates Index 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 das gleiche wie im ersten Beispiel, mit kleinen Ausnahmen. Ü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. Der Zugriff erfolgt über den exp-Index, während das Löschen über den Primary erfolgt. Wir erinnern uns, dass früher sowohl das eine als auch das andere nur mithilfe des Primary-Indexes gemacht wurde.
Nach den vorbereitenden Arbeiten 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})Wir werden erneut 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}Nach 30 Sekunden fügen wir analog weitere 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}Lass uns überprüfen, ob der Eintrag 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]
...Lass uns die Auswahl nach mehr als 60 Sekunden wiederholen (gerechnet vom Beginn des ersten Eintrags) und sehen, 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 verbleiben Tupel, die noch etwa 30 Sekunden leben. Darüber hinaus wurde das Scannen beim Übergang vom Tupel mit der ID 2 und der Lebensdauer 1576421257 zum Tupel mit der ID 3 und der Lebensdauer 1576421287 gestoppt. Tupel mit einer Lebensdauer von 1576421287 und mehr wurden aufgrund der Ordnung der Schlüssel des exp-Indexes nicht betrachtet. Das ist die Einsparung, die wir ursprünglich erreichen wollten.
Lass uns die Aufgabe beenden:
capped_connection:call('libcapped-expirationd.kill', {'indexed'})Implementierung
Am besten erzählt die Ausgangsdatei immer von allen Besonderheiten des Projekts. ! Im Rahmen der Veröffentlichung werden wir uns nur auf die wichtigsten Punkte konzentrieren, nämlich auf die Algorithmen zur Durchquerung des Spaces.
Die Argumente, die wir in 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 Indexes, nach dem Tupel gelöscht werden. Das Attribut it_index_id ist die Identifikation des Indexes, nach dem Tupel durchlaufen werden. Das Attribut it_index_type ist der Typ des Indexes, nach dem Tupel durchlaufen werden. Das Attribut filed_no ist die Nummer des Tupelfeldes mit der Lebensdauer. Das Attribut scan_size ist die maximale Anzahl der Tupel, die innerhalb einer Transaktion betrachtet werden. Das Attribut scan_time ist die Zeit für einen vollständigen Scan in Sekunden.
Wir werden das Parsen der Argumente nicht betrachten. Dies ist eine mühsame, aber nicht schwierige Aufgabe, bei der Ihnen die Bibliothek . Schwierigkeiten können nur bei den Indizes auftreten, die als komplexe Datenstruktur vom Typ mp_map aus Lua übergeben werden, und nicht in Form einfacher Typen wie mp_bool, mp_double, mp_int, mp_uint und mp_array. Aber es ist nicht nötig, den gesamten Index zu parsen. Es reicht aus, seine Einzigartigkeit zu überprüfen, den Typ zu bestimmen und die Identifikation zu extrahieren.
Lassen Sie uns die Prototypen aller Funktionen auflisten, die zum 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);Jetzt kommen wir zum Wichtigsten — zur Logik der Durchsuchung des Raums und zur Löschung von Tupeln. Jeder Block von Tupeln mit einer Größe von höchstens scan_size wird in einer einzigen Transaktion durchsucht und geändert. Bei Erfolg wird diese Transaktion bestätigt, im Fehlerfall wird sie zurückgerollt. Das letzte Argument der Funktion expirationd_iterate ist ein Zeiger auf den Iterator, von dem aus die Durchsuchung beginnt oder fortgesetzt wird. Dieser Iterator wird inkrementiert, bis ein Fehler auftritt, der Speicher ausgeht oder die Möglichkeit besteht, den Prozess vorzeitig zu stoppen. Die Funktion expirationd_expired überprüft die Lebensdauer des Tupels, expirationd_delete entfernt 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;
}generierten Wortliste wurde sichergestellt.
Der Quellcode kann eingesehen werden unter !
Quelle: habr.com
