Communication inter thread.

Bonjour a tous!

Avant tout, je précise, je suis sur du linux, j'utilise les pthread POSIX, et je travail sur de l'embarquer, ce qui me coupe de la plupart des grosses librairies. Passons aux problèmes :)

J'ai des petits soucis de conception pour le programme que je suis en train de faire. Pour faire rapide, j'ai un processus principale (issu d'un fork, mais c'est pas l'essentiel ici) qui lance des threads avec chacun une tache bien definie. Jusque la, ca va. Ces threads sont en fait a la chaine, et doivent traiter des trames, puis les stacker dans la FIFO du prochain thread (pas toujours dans le même ordre).

Le truc, c'est que dans ma conception, je me suis dit que j'allais avoir une série de listes (qui représenteraient mes FIFO) dans mon thread principal, et que grâce a la mémoire partager entre threads, je pourrai faire des push_back pour remplir les bonnes fifos, qui seront pop_front dans le thread qui les traitent. (Je sais pas si je suis bien clair)

Je suis passe en phase de tests pour essayer de faire fonctionner l'ensemble. Et je me retrouve avec des gros problèmes pour faire passer mes listes entre mes threads! En effet, on ne peut faire passer qu'un argument a la fonction a exécuter a la création du thread (pthread_create), et ca me bloque énormément (sans compter les problèmes entrainer par l'obligation de passer par du void*).
Je dois surement m'y prendre très mal... J'ai trouver très peu de doc sur la communication inter-thread linux pour m'orienter.


Est ce que quelqu'un serait susceptible de me donner des pistes pour faire communiquer plus facilement mes threads? Un tuto, une lib vraiment légère?
Je suis preneur de n'importe quelles aides ou infos que vous pourriez me donner!

Merci d'avance!

(Si vous voyez des fautes de francais aussi, corrigez moi, j'essaye de plus en faire ^^)

Salut,

En effet, tu ne peux faire passer qu'un seul argument lorsque tu crée un thread, mais ça ne pause aucun soucis car tu peux très bien passer une structure, une classe...

D'après ce que je comprend de ton problème je pense que le mieux serait de faire une classe "Liste" par exemple qui contient un conteneur (std::vector ou autre selon ton besoin) et tu défini des méthodes Thread-Safe push pop etc..

Pour cela dans ta classe tu déclare un pthread_mutex, que tu vérouille quand tu rentre dans une méthode et que tu libères quand tu en sort.

Ainsi dans tes threads tu fait passer un pointeur sur cette classe, et tu peux faire toutes les opérations que tu veux sans soucis.

J'espère avoir bien saisis ton problème.

Cordialement,
NeoKript

Ce serait une solution, en effet! Et que penserais tu de declarer les std::list en global, et de locker leur utilisation avec les mutex? Je sens qu'il y a un danger a faire comme ca, mais ca pourrait aussi marcher, non?

C'est une solution envisageable..

Mais je pense qu'elle est en effet plus dangereuse et puis je n'aime pas trop les variables globales, tant qu'on peut éviter vaut mieux le faire.

Surtout que si tu déclare ta variable en globale, si tu as qu'une liste ça va encore mais si tu en a plusieurs tu va avoir 2 variables globales par liste (une pour le mutex et une pour la liste)...

Vu que d'après ce que tu dis, tu à plusieurs listes, je penses que le mieux est de faire une classe.

Apres le truc, c'est que les threads sont lancer pour longtemps. J'ai un controller, puis je lance 4 thread, qui vont se faire passer des trames entre eux, avec une FIFO (list) pour chaque thread, pour ne perdre aucune information.

Chaque thread doit aller chercher une trame dans sa FIFO, la traiter, et l'envoyer dans la FIFO du bon thread pour continuer le traitement (ca serait plus clair avec un schema). Et ca jusqu'a ce que la trame soit envoyer a un autre processus.

Dans ce modele la, j'ai un peu du mal a voir comment je vais pouvoir me faire passer ma classe de traitement de liste, et que les thread fonctionne entre eux de manière dynamique. Etant donner qu'une fois que les thread sont lancer, il font leurs traitements en continu, sans repasser par le controller.

C'est un cas d'utilisation un peu particulier, par ce qu'on lance pas un thread pour continuer tranquil sur le thread principal, mais on a un thread principale qui lance les threads de traitements et qui verifie que tout ce passe bien. Il ne fait pas ou peu de traitement lui meme, en dehors des threads qu'il lance.

Je veux bien un peu plus de details sur comment tu ferais ^^ Ou d'autres personnes, si jamais :)

Merci pour les reponses en tout cas!

Alors en fait si j'ai bien compris, ton thread principal lance plusieurs threads, chaque thread est dédié à une tâche précise et donc une trame est par exemple traitée dans le premier threads, une fois que le threads à traiter la trame il l'envoi au thread 2 qui à son tour traite la trame...Etc.. Et ce jusqu'à que la trame soit envoyer à un autre procesus ? Je suppose qu'une trame c'est un buffer de char ?

C'est ça dans l'idée ?

Du coup si c'est ça, il faut seulement faire passer la trame d'un thread à l'autre,
une méthode simple consisterait :
- Avoir une classe pour chaque traitement, toute les classes de traitements aurait une FIFO contenant les trames à traiter ainsi qu'un Mutex pour que la communication soit Thread-Safe, ainsi tu aurait juste une méthode pour ajouter une trame à la FIFO.

Code:

---

1 2 3 4 5 6 7 8 9 10

class Traitment { private: pthread_mutex_t _mutex; std::vector<String> _fifo;   public: void AddTrame(String Trame); };

---

Donc ça, ca pourrait être un exemple de classe type.
- Avoir une classe unique (utiliser le Design Patter Singleton). Cette classe gérerait les threads, c'est à dire qu'elle créerai les classes de traitement et les exécuterait dans un Thread, elle stockerait également les pointeurs vers les classes de traitement dans une Map par exemple au moins tu pourrais faire quelque chose comme ca MaClasseSingleton::Instance.DispatchTrame(String Trame, String TreatmentUnit); En dans cette méthode, tu aurais juste à faire this._map[TreatmentUnit].AddTrame(Trame).

Je ne sais pas si je suis très clair.

Mais ainsi à partir de n'importe quel thread (classe de traitment), tu pourrais communiquer avec n'importe quel autre, ajouter des élément dans ça FIFO etc...

Si tu ne comprend pas dis moi, je te ferais un exemple si j'ai le temps.

Salut,
Les variables globales vont introduire un fort couplage entre tes différents threads ce qui à la longue se révèle toujours être problématique et handicapant. Donc un conseil : oublis.

De ce que je comprends, tu fais une sorte de chaînage (un peu comme un pipeline) avec n thread qui reçoivent une commande, la traîte et envoi le résultat au suivant.
Le mieux serait d'encadrer tout ça dans des classes au bon niveau de services.
Il est dommage que tu ne puisses (veuilles?) utiliser des bibliothèques car cela va impliquer de refaire à la main des choses existantes. Si tu es sur linux, c'est déjà que ton embarqué n'est pas trop contraint (sinon, tu aurais un os réduit à un ordonnanceur type OS temps réel).
Ce que je vois :
=> une classe pour encapsuler les thread posix : soit elle est abstraite et les classes dérivées implémentent une fonction virtuelle pure de type Run soit elle est générique et prend un 'callable' pour l'exécution. Construire une classe pour tes threads te permettra d'avoir un contexte plus riche pour ceux-ci et t'éviteras de te poser des question sur des void* ;)

=> probablement mettre en place un design pattern de type command pour transiter tes traitements et/ou données d'un thread à l'autre

=> construire un système de boîtes aux lettres avec des concurrent queues. Cherche sur ces mots clés pour plus de détails. Autres mots clés intéressant : pattern producer/consumer ou reader/writter; work-queues ; active objects.

Bref, commence par faire une bonne dose de conception avant de te lancer là dedans histoire de partir sur des bases solides.

Quelques bibliothèques intéressantes à potasser pour au - t'inspirer de ce qu'elles peuvent proposer : Boost.Thread, ACE, POCO, ICE. Et bien sûr de fouiller du côté du blog de sutter.

Je vais regarder toute vos reponses a tete reposer.

@neokript: oui c'est ca dans l'idee. J'ai plusieurs processus qui communiquent entre eux. Un de ces processus recoit des trames a traiter a travers un named pipe. Il les transforme pour les uniformiser, et les passe au bon thread pour les traiter, et ca termine sur un dernier thread charger de les retransformer pour les renvoyer a un autre processus. Et effectivement, les trames sont des string.

Et tu es assez clair, même si il faut relire plusieurs fois pour être bien sur de comprendre ^^ Le seul point noir, c'est comment est ce que tu fait passer ces classes entre les thread? Etant donner qu'avec le POSIX, tu peux seulement faire passer 1 arguments (et en le castant en void* en plus), j'ai du mal a voir comment faire pour tout faire transiter correctement!

@3DArchi: C'est pas que je veux pas, sinon j'aurais utiliser boost pour la gestion des threads. Je travail sur un linux embarquer (une busybox) le tout reduit a son minimum pour prendre le moins de place possible. Donc je ne ferme pas la porte a une lib légère et facile a intégrer, mais de base, je pars du principe que je n'en utiliserais pas. Donc boost pas possible ^^ Apres si tu connais une ptite lib pour la gestion des threads, je suis plus que preneur!

Je vais regarder plus en avant les pattern dont tu me parle en tout cas. Mais comme pour neokript, j'ai du mal du mal a voir comment faire transiter toutes ces infos entre les threads...


Merci de vos interventions en tout cas!

Citation:

---

Envoyé par Baradhur

Je vais regarder toute vos reponses a tete reposer.

Etant donner qu'avec le POSIX, tu peux seulement faire passer 1 arguments (et en le castant en void* en plus), j'ai du mal a voir comment faire pour tout faire transiter correctement!

---

Il y a quelque chose que tu n'as pas du saisir dans mon explication.

Donc en fait le principe :
- Toutes tes classes de traitement hériterais de la classe TreatmentBase qui serait par exemple :

Code:

---

1 2 3 4 5 6 7 8 9 10

class TraitmentBase { private: pthread_mutex_t _mutex; std::vector<std::tring> _fifo;   public: void Run() = 0; void AddTrame(String Trame) = 0; };

---

Ainsi la méthode Run lancerais un while infini qui vérifiais si la pile contient une données si c'est le cas, elle dépile un élément le traite l'envoi à une autre unité de traitement. Si la pile est vide, tu peux faire un sleep de quelque milliseconde afin de ne pas surcharger le CPU.
- Ensuite, tu aurais une classe qui gères les Threads, c'est classe serait un Singleton, (une seule instance) et contiendrais une std::map<std::string, TreatmentBase *> _map; ainsi lorsque tu instancie ton unité de traitement, tu la rajoute à la map comme ceci :

Code:

---

1 2 3 4

TraitmentBase *T = new TraitementSpecifique1(); this->_map["Mon traitement 1"] = T; T.Run();

---

Il ne reste plus qu'à faire une méthode qui te permet de récupérer une unité de traitement TreatmentBase *GetTreatment(std::string TreatmentName) qui te retourne le TreatmentBase associé au nom TreatmentName.

Ainsi lorsque tu est dans l'unité de traitment1 et que tu veux envoyer une trame à l'unité de traitement 2, il suffit de faire quelque chose du genre :

Citation:

---

GestionTreatmentUnit::Instance->GetTreatment("Traitment 2")->AddTrame("ta trame a envoyer a l'unite de traitment 2")

---

Voilà, j'espère que tu comprend mieux.
Par contre, je suis d'accord avec 3DArchi, utiliser un des design pattern qu'il a mentionné serait surement plus propre, après l'utilisation d'un classe singleton, ce n'est pas comme ci tu avais des variables globales de partout.

Rapidement :

Citation:

---

Envoyé par NeoKript

tu peux faire un sleep de quelque milliseconde afin de ne pas surcharger le CPU.

---

==> à éviter. C'est une attente active même si tu fais un sleep. On peut construire des FIFO concurrente avec attente bloquante en utilisant les objets du multithreading/processing

@Baradhur : as-tu fait un bench de boost.thread pour voir son impact mémoire ? Juste tu fait un petit proto avec boost.thread de ce que tu veux faire, tu génères et tu regardes ce que ça fait.

Citation:

---

Envoyé par 3DArchi

Rapidement :

==> à éviter. C'est une attente active même si tu fais un sleep. On peut construire des FIFO concurrente avec attente bloquante en utilisant les objets du multithreading/processing

---

Oui mais du coup, obliger de passer par une bibliothèques (ce qu'il souhaite éviter visiblement), car la lib pthread ne contient pas d'implémentation de piles concurrentes, où alors je ne suis pas au courant.

Citation:

---

Envoyé par NeoKript

Oui mais du coup, obliger de passer par une bibliothèques (ce qu'il souhaite éviter visiblement), car la lib pthread ne contient pas d'implémentation de piles concurrentes, où alors je ne suis pas au courant.

---

Je connais mal POSIX mais il me semble qu'avec des conditions variables tu dois pouvoir éviter les sleep.

Un tutoriel sur les threads POSIX : Initiation à la programmation multitâche en C avec Pthreads, par Franck Hecht

Pour ce qui est des lib et de la memoire, j'ai une limitation sur la RAM, et sur l'espace memoire sur le disque. Par ce que j'ai pas que des binaires C++ qui tournent dessus, j'ai aussi un serveur lighttpd (avec des cgi en c++) et une bdd SQLite (sans compter l'instal de la busybox aussi). Et je dois faire rentrer ca dans un pti trou de souris ^^
Donc j'evite au maximum tout ce qui peut prendre la place, et tout ce qui pourrait etre trop "RAMovore". Sinon, c'est sur qu'avec boost ca serait plus simple.

Pour tout ce que vous m'avez propose, ca va me prendre un peu de temps pour regarder tout, par ce que j'ai que 2 jours par semaines a y consacre (alternance powa) et c'est le jeudi vendredi. D'ici la, je pourrais pas m'y plonger a 100%. Mais je reste tout de meme tres attentif :)

Merci a vous!

Citation:

---

Envoyé par Baradhur

Pour ce qui est des lib et de la memoire, j'ai une limitation sur la RAM, et sur l'espace memoire sur le disque. Par ce que j'ai pas que des binaires C++ qui tournent dessus, j'ai aussi un serveur lighttpd (avec des cgi en c++) et une bdd SQLite (sans compter l'instal de la busybox aussi). Et je dois faire rentrer ca dans un pti trou de souris ^^
Donc j'evite au maximum tout ce qui peut prendre la place, et tout ce qui pourrait etre trop "RAMovore". Sinon, c'est sur qu'avec boost ca serait plus simple.

Pour tout ce que vous m'avez propose, ca va me prendre un peu de temps pour regarder tout, par ce que j'ai que 2 jours par semaines a y consacre (alternance powa) et c'est le jeudi vendredi. D'ici la, je pourrais pas m'y plonger a 100%. Mais je reste tout de meme tres attentif :)

Merci a vous!

---

Attention ( pour l'avoir souvent vu) au raisonnement : je n'utilise pas WonderFullLibrary parce que ça prend trop de place. Donc je refais à la main. Au final : sizeof(ma_solution)> sizeof(WonderFullLibrary) && bug(ma_solution)> bug(WonderFullLibrary) && maintenance(ma_solution)==:cry:
C'est pour ça que je te conseille de faire un petit bench pour voir si vraiment tu ne peux t'appuyer sur quelque chose d'existant.

Par rapport aux bibliothèques proposées par 3DArchi, j'ajouterai TBB, qui possède aussi des structures de données adaptées à ce cas (une concurrent queue entre autres).

Je reviens vers vous avec la solution que je vais utilise pour l'ensemble de mon programme. Et c'est vraiment simple, je me suis pris la tete pour pas grand chose.

Je vais avoir une structure qui va contenir mes listes et mutex associes (je vais surement en faire plusieurs, pour chaque cas, histoire de pas trop mixer la memoire de partout). Et cette structure, je la malloc dans mon main, et je la passe au thread qui en ont besoin, et je la recast derriere.

Ainsi, je me retrouve avec mes fifos dans les threads qui en ont besoin, et avec une gestion des mutex intelligente, tout se passe bien. Avec toutes les classes imbriques les unes dans les autres, ca sera surement un peu plus complexe, mais le principe est a et il marche. Ca rejoins ce que neokript proposait.

Merci a tous pour vos contributions!


Au passage, si quelqu'un repasse par la, a votre avis quel IPC est le mieux pour une fifo inter-process contenant des strings? Pipe, SHM et semaphores,etc? Je suis en train de pencher pour les shm (gestion similaire a la memoire des threads), mais j'aimerais tout de meme avoir vos avis!

edit: en sachant que tout les processus seront issu d'un seul et meme processus (a coup de fork + exec) qui servira a controller que les processus marche bien, et les relancer si jamais l'un d'eux plante. voila :)

Salut,

Pour la communication inter-processus, perso j'utiliserais les pipes surtout si c'est le même processus qui fait des fork... l'utilisations de cannaux me semble plus adéquate après j'ai surtout utiliser les shm donc bon je sais pas trop. mais les pipes pour les fork ça semble pas mal.

Je sais pas si tu sais comment ça fonctionne mais en gros tu as 2 FileDescriptor (un pour l'écriture, un pour la lecture), du coup si tu connais les fonction read write close ... (que tu as peux être utiliser pour les socket ou les fichier) ba tu peux les utiliser pour envoyer et recevoir des données dans tes canaux.

Voilà bonne soirée

PS : Par rapport à ce que tu dis, on à du mal à voir si tu fais du C où du C++, parceque malloc c'est du C et là on est dans la section c++ ^^

Oui, quand je dis malloc, ca porte a confusion, c'est sur ^^ Mais je prefere dire malloc que new. Ce qui revient dans les 2 a de l'allocation dynamique. Bref ^^

Merci pour vos contributions :)

Non, attention malloc et new ne font pas la même chose, je ne te permet pas de dire ca ^^

Pourquoi ne pas passer par des events ? Ou une file de message linux ?

Citation:

---

Envoyé par Baradhur

Oui, quand je dis malloc, ca porte a confusion, c'est sur ^^ Mais je prefere dire malloc que new. Ce qui revient dans les 2 a de l'allocation dynamique. Bref ^^

Merci pour vos contributions :)

---

Et new provoque l'exécution d'un constructeur. Ce qui fait que malloc fait de l'allocation, et que new fait de l'allocation ET de l'initialisation. Le résultat n'est pas tout à fait similaire :)