Spécifications d'une file d'attente sur disque.

Tout d'abord, nous usionnons les 3 blocs analyseur, filtre et rédacteur car ces 3 blocs peuvent fonctionner de manière sychrone (un même courriel à un instant donné pour l'ensemble de cette chaîne de traitemenst). Nouis donnons à cet ensemble le nom de "sélecteur".

Le lecteur et le rédacteur en tant que producteurs ont besoin de stocker le plus rapidement possible un volume de données qui peut être important. L'archiveur, l'analyseur et l'émetteur en tant que consommateurs, ont besoin d'être pévenus de la production effectuée par le lecteur ou le rédacteur. Comme l'ensemble des données produites présente un rsique de saturation de la mémoire, le stockage de la production se fait sur disque. Il nous faut donc mettre entre producteurs et consommateurs des files d'attente qui gèrent ces données sur disque. Cela nous permet également de définir l'organisation des traitements en termes de threads :

Choix d'une solution.

Une première solution pourrait être de parcourir la file d'attente disque à intervalle de temps régulier pour en observer les changements d'état. Ce mécanisme de scrutation est consommateur de ressources processeur qui peut devenir prohibitive dès lors que l'on souhaite une fréquence de parcours élevée.

Mécanisme QFileSystemWatcher.

A la place, nous allons utiliser un mécanisme simple et éprouvé : la gestion d'une file d'attente sur disque surveillée par un observateur du système de fichier (file watcher). Un observateur du système de fichier surveille l'activité disque et adresse un signal au processus qui l'a invoqué dès qu'un changement d'état de l'activité disque surveillée est détecté. Un observateur du système de fichier est un objet délégué du système d'exploitation et Linux comme Windows ou d'autres systèmes les mettent en oeuvre. Qt fait une abstraction de ce mécanisme via la classe QFileSystemWatcher. Un objet de cette classe émet un des signaux fileChanged ou directoryChanged selon la nature de l'entité du système de fichier surveillé.

fileChanged est émis lorsque le fichier surveillé a été modifié, renommé ou supprimé du disque. Qt ne permet pas de savoir laquelle de ces actions est ç l'origine du signal ;
directoryChanged est émis lorsqu'un répertoire ou son contenu est modifié ou supprimé.

Le principal intérêt de la classe QFileSystemWatcher est de déléguer au système d'exploitation la surveillance. Cela évite la mise en place d'un boucle de scrutation consommatrice de temps processeur. Elle a cepandant plusieurs limites importantes :

la surveillance d'un répertoire n'est pas récursive ;
on ne peut pas mettre en pause la surveillance ;
on ne peut pas savoir quelle est l'action à l'origine du signal émis.

En fin la lecture d'une file est nécessaire pour au moins 2 consommateurs. On ne peut donc pas purger un fichier en file d'attente tant que l'ensemble des consommateurs qui y accèdent ne l'ait lu. Cela impose de gérer la file d'attente via une table qui référence les consommateurs et met à jour, pour chaque fichier, leur consommation par ces derniers. C'est donc la file qui gère cette table et la purge des fichiers. Observons alors le cycle de vie d'un fichier :

Le producteur, lui, n'a pas le temps de gérer la file. Il se contente de déposer les fichiers sur le répertoire qu'elle gère. La file est prévenue par l'observateur du système de fichier d'une modification de son contenu. A la réception du signal directoryChanged elle liste les fichiers du répertoire, et ajoute ceux qui n'existe pas dans la table.

Protection des fichiers déposés.

Notre file d'attente sera un objet issu d'une classe que l'on peut nommer FileQueue. Cette classe dispose d'un, objet QFileSystemWatcher à l'écoute du signal directoryChanged qui est le seul qui nous intéresse ici.

Le premier risque que nous rencontrons est l'effacement accidentel de la liste d'attente sur disque. Cela peut être réalisé non seulement par l'application elle-même mais aussi par une application tierce dont une commande rm ou unlink.

Comme l'application est destinée en priorité à des systèmes Linux, nous allons en premier lieu nous intéresser à ce système. Linux propose une grande variété de solutions pour protéger les fichiers d'un répertoire :

une politique de droits mais cela suppose d'ajouter un compte spécifique à l'application ;
la mise en place du sticky bit qui interdit toute suppression autre que par l'utilisateur propriétaire du répertroire, l'utilisateur propriétaire du fichier ou du compte root. Là encore, il faut ajouter un compte spécifique à l'application qui devient "propriétaire" du répertoire de la file d'attente et de ses fichiers ;
rendre le répertoire immutable via la commande chattr +i. Toutefois, cette commande nécessite généralement des droits root ce qui peut la rendre dangereuse (le prototype C immutablefile) montre que ce droit est attaché à la fonction ioctl(fd, FS_IOC_SETFLAGS, &flags) lorsque flags contient FS_IMMUTABLE_FL.

Pour éviter d'écrire du code spécifique à une plate-forme, nous allons choisir une protection "externe" du répertoire via le mécanisme de sticky bit. Pour cette raison, l'application devra être lancée avec un compte spécifique (par exemple pimail).

Mise en oeuvre de la file d'attente disque.

Nous devons donc gérer la fille d'attente disque via une classe que nous nommons FileQueue. Son rôle est de maintenir une table dont chaque entrée est un nom de fichier du répertoire surveillé associé à une liste de booléens appelés "booléens de consommation". En effet, l'état de chaque booléen reflète la consommation ou non du fichier par l'objet consommateur de même indice que le booléen.

Dès que le signal directoryChanged est reçu de l'OS via la classe QFileSystemWatcher, la classe compare le contenu du répertoire à cette table et la met à jour en conséquence :

Chaque nouveau fichier est ajouté à la table avec l'ensemble des booléens de consommation à false. Le signal filesInserted(const QStringList&) est émis (la liste est celle des nouveaux fichiers).
La liste des fichiers supprimés est créée. Chaque fichier de cette liste est ôté de la table quelque soit l'état des booléens de consomation. Enfin, le signal filesDeleted(const QStringList&) est émis.
La liste des fichiers consommés par l'ensemble des consommateurs est dressée. Chaque fichier de cette liste est supprimé. Lee signal filesUsed(const QStringList&) est émis.

Plusieurs remarques sont à faire ici pour implémenter les slots des objets consommateurs :

Le nom des fichiers de la file d'attente ne sont transmis qu'une seule fois par voie de signal mais il est toujours possible de demander à la classe FileQueue de transmettre le contenu de sa table des fichiers non consommés par l'objet consommateuyr qui l'invoque.

Le slot consommateur doit s'assurer avant de la traiter que le fichier existe encore. Même si des protections du niveau OS sont mises en place pour protéger les fichiers de la file d'une suppression accidentelle, elle reste néanmoins possible.

Fonctionnement du mécannisme signal/slot en environnement multi-thread.

Le grand intérêt de l'utilisation des signaux Qt et leur capacité à pouvoir être émis et reçus par des threads différents. Il ya cepandant des précautions à prendre.

Chaque objet dérivé de la classe classe QObject s'exécute sur un thread précis. La relation entre l'objet et le thread est appelée thread affinity. La méthode QObject::thread() permet de connaître le thread sur lequel s'exécute l'objet. La méthode QObject::moveToThread() permet de changer cette affinité en affectant l'exécution des méthodes de l'objet à un autre thread.

ATTENTION : l'affinité de l'objet à un thread n'implique pas que les méthodes s'exécutent nécessairement dans ce thread : un objet ne s'exéute pas dans un thread, il a une affinité avec un thread. Dans l'exemple ci-dessous :

obj->moveToThread(thread);
obj->doSomething();

l'objet ne s'exécute pas dans le thread de l'affinité fixe par moveToThread() mais dans le thread qui invoque la méthode doSomething().

L'affinité intervient dans différents modes d'utilisation :

le couples signaux/slots : connect(pA, &a::sigA, pB, &b::slotB), fait que le slot p::slotB s'exécute dans le thread pB->thread() ;
les évènements : QCoreApplication::postEvent(obj, event);, fait que l'évènement event est traité dans le thread de l'affinité de obj (obj->threda()) ;
Les chronomètres (QTimer) : QTimer* pT = new QTimer(pObj);, fait que le timer s'exécute dans l'affinité de pObj (pObj->thread()).

Ceci explique le modèle de conception des threads recommandé par Qt.

// Création
QThread* pThread = new QThread;
Worker* pWorker = new Worker;     // Affinité de pWorker : QThread::currentThread()
// Déplacement (changement de l'affinité)
pWorker->moveToThread(pThread);   // Affinité de pWorker : pThread
// Traitement Worker::process() exécuté dans le thread pThread
connect(pThread, &QThread::started, pWorker, &Worker::process);

Un objet enfant a obligatoirement l'affinité de son thread parent : il ne peut pas en chnager.

La méthode pObj->deleteLater() poste un évènement de destruction dans le thread d'affinité de pObj. Cela évite la destruction redondante (cross-thread) et les bugs liés à la concurrence.

Le plus simple est de garder à l'esprit que QObject::thread() signifie Dans quel thread cet objet reçoit-il ses évènements et exécute-t-il ses slots ?

Voici un exemple de communication par signal inter-thread :

class Worker : public QObject {
    Q_OBJECT

signals:
    void resultReady(QString text);

public slots:
    void doWork() {
        emit resultReady("Travail terminé");
    }
};

QThread* pThread = new QThread;

Worker* pWorker = new Worker;       // affinité pWorker = thread principal
worker->moveToThread(pThread);      // affinité pWorker = pThread
Receiver* pReceiver = new Receiver; // affinité pReceiver = thread principal
// slot exécuté dans le thread principal
QObject::connect(pWorker, &Worker::resultReady, pReceiver, &Receiver::handleResult);
pThread->start();
// Exécuté dans le thread pThread (signal émis depuis pThread)
QMetaObject::invokeMethod(pWorker, "doWork");

Notez que par défaut, le mode de connexion des signaux est Qt::AutoConnection. Dans ce mode, Qt décide automatiquement :

Dans le cas inter-thread, le slot sera exécuté dans le thread du receveur de signal. Lorsque l'émission et la réception ne sont pas dans le même thread (Qt::QueuedConnection), le signal est émis via un évènement posté dans la boucle événmentielle du thread récepteur. De ces connaissances on extrait 2 conditions importantes pour le fonctionnement des signaux inter-thread :

Après cette présentation du fonctionnement des signaux en contexte multi-thread, on peut s'intérresser à la répartition des traitements lors de la mise en oeuvre de notre file d'attente.

Organisation des traitements.

schématiquement, la file d'attente s'insère entre un et plusieurs producteurs et un ou plusiseurs consommateurs comme indiqué sur le schéma qui suit :

Cette vision rend compte de l'organisation fonctionnelle mais pas de celle des traitements. Ici, cette organisation est importante puisque l'on créé la file d'attente pour "lisser" les différences de vitesses de traitements entre producteurs et consommateurs (vitesse qui pour les uns comme pour les autres paut également être variable). Le schéma qui suit montre l'organisation des traitements lorsque file d'attente, producteur et consommateur appartiennent au même processus :

Un processus multithread est à la fois compliqué à concevoir et encore plus à tester et mettre au point. Nous avons donc créer une application prototype filequeue où le producteur créé et dépose de petits fichiers après le déclenchement d'un tic horloge.

Dans la réalité de notre application, le producteur sera le lecteur et le consommateur l'analyseur.