Discussion :

[poukill]

Bonjour à tous,

j'aimerai lancer plusieurs traitements répétés de complexité identiques en parallèle (en général 1, 2 ou 3) pour une appli temps-réel.
La création de thread étant couteuse, j'imagine qu'il vaut mieux tout créer au début du programme !

Supposons que j'aie dans ma classe de base de calcul un vector de thread.

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
class Engine
{
//
private:
	CFPSCounter                                   m_fps_counter;
	std::vector<boost::shared_ptr<boost::thread>> m_execution_threads;
};

Je créé ensuite autant de threads que de traitements (2 par exemple).
Toutes les 20 ms, il faut que je relance les 2 traitements en parallèle.
Exemple :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
 
m_execution_threads[0].reset(new boost::thread(
		boost::bind(&IVisionModule::Apply, m_vision_chains[0])));
m_execution_threads[1].reset(new boost::thread(
		boost::bind(&IVisionModule::Apply, m_vision_chains[1])));

Du coup, c'est pas terrible : ça risque d'être lent mon affaire... Le new me recréé un thread à chaque fois...

Vous avez une idée pour donner juste du travail à des threads?

P.S : J'ai pensé à boost::asio qui émule l'asynchronicité via un pool de thread (boost::asio::post() ), mais je sais pas si c'est bien performant tout ça. J'ai une limite de "souplesse" à échanger contre de la performance...

[3DArchi]

Salut,
Il me semble qu'il y a un pattern dans ACE qui pourrait faire ton affaire. Je crois que c'est l'active object pattern.
En version hyper simplifié, l'idée c'est que ton thread fonctionne comme une boucle de message. Il est en attente qu'une commande soit enfilée dans la boucle. Puis il la sort, l'exécute et se remet en attente d'une nouvelle commande. En fait, le thread n'est jamais quitté.

[Mac LAK]

Pour ça, j'associe en général un sémaphore à chaque thread, plus une adresse mémoire de passage d'un bloc de données.
Au signalement, le thread travaille sur le bloc fourni, retourne le résultat dans une zone prévue et lève un signal quelconque (flag, sémaphore, tout dépend du contexte). Il est assez difficile de faire plus rapide, vu que l'on réduit les opérations inutiles à un maximum presque absolu.

Une autre variante est d'alimenter par une FIFO un pool de threads avec des données à traiter, les résultats étant renvoyés dans une autre FIFO. Ce mécanisme fonctionne plutôt bien avec des volumes importants de données.

[poukill]

Salut,
Il me semble qu'il y a un pattern dans ACE qui pourrait faire ton affaire. Je crois que c'est l'active object pattern.
En version hyper simplifié, l'idée c'est que ton thread fonctionne comme une boucle de message. Il est en attente qu'une commande soit enfilée dans la boucle. Puis il la sort, l'exécute et se remet en attente d'une nouvelle commande. En fait, le thread n'est jamais quitté.

Salut 3DArchi, j'étais justement tombé sur ce pattern quelques jours auparavant : http://blog.cplusplus-soup.com/2006/12/boost.html
Cet exemple utilise Boost.Asio. Je sais pas si c'est le plus indiqué pour du temps réel, mais bon.

Pour ça, j'associe en général un sémaphore à chaque thread, plus une adresse mémoire de passage d'un bloc de données.
Au signalement, le thread travaille sur le bloc fourni, retourne le résultat dans une zone prévue et lève un signal quelconque (flag, sémaphore, tout dépend du contexte). Il est assez difficile de faire plus rapide, vu que l'on réduit les opérations inutiles à un maximum presque absolu.

Une autre variante est d'alimenter par une FIFO un pool de threads avec des données à traiter, les résultats étant renvoyés dans une autre FIFO. Ce mécanisme fonctionne plutôt bien avec des volumes importants de données.

Pour associer un sémaphore à chaque thread, j'imagine que tu es dépendant de l'OS? Je connais encore mal cette façon de programmer (mes souvenirs de sémaphore remonte 6 ans en arrière quand j'avais fait un peu de VxWorks à l'école, je n'ai jusqu'à présent utilisé que les mutex dans mon boulot professionel), alors j'ai un peu de mal à me faire une idée. Si tu pouvais m'envoyer un mot clé ou un lien vers lequel je puisse me tourner pour me documenter ce serait sympa.
Le pool de threads avec FIFO, je vois beaucoup mieux ce que ça représente. J'ai trouvé une implémentation avec threadpool (Boost non officielle). Le principe est de choisir un nombre de threads de travail à l'initialisation. Puis de leur donner du boulot. La politique peut etre changé (FIFO, LIFO, priority). Moi ce serait plutôt FIFO ici, comme tout va vraiment se passer en parallèle.
L'active object avec Boost.Asio pourrait faire l'affaire. J'avoue ne pas avoir d'idées concernant la performance.

Surtout, si vous avez des idées, des remarques sur ce que je viens de dire, n'hésitez pas !

[Mac LAK]

Pour associer un sémaphore à chaque thread, j'imagine que tu es dépendant de l'OS? Je connais encore mal cette façon de programmer (mes souvenirs de sémaphore remonte 6 ans en arrière quand j'avais fait un peu de VxWorks à l'école, je n'ai jusqu'à présent utilisé que les mutex dans mon boulot professionel), alors j'ai un peu de mal à me faire une idée. Si tu pouvais m'envoyer un mot clé ou un lien vers lequel je puisse me tourner pour me documenter ce serait sympa.

Non, pas du tout au contraire. J'utilise des librairies d'abstraction à ce niveau, comme ACE ou POCO, qui fournissent une interface commune quel que soit l'OS. Voilà la page des sémaphores pour POCO, par exemple.

En fait, je définis dans une structure toutes les données nécessaires au thread pour pouvoir bosser (sémaphores, flags, buffers, instances de classes, etc.), sans recopies bien entendu, et c'est cette structure que je passe au thread en paramètre. Chaque thread a besoin de deux sémaphores : un qui lui est spécifique, et un global pour la distribution du boulot.
A l'initialisation, le thread-père va initialiser la structure, lancer le thread et attendre le sémaphore propre du thread.
Le thread, lui, va si nécessaire copier la structure (qui peut être sur la pile du thread-père, donc "volatile") dans ses propres variables locales, finir son initialisation, signaler son propre sémaphore (=> libération du père), et ensuite attendre le sémaphore global (= du boulot vient d'arriver).

Cette technique est un peu lente au niveau de la création des threads (enfin... tout est relatif, hein, on peut en créer quelques milliers en une seule seconde ! ), mais permet ensuite d'avoir des threads immédiatement disponibles pour le travail. En général, j'arrête les threads via un flag global (demande "propre" de terminaison), suivi d'un signalement de tous les sémaphores. Ne reste plus qu'à attendre la fin des threads, avec flinguage brutal après un certain délai si c'est nécessaire.

Pour l'attente du boulot, c'est très simple : le sémaphore contient autant de jetons que de données en attente, donc si tu pousses 200 "paquets" vers le pool de threads, tu "crédites" le sémaphore de 200. Chaque thread va "consommer" une unité en se libérant, traiter un paquet, puis de nouveau attendre le sémaphore. S'il y a des données, l'attente est nulle. Sinon, le thread va être bloqué jusqu'à réception de nouvelles données. Quand le sémaphore contient zéro jetons, tous les threads sont en attente, et la FIFO source est vide.

Le pool de threads avec FIFO, je vois beaucoup mieux ce que ça représente. J'ai trouvé une implémentation avec threadpool (Boost non officielle). Le principe est de choisir un nombre de threads de travail à l'initialisation. Puis de leur donner du boulot. La politique peut etre changé (FIFO, LIFO, priority). Moi ce serait plutôt FIFO ici, commei tout va vraiment se passer en parallèle.

Le gros problème est de savoir si le résultat des threads doit être "ordonné" ou non : en effet, tu ne peux pas savoir à l'avance quel thread aura terminé en premier.
Donc, si les paquets "source" doivent être traités de façon à ce que les paquets "résultat" aient le même ordre, il faut parfois utiliser un étage de plus à la sortie des threads pour rétablir l'ordre.
Pour ma part, j'utilise le plus souvent un numéro séquentiel unique généré lors de la récupération d'un paquet "source", plus une FIFO unitaire de sortie pour chaque thread. Ensuite, un sérialiseur connecte toutes les FIFO des threads, attends "le" bon numéro, et sérialise dans le bon ordre vers la FIFO finale.

L'active object avec Boost.Asio pourrait peut être aussi faire l'affaire. J'avoue ne pas avoir d'idées concernant la performance.

Je sais qu'en général, pour ce genre de choses, je prévois des systèmes très optimisés pour garantir les performances. Les FIFO, par exemple, sont gérées sans mutex, j'use et abuse des fonctions atomiques (type InterlockedExchange, InterlockedIncrement, etc.), et je ne fais bien entendu passer que des pointeurs. Je rends les données indissociables entre elles par des ajouts systématiques de structures d'encapsulation, afin de ne toujours avoir besoin que d'un seul pointeur à manipuler.
En général, si tu ne mets pas de mutex inutiles, tes performances ne sont conditionnées que par peu d'éléments :

• Vitesse de commutation de contexte après signalement du sémaphore.
  Bien entendu, les OS temps réel s'en sortent mieux que les OS généralistes à ce niveau. Toutefois, Windows (kernel NT) et Linux (kernel 2.6) offrent en général des performances correctes.
• Gestion réelle des threads : dépend de l'OS, bien sûr, et aussi du nombre de cœurs disponibles.
  En général, je dimensionne mon pool de threads par rapport au nombre de cœurs, de façon à ne pas avoir de threads "inutiles" tout en ayant une utilisation maximale du CPU.

Côté objets, un "vrai" pool de threads n'est pas forcément nécessaire (voire souhaitable...) dans un tel contexte : contrairement à l'usage possible sur un serveur, par exemple HTTP ou BDD, il n'y a pas de "pics" de charge à assurer, ni de fluctuations très importantes dans la répartition de la charge. Soit tes threads bossent, soit ils ne foutent rien, mais tu n'as pas de grosses contraintes de répartition à faire. Du coup, un simple tableau d'objets-threads suffit en général amplement.

L'autre point critique, ce sont les FIFO : il est préférable de bien travailler leurs performances, et de tenter au maximum d'éviter l'utilisation de mutex. Souvent, on peut se passer de mutex sur une des extrémités de la FIFO (soit côté application, soit côté threads de travail) : c'est toujours bon à prendre, et de préférence côté threads de travail.
Toutefois, cela demande pas mal de boulot de faire ça, et les FIFO "toutes prêtes" sont rarement idéales.

[poukill]

Tout d'abord, merci beaucoup pour ta réponse, il m'a fallut un peu de temps pour la "digérer". Tant mieux.

Non, pas du tout au contraire. J'utilise des librairies d'abstraction à ce niveau, comme ACE ou POCO, qui fournissent une interface commune quel que soit l'OS. Voilà la page des sémaphores pour POCO, par exemple.

J'y ai jeté un coup d'oeil, c'est excellent. J'en avais juste entendu parlé, mais jamais utilisé. J'ai quelques frissons à introduire de nouvelles lib dans le projet (Qt, Vigra, Xeumeuleu, Boost, ...). Celà dit, la qualité étant au rendez-vous, ça ne posera pas non plus trop de problèmes.

En fait, je définis dans une structure toutes les données nécessaires au thread pour pouvoir bosser (sémaphores, flags, buffers, instances de classes, etc.), sans recopies bien entendu, et c'est cette structure que je passe au thread en paramètre. Chaque thread a besoin de deux sémaphores : un qui lui est spécifique, et un global pour la distribution du boulot.
A l'initialisation, le thread-père va initialiser la structure, lancer le thread et attendre le sémaphore propre du thread.
Le thread, lui, va si nécessaire copier la structure (qui peut être sur la pile du thread-père, donc "volatile") dans ses propres variables locales, finir son initialisation, signaler son propre sémaphore (=> libération du père), et ensuite attendre le sémaphore global (= du boulot vient d'arriver).

OK, je vois l'esprit. Je connaissais pas du tout, alors je prétend pas tout saisir. C'est peut être un peu "compliqué" pour ce que j'essaye de faire, mais je le note dans un coin de ma tête.

Le gros problème est de savoir si le résultat des threads doit être "ordonné" ou non : en effet, tu ne peux pas savoir à l'avance quel thread aura terminé en premier.
Donc, si les paquets "source" doivent être traités de façon à ce que les paquets "résultat" aient le même ordre, il faut parfois utiliser un étage de plus à la sortie des threads pour rétablir l'ordre.
Pour ma part, j'utilise le plus souvent un numéro séquentiel unique généré lors de la récupération d'un paquet "source", plus une FIFO unitaire de sortie pour chaque thread. Ensuite, un sérialiseur connecte toutes les FIFO des threads, attends "le" bon numéro, et sérialise dans le bon ordre vers la FIFO finale.

Dans mon cas, l'ordre n'est pas important. Les traitements sont indépendants. De complexité équivalente, c'est tout. Après la fin des traitements, je dois pouvoir accéder aux résultats de tous les traitements, mais ça, ça ne pose pas problèmes a priori.

Je sais qu'en général, pour ce genre de choses, je prévois des systèmes très optimisés pour garantir les performances. Les FIFO, par exemple, sont gérées sans mutex, j'use et abuse des fonctions atomiques (type InterlockedExchange, InterlockedIncrement, etc.), et je ne fais bien entendu passer que des pointeurs. Je rends les données indissociables entre elles par des ajouts systématiques de structures d'encapsulation, afin de ne toujours avoir besoin que d'un seul pointeur à manipuler.
En général, si tu ne mets pas de mutex inutiles, tes performances ne sont conditionnées que par peu d'éléments :

• Vitesse de commutation de contexte après signalement du sémaphore.
  Bien entendu, les OS temps réel s'en sortent mieux que les OS généralistes à ce niveau. Toutefois, Windows (kernel NT) et Linux (kernel 2.6) offrent en général des performances correctes.
• Gestion réelle des threads : dépend de l'OS, bien sûr, et aussi du nombre de cœurs disponibles.
  En général, je dimensionne mon pool de threads par rapport au nombre de cœurs, de façon à ne pas avoir de threads "inutiles" tout en ayant une utilisation maximale du CPU.

Côté objets, un "vrai" pool de threads n'est pas forcément nécessaire (voire souhaitable...) dans un tel contexte : contrairement à l'usage possible sur un serveur, par exemple HTTP ou BDD, il n'y a pas de "pics" de charge à assurer, ni de fluctuations très importantes dans la répartition de la charge. Soit tes threads bossent, soit ils ne foutent rien, mais tu n'as pas de grosses contraintes de répartition à faire. Du coup, un simple tableau d'objets-threads suffit en général amplement.

L'autre point critique, ce sont les FIFO : il est préférable de bien travailler leurs performances, et de tenter au maximum d'éviter l'utilisation de mutex. Souvent, on peut se passer de mutex sur une des extrémités de la FIFO (soit côté application, soit côté threads de travail) : c'est toujours bon à prendre, et de préférence côté threads de travail.
Toutefois, cela demande pas mal de boulot de faire ça, et les FIFO "toutes prêtes" sont rarement idéales.

C'est vrai qu'un simple tableau d'objets-threads suffirait dans mon cas. Ma question initiale était : comment les garder actifs et en attente avant de leur donner du boulot à nouveau. Du coup, ThreadPool me convient bien a priori puisque elle implémente déjà ce pool, autant s'en servir.
Une fois que j'aurai terminé de construire mon système en entier, je verrai si je suis dans les timings ou pas. Et donc si des optimisations seront nécessaires !

Merci Mac Lak pour ces réponses pointues.