Discussion :

[Adr.d]

Bonjour,

je réalise une plateforme qui applique des traitements à des fichiers.
Pour faire simple, lors du lancement, on récupère le répertoire des fichiers à traiter qui sera parcouru avec un walk, un gestionnaire prend le fichier courant et lui applique les traitements sélectionnés, pour finir on stocke les résultats.

Pour l'instant le traitement est sériel, pour un répertoire de 5 fichiers, A->E et si on applique les traitements 2,5 et 8 on aura :
traitement 2 sur A, tr5 sur A, tr8 sur A puis tr2 sur B, ... jusqu'à tr8 sur E

Dans l'optique de rendre cette plateforme plus rapide, on m'a demandé de faire une version parallèle.

Après avoir regardé des tutos, de la doc (thread, threading, queue, multiprocessing, ...), je voyais les choses comme ça. Au fur et à mesure du parcours du répertoire cible, le gestionnaire va remplir une queue (fifo du coup) avec les traitements sur les fichiers sous forme de tâches. Quand la queue est pleine, on attend. Et pendant ce temps on a les différents threads qui font les tâches en parallèle. Alors déjà, est-ce que cette démarche est correcte ?

Les traitements sont sous forme de classes, qui sont instanciées à la sélection et qui ont entre autres une fonction action(path).
Le problème que je rencontre est que, si j'ai bien compris, au moment de la création du thread, threading.Thread(), il faut lui fournir dans target un objet exécutable. Dans mon cas je devrais lui passer la fonction action() du traitement, ce qui m'embête puisque cela contraint un thread à ne travailler qu'avec un type de traitement (ou alors j'ai mal compris).

Ma question est donc : est-ce qu'il y a un moyen d'attribuer différentes actions à un thread alternativement comme ce que je voudrais faire ?

Sinon, peut-être qu'il faut placer l'action parallèle à un autre endroit ?

Merci

[Sve@r]

Bonjour

Ma question est donc : est-ce qu'il y a un moyen d'attribuer différentes actions à un thread alternativement comme ce que je voudrais faire ?

Pourquoi faire ? D'après ce que je lis, tu n'as qu'une seule action à faire (traiter le fichier).
Te suffit de créer un thread par traitement (un pour trt2, un pour trt5 et un pour trt8) et tu lances tes 3 threads sur ficA (en présumant évidemment que trt5 n'a pas besoin des données générées par trt2) et start les thread...

Ici un petit exemple illustratif

Code python :

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

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
#!/usr/bin/env python3
# coding: utf-8
 
import random
import sys
import time
from threading import Thread
from threading import RLock
 
class cTraitement(Thread):
	def __init__(self, mot, verrou, count=5):
		super().__init__()
		self.__mot=mot
		self.__verrou=verrou
		self.__count=count
	# __init__()
 
	def run(self):
		super().run()
		for i in range(self.__count):
			with self.__verrou:
				for l in self.__mot:
					sys.stdout.write(l)
					sys.stdout.flush()
				# for
			# with
			time.sleep(random.randint(1, 60) / 100.0 + 0.2)
		# for
	# run()
# class cTraitement
 
verrou=RLock()
thread=tuple(cTraitement(x, verrou) for x in ("canard", "tortue", "lapin"))
for t in thread: t.start()
for t in thread: t.join()

Chaque thread gère son animal. Plus un lock pour ne pas avoir de mélange à l'écran (pas de "capin" par exemple) mais qui te sera peut-être inutile...

[wiztricks]

Salut,

Ma question est donc : est-ce qu'il y a un moyen d'attribuer différentes actions à un thread alternativement comme ce que je voudrais faire ?

Vous pouvez utiliser des works queue mais avant de coder quoi que ce soit jetez un œil au module concurrent.futures qui est livré en standard.

- W

[Adr.d]

Bonjour,

merci pour vos réponses !

Te suffit de créer un thread par traitement

En effet, mais c'est justement ce que je voudrais éviter. Pourquoi ? Parce que ! Et plus sérieusement, au cas où. Je ne maitrise pas encore tout ça et je préfère ne pas me fermer de porte. De plus, ça fonctionne si c'est moi qui décide du nombre de thread. Or peut-être qu'on va m'imposer de laisser l'utilisateur choisir. Donc si j'ai trois threads et que deux traitements à effectuer, un thread sera inactif ? Ou à l'inverse, on m'impose quatre threads et j'ai 5 types de traitement à réaliser par fichier.
Bref, c'est pour ça que je cherche à rendre variable l'action réalisée par un thread.
En tout cas merci pour l'exemple, ça m'a montré l'utilisation des verrous, que j'utiliserai peut-être. Et au passage, pour l'instant les résultats des traitements ne dépendent pas les uns des autres.

jetez un œil au module concurrent.futures

Oui j'ai vu ça aussi mais la syntaxe qui est présentée me semble avoir le même problème.

Code :

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

1
2
3
with concurrent.futures.ThreadPoolExecutor(max_workers=NUM_WORKERS) as executor:
    futures = {executor.submit(check_website, address) for address in WEBSITE_LIST}
    concurrent.futures.wait(futures)

De ce que je vois, je dois fixer l'action qui sera exécutée dans le run dès le début.
Mais je vais regarder plus en détails les works queue et concurrent.futures

Encore merci pour les réponses

[wiztricks]

De ce que je vois, je dois fixer l'action qui sera exécutée dans le run dès le début.
Mais je vais regarder plus en détails les works queue et concurrent.futures

On crée un pool d'exécuteur/workers d'un côté et leur soumettez les actions à exécuter quand il y a besoin.
Mais passez du temps dans la documentation et à jouer avec des exemples: c'est pas si trivial.

- W

[Adr.d]

Bonsoir,

je pense avoir trouvé ce qui va correspondre à mon besoin. J'ai utilisé une queue et des threads.

Code :

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

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 threading
#import concurrent.futures
 
 
from process_one import *
from process_two import *
from queue import Queue
from threading import Thread
 
NUM_WORKERS = 4
task_queue = Queue()
 
 
class Manager:
    def __init__(self,listF):
        self.nameF = ["process_one","process_two"]
        self.threads = [Thread(target=self.exec_filtr) for _ in range(NUM_WORKERS)]
        [thread.start() for thread in self.threads]
 
    def exec_filtr(self):
        while True:
            tr,path = task_queue.get()
            print(str(tr.action(path)))
 
    def action(self,path):
        for tr in self.listF:
            task_queue.put((filtr,path))

Pour le pool de thread (concurrent.futures.ThreadPoolExecutor), je n'ai pas compris comment on rajoute des tâches à la liste après coup.

Ce n'est qu'une première version, pour vérifier le fonctionnement, il reste donc plusieurs points à retravailler.
Pour l'instant, une fois tous les fichiers traités, le programme ne s'arrête pas, est-ce qu'il y a une façon d'arrêter les threads quand ils n'ont plus rien à traiter dans la queue ?
Peut-être en utilisant Queue.get() avec un timeout et en gérant l'exception ?

Je n'ai pas précisé de maxsize pour la queue. C'est bien pratique car les tâches s'accumulent sans que je n'ai rien à gérer. Je l'ai testé sur une centaine de fichiers. Est-ce que ça peut poser un problème s'il y a plusieurs milliers, voir dizaines/centaines de milliers de fichiers à traiter ? Et si c'est le cas, quelle taille max une queue devrait avoir ?

Merci pour l'aide !

[wiztricks]

Pour le pool de thread (concurrent.futures.ThreadPoolExecutor), je n'ai pas compris comment on rajoute des tâches à la liste après coup.

Ça se fait avec .submit et il y a des exemples dans la documentation.

Ce n'est qu'une première version, pour vérifier le fonctionnement, il reste donc plusieurs points à retravailler.

Parfois coder par soi même va plus vite que de comprendre comment fonctionne une bibliothèque mais on profite aussi des solutions des anciens qui ont balayés les différents cas d'utilisation. Et ici, c'est très utile car la programmation "parallèle", c'est plein de pièges.
Après c'est vous qui voyez mais ré-inventer le fil à couper le beurre...

Est-ce que ça peut poser un problème s'il y a plusieurs milliers, voir dizaines/centaines de milliers de fichiers à traiter ? Et si c'est le cas, quelle taille max une queue devrait avoir ?

Soumettre un peu moins vite les taches à effectuer (en fonction de la longueur de la queue) est aussi une solution (et qui évite de se poser la question sur la dimension du backlog: quand il y a 10/20 jobs en attente, je me mets en pause plutôt qu'en ajouter).

- W

[Sve@r]

Et plus sérieusement, au cas où. Je ne maitrise pas encore tout ça et je préfère ne pas me fermer de porte.

Rien ne t'oblige à modifier ton programme qui fonctionne. Tu peux très bien partir sur une copie de test...

De plus, ça fonctionne si c'est moi qui décide du nombre de thread. Or peut-être qu'on va m'imposer de laisser l'utilisateur choisir. Donc si j'ai trois threads et que deux traitements à effectuer, un thread sera inactif ? Ou à l'inverse, on m'impose quatre threads et j'ai 5 types de traitement à réaliser par fichier.

Ah ben il faut être un peu souple dans la vie (justement ne pas se fermer de porte). sys.argv n'est pas fait que pour faire joli...

Imagine par exemple

Code python :

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def trt1(fic): ...
def trt2(fic): ...
def trt3(fic): ...
... (etc autant de fonctions que de traitements différents)...
 
work={
	"trt1" : trt1,
	"trt2" : trt2,
	"trt3" : trt3,
	...
}
 
for arg in sys.argv[1:]:
	work[arg](fichier)

Tu lances ce programme en lui passant la chaine "trt1 trt7" en paramètre, il n'exécutera que trt1() et trt7(). Si maintenant tu le relances en lui passant "trt2" il n'exécutera que trt2() et etc. Et tu associes chaque traitement demandé à un thread exactement comme avec mon ("canard", "tortue", "lapin")...