Perl 5.8.8 Threads – Benchmark d’insertion de masse MySQL

24 juil 2008

Perl 5.8.8 Threads – Benchmark d’insertion de masse MySQL

Catégorie : Perl — Celogeek @ 17:36

Hello, voici un petit tuto sur les threads en perl 5.8.8.

L’implémentation des threads en perl 5.8.8 est assez légère, cependant suffisamment complète pour pouvoir faire ce que l’on veut.

Le problème rencontré est l’absence d’indicateur de fin d’une thread. Une des solutions possibles consiste à créer une file d’attente (Thread::Queue) et d’y empiler les numéro (TID) de threads, quand celles-ci se termine. Le programme principal n’a plus qu’à nettoyer les threads terminées qui s’empilent dans la file d’attente. Le tout reste très propre et facilement contrôlable.

On peut imaginer un ajout automatique dans cette file d’attente par capture de signal ( KILL ), pour prévenir les DIE inopinés … à tester vous-même.

Voici un exemple de code, effectuant une insertion de masse de 1 000 000 de lignes dans MySQL.

use strict;
use threads;
use Thread::Queue;
use DBI;

my $stream = new Thread::Queue;
my $threadFinish = new Thread::Queue;

my $dsn = "DBI:mysql:database=test";
my $usr = "test";
my $pwd = "test";

for(my $i=0; $i<8; $i++) {
   new threads(\&disp, $stream, $threadFinish, $dsn, $usr, $pwd, sprintf("t%d",$i%8+1));
}
for my $i ( 1 .. 1000000 ) {
   $stream->enqueue($i);
}

for(my $i=0; $i
list; $i++) {
 $stream->enqueue(undef);
 my $tid = $threadFinish->dequeue;
 print "Dequeue ",$tid,"\n";
 threads->object($tid)->join;
}

sub disp {
 my ($stream, $finish, $dsn, $usr, $pwd, $table) = @_;
 my $conn = DBI->connect($dsn,$usr,$pwd);
 if (!$conn) {
   $finish->enqueue(threads->self->tid);
   return;
 }
 #$conn->do("TRUNCATE TABLE ".$table);
 my $sth = $conn->prepare("INSERT INTO ".$table." (data) VALUES (NOW())");
 while (my $num = $stream->dequeue) {
   #print threads->self->tid, " - ",$num,"\n" if $num % 1000 == 0;
   $sth->execute;
 }
 $sth->finish;
 $finish->enqueue(threads->self->tid);
 return;
}

Ce programme lance les threads qui vont préparer l’insertion dans la base. Celles-ci lisent une file d’attente qui leur donne le top départ et aussi l’arrêt. Chaque thread ainsi créée effectue son travail et dépile de la file d’attente un élément pour lui dire de continuer.

Le programme principal empile 1 000 000 d’éléments dans la file d’attente, ceux-ci sont récupérés par l’ensemble des threads au fur et à mesure des insertions.

Une fois la tâche terminée, la thread empile son numéro d’TID dans la file d’attente de nettoyage, ainsi le programme principal peut les libérer sans crainte d’avoir un lock ou autre chose d’inattendu.

Voilà !

Celogeek !

Amusez-vous bien !