Conflit Parallel.ForEach et Threads

Bonjour,

Je développe un logiciel qui est interfacé avec un capteur laser. Celui-ci m'envoie la distance entre le capteur et l'objet qui se situe devant lui. Il travaille à 2000Hz. Je communique avec lui à travers une API fournit.
Pour faire simple, j'ai créé une thread qui est en interface avec ce capteur. Les trames reçues sont mises dans une Queue pour traiter aussi vite que possible et ne pas perdre de trame. Ensuite, une autre Thread vient prendre les trames et les décoder.
Pour un objet donné, je constitue un ensemble de points mesurés avec le capteur. Je fais ensuite des calculs sur cet ensemble de points.

Dans le module de calcul, pour améliorer les performances j'ai utilisé un Parallel.ForEach(). Cela fonctionne bien et j'ai amélioré mes temps de calcul.
Tout va bien tant que je fonctionne par étapes : 1) mesure d'un objet puis 2) calcul sur les points obtenus.

Pour encore gagner en productivité, j'aimerai : mesuré la pièce "n+1" pendant que je calcule la pièce "n". Et là, les problèmes commencent. J'ai régulièrement des pertes de trames. J'en conclu qu'il y a un conflit entre la fonction Parallel.ForEach() et les Threads d'acquisition de points.

Qu'en pensez-vous ? Comment améliorez les choses ?

J'ai essayé de mettre les Threads dans une priorité plus haute, voir même critique. Soit je n'ai pas réussi, soit ca ne fonctionne pas ...

il faudrait quelques bouts de code pour comprendre l'origine du problème

mais quand plusieurs threads doivent accéder aux même objets il faut que ceux ci soient dit thread safe
la classe queue par exemple ne l'est pas
quand on utilise des objets non thread safe (enfin c'est plutôt une granularité au niveau des méthodes) il faut alors utiliser des mécanismes de verrou (lock, readerwriterlockslim ...)

Bonjour,
Tu peux aussi utiliser une ConcurrentQueue, qui elle est thread-safe.

Tatayo.

Merci pour vos réponses. C'est difficile de mettre du code car le projet est plutôt volumineux.

Tout se passe bien "entre les treads". J'ai mis en place des lock, pour que la thread "acquisition" puisse ajouter une trame dans la file, et que la thread "decodage" puisse retirer une trame tranquillement.

Le problème est vraiment lié à l'utilisation en même temps du Parallel.ForEach. D'ailleurs, si à la place j'utilise un ForEach basique, plus de problème. Mon temps de calcul est plus long, mais il ne perturbe pas l'acquisition.

dans ce cas c'est peut etre votre lock qui est trop long en traitement, que celui ci est aussi utilisé par l'acquisition faisant que l'acquisition rate des infos ...
comme se passe l'acquisition ? c'est une demande de données envoyé ? par tcp/serial ? ou juste un flux à lire ?

si c'est une demande il faut en effet que celle ci ne soit pas bloquée longtemps par un lock
il faudrait l'acquérir vite, juste enqueue pendant le lock
de l'autre côté seul le dequeue doit être dans un lock, le traitement devrait être hors lock et avec la donnée acquise par le dequeue

C'est une bonne piste. Je vais regarder de près le contenu des lock. Quelle serait le rapport avec le Parallel.Foreach ? car si je le retire, je n'ai pas de trous ...

avec des temps improvisés :
si sans parallel tu as 1sec de verrou puis 3sec de traitement, ca laisse 3sec à l'acquisition pour utiliser le verrou autant de fois qu'elle le veux
avec parallel plusieurs threads vont tenter de faire ca
donc le 1er thread lock 1sec puis va faire son traitement
le 2eme thread attend déjà 1sec pour avoir le verrou puis garde le verrou 1sec puis va faire son traitement
le 3eme thread va attendre 2 sec pour avoir le verrou et le garder 1 sec
donc le thread d'acquisition quand il demande le verrou il attend son tour pendant plusieurs secondes aussi, c'est beaucoup plus long si plusieurs personnes le demande que si une seule le demande

Bonjour,

Comme quoi, on n'est jamais assez précis dans les explications ...

Le Parallel.ForEach() travaille sur les données acquises à la pièce précédente. Donc il ne fait pas de lock sur la Queue de réception. Il travaille vraiment dans un module à part.
Mon sentiment, si j'ai bien compris, le Parallel.ForEach() permet d'utiliser en parallèle plusieurs CPU pour faire des choses. Je me demande si il ne "s'accapare" le temps CPU pour lui, et du coup, les Threads qui ont besoin aussi de temps ne sont pas "activées". Est-ce comme cela que ca fonctionne ? quelle est la priorité d'execution entre une thread et un Parallel.ForEach ?

sans code on peut juste supposer des choses oui ...

parallel.foreach utilise le threadpool, donc le nombre de thread est assez limité par défaut je crois
donc une appli ca peut encaisser quelques threads en plus sans trop ralentir le reste
car le principe du threadpool c'est que si tu as 100 objets dans la collection, ca ne va pas lancer 100 threads, mais chaque thread créé va traiter n objets (dépendant du nombre de coeur sur la machine)
déjà windows de base il doit exécuter plein d'applis, et dans chaque appli il y a plusieurs threads (certaines apps ont des centaines de threads), donc dans le principe c'est chaque coeur de processeur se voit attribuer un thread pendant un petit temps d'exécution, puis c'est le thread suivant qui vient prendre la place sur le coeur
en jouant sur la priorité (c'est possible) certains s'exécutent plus souvent que d'autres, mais on en a jamais eu besoin malgré plein de threads et de l'acquisition aussi ...

donc sans lock commun entre l'acquisition et ce code contenu dans le parallel.foreach je ne vois pas trop d'où ca pourrait venir...

On est d'accord alors. Les deux "modules" ne devraient pas être en conflit. Et pourtant, dans les faits, il y a un impact notable lorsque j'active ou non le Parallel.ForEach.

Les threads sont difficiles à debugger car impossible de mettre des points d'arrêts sans "fausser" le comportement... J'utilise un système de "trace" rapide, puis je lie le contenu de mes traces après coups. Pour ne pas perturber. J'aimerai tracer le temps passer dans les locks pour voir si effectivement je perd des trames. Connais tu un moyen de faire ca ?
J'ai utiliser le class StopWatch, mais je ne suis pas sûr des mesures car les temps sont très court j'imagine...

le stopwatch est bien mais peu lisible pour des temps courts
par contre tu peux le cumuler, à savoir le déclarer sur la classe (meme en mettre plusieurs, genre un pour tel temps et un autre pour tel autre)
puis faire juste des .Start et .stop
tu attends quelques secondes et tu mets un point d'arret (après un .stop) pour verifier le elapsedmilliseconds
avec un compteur tu peux meme connaitre les nanosecondes en divisant

oui d'accord. Ca te fait un temps "moyen". Je peux deja voir si les threads ont autant de temps avec et sans le Parallel.Foreach(). Et peut-être voir a quel periode elles sont "reveillées".
Par rapport à l'acquisition je n'ai pas répondu ... L'API fonctionne avec un event pour la réception de trame. Donc je suis "réveillé" chaque fois qu'une trame est recu (d'ailleurs il faudrait voir dans quel thread tourne ce fameux reveille. Car elle est peut-être bloqué pendant le Parallel.ForEach()). Je n'ai pas accès au code dans l'API pour savoir comment ils ont codés ... Cet event me donne en paramètre la nouvelle trame, que je met dans la Queue le plus vite possible.

J'ai repensé au multi-thread. J'ajoute des informations au fur et à mesure ... J'ai aussi d'autres thread qui tourne pour le filtrage des points. En effet, les points brutes contiennent des artefacts. J'ai une gestione en "flux de données".

Thread :
1 - acquisition trame
2- décodage de la trame
3- lorsque j'ai n points => filtrage "bloc
4- Lorsque j'ai 1 "bloc" filtré => Compression "bloc" => Cloud

Les threads 2 - 3 et 4 devrait tourner avec des priorités moindres. Le plus critique étant l'acquisition. Mais je n'ai pas su modifier les niveaux de priorités. J'ai l'impression qu'en C++ (il y a bien longtemps) j'ai dejà fait ce genre de truc ...

Voici le code "simplifier" de la partie Acqusition.


Fonction de reception des trames =>

Code:

---

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20

private void _OnUpdateDataSample(cDataSample dataSample) { bool enable = false;   lock (_lockAcquisition) enable = (_currentStatus == EStatus.MesureSingleInProgress || _currentStatus == EStatus.MesureMultipleInProgress);   // enable == false => On jette les trames pour le moment if (!enable) return;   // ajoute les trames dans la fifo lock (_lockFifoDataSample) { if (_fifoDataSample != null) { _fifoDataSample.Enqueue(dataSample); } } }

---

La thread de decodage :

Code:

---

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38

private void ThreadDataSampleLoop() { try { const int timeoutDefaultValue = 40; int timeoutValue = timeoutDefaultValue; Int32 startPositionXCount = 0;   while (_threadDataSample.EventExitProcessThread.WaitOne(timeoutValue) == false) { #region Data Sample lock (_lockFifoDataSample) { if ((_client != null) && (_client.ClientIsConfigured)) { while ((_fifoDataSample != null) && (_fifoDataSample.Count > 0)) { cDataSample dataSampleGui = (cDataSample)_fifoDataSample.Dequeue(); if (dataSampleGui != null) { // décodage des trames qui sont dans la file d'attente } } } } #endregion } _threadDataSample.EventExitProcessThreadDo.Set(); } catch (Exception ex) { if (OnError != null) { OnError(this); } } }

---

La partie "fifo" est bien protégée par le lock !

Merci de votre aide.

en tout cas les 2 sont bien sur le même lock (_lockFifoDataSample) donc ca ne peut pas être en même temps
donc si // décodage trame prend du temps ca bloque l'acquisition

par contre si c'est un event ca vient peut être de la dll fournie, aucun event ne peut être raté quand il est levé, ils sont empilés jusqu'à traitement (ce qui peut causer des problèmes à ceux qui n'arrivent pas à traiter assez vite des fois ^^)
auquel cas il faudrait contacter le fournisseur, pour lui dire relayer les events sur un thread séparé pour ne pas bloquer (ou éviter de bloquer de ton côté, ca irait aussi)

dans le principe d'une acquisition avec event discutable c'est

Code:

---

1 2 3 4 5 6

while (true) { // lecture des données // levée de l'event // thread.Sleep(x) }

---

or levée de l'event s'il est fait bêtement est bloquant jusqu'à que tous les abonnés ai traité l'event, donc [lecture des données] n'est pas refait dans la foulée
quand c'est bien fait c'est

Code:

---

1 2 3 4 5 6

while (true) { // lecture des données // levée de l'event en begininvoke/thread dédié => donc non bloquant // thread.Sleep(x) }

---

d'ailleurs en bien fait c'est pas forcément un event d'ailleurs, comme ca ca peut être désynchronisé sans s'empiler

oui je crois que je vois le problème. Mais alors il faudrait que j'utilise une deuxième Queue. et que je fasse le decodage dans cette 2ieme Queue avec une Thread2 ? Si c'est trop lent, c'est pas gave car je vais empiler les trames dans la première Queue.

J'aurais :

- Event => FifoDataSample.Enqueu

- Thread1 => FifoDecode.Enqueue(FifoDataSample.Dequeue())

- Thread2 => Decodage (FifoDecode.Dequeu())

Ca marche là ?

et donc 2 lock different, un pour l'accès à FifoDataSample et l'autre pour FifoDecode.

il faut bien le meme lock entre enqueue et dequeue
mais il ne faut faire que le dequeue dedans

et thread 1 peut etre infini et unique
genre

Code:

---

1 2 3 4 5 6 7 8 9

while (true) { if queue.count > 0 { // lock et dequeue // lancer thread de traitement du truc dequeué (c'est ce que je disais à un moment) } thread.sleep(5) // un while infini doit souvent avoir un sleep pour ne pas bouffer 100% de ressources, déjà 1ms ca évite le problème, après c'est à choisir selon l'acquisition, genre la moitié ou un tiers du temps prévu par l'acquisition }

---

comme ca le lock du dequeue est très court et ne bloque pas l'acquisition car il ne contient pas le traitement

Oui, finalement je ne suis pas si loin ...
A la place du lock qui englobe tout le code, je limite seulement au dequeu :

Code:

---

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39

private void ThreadDataSampleLoop() { try { const int timeoutDefaultValue = 40; int timeoutValue = timeoutDefaultValue; Int32 startPositionXCount = 0;   while (_threadDataSample.EventExitProcessThread.WaitOne(timeoutValue) == false) { #region Data Sample if ((_client != null) && (_client.ClientIsConfigured)) { while ((_fifoDataSample != null) && (_fifoDataSample.Count > 0)) { cDataSample dataSampleGui = null;   lock (_lockFifoDataSample) (cDataSample)_fifoDataSample.Dequeue();   if (dataSampleGui != null) { // décodage des trames qui sont dans la file d'attente } } } #endregion } _threadDataSample.EventExitProcessThreadDo.Set(); } catch (Exception ex) { if (OnError != null) { OnError(this); } } }

---

Dans ce cas, je bloque uniquement l'acquisition le temps du "dequeu". Je comprends.

Faut-il mettre un lock pour faire un test du type ?

Code:

---

1 2	if ((_fifoDataSample != null) && (_fifoDataSample.Count > 0))

---

dans ton cas je pense pas
mais dans le principe le lock évite que des données changent entre plusieurs lignes exécutées

genre si tu fais if .count > 0 ca peut te répondre count vaut 1 mais si un autre thread sans lock dequeue l'item alors au moment où tu vas faire dequeue ca va planter
toi c'est le même thread qui est le seul à pouvoir dequeuer donc si le test du count passe il y aura forcément un item

l'autre type de problème qui peut arriver c'est

Code:

---

1 2 3 4 5 6

if _queue != null return _queue lock (_l) { _queue = new queue(); return _queue; }

---

ici c'est un bug potentiel car si 3 threads rentrent et que les 3 voient null il vont aller sur le lock et 3 instances vont être créés et tout le monde ne va pas travailler sur la même queue.
on fait alors

Code:

---

1 2 3 4 5 6 7

if _queue != null return _queue lock (_l) { if _queue != null return _queue _queue = new queue(); return _queue; }

---

cet exemple ne te concerne peut etre pas mais il aide à voir ce qu'il faut locker pour prendre des décisions sans changement de données intermédiaire

tu peux quand même jeter un oeil où ton _fifoDataSample est instancié

oui c'est bien clair. J'évite le "cas par cas" en général. Car on ne sait jamais qui va modifier le code dans le futur. Donc je vais prendre "ceinture et bretelle" ;).

Merci pour ton aide. Je vais faire des essais maintenant.

Cordialement,

Bonjour,
Excuser mon intrusion dans cette discussion déjà résolue mais j'ai cru comprendre que l'alimentation d'une queue était indépendante de son dépilage. D'ailleurs je ne sais pas si l'objet ConcurrentQueue met une protection particulière par rapport à Queue (nom de méthode identique : Enqueue). Par contre le dépilage oui avec TryDequeue (et TryPeek sans dépiler).

Pour moi, vu que tu veux faire du parallèle, utilise le ConcurrentQueue.

Citation:

---

Envoyé par ypelissier

Bonjour,
Excuser mon intrusion dans cette discussion déjà résolue mais j'ai cru comprendre que l'alimentation d'une queue était indépendante de son dépilage. D'ailleurs je ne sais pas si l'objet ConcurrentQueue met une protection particulière par rapport à Queue (nom de méthode identique : Enqueue). Par contre le dépilage oui avec TryDequeue (et TryPeek sans dépiler).

Pour moi, vu que tu veux faire du parallèle, utilise le ConcurrentQueue.

---

Bonjour,

Oui c'est une possibilité. En faite, j'utilise une Queue fournit dans l'API avec le capteur. Du coup, j'utilise des lock pour protéger l'accès.
Suite aux conseils j'ai bien relu mon code et effectivement, je faisais le décodage de la trame "a l'intérieur" du lock. Donc je bloquais l'acquisition.
Cordialement