multiprocessing et QProcess

Bonsoir,

bon je vais être honnête, j'ai pas tout piger à ce que je vais vous demander :p

je veux permettre le multiprocess sur une liste de travaille très longue...

j'ai plusieurs milliers d'images que je transforme en texte automatiquement via la commande :

Code:

---

1 2	for File in self.SubImgFiles: process.start('tesseract -l {0} "{1}" "{1}"'.format(self.Lang, File))

---

je voudrais donc accelerer le traitement en séparant le travail en plusieurs parties pour le partager.

je me dis qu'il faudrait créer X listes de fichiers et les donner aux différents processeurs.

j'ai vu multiprocessing qui semble le permettre mais les exemples que j'ai trouvé ne sont pas forcement très parlant.

Avez vous une idée de la façon de s'y prendre ?


J'ai posté ici car je ne pense pas que le QProcess soit déterminant dans la façon de s'y prendre.


Merci et bonne soirée à vous.

Salut,

Je ne vois pas trop l'intérêt de multiprocessing: il n'y a pas de code/données Python à partager juste la commande/programme externe à lancer avec différents paramètres.
Dans tous les cas, on va avoir à lancer N process en parallèle (avec un N qui dépend du nombre de CPU disponibles) et leur soumettre le fichier suivant à traiter.
Peut être regarder du côté de concurrent.futures qui fait déjà une partie du boulot au dessus des Threads ou de subprocess (au choix).

- W

Citation:

---

Dans tous les cas, on va avoir à lancer N process en parallèle (avec un N qui dépend du nombre de CPU disponibles) et leur soumettre le fichier suivant à traiter.

---

C'est tout à fait l'idée que j'en avais :)

Je vais regarder ce que tu me conseilles, j'espere juste que ça lancera pas toutes les commandes sur le meme processeur.

je reviendrai vers vous pour vous dire ce qu'il en est.

Merci à toi.

EDIT : C'est parfait !
je suis passé de 770s à 215s !
Franchement pratique !

voici ma version

Code:

---

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

from concurrent.futures import ThreadPoolExecutor # Permet de le multi calcul from psutil import cpu_count # Indique le nombre de cpu     def TesseractConvert(self, File): ### Création du QProcess avec unification des 2 sorties (normale + erreur) process = QProcess()   ### Reconnaissance de l'image suivante process.start('tesseract -l {0} "{1}" "{1}"'.format(self.Lang, File)) process.waitForFinished()   # Décompte du nombre de fichier traité avec création du pourcentage self.FilesDone += 1 self.Pourcentage = int((self.FilesDone * 100) / self.TotalSubtitles) print(str(self.Pourcentage))     with ThreadPoolExecutor(max_workers=cpu_count()) as executor: for File in self.SubImgFiles: executor.submit(self.TesseractConvert, File)

---

Bonjour,

je me permets de relancer vite fait ce topic,

en effet mes threads modifient tous une QProgressBar, ce que qt n'aime pas :)

J'ai testé différents trucs mais qui n'ont pas amélioré le systeme...

Je pense qu'il faut jouer avec les pyqtSlot et pyqtSignal mais ça me parait un peu lourdingue de programmer des centaines voire des milliers de signaux (1 pour chaque thread de mon multiproces)...

Avez vous une idée pour simplifier le truc ?

Le code :

Code:

---

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

from concurrent.futures import ThreadPoolExecutor # Permet de le multi calcul   def TesseractConvert(File): Reply = LittleProcess('tesseract -l {0} --tessdata-dir "{2}" "{1}" "{1}"'.format(GlobalVar["TesseractLanguage"], File, GlobalVar["TesseractFolder"]))   GlobalVar["FilesDone"] += 1   GlobalVar["ProgressBar"].setValue(GlobalVar["FilesDone"])   ### Si c'était le dernier fichier, on cache la fenetre if GlobalVar["FilesDone"] == GlobalVar["TotalSubtitles"]: GlobalVar["ProgressDialog"].accept()   with ThreadPoolExecutor(max_workers=GlobalVar["NbCPU"]) as executor: for ImageFile in GlobalVar["SubImgFiles"]: executor.submit(TesseractConvert, ImageFile)   GlobalVar["ProgressDialog"].exec()

---

Bonjour hizoka,

Il n'est pas facile de juger de la validité d'une solution quand on ne connait pas bien le problème à résoudre. Mais imaginons que tu ais plusieurs milliers de tâches indépendantes de durées différentes à traiter par du code Python, voilà la solution qui me plait le plus et que j'ai déjà utilisée:

- On lance au début du programme et d'un seul coup autant de Process (du module multiprocessing) qu'il y a de CPU, avec 2 arguments: "qin" = Queue d'entrée et "qout" = Queue de sortie. Les 2 queues (du module multiprocessing) seront communes à tous les process ainsi qu'au programme principal et pourront donc être lues et écrites par n'importe lequel d'entre eux (le système de verrous est transparent).

- le programme principal rentre dans la queue d'entrée "qin" la totalité des travaux à réaliser (p. ex. les fichiers à traiter)

- Chaque process a un fonctionnement permanent! Au début de chaque boucle while, il attend qu'il y ait un travail en attente dans la queue d'entrée "qin" (travail=qin.get()), le prend, le traite et rentre son résultat dans la queue de sortie "qout" (qout.put(resultat)). Ceci fait, il revient en début de la boucle pour prendre le travail suivant s'il y en a encore.

- le programme principal compte la longueur de la queue de sortie "qout" et sait que tout est terminé quand le nombre total de travaux réalisés est atteint. Lorsque c'est fait, il peut détruire les process qui sont devenus inutiles. A noter que le suivi par le programme principal de ce nombre de travaux réalisés peut alimenter une barre de progression. Pour ne pas figer la partie graphique, le comptage peut être fait par un timer (il y a d'autres solutions).

A mon avis, le grand avantage de cette méthode est que ce n'est pas le programme principal qui alimente chaque process: c'est chaque process qui vient chercher du travail quand il a terminé le précédent. Ce qui fait que le process qui est tombé sur un travail de courte durée en prendra plus! Cela veut dire que les ressources des CPU sont complètement exploitées.

A noter qu'avec ajout de l'instruction "freeze_support()", cette solution est compatible avec cx_freeze.

Si cette solution t'intéresse, je peux te proposer du code.

Salut,

Citation:

---

Envoyé par hizoka

en effet mes threads modifient tous une QProgressBar, ce que qt n'aime pas :)

J'ai testé différents trucs mais qui n'ont pas amélioré le systeme...

Je pense qu'il faut jouer avec les pyqtSlot et pyqtSignal mais ça me parait un peu lourdingue de programmer des centaines voire des milliers de signaux (1 pour chaque thread de mon multiproces)...

---

D'après ce que j'en lis TesseractConvert(File) est un thread qui va créer un sous process. On va donc lancer autant de threads que de fichiers et créer de s/process... Ce qui effectivement distribue le contrôle sur beaucoup d'agents autonomes.
Comment simplifier? En virant les agents et en le remplaçant par un "manager" En créant un "manager" qui aura la responsabilité de pousser threads et s/process et de mettre à jour la barre de progression à la fin de chaque s/process.
Si on reste avec le module concurrent.futures, on peut appeler un callback "à la fin de..." et l'architecture du bazar ressemble à:

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 40 41 42 43 44 45 46 47 48 49 50 51 52

from threading import Thread from concurrent.futures import ThreadPoolExecutor as Pool import time   def launcher(count, on_update, on_done):   left = count step = 100 / count def callback(future): nonlocal left left -= 1 on_update(step) if left == 0: on_done()   with Pool(max_workers=2) as pool: for z in range(count): future = pool.submit(subprocess.call, 'python -c "import time; time.sleep(0.5)"') future.add_done_callback(callback)   if __name__ == '__main__':   import tkinter as tk from tkinter.constants import * from tkinter import ttk   root = tk.Tk()   def updater(root, pb): def _do_update(x): root.after_idle(lambda: pb.step(x)) return _do_update   def on_done(): def reset(): pb.pack_forget() start_btn['state'] = NORMAL root.after_idle(reset)   def do_launch(): start_btn['state'] = DISABLED pb.pack() update = updater(root, pb) p = Thread(target=launcher, args=(150, update, on_done)) p.daemon = True p.start()   start_btn = tk.Button(root, text='start', command=do_launch) start_btn.pack() pb = ttk.Progressbar(root, orient=HORIZONTAL, length=200, mode='determinate')   tk.mainloop()

---

note: tkinter ou Qt, c'est la même chose côté répartition des tâches et de toutes façons ce qui vous intéresse c'est la partie au dessus de "if __name__ == '__main__':"

Citation:

---

Envoyé par tyrtamos

Il n'est pas facile de juger de la validité d'une solution quand on ne connait pas bien le problème à résoudre. Mais imaginons que tu ais plusieurs milliers de tâches indépendantes de durées différentes à traiter par du code Python, voilà la solution qui me plait le plus et que j'ai déjà utilisée:

- On lance au début du programme et d'un seul coup autant de Process (du module multiprocessing) qu'il y a de CPU, avec 2 arguments: "qin" = Queue d'entrée et "qout" = Queue de sortie. Les 2 queues (du module multiprocessing) seront communes à tous les process ainsi qu'au programme principal et pourront donc être lues et écrites par n'importe lequel d'entre eux (le système de verrous est transparent).

- le programme principal rentre dans la queue d'entrée "qin" la totalité des travaux à réaliser (p. ex. les fichiers à traiter)

- Chaque process a un fonctionnement permanent! Au début de chaque boucle while, il attend qu'il y ait un travail en attente dans la queue d'entrée "qin" (travail=qin.get()), le prend, le traite et rentre son résultat dans la queue de sortie "qout" (qout.put(resultat)). Ceci fait, il revient en début de la boucle pour prendre le travail suivant s'il y en a encore.

- le programme principal compte la longueur de la queue de sortie "qout" et sait que tout est terminé quand le nombre total de travaux réalisés est atteint. Lorsque c'est fait, il peut détruire les process qui sont devenus inutiles. A noter que le suivi par le programme principal de ce nombre de travaux réalisés peut alimenter une barre de progression. Pour ne pas figer la partie graphique, le comptage peut être fait par un timer (il y a d'autres solutions).

A mon avis, le grand avantage de cette méthode est que ce n'est pas le programme principal qui alimente chaque process: c'est chaque process qui vient chercher du travail quand il a terminé le précédent. Ce qui fait que le process qui est tombé sur un travail de courte durée en prendra plus! Cela veut dire que les ressources des CPU sont complètement exploitées.

A noter qu'avec ajout de l'instruction "freeze_support()", cette solution est compatible avec cx_freeze.

Si cette solution t'intéresse, je peux te proposer du code.

---

Très bonne idée, mais hizoka utilise déjà concurrent.futures qui fait déjà tout çà "en standard".
*edit* J'ai remis ThreadPoolExecutor à la place de Process, car si l'idée est de lancer une commande externe, ce n'est pas utile. Qt dispose d'une bibliothèque semblable (QtConcurrent). Elle ne sait lancer que des Threads (ils n'ont pas de soucis de GIL en C++) mais je ne sais même pas si elle est dans PyQt.

- W

Bonjour wiztricks,

Citation:

---

Envoyé par wiztricks

Très bonne idée, mais hizoka utilise déjà concurrent.futures qui fait déjà tout çà "en standard"

---

Merci pour l'info! Je n'ai pas encore suffisamment compris ce que faisait "concurrent.futures" pour l'utiliser. Je vais creuser le sujet. :D

Je viens de faire quelque chose d'amusant: comparer les 2 solutions: celle avec multiprocessing et celle avec concurrent.futures.

Le principe que j'ai utilisé est le suivant: je crée une liste de 5000 tâches avec pour chacune une durée tirée au hasard entre 10 et 100 millisecondes. Cela donne en gros un temps total de 270 secondes. Chaque processus fait chaque travail en attendant en fait la durée transmise. Je mesure ensuite le temps de traitement avec plusieurs processus, et je compare les 2 valeurs.

Solution 1 avec multiprocessing:

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

import time from random import randint from multiprocessing import Process, Queue   #============================================================================ def traitement(qin, qout): while True: n = qin.get() time.sleep(n) qout.put(n)   #============================================================================ if __name__ == '__main__':   # création de la liste de travaux ttot = 0 # pour le cumul des durées des travaux T = Queue() # pile des travaux à réaliser ttot = 0 # pour le cumul des durées des travaux for i in range(0, 5000): duree = randint(10,100)/1000 # tirage au hasard d'un temps de travail T.put(duree) ttot += duree   # création et lancement des process de traitement R = Queue() # pile des travaux réalisés t = time.clock() for cpu in range(0,8): process = Process(target=traitement, args=(T, R)) process.daemon = True process.start() #attente que tout soit fini while R.qsize()<5000: pass t = time.clock()-t

---

Solution 2 avec concurrent.futures

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

import time from random import randint from concurrent.futures import ProcessPoolExecutor   #============================================================================ def traitement(n): time.sleep(n) return n   #============================================================================ if __name__ == '__main__':   # création de la liste des 5000 travaux ttot = 0 # pour le cumul des durées des travaux T = [] # liste des travaux for i in range(0, 5000): duree = randint(10,100)/1000 # tirage au hasard d'un temps de travail T.append(duree) ttot += duree   # exécution des travaux et enregistrement des résultats dans R R = [] t = time.clock() with ProcessPoolExecutor(max_workers=8) as executor: for travail, resultat in zip(T, executor.map(traitement, T)): R.append(resultat) t = time.clock()-t

---

Comparaison: les 2 solutions donnent à peu près le même résultat: le travail total de 270 secondes est fait en 34 secondes. Cela veut dire qu'on fait le travail 8 fois plus vite sur un PC qui a un CPU à 8 cœurs (en fait 4 cœurs physiques * 2 avec l'hyperthreading). Mais, bien sûr, la solution 2 avec "concurrent.futures" est plus simple à mettre en œuvre. Dans la mesure où concurrent.futures est construit sur la base de multiprocessing, c'est assez rassurant de voir la similitude des résultats.


Mais la suite est bizarre. Avec 8 cœurs, que les 8 processus permettent de diviser le travail à réaliser par 8 parait logique, mais si je crée 16 processus, je divise le temps par presque 16. Je peux essayer aussi avec 32 et même plus: avec 50 processus, je divise le temps par environ 40! Ceci avec les 2 solutions. Au delà, je ne gagne plus rien et ça augmente même.

J'aimerais bien comprendre ça...

Salut,

Citation:

---

Envoyé par tyrtamos

Mais la suite est bizarre. Avec 8 cœurs, que les 8 processus permettent de diviser le travail à réaliser par 8 parait logique, mais si je crée 16 processus, je divise le temps par presque 16. Je peux essayer aussi avec 32 et même plus: avec 50 processus, je divise le temps par environ 40! Ceci avec les 2 solutions. Au delà, je ne gagne plus rien et ça augmente même.

J'aimerais bien comprendre ça...

---

Si le travail se réduit à:

Code:

---

1 2 3

def traitement(n): time.sleep(n) return n

---

Il va effectivement en lancer beaucoup pour charger un CPU.

- W

onsoir, merci pour vos retours.

En fait mon systeme marche tres bien mais j'ai le droit en fait à des retours

Code:

---

1 2 3

QObject::startTimer: Timers cannot be started from another thread QObject::killTimer: Timers cannot be stoped from another thread ...

---

et avec ta proposition wiztricks, j'ai également les messages.

Je pensais que ta soluce allait me régler le souci, désolé de ne pas l'avoir signalé.

Salut,

Citation:

---

Envoyé par hizoka

et avec ta proposition wiztricks, j'ai également les messages.

Je pensais que ta soluce allait me régler le souci, désolé de ne pas l'avoir signalé.

---

Ça c'est un problème entre le thread qui gère la meute de process et le GUI graphique. C'est au dessous du "if __name__ == '__main__':" entre cette thread et la progress bar. Mais vous avez plein d'exemples faits par Tyrtamos comme celui-ci. Il suffit de comprendre les 2 logiques et de les coller ensemble.

Si vous ne vous en sortez pas, je verrais ce que je peux faire ce week end.

- W

Bonjour,

c'est en effet ce à quoi j'avais pensé en 1er lieu (je lis dès que possible Tyrtamos) mais le truc c'est que je vois pas trop comment l'utiliser dans le cas présent...

pour éviter que je fasse des milliers de connections, dois je imaginer de lancer le multi thread via un thread et que ce soit lui qui émette les retours d'infos ?
lancement : thread principal => thread secondaire => multi threads
reccup infos : multi threads => thread secondaire via la fonction callback => thread principal via l'emit

cela permettrait il d'éviter que ce soir trop lourd et de créer des milliers de connexions ?

J'attends vos conseils avisés :)

Citation:

---

Envoyé par hizoka

pour éviter que je fasse des milliers de connections, dois je imaginer de lancer le multi thread via un thread et que ce soit lui qui émette les retours d'infos ?

---

C'est ce que fait mon exemple: il suffit de remplacer le thread principal par un Thread Qt.

Citation:

---

Envoyé par hizoka

cela permettrait il d'éviter que ce soir trop lourd et de créer des milliers de connexions ?

---

Le gros soucis dans le code que vous avez montré est ici:

Code:

---

1 2 3 4

def TesseractConvert(File): ... GlobalVar["FilesDone"] += 1 ...

---

Si vous voulez que çà fonctionne, il faudrait verrouiller l'opération avec un mécanisme de lock... Sinon, parfois vous n'arriveriez jamais à terminer. Mais c'est quand même plus simple de faire ce comptage dans un seul Thread Qt qui pourra s'interfacer assez simplement avec la Progress Bar plutôt que d'avoir à coder à la main cette interface avec des threads non-Qt.
In fine, le résultat est le même mais on essaie d'utiliser au mieux l'existant pour écrire moins de lignes.

- W

Oups, mauvais lecture de votre code !
Ce qui me rassure c'est que j'avais une idée correcte :)

Je viens de mettre le systeme en place et tout fonctionne... presque... j'ai toujours ces messages :

Code:

---

1 2	QObject::killTimer: Timers cannot be stopped from another thread QObject::startTimer: Timers cannot be started from another thread

---

mais je ne pige vraiment pas pourquoi...
mon QThread émet bien la progression, qui est récupérer par une fonction dans le thread principal et la bar est mise à jour.
J'ai essayé en passant par une fonction dans la classe de la fenêtre de la progressbar mais le résultat est le même...

Mais il faut que je reparte de zéro car les exemples de Tyrtamos fonctionnent...

Je reviens vers vous si je n'y arrive pas !

Bonjour,

Juste un petit complément au cas où.

Un thread utilisant QThread est nécessaire pour qu'il soit capable d'envoyer (.emit(...)) des messages au graphique pour, par exemple, mettre à jour une barre de progression.

Mais il arrive qu'on préfère utiliser threading de Python pour le thread. Si c'est le cas, il est possible de lui faire émettre quand même des messages au graphique PyQt de la façon suivante: on utilise une classe qui hérite en même temps de threading et QObject (héritage multiple)!

Salut,

Avec Qt, çà donnerait un truc comme çà:

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 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69

from concurrent.futures import ThreadPoolExecutor as Pool import subprocess from PyQt5.QtCore import QThread, pyqtSignal   class Launcher(QThread): on_update = pyqtSignal(float) on_done = pyqtSignal() _abort = False   def __init__(self, count=10): super().__init__() self.count = count   def run_item(self): if not self._abort: subprocess.call ('python -c "import time; time.sleep(0.5)"')   def run(self):   step = 100 / self.count done = 0   def callback(future): nonlocal done if not self._abort: done += step print('callback', done) self.on_update.emit(done) if done >= 100 : self.on_done.emit()     self.futures = [] pool = self.pool = Pool(max_workers=2)   for z in range(self.count): future = pool.submit(self.run_item) future.add_done_callback(callback) self.futures.append(future)   def shutdown(self): # ** too much ** self._abort = True for f in self.futures: f.cancel() self.pool.shutdown(wait=True)       if __name__ == '__main__': from PyQt5.QtWidgets import QApplication, QProgressDialog, QPushButton   def do_launch(): launcher = Launcher() dialog = QProgressDialog("En cours...", "Annuler", 0, 100) launcher.on_update.connect(dialog.setValue) launcher.on_done.connect(dialog.reset) launcher.start()   dialog.exec_() launcher.shutdown()     app = QApplication([])   btn = QPushButton('start') btn.clicked.connect(do_launch) btn.show()   exit(app.exec_())

---

fonctionnellement, çà fait la même chose que la mouture précédente excepté l'ajout d'un arrêt/shutdown pour gérer l'option qui va avec dans la QProgressDialog.

Citation:

---

Envoyé par tyrtamos

Un thread utilisant QThread est nécessaire pour qu'il soit capable d'envoyer (.emit(...)) des messages au graphique pour, par exemple, mettre à jour une barre de progression.

Mais il arrive qu'on préfère utiliser threading de Python pour le thread. Si c'est le cas, il est possible de lui faire émettre quand même des messages au graphique PyQt de la façon suivante: on utilise une classe qui hérite en même temps de threading et QObject (héritage multiple)!

---

Dans ce bazar, on peut choisir la mouture du thread principal mais pas celles utilisées par concurrent.future (qui sont des threads standards Python). Donc on peut faire simple.

- W

Merci beaucoup à vous deux !

mes tests fonctionnent, il ne me reste plus qu'à intégrer tout ça dans mon soft :)

Bonjour à vous,

je réouvre ce topic car j'ai une question annexe.

Si je lance mon soft, le QThread puis le multiprocess et que je veux le kill le tout depuis la console avec "ctrl + c", ça ne fait que killer l'un des multiple process.

Avez vous une idée pour résoudre le problème ?

J'ai pensé à pas mal de trucs mais ça ne le fait pas trop, genre :

Code:

---

except KeyboardInterrupt:

---

bien que je l'ai mis à différents endroits, je n'ai pas réussis à déclencher cet except.

Est-ce possible ? et si oui, où faut il le placer ?
Une autre idée ou solution ?

Merci d'avance à vous !

Salut,

Si CTRL+C interrompt le programme c'est en général avec un traceback qui t'indique où, dans le code, le signal a été intercepté.

De toutes façons c'est dans la boucle principale, pas dans un thread.


Autre chose, tu as vu ceci:
https://pypi.python.org/pypi/pytesseract
https://pypi.python.org/pypi/tesserocr

En fait les traceback ne sont pas toujours au même endroit... 8O
j'ai essayé à divers endroits mais pas trouvé la soluce...

Ce que je propose diffère de ce qui existe déjà (du moins je m'en convaincs :p )


Il me dit que ça vient de ma fonction qui gere la progression de la progressbar
La fonction contient :

Code:

---

1 2 3 4 5 6

### Envoie de la valeur self.ProgressBar.setValue(value)     ### Force la mise à jour graphique QCoreApplication.processEvents()

---

j'ai essayé d'y mettre le try et l'except KeyboardInterrupt
...

Bonjour,

je me permets de relancer le sujet car j'ai fait pas mal de tests avec les try et except, j'en ai mis dans toutes les fonctions mais rien...

j'ai testé exception de concurrent.futures mais pas réussis non plus...

et mis à part ça je pense avoir fait à peu prés tout ce que je voulais :)

merci à vous.

Citation:

---

Envoyé par hizoka

je me permets de relancer le sujet car j'ai fait pas mal de tests avec les try et except, j'en ai mis dans toutes les fonctions mais rien...

---

Si on a crée une Dialog box avec un Button "Cancel" écrit dessus, pourquoi s'em... à attraper des Ctrl-C?

- W

parce que cette fameuse fenêtre est facultative :p
et que c'est soit dans une fenêtre de progression soit dans la console :)

C'est bien joli de vouloir toutes les options possibles mais si vous ne précisez pas le contexte (en postant un code "fonctionnel"), si ce n'est plus une application graphique, vous n'avez plus besoin de Qt et donc une "logique":

Code:

---

1 2 3 4 5 6 7

while True: try: ...on attend la fin du thread avec un timeout. ...if done: break (on sort normalement). except KeyboardInterrupt: ...on dégage les activités en cours ... et voilà...

---

devrait fonctionner (sauf à poster un code qui montre que ce n'est pas le cas sous l'environnement X avec la version de Python Z).

- W

Voila mon code épuré :

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 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124

GlobalVar["ProgressDialog"] = ProgressDialog(None)   ############################################################################# ### Classe gérant la fenetre de progression class ProgressDialog(QDialog): def __init__(self, Parent=None): ### Création de la fenetre de progression. ### ...     ### Affiche si besoin la fenêtre if GlobalVar["GuiLevel"] == 2: self.show()     ### Conversion des images en textes ## Création du thread GlobalVar["MonProgressThread"] = ProgressThread()   ## Connexion des signaux de progression et de fin du thread GlobalVar["MonProgressThread"].ProgDuThread.connect(self.WorkProgression) GlobalVar["MonProgressThread"].FinDuThread.connect(self.Next)   ## Démarrage du thread GlobalVar["MonProgressThread"].start()       #======================================================================== def WorkProgression(self, value): """Fonction de progression de la barre.""" ### Envoie de la valeur self.ProgressBar.setValue(value)     ### Force la mise à jour graphique QCoreApplication.processEvents()       #======================================================================== def Next(self, value): """Fonction lancée en fin de conversion des images en textes.""" ## Fermeture via validation de la fenêtre de progression self.accept()   ## ...   ############################################################################# ### Classe de conversion des images en textes class ProgressThread(QThread): ### Signaux et variable blo ## Création d'un nouveau signal pour informer de la fin du thread FinDuThread = pyqtSignal(str)   ## Création d'un nouveau signal pour info du % d'avancement ProgDuThread = pyqtSignal(float)     #======================================================================== def __init__(self): super().__init__()     #======================================================================== def shutdown(self): """Fonction de nettoyage lancée lors de la fermeture de la fenêtre de progression ou de la création d'un fichier Stop.""" ### Variable bloquante servant aux commandes en cours d'execution self.abort = True     ### Annulation du travail restant for f in GlobalVar["WorkList"]: f.cancel()   self.pool.shutdown(wait=True)       #======================================================================== def run(self): """Fonction lancée lors de l'execution du QThread.""" ### Création de variables pour le QThread ## Valeur de progression self.done = 0   ## Variable bloquante self.abort = False     ### Création de la liste de travail GlobalVar["WorkList"] = [] pool = self.pool = ThreadPoolExecutor(max_workers=QThread.idealThreadCount())     ### Remplissage de la liste for ImageFile in GlobalVar["SubImgFiles"]: ## Chaque commande créée lance la fonction self.Work avec l'argument ImageFile future = pool.submit(self.Work, ImageFile)   ## Ajout de la commande à la liste de travail GlobalVar["WorkList"].append(future)     #======================================================================== def Work(self, File): """Fonction éxécutée pour chaque image à convertir.""" ### Si la variable bloquante est active, c'est que la fenêtre est fermée ou un fichier stop est présent if self.abort: return     ### ...     ## Progression if GlobalVar["GuiLevel"] < 2: print(self.done) else: self.ProgDuThread.emit(self.done)   ### Si c'était le dernier fichier à traiter if self.done == MaxValue: self.FinDuThread.emit(QCoreApplication.translate("main", "The conversion is finished."))

---

La variable GlobalVar["GuiLevel"] permet de savoir s'il faut ou non afficher la fenêtre et donc la façon de faire connaître l'avancement

Pour faire simple, je créé tout la fenêtre quelque soit le cas mais je l'affiche ou non...

Citation:

---

Envoyé par hizoka

Voila mon code épuré :

---

On ne peut pas faire grand chose avec çà...

Un code épuré est par exemple comme celui que je vous ai posté tantôt: il fonctionne et on s'attend à y trouver try...except KeyboardInterrupt dedans.

Par expérience, je ne pense pas qu'il y ait de solution simple à votre problème. Et je ne sais pas si j'aurais le temps de vous montrer comment le poser à défaut de le résoudre.

- W

Bonjour,

Si je comprends bien, tu veux arrêter sur demande le travail réalisé par le QThread et les thread de concurrent.futures qu'il a lancé?

Malheureusement, si j'ai bien compris la notice, les threads de concurrent.futures ne peuvent pas s'arrêter comme ça. En fait, tu es dans une contradiction: tu veux un truc simple et automatique (concurrent.futures), mais tu veux aussi continuer à maîtriser le détail pour remplir des fonctions particulières. J'ai bien peur que ce soit une impasse.

Si tu tiens vraiment à arrêter sur demande le travail réalisé par les threads qui font le travail, je vois 2 solutions:

- Au lieu de concurrent.futures, utiliser la solution avec multiprocessing et les queues que j'ai décrites plus haut, mais permettre aux process (ça marcherait aussi avec des threads) qui font le travail, de se suicider si, par exemple, ils n'ont pas trouvé de travail dans la seconde. Il suffirait alors de vider la queue des travaux à réaliser et d'attendre une seconde pour que tout s'arrête.

- Passer à une solution QThread ou QProcess de PyQt au lieu du multiprocessing de Python, parce que eux peuvent d'arrêter sur demande avec .terminate.

Cette méthode terminate n'est pas recommandée et constitue un arrêt "sauvage" du travail en cours, mais elle est homogène avec la brutalité d'un "Cle-C"! Pour l'utiliser de temps en temps, je peux témoigner que ça marche, à condition de désactiver juste avant les liens (disconnect) établis correspondant aux signaux, et de ne pas vouloir récupérer les résultats des travaux en-cours qu'on a ainsi arrêté!

Dans le principe, on doit pouvoir monter un pool de QThread (ou QProcess) qui s'alimentent en travail à une QQueue, et donne le résultat du travail à une autre QQueue, en protégeant les accès à ces 2 queues avec des QMutex.

Mais bien sûr, c'est un peu plus compliqué que concurrent.futures...

j'ai testé ça qui fonctionne (pas très propre) mais cela augmente la durée de travail de 25% :
je lance mon QThread et je fous une boucle qui utilise une variable bloquante... qui se débloque une fois le travail fini...

Code:

---

1 2 3 4 5 6 7 8 9

GlobalVar["MonProgressThread"].start()   ## Démarrage du thread try: while not GlobalVar["Termine"]: pass   except KeyboardInterrupt: self.Stop()

---

en fait ce qui pose soucis, c'est seulement la récupération de l'info KeyboardInterrupt, le travail en cours, je peux tout arrêter sans problème.

Pensait vous qu'il aurait moyen de lancer un QThread qui surveille stderr dans son coin (et qui enverrait un message d'arret du taf en cours) sans me bouffer trop de processeur ?

Si y a rien qui permettrait de faire ce genre de chose, je me pencherais sur ta proposition tyrtamos.

En tout cas, merci beaucoup à vous pour vos éclairages et vos partages !!

EDIT :
avec un sleep de 0.2s à la place du pass, ça semble ok niveau temps de travail

Code:

---

1 2 3 4 5 6 7 8 9

GlobalVar["MonProgressThread"].start()   ## Démarrage du thread try: while not GlobalVar["Termine"]: slep(0.2)   except KeyboardInterrupt: self.Stop()

---

Alors évidemment, la boucle ne serait lancé qu'en mode console

Salut,

Toujours dans l'idée de montrer la difficulté, j'ai "arrangé" le bricolage précédent en virant tout ce qui avait trait à Qt en gardant le minimum pour que çà fonctionne lancé depuis la console *et* en y collant le fameux try...except KeyboardInterrupt.
Ca donne:

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 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70

from concurrent.futures import ThreadPoolExecutor as Pool import subprocess import time import sys   class Launcher:   _abort = False _done = False   def __init__(self, count=10): self.count = count   def run_item(self, ident): if not self._abort: subprocess.call ('python -c "import time; time.sleep(0.5)"') return ident   def run(self):   step = 100 / self.count done_ratio = 0   def callback(future): nonlocal done_ratio if not self._abort: done_ratio += step #print('callback', done_ratio, 'ix', future.result()) self.on_update(done_ratio) if done_ratio >= 100 : self._done = True   pool = self.pool = Pool(max_workers=2) for z in range(self.count): future = pool.submit(self.run_item, z) future.add_done_callback(callback)   def shutdown(self): # ** too much ** self._abort = True self.pool.shutdown(wait=True)   def on_update(self, ratio): pass   class ConsoleLauncher(Launcher):   def on_update(self, ratio): sys.stdout.write('\r' + '*' * int(ratio/5)) sys.stdout.flush()     def wait(self, delay): if not self._done: time.sleep(delay) return self._done     if __name__ == '__main__': launcher = ConsoleLauncher()   launcher.run() while True: try: if launcher.wait(0.1): break except KeyboardInterrupt: print('***shutdown***') time.sleep(0.01) launcher.shutdown() break

---

çà raconte quoi?
En mode console, pas besoin de créer un Thread pour ne pas bloquer le GUI. Launcher se gère son pool de threads qui lancent les subprocess.call exécutant une commande externe. ConsoleLauncher "adapte" le Launcher au code principal: la logique est différente puisqu'on lance les différentes tâches, on affiche la progression et on attend que çà se termine ou un CTRL-C.
Dit autrement, si on veut que ce code sache se lancer avec un GUI ou avec une console, on a des logiques différentes qui s'appuie sur des adaptateurs adhoc qui iront titiller au bout du bout Launcher.
note: pour sûr, c'est plus compliqué que .show/.hide.
A partir de là:
1 - vérifier que ce code fonctionne sur vos OS: je me suis contenté de tester çà sous Windows,
2 - Je pourrais m'amuser à coder les bouts qui manquent pour le fun (lentement, j'ai plein d'autres truc à faire).
3 - à explorer: Qt est une bibliothèque pour le programmeur C++ et le programmeur C++ réalise aussi des programmes "console" (i.e. sans le GUI Qt). Donc plutôt qu'avoir un mode console sans Qt et un mode graphique avec Qt, il doit y avoir une ligne médiane à explorer (l'intérêt étant d'avoir quelque chose de plus cohérent et un peu moins de code).

Citation:

---

Envoyé par tyrtamos

Malheureusement, si j'ai bien compris la notice, les threads de concurrent.futures ne peuvent pas s'arrêter comme ça. En fait, tu es dans une contradiction: tu veux un truc simple et automatique (concurrent.futures), mais tu veux aussi continuer à maîtriser le détail pour remplir des fonctions particulières. J'ai bien peur que ce soit une impasse.

---

Les exemples que j'ai posté montrent qu'on peut arrêter ce bazar non?
La question telle que je l'ai comprise est un problème de design: comment avoir des fonctionnalités semblables que l'application soit lancée en mode graphique ou en mode console... C'est pas tellement le code qui est compliqué dans ce cas là mais la définition des différents composants et le rôle/responsabilité qu'on va pouvoir leur attribuer.

- W

Et voilà...

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 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112

import concurrent.futures as futures from concurrent.futures import ThreadPoolExecutor as Pool from PyQt5.QtCore import QObject, pyqtSignal   class Launcher(QObject):   update_done_ratio = pyqtSignal(float) finished = pyqtSignal()   def __init__(self, parent=None, max_workers=2): super().__init__(parent) self.pending_fs = set() self.done_fs = set() self.pool = Pool(max_workers=max_workers) self.finished.connect(lambda: self.pool.shutdown(wait=True))   def fs_callback(self, fs): self.pending_fs.remove(fs) self.done_fs.add(fs) done = len(self.done_fs) left = len(self.pending_fs) self.update_done_ratio.emit(done / (done + left)) if not self.pending_fs: self.finished.emit()   def submit(self, func): fs = self.pool.submit(func) self.pending_fs.add(fs) fs.add_done_callback(self.fs_callback) return fs   def abort(self): for fs in self.pending_fs.copy(): fs.cancel()   def console_app(argv, iterfunc=None):   import sys import signal import time   from PyQt5.QtCore import QCoreApplication   class Helper(Launcher): def __init__(self, max_workers=2): super().__init__(max_workers=max_workers) self.update_done_ratio.connect(self.update) self.finished.connect(app.quit) signal.signal(signal.SIGINT, self.signal_handler)   def update(self, ratio): size = 50 sys.stdout.write('\r' + '%-*s|(%2.1f)' % ( size, '*' * int(ratio * size), ratio * 100)) sys.stdout.flush()   def signal_handler(self, signal, frame): time.sleep(0.1) print ('ctrl-C') self.abort()   app = QCoreApplication([])   launcher = Helper()   for func in iterfunc: launcher.submit(func)   app.exec_()   def qt_app(argv, iterfunc=None):   from PyQt5.QtWidgets import QApplication, QProgressDialog, QPushButton     def do_launch(functions): launcher = Launcher() dialog = QProgressDialog("En cours...", "Annuler", 0, 100) launcher.update_done_ratio.connect(lambda x: dialog.setValue(100*x)) launcher.finished.connect(dialog.reset)   for func in functions: launcher.submit(func)   dialog.exec_()   app = QApplication([]) functions = list(iterfunc) # histoire de pouvoir faire X start btn = QPushButton('start') btn.clicked.connect(lambda: do_launch(functions)) btn.show()   exit(app.exec_())     if __name__ == '__main__': import time import sys   COUNT = 10   def gen_func(): def task(x): time.sleep(1) return(x) for z in range(COUNT): yield lambda x=z: task(x)   if len(sys.argv) > 1: console_app([], iterfunc=gen_func()) else: qt_app([], iterfunc=gen_func())

---

Dans cette mouture, le Launcher est un QObject qui permet d'adapter concurrent.futures aux besoins. Les différentes portions ont été pensées modules mais écrits sous la forme de fonctions (avec des imports qui traînent dans les premières instructions).

- W