Il y a quelque temps, nous avons rencontré un problème de nettoyage des tuples dans les espaces. . Le nettoyage devait être lancé non pas lorsque tarantool manquait déjà de mémoire, mais à l'avance et avec une certaine périodicité. Pour cela, tarantool possède un module, écrit en Lua, appelé . Après une utilisation prolongée de ce module, nous avons réalisé qu'il ne nous convenait pas : les nettoyages fréquents de grandes quantités de données provoquaient des blocages de Lua dans le GC. Nous avons donc envisagé de développer notre propre module capped expirationd, espérant que le code écrit dans un langage de programmation natif résoudrait nos problèmes de manière optimale.
Un bon exemple pour nous a été le module tarantool nommé . L'approche utilisée dans celui-ci repose sur le fait qu'un champ distinct est créé dans l'espace, indiquant le temps de vie du tuple, autrement dit, ttl. Le module scanne en arrière-plan l'espace, compare ttl avec l'heure actuelle et décide s'il faut supprimer le tuple ou non. Le code du module memcached est simple et élégant, mais trop général. Premièrement, il ne prend pas en compte le type d'index utilisé pour le parcours et la suppression. Deuxièmement, à chaque passage, tous les tuples sont scannés, dont le nombre peut être assez élevé. Si dans le module expirationd le premier problème a été résolu (l'index de type arbre est isolé dans une classe distincte), le second n'a toujours pas reçu d'attention. Ces trois points ont prédisposé notre choix en faveur de l'écriture de notre propre code.
Description
La documentation de tarantool contient un très bon sur la façon d'écrire vos propres procédures stockées en C. Je vous recommande tout d'abord de vous y familiariser afin de comprendre les insertions avec commandes et code qui apparaîtront ci-dessous. Il convient également de prêter attention au des objets disponibles lors de l'écriture de votre propre module capped, à savoir , , et .
Commençons par le début et voyons à quoi ressemble le module capped expirationd de l'extérieur :
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)Pour simplifier, lançons tarantool dans le répertoire où se trouve notre bibliothèque libcapped-expirationd.so. Deux fonctions sont exportées à partir de la bibliothèque : start et kill. Tout d'abord, il est nécessaire de rendre ces fonctions accessibles depuis Lua à l'aide de box.schema.func.create et box.schema.user.grant. Ensuite, créez un espace dont les tuples contiendront seulement trois champs : le premier est un identifiant unique, le deuxième est une adresse électronique, le troisième est la durée de vie du tuple. Au-dessus du premier champ, nous construisons un index tree et l'appelons primary. Ensuite, nous obtenons l'objet de connexion à notre bibliothèque native.
Après les travaux préparatoires, lançons la fonction 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})Cet exemple fonctionnera en effectuant un scan exactement comme le module expirationd, qui est écrit en Lua. Le premier argument de la fonction start est un nom de tâche unique. Le deuxième est l'identifiant de l'espace. Le troisième est un index unique par lequel les tuples seront supprimés. Le quatrième est l'index par lequel les tuples seront parcourus. Le cinquième est le numéro du champ du tuple avec la durée de vie (la numérotation commence à 1 et non à 0 !). Le sixième et le septième sont les paramètres de scan. 1024 est le nombre maximum de tuples qui seront examinés dans une seule transaction. 3600 est le temps de scan complet en secondes.
Notez que dans l'exemple, le même index est utilisé à la fois pour le parcours et la suppression. Si c'est un index tree, le parcours se fait du plus petit clé au plus grand. S'il s'agit d'un autre type, par exemple un index hash, le parcours se fait généralement dans un ordre aléatoire. Toutes les tuples de l'espace sont examinées en une seule opération de scan.
Insérons quelques tuples dans l'espace avec une durée de vie de 60 secondes :
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}Vérifions que l'insertion s'est bien déroulée :
tarantool> box.space.tester.index.primary:select()
---
- - [0, 'user0@tarantool.io', 1576418976]
- [1, 'user1@tarantool.io', 1576418976]
- [2, 'user2@tarantool.io', 1576418976]
...Répétons la sélection après 60+ secondes (en comptant depuis le début de l'insertion du premier tuple) et nous verrons que le module capped expirationd a déjà fonctionné :
tarantool> box.space.tester.index.primary:select()
---
- []
...Arrêtons la tâche :
capped_connection:call('libcapped-expirationd.kill', {'non-indexed'})Examinons le deuxième exemple, où un index distinct est utilisé pour le parcours :
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)Ici, tout est similaire au premier exemple, à une petite exception près. Nous construisons un index d'arbre sur le troisième champ et l'appelons exp. Cet index n'a pas besoin d'être unique, contrairement à l'index appelé primary. Le parcours se fera selon l'index exp, alors que la suppression s'effectuera avec l'index primary. Nous nous souvenons qu'auparavant, les deux étaient faits uniquement avec l'index primary.
Après les travaux préparatoires, nous lançons la fonction start avec de nouveaux arguments :
capped_connection:call('libcapped-expirationd.start', {'indexed', box.space.tester.id, box.space.tester.index.primary, box.space.tester.index.exp, 3, 1024, 3600})Nous allons à nouveau insérer plusieurs tuples dans l'espace avec une durée de vie de 60 secondes :
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}Dans 30 secondes, nous ajouterons plusieurs autres tuples par analogie :
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}Vérifions que l'insertion s'est bien déroulée :
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]
...Répétons la sélection après 60+ secondes (en comptant depuis le début de l'insertion du premier tuple) et nous verrons que le module capped expirationd a déjà fonctionné :
tarantool> box.space.tester.index.primary:select()
---
- - [3, 'user3@tarantool.io', 1576421287]
- [4, 'user4@tarantool.io', 1576421287]
- [5, 'user5@tarantool.io', 1576421287]
...Il reste des tuples dans l'espace qui ont encore environ 30 secondes à vivre. De plus, le scan s'est arrêté lors du passage du tuple avec l'identifiant 2 et une durée de vie de 1576421257 au tuple avec l'identifiant 3 et une durée de vie de 1576421287. Les tuples avec une durée de vie de 1576421287 ou plus n'ont pas été consultés grâce à l'ordre des clés de l'index exp. Cela représente l'économie que nous souhaitions atteindre dès le départ.
Arrêtons la tâche :
capped_connection:call('libcapped-expirationd.kill', {'indexed'})Mise en œuvre
Le meilleur moyen d'apprendre toutes les spécificités d'un projet est toujours de consulter son code source ! Dans le cadre de cette publication, nous nous arrêterons uniquement sur les points les plus importants, à savoir les algorithmes de parcours de l'espace.
Les arguments que nous passons à la méthode start sont conservés dans une structure appelée 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;
};L'attribut name est le nom de la tâche. L'attribut space_id est l'identifiant de l'espace. L'attribut rm_index_id est l'identifiant de l'index unique utilisé pour la suppression des tuples. L'attribut it_index_id est l'identifiant de l'index utilisé pour parcourir les tuples. L'attribut it_index_type est le type d'index utilisé pour parcourir les tuples. L'attribut field_no est le numéro du champ du tuple avec la durée de vie. L'attribut scan_size est le nombre maximal de tuples examinés dans une seule transaction. L'attribut scan_time est le temps nécessaire pour un scan complet en secondes.
Nous ne traiterons pas le parsing des arguments. C'est un travail minutieux, mais simple, pour lequel vous pouvez utiliser la bibliothèque . Les difficultés peuvent survenir uniquement avec les index qui sont transmis de Lua sous la forme d'une structure de données complexe de type mp_map, et non via des types simples tels que mp_bool, mp_double, mp_int, mp_uint et mp_array. Mais il n'est pas nécessaire de parser l'ensemble de l'index. Il suffit de vérifier son unicité, de déterminer son type et d'en extraire l'identifiant.
Enumérons les prototypes de toutes les fonctions utilisées pour le parsing :
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);Passons maintenant à l'essentiel — la logique de parcours de l'espace et de suppression des tuples. Chaque bloc de tuples, ne dépassant pas la taille de scan_size, est examiné et modifié dans une seule transaction. En cas de succès, cette transaction est validée, en cas d'erreur, elle est annulée. Le dernier argument de la fonction expirationd_iterate est un pointeur vers l'itérateur à partir duquel commence ou se poursuit le scan. Cet itérateur est incrémenté jusqu'à ce qu'une erreur se produise, que l'espace soit épuisé, ou qu'il soit possible d'interrompre le processus à l'avance. La fonction expirationd_expired vérifie la durée de vie du tuple, expirationd_delete supprime le tuple, et expirationd_breakable vérifie s'il faut continuer.
Code de la fonction 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 de la fonction 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 de la fonction 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 de la fonction 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;
}L'application
Vous pouvez consulter le code source sur !
Source : habr.com
