Threads / EventManager

On Apr 28, 5:18 pm, Merwin <merwin....@gmail.com> wrote:

Bon je vous explique un peu ce que j'essaye de faire:

1. J'ai un programme qui contient plusieurs Threads (avec pthread)
2. Il y a un Thread qui gère la réception de données réseaux via des
sockets Tcp et select().
3. Il a X autres Threads qui traite chacun les données réseaux leur
appartenant, ainsi que des évènements envoyés par d'autres Threads.

Je cherche donc à créer un 'EventManager', l'idée c'est que
mon Thread qui gère le réseau reçoit des données, il envoi
alors un un Evenement au thread concerné par ces données pour
lui dire qu'il a quelquechose a traiter.

En gros:

<Code éxécuté dans chaque Thread>:
{
while( true )
{
Event = EventManager.somethingHappened();
switch(Event.type())
{
case DATA_AVALAIBLE:
// Traitement des paramètres de l'év ènement etc
case ...:
...
case ...:
...
}
}
}

<Code éxécuté dans le thread des connections>:
{
...
// Données réçue sur un socket
Event e(DATA_AVALAIBLE, params...)
EventManager.postEvent(e);
...
}

J'ai vraiment du mal à gérer ça, je ne vois pas comment faire,
je pensais utiliser une variable de condition (avec
pthread_cond_wait()), mais tous les Threads ne peuvent pas
utiliser la même puisque un Evenement n'est pas forcément
adressé à tous les threads...

De plus, la gestion des paramètres est compliqué, un évènement
peut avoir un nombre de paramètres et un type de paramètres
variable selon l'événement donné...

Si vous avez des idées ! Je précise je débute et mon
implémentation est peut-être une mauvaise idée, j'attends vos
suggestions dans ce cas, merci.

Il y a plusiers aspects à considérer, comment faire
l'aiguillage, comment représenter les donneées, comment
communiquer entre les threads, etc. Ce dernier est sans doute le
plus facile : il suffit d'utiliser une queue de messages, ce
qui est facile à implémenter avec des pthread et la bibliothèque
standard :

template< typename T >
class MessageQueue
{
public:
void send( std::auto_ptr< T > message )
{
ScopedLock l( mutex ) ;
queue.push_back( message.get() ) ;
message.release() ;
pthread_cond_signal( &cond ) ;
}

std::auto_ptr< T > receive()
{
ScopedLock l( mutex ) ;
while ( queue.empty() ) {
pthread_cond_wait( &cond, &mutex ) ;
}
std::auto_ptr< T > result( queue.front() ) ;
queue.pop_front() ;
return result ;
}
private:
pthread_mutex_t mutex ;
pthread_cond_t cond ;
std::deque< T* > queue ;
} ;

(Il faut évidemment un constructeur, un destructeur, et
peut-être de la logique pour assurer un arrêt propre.)

La question ensuite, c'est où et comment tu fais l'aiguillage.
Mais chaque thread worker attend sur un MessageQueue différent,
avec éventuellement un type de message différent (mais les
messages peuvent être polymorphique). Ce que j'ai souvent fait,
c'est que c'est le message même qui fait le traitement, au moyen
d'une fonction virtuelle processMessage ; le thread worker
resemble alors :

std::auto_ptr< AbstractMessage >
message = myQueue.receive() ;
while ( message.get() != NULL ) {
message->processMessage() ;
message = myQueue.receive() ;
}

(Pour l'arrêt propre, alors, il suffit de vider la queue, puis
envoyer un pointeur null, c-à-d :

template< typename T >
void
MessageQueue::shutdown()
{
ScopedLock l( mutex ) ;
while ( ! queue.empty ) {
std::auto_ptr< T > tmp( queue.front() ) ;
queue.pop_front() ;
}
queue.push_back( NULL ) ;
pthread_cond_signal( &cond ) ;
}

Quant à l'aiguillage, sans savoir plus sur l'application, c'est
difficile à dire. En gros, il faut que le thread qui aiguille
(soit le thread qui gère la connexion, soit un thread dédié)
décide quel thread va traiter le message, et quel est le type du
message à envoyer ; ensuite, il construit le message, et
l'envoie où il faut.

--
James Kanze (GABI Software) email:james.kanze@gmail.com
Conseils en informatique orientée objet/
Beratung in objektorientierter Datenverarbeitung
9 place Sémard, 78210 St.-Cyr-l'École, France, +33 (0)1 30 23 00 34

Merwin <merwin.irc@gmail.com> writes:

Bonjour à tous,

Bon je vous explique un peu ce que j'essaye de faire:

1. J'ai un programme qui contient plusieurs Threads (avec pthread)
2. Il y a un Thread qui gère la réception de données réseaux via des
sockets Tcp et select().
3. Il a X autres Threads qui traite chacun les données réseaux leur
appartenant, ainsi que des évènements envoyés par d'autres Threads.

Je cherche donc à créer un 'EventManager', l'idée c'est que mon Thread
qui gère le réseau reçoit des données, il envoi alors un un Evenement
au thread concerné par ces données pour lui dire qu'il a quelquechose
a traiter.
[...]
J'ai vraiment du mal à gérer ça, je ne vois pas comment faire, je
pensais utiliser une variable de condition (avec pthread_cond_wait()),
mais tous les Threads ne peuvent pas utiliser la même puisque un
Evenement n'est pas forcément adressé à tous les threads...

Il faut choisir entre deux:

- avoir une seule queue d'évènement ; dans ce cas, mettre un évènement
dans la queue est simple, le fil réseau n'a pas besoin de connaitre
la correspondance entre les évènements qu'il envoit et le fil qui
doit le traiter. Mais la collecte des évènements est plus
compliquée car chaque fil devra sélectionner les évènements dont il
a besoin (il faudrait ajouter un paramètre à getNextEvent
(somethingHappened)).

- avoir une queue par fil ; dans ce cas, envoyer un évènement est un
peu plus compliqué car il faut pouvoir déterminer la correspondance
entre l'évènement et le fil qui va le traiter (mais ça s'apparente
à l'envoi d'un message à un objet). Mais la collecte est simple,
car chaque fil a sa propre queue.


Si la correspondance entre les évènements et les fils est simple,
cette dernière solution aurait ma préférence. Un avantage d'avoir une
queue par fil est qu'il y a plusieurs queues, et donc moins de
contention.


Pour ce qui est des conditions et des mutex, ils ne dépendent pas des
fils, mais des conditions et des zone à exclusion mutuelle. En gros,
chaque queue doit être gérée indépendament.

De plus, la gestion des paramètres est compliqué, un évènement peut
avoir un nombre de paramètres et un type de paramètres variable selon
l'événement donné...

Les évènements sont typiquement classifiables avec une belle
hierarchie: fais des sous-classes à volonté!
(en C, on aurait utilisé une union).

--
__Pascal Bourguignon__

Pascal J. Bourguignon a écrit :

Merwin <merwin.irc@gmail.com> writes:

Bonjour à tous,

Bon je vous explique un peu ce que j'essaye de faire:

1. J'ai un programme qui contient plusieurs Threads (avec pthread)
2. Il y a un Thread qui gère la réception de données réseaux via des
sockets Tcp et select().
3. Il a X autres Threads qui traite chacun les données réseaux leur
appartenant, ainsi que des évènements envoyés par d'autres Threads.

Je cherche donc à créer un 'EventManager', l'idée c'est que mon Thread
qui gère le réseau reçoit des données, il envoi alors un un Evenement
au thread concerné par ces données pour lui dire qu'il a quelquechose
a traiter.
[...]
J'ai vraiment du mal à gérer ça, je ne vois pas comment faire, je
pensais utiliser une variable de condition (avec pthread_cond_wait()),
mais tous les Threads ne peuvent pas utiliser la même puisque un
Evenement n'est pas forcément adressé à tous les threads...

Il faut choisir entre deux:

- avoir une seule queue d'évènement ; dans ce cas, mettre un évènement
dans la queue est simple, le fil réseau n'a pas besoin de connaitre
la correspondance entre les évènements qu'il envoit et le fil qui
doit le traiter. Mais la collecte des évènements est plus
compliquée car chaque fil devra sélectionner les évènements dont il
a besoin (il faudrait ajouter un paramètre à getNextEvent
(somethingHappened)).

- avoir une queue par fil ; dans ce cas, envoyer un évènement est un
peu plus compliqué car il faut pouvoir déterminer la correspondance
entre l'évènement et le fil qui va le traiter (mais ça s'apparente
à l'envoi d'un message à un objet). Mais la collecte est simple,
car chaque fil a sa propre queue.


Si la correspondance entre les évènements et les fils est simple,
cette dernière solution aurait ma préférence. Un avantage d'avoir une
queue par fil est qu'il y a plusieurs queues, et donc moins de
contention.


Pour ce qui est des conditions et des mutex, ils ne dépendent pas des
fils, mais des conditions et des zone à exclusion mutuelle. En gros,
chaque queue doit être gérée indépendament.

De plus, la gestion des paramètres est compliqué, un évènement peut
avoir un nombre de paramètres et un type de paramètres variable selon
l'événement donné...

Les évènements sont typiquement classifiables avec une belle
hierarchie: fais des sous-classes à volonté!
(en C, on aurait utilisé une union).

Merci, je pense que je vais en effet partir sur le principe une
queue/objet. Pour les évènements en sous classe je veux bien, mais
comment je peux déterminer le type de l'évènement quand je le récupère?

Merwin <merwin.irc@gmail.com> writes:

Merci, je pense que je vais en effet partir sur le principe une
queue/objet. Pour les évènements en sous classe je veux bien, mais
comment je peux déterminer le type de l'évènement quand je le
récupère?

Une première solution c'est de mettre les methodes de traitement de
l'évènement dans l'évènement lui-même (comme méthode virtuelle).

Event* e=GetNextEvent();
e->process(this);

(et la méthode process de chaque sous-classe d'Event peut se réduire à
faire suivre un message à son argument:

OpenEvent::process(Recipient* r){ r->processOpenEvent(this); }
CloseEvent::process(Recipient* r){ r->processCloseEvent(this); }
SendEvent::process(Recipient* r){ r->processSendEvent(this); }


Sinon, on peut utiliser dynamic_cast, mais c'est moins beau:

Event* e=GetNextEvent();
OpenEvent* oe=dynamic_cast<OpenEvent*>e;
if(oe){ processOpen(oe); }else{
CloseEvent* ce=dynamic_cast<OpenEvent*>e;
if(ce){ processClose(ce); }else{
SendEvent* se=dynamic_cast<OpenEvent*>e;
if(se){ processSend(se); }else{
throw std::exception("Unexpected event class");}}}

On pourrait utiliser un enum et un switch, ou un index et une table
pour accéler la sélection de la méthode à envoyer, mais il faudra
quand même utiliser un dynamic_cast pour revenir à la classe exacte de
l'évènement afin d'accéder à ses champs spécifiques.

--
__Pascal Bourguignon__

Pascal J. Bourguignon a écrit :

Merwin <merwin.irc@gmail.com> writes:

Merci, je pense que je vais en effet partir sur le principe une
queue/objet. Pour les évènements en sous classe je veux bien, mais
comment je peux déterminer le type de l'évènement quand je le
récupère?

Une première solution c'est de mettre les methodes de traitement de
l'évènement dans l'évènement lui-même (comme méthode virtuelle).

Event* e=GetNextEvent();
e->process(this);

(et la méthode process de chaque sous-classe d'Event peut se réduire à
faire suivre un message à son argument:

OpenEvent::process(Recipient* r){ r->processOpenEvent(this); }
CloseEvent::process(Recipient* r){ r->processCloseEvent(this); }
SendEvent::process(Recipient* r){ r->processSendEvent(this); }


Sinon, on peut utiliser dynamic_cast, mais c'est moins beau:

Event* e=GetNextEvent();
OpenEvent* oe=dynamic_cast<OpenEvent*>e;
if(oe){ processOpen(oe); }else{
CloseEvent* ce=dynamic_cast<OpenEvent*>e;
if(ce){ processClose(ce); }else{
SendEvent* se=dynamic_cast<OpenEvent*>e;
if(se){ processSend(se); }else{
throw std::exception("Unexpected event class");}}}

On pourrait utiliser un enum et un switch, ou un index et une table
pour accéler la sélection de la méthode à envoyer, mais il faudra
quand même utiliser un dynamic_cast pour revenir à la classe exacte de
l'évènement afin d'accéder à ses champs spécifiques.

En lisant la doc de la STL, je suis tombé sur typeinfo:
http://www.cplusplus.com/reference/std/typeinfo/type_info/

Est-ce que ça ne serait pas mieux ?

Merwin <merwin.irc@gmail.com> writes:

Pascal J. Bourguignon a écrit :

Merwin <merwin.irc@gmail.com> writes:

Merci, je pense que je vais en effet partir sur le principe une
queue/objet. Pour les évènements en sous classe je veux bien, mais
comment je peux déterminer le type de l'évènement quand je le
récupère?

Une première solution c'est de mettre les methodes de traitement de
l'évènement dans l'évènement lui-même (comme méthode virtuelle).

Event* e=GetNextEvent();
e->process(this);

(et la méthode process de chaque sous-classe d'Event peut se réduire à
faire suivre un message à son argument:

OpenEvent::process(Recipient* r){ r->processOpenEvent(this); }
CloseEvent::process(Recipient* r){ r->processCloseEvent(this); }
SendEvent::process(Recipient* r){ r->processSendEvent(this); }


Sinon, on peut utiliser dynamic_cast, mais c'est moins beau:

Event* e=GetNextEvent();
OpenEvent* oe=dynamic_cast<OpenEvent*>e;
if(oe){ processOpen(oe); }else{
CloseEvent* ce=dynamic_cast<OpenEvent*>e;
if(ce){ processClose(ce); }else{
SendEvent* se=dynamic_cast<OpenEvent*>e;
if(se){ processSend(se); }else{
throw std::exception("Unexpected event class");}}}

On pourrait utiliser un enum et un switch, ou un index et une table
pour accéler la sélection de la méthode à envoyer, mais il faudra
quand même utiliser un dynamic_cast pour revenir à la classe exacte de
l'évènement afin d'accéder à ses champs spécifiques.

En lisant la doc de la STL, je suis tombé sur typeinfo:
http://www.cplusplus.com/reference/std/typeinfo/type_info/

Est-ce que ça ne serait pas mieux ?

C'est le mécanisme sous-jacent à dynamic_cast.

Le mieux, c'est la méthode virtuelle, avec retour sur l'objet
appelant.


--
__Pascal Bourguignon__

Pascal J. Bourguignon a écrit :

Merwin <merwin.irc@gmail.com> writes:

Pascal J. Bourguignon a écrit :

Merwin <merwin.irc@gmail.com> writes:

Merci, je pense que je vais en effet partir sur le principe une
queue/objet. Pour les évènements en sous classe je veux bien, mais
comment je peux déterminer le type de l'évènement quand je le
récupère?

Une première solution c'est de mettre les methodes de traitement de
l'évènement dans l'évènement lui-même (comme méthode virtuelle).

Event* e=GetNextEvent();
e->process(this);

(et la méthode process de chaque sous-classe d'Event peut se réduire à
faire suivre un message à son argument:

OpenEvent::process(Recipient* r){ r->processOpenEvent(this); }
CloseEvent::process(Recipient* r){ r->processCloseEvent(this); }
SendEvent::process(Recipient* r){ r->processSendEvent(this); }


Sinon, on peut utiliser dynamic_cast, mais c'est moins beau:

Event* e=GetNextEvent();
OpenEvent* oe=dynamic_cast<OpenEvent*>e;
if(oe){ processOpen(oe); }else{
CloseEvent* ce=dynamic_cast<OpenEvent*>e;
if(ce){ processClose(ce); }else{
SendEvent* se=dynamic_cast<OpenEvent*>e;
if(se){ processSend(se); }else{
throw std::exception("Unexpected event class");}}}

On pourrait utiliser un enum et un switch, ou un index et une table
pour accéler la sélection de la méthode à envoyer, mais il faudra
quand même utiliser un dynamic_cast pour revenir à la classe exacte de
l'évènement afin d'accéder à ses champs spécifiques.

En lisant la doc de la STL, je suis tombé sur typeinfo:
http://www.cplusplus.com/reference/std/typeinfo/type_info/

Est-ce que ça ne serait pas mieux ?

C'est le mécanisme sous-jacent à dynamic_cast.

Le mieux, c'est la méthode virtuelle, avec retour sur l'objet
appelant.

Merci pour ton aide, je vais essayer tout ça ;-)

On Apr 29, 10:40 am, p...@informatimago.com (Pascal J. Bourguignon)
wrote:

Merwin <merwin....@gmail.com> writes:
> Bon je vous explique un peu ce que j'essaye de faire:

> 1. J'ai un programme qui contient plusieurs Threads (avec pthread)
> 2. Il y a un Thread qui gère la réception de données réseaux vi a des
> sockets Tcp et select().
> 3. Il a X autres Threads qui traite chacun les données réseaux leur
> appartenant, ainsi que des évènements envoyés par d'autres Thread s.

> Je cherche donc à créer un 'EventManager', l'idée c'est que mon T hread
> qui gère le réseau reçoit des données, il envoi alors un un Eve nement
> au thread concerné par ces données pour lui dire qu'il a quelquecho se
> a traiter.
> [...]
> J'ai vraiment du mal à gérer ça, je ne vois pas comment faire, je
> pensais utiliser une variable de condition (avec pthread_cond_wait()),
> mais tous les Threads ne peuvent pas utiliser la même puisque un
> Evenement n'est pas forcément adressé à tous les threads...

Il faut choisir entre deux:

- avoir une seule queue d'évènement ; dans ce cas, mettre un évèn ement
dans la queue est simple, le fil réseau n'a pas besoin de connaitre
la correspondance entre les évènements qu'il envoit et le fil qui
doit le traiter. Mais la collecte des évènements est plus
compliquée car chaque fil devra sélectionner les évènements don t il
a besoin (il faudrait ajouter un paramètre à getNextEvent
(somethingHappened)).

- avoir une queue par fil ; dans ce cas, envoyer un évènement est un
peu plus compliqué car il faut pouvoir déterminer la correspondance
entre l'évènement et le fil qui va le traiter (mais ça s'apparen te
à l'envoi d'un message à un objet). Mais la collecte est simple,
car chaque fil a sa propre queue.

Tôt ou tard, il va bien falloir une queue par thread de
traitement : si j'ai bien compris, l'aiguillage vers les
threads dépend du type de messages, avec chaque thread
spécialisé dans un type particulier. En revanche, si
l'aiguillage est coûteux en temps, il pourrait être intéressant
de le mettre dans un thread à part, pour ne pas bloquer la
connexion : il y aurait alors un queue de messages pour la
communication entre le thread de connexion et le thread
d'aiguillage, et une queue de messages par thread de traitement.

[...]

Les évènements sont typiquement classifiables avec une belle
hierarchie: fais des sous-classes à volonté!
(en C, on aurait utilisé une union).

En C, on aurait utilisé un void*, suivi des casts à volenté:-).
(En fait, en C, moi, j'aurais défini une struct qui représente
la classe de base, avec un plus un champs qui spécifie le type,
suivi des struct pour chaque type « dérivé », avec la struct
de base comme premier élément. Il y aurait bien eu des casts,
mais uniquement entre le type de base et des types
« dérivés ». Mais les void* pour faire ceci, j'en ai vu pas
mal.)

Il reste encore une question par rapport à la hièrarchie :
est-ce que les types ne représentent que les données du
messages, ou est-ce qu'ils incorporent déjà le comportement du
traitement. La solution classique (je crois), c'est que les
messages ne représentent que les données, plus ou moins, et que
le traitement se trouve dans la boucle du thread traitant. Mais
on pourrait bien imaginer que le type envoyé en message sache se
traiter lui-même ; que la boucle dans les threads soit toujours
identique. On revient aux deux altérnatifs pour la boucle dans
les threads :

Spécialisée :

std::auto_ptr< Event >
event = myQueue.receive() ;
while ( event.get() != NULL ) {
EventTypeA* eventA
= dynamic_cast< EventTypeA* >( event.get() ) ;
// traitement de l'eventA...
// unique à chaque thread.
event = myQueue.receive() ;
}

Générique :

std::auto_ptr< Event >
event = myQueue.receive() ;
while ( event.get() != NULL ) {
event->process() ;
event = myQueue.receive() ;
}

Dans cet alternatif, tous les threads traitants partage le même
code. On pourrait même imaginer l'utilisation d'un pool de
threads ; qu'il n'y a qu'une queue de messages, sur laquelle
tous les threads traitants attendent, et qu'un thread donné
traite une fois un type, la prochaine fois une autre.
(Évidemment, je ne sais pas si cette solution conviendrait à ses
besoins.)

Quant à l'aiguillage, j'essaierais d'extraire une « clé » du
message, afin d'utiliser std::map, ou quelque chose de
semblable, pour recupérer un objet avec un pointeur vers une
usine, pour créer l'objet, et un pointeur vers la queue où
il faut l'envoyer. Mais là aussi, je ne sais pas assez des
critères d'aiguillage pour dire si c'est faisable ou non.

--
James Kanze (GABI Software) email:james.kanze@gmail.com
Conseils en informatique orientée objet/
Beratung in objektorientierter Datenverarbeitung
9 place Sémard, 78210 St.-Cyr-l'École, France, +33 (0)1 30 23 00 34

On Apr 29, 2:29 pm, p...@informatimago.com (Pascal J. Bourguignon)
wrote:

Merwin <merwin....@gmail.com> writes:
> Merci, je pense que je vais en effet partir sur le principe
> une queue/objet. Pour les évènements en sous classe je veux
> bien, mais comment je peux déterminer le type de l'évènement
> quand je le récupère?

Une première solution c'est de mettre les methodes de
traitement de l'évènement dans l'évènement lui-même (comme
méthode virtuelle).

Event* e=GetNextEvent();
e->process(this);

(et la méthode process de chaque sous-classe d'Event peut se
réduire à faire suivre un message à son argument:

OpenEvent::process(Recipient* r){ r->processOpenEvent(this); }
CloseEvent::process(Recipient* r){ r->processCloseEvent(this); }
SendEvent::process(Recipient* r){ r->processSendEvent(this); }

Sinon, on peut utiliser dynamic_cast, mais c'est moins beau:

Event* e=GetNextEvent();
OpenEvent* oe=dynamic_cast<OpenEvent*>e;
if(oe){ processOpen(oe); }else{
CloseEvent* ce=dynamic_cast<OpenEvent*>e;
if(ce){ processClose(ce); }else{
SendEvent* se=dynamic_cast<OpenEvent*>e;
if(se){ processSend(se); }else{
throw std::exception("Unexpected event class");}}}

Si chaque thread est spécialisé pour un type donné, il n'y a
qu'un dynamic_cast par thread :

std::auto_ptr< Event > event = myQueue.receive() ;
OpenEvent* open = dynamic_cast< OpenEvent* >( event.get() ) ;
assert( open != NULL ) ;

A ce moment, grosso modo, chaque thread créer sa propre queue,
puis inscrira la queue et une fonction « factory » pour les
types d'évenements qu'il puisse traiter.

Dans ce scenario, il faudrait aussi prévoir la synchoronisation
de l'initialisation ; il ne faut pas que le thread de connexion
commence à travailler avant que tous les threads worker soient
inscris.

On pourrait utiliser un enum et un switch, ou un index et une
table pour accéler la sélection de la méthode à envoyer, mais
il faudra quand même utiliser un dynamic_cast pour revenir à
la classe exacte de l'évènement afin d'accéder à ses champs
spécifiques.

La sélection pourrait bien se faire au niveau du choix de la
queue de messages, de façon à ce que chaque thread ne recoit
qu'un seul type de message. (Ceci dit, en général, je préfère la
solution avec la fonction virtuelle dans l'objet du message.)

--
James Kanze (GABI Software) email:james.kanze@gmail.com
Conseils en informatique orientée objet/
Beratung in objektorientierter Datenverarbeitung
9 place Sémard, 78210 St.-Cyr-l'École, France, +33 (0)1 30 23 00 34

[snip]

Merci pour tous tes conseils, je n'avais pas du tout pensé à utiliser
les templates et autres astuces, je vais continuer à regarder dans cette
direction, merci beaucoup ;-)

--
Thibaut