Discussion :

[Julien N]

Salut.

Actuellement mon architecture est la suivante:

1. Une interface (point d'entrée du programme)
2. Un thread "principal", crée depuis l'interface, lance une simulation. Cette simulation correspond à l'exécution d'un certain nombre de tâches. Ces tâches peuvent être soit nombreuses et rapides, soit peu nombreuses mais très longues. Afin d'implémenter un mécanisme de pause/resume/stop, ces tâches sont à leur tour exécutées dans des threads (en réalité un seul traitant chaque tâche au fur et à mesure).

Tout cela marche assez bien. Mais comme indiqué ci-dessus, certaines simulations font appels à des tâches longues (je parle ici d'heures). Je souhaite désormais en exécuter plusieurs en parallèle. Si j'augmente le nombre de threads en charge de traiter les tâches, je ne gagne rien en terme de temps calcul, j'en perds même car les calculs en question sont gournands. Je pense donc me tourner vers du multiprocessing.

Le problème c'est que je sèche complètement quand il s'agit de coupler le multithreading et le multiprocessing.

J'ai écrit un démonstrateur (ci-dessous) où j'ai enlevé le surplus (dont les méthodes permettant de mettre en pause les calculs):

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
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
import threading, time
from multiprocessing import Process
import queue
import random
 
random.seed(1234)
 
 
class Worker(threading.Thread):
 
    def __init__(self, target, tasks, results, name='', **kwgs):
 
        super(Worker, self).__init__()
 
        self.target = target
        self.tasks = tasks
        self.results = results
        self.name = name
 
        self._kwgs = kwgs
 
    def run(self):
 
        while not self.tasks.empty():
 
            task = self.tasks.get()
            res = self.target(*task, **self._kwgs)
            print('  WORKER-%s : %s' % (self.name, res))
            self.results.put((task, res))
 
 
class ThreadedTasks(threading.Thread):
 
    def __init__(self, target, tasks, nprocs=1, **kwgs):
 
        super(ThreadedTasks, self).__init__()
 
        self.target = target
 
        # Create a queue where tasks will be put and shared with workers
        self._generator_tasks = tasks
        self._qtasks = queue.Queue(maxsize=nprocs)
 
        # Create a queue to share resutls between threads
        self._qresults = queue.Queue()
 
        # Create workers and put the same number of tasks in the queue,
        # otherwise they will finish their "job" immediately.
        self._workers = []
        for i in range(nprocs):
 
            task = next(self._generator_tasks)
            self._qtasks.put(task, block=True)
 
            # Build worker
            worker = Worker(
                target, self._qtasks, self._qresults, name=str(i), **kwgs,
                )
            self._workers.append(worker)
 
    def run(self):
 
        # Start workers
        for worker in self._workers:
            worker.start()
 
        for task in self._generator_tasks:
            self._qtasks.put(task, block=True)
 
        # Wait for workers to finish
        for worker in self._workers:
            worker.join()
 
class Solver(object):
 
    def __init__(self, name):
        self.name = name
 
    def __repr__(self):
        return self.name
 
    def __str__(self):
        return str(self.name)
 
    def solve(self, *args, **kwgs):
        timelaps = round(random.random(), 3)
        time.sleep(timelaps)
        return timelaps
 
 
def func(solver, *args, **kwgs):
    return solver.solve(*args, **kwgs)
 
def main():
 
    # Create a generator of tasks. The main advantage of the generator is
    # its light memory footprint. Especially for large number of tasks
    tasks = ((Solver(i), i) for i in range(50))
 
    # Create main thread
    thread = ThreadedTasks(func, tasks, nprocs=3)
 
    # Start main thread
    thread.start()
 
    # Wait until the thread exits
    thread.join()
 
if __name__ == '__main__':
 
    main()

Le code ci-dessus est la version sans multi-processing fonctionnant bien pour moi (commentaires acceptés). Naivement je pensais ajouter ceci pour lancer la fonction target dans un process:

Code :

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

1
2
3
4
5
6
 
q= queue.Queue()
p = Process(target=my_function, args=(q, 1))
p.start()
p.join()
res = q.get()

Ce qui mis ensemble ferait (juste la classe Worker:

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
28
class Worker(threading.Thread):
 
    def __init__(self, target, tasks, results, name='', **kwgs):
 
        super(Worker, self).__init__()
 
        self.target = target
        self.tasks = tasks
        self.results = results
        self.name = name
 
        self._kwgs = kwgs
 
    def run(self):
 
        def process_func_wrapper(func, q, *args, **kwargs):
            q.put(func(*args, **kwargs))
 
        while not self.tasks.empty():
 
            task = self.tasks.get()
            q = queue.Queue()
            p = Process(target=process_func_wrapper, args=(self.target, q, *task), kwargs=self._kwgs)
            p.start()
            p.join()
            res = q.get()
            print('  WORKER-%s : %s' % (self.name, res))
            self.results.put((task, res)

J'ai alors un beau problème de méthode non "pickable". Je ne vois pas comment le contourner. Et je me dis que je suis en train de constuire une machine à gaz. Quel serait selon vous la meilleure approche pour réaliser ce que je souhaite?

Exception in thread 0:
Traceback (most recent call last):
File "C:\users-apps\Anaconda3\lib\threading.py", line 914, in _bootstrap_inner

self.run()
File "pause_thread_ter.py", line 33, in run
p.start()
File "C:\users-apps\Anaconda3\lib\multiprocessing\process.py", line 105, in st
art
self._popen = self._Popen(self)
File "C:\users-apps\Anaconda3\lib\multiprocessing\context.py", line 212, in _P
open
return _default_context.get_context().Process._Popen(process_obj)
File "C:\users-apps\Anaconda3\lib\multiprocessing\context.py", line 313, in _P
open
return Popen(process_obj)
File "C:\users-apps\Anaconda3\lib\multiprocessing\popen_spawn_win32.py", line
66, in __init__
reduction.dump(process_obj, to_child)
File "C:\users-apps\Anaconda3\lib\multiprocessing\reduction.py", line 59, in d
ump
ForkingPickler(file, protocol).dump(obj)
TypeError: can't pickle _thread.lock objects

Ciao

Julien

[wiztricks]

Salut,

Le problème c'est que je sèche complètement quand il s'agit de coupler le multithreading et le multiprocessing.

Je ne vois pas l'intérêt d'avoir un thread pour chaque process. Il est au plus nécessaire d'avoir un thread pour lancer, récupérer les résultats i.e. gérer l'ensemble des "process".

De plus pour distribuer le boulot entre différents process, le module concurrent.futures devrait vous éviter d'avoir à recoder tout çà vous même.

- W

[Julien N]

Je ne vois pas l'intérêt d'avoir un thread pour chaque process.

Moi non plus, j'avoue. Initialement je suis parti sur l'idée de créer un thread (dans mon cas la classe ThreadedTasks) afin d'avoir les fonctionalités suivantes:

• Traiter une liste (ou ici un générateur) de tâches
• Etre en mesure de mettre en pause ou stopper le traitement des tâches (que je réalise à l'aide d'events)

J'utilise souvent concurrent.futures mais les exemples sur lesquels je me suis basé traitaient tous du threading.

D'un point de vue architecture, vous me conseillerez de remplacer complètement les classes ThreadedTasks et Worker au profit d'une nouvelle classe entièrement basée sur concurrent.futures?
Lorsque je fais du multiprocessing j'utilise le snippet suivant:

Code :

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

1
2
3
4
5
6
7
8
9
10
from concurrent.futures import ProcessPoolExecutor
 
def concurrent_process(func, jobs_to_do, nprocs=None):
    # Let the executor divide the work amongst processes by using 'map'.
    jobs_done = {}
    with ProcessPoolExecutor(max_workers=nprocs) as executor:
        for job, res in [(job, executor.submit(func, *job)) for job in jobs_to_do]:
            res.result()
            pass
    return jobs_done

J

[wiztricks]

D'un point de vue architecture, vous me conseillerez de remplacer complètement les classes ThreadedTasks et Worker au profit d'une nouvelle classe entièrement basée sur concurrent.futures?

C'est çà (modulo le fait que pour moi "class" = organisation du code: concurrent.futures ne vous impose pas d'utiliser des "class" juste de choisir un Pool de threads ou de Process).

- W

[Julien N]

Loin d'être trivial cette affaire.

Je suis parvenu tant bien que mal à lancer en parallèle des tâches depuis un thread, tout en conservant la possibilité de les annuler ou de faire une pause. Voici ce que cela donne (code complet):

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
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
import random
import time
import threading
import queue
from concurrent.futures import ProcessPoolExecutor, as_completed
 
class TasksProcessor(threading.Thread):
 
    def __init__(self, target, tasks, nprocs=1, **kwgs):
 
        super(TasksProcessor, self).__init__()
 
        self.target = target
        self._generator_tasks = tasks
 
        # List of futures
        # TODO: find a way to use generators instead
        self._futures = []
 
        self._max_workers = nprocs
 
        self._results = []
 
        # Flag to state if the processes shall be paused or stopped
        self._stopped = False
        self._paused = False
 
    def run(self):
 
        def done(f):
            while self._paused:
                pass
 
        with ProcessPoolExecutor(max_workers=self._max_workers) as executor:
 
            for idx, task in enumerate(self._generator_tasks):
                future = executor.submit(self.target, *task)
                future.add_done_callback(done)
                self._futures.append((idx, future))
 
            for idx, f in self._futures:
                if f.cancelled():
                    self._results.append((idx, None))
                else:
                    self._results.append((idx, f.result()))
 
    def pause(self):
        self._paused = True
 
    def resume(self):
        self._paused = False
 
    def stop(self):
        self._stopped = True
        for _, f in reversed(self._futures):
            f.cancel()
 
    @property
    def stopped(self):
        return self._stopped
 
    @property
    def results(self):
 
        _results = sorted(self._results, key=lambda res: res[0])
        results = [res[1] for res in _results]
 
        return results
 
class Solver(object):
 
    def __init__(self, name):
        self.name = name
 
    def __repr__(self):
        return self.name
 
    def __str__(self):
        return str(self.name)
 
    def solve(self, *args, **kwgs):
        timelaps = round(random.random(), 3)
        time.sleep(timelaps)
        print('SOLVER %s: %s' % (self.name, timelaps))
        return timelaps
 
 
def func(solver, *args, **kwgs):
    return solver.solve(*args, **kwgs)
 
def main():
 
    # Create a generator of tasks. The main advantage of the generator is
    # its light memory footprint. Especially for large number of tasks
    tasks = ((Solver(i), i) for i in range(50))
 
    processor = TasksProcessor(func, tasks, nprocs=3)
 
    print('start')
    processor.start()
    # Wait for 3 secs before sending the pause request
    for i in range(3):
        time.sleep(1)
    print('Pause')
    processor.pause()
    time.sleep(10)
    print('Resume')
    processor.resume()
    print('Want to interrupt NOW!')
    processor.stop()
 
    processor.join()
 
    print(processor.results)
 
if __name__ == '__main__':
 
    main()

A mon avis, c'est loin d'être safe. Faut que je réfléchisse à comment m'assurer que les processus soient tués proprement si le thread est stoppé brutalement.

Autre chose, je ne parviens pas à utiliser un générateur de futures. En effet, en faisant cela:

Code :

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

1
2
 for idx, task in enumerate(self._generator_tasks):
    future = executor.submit(self.target, *task)

Je perds tout l'intérêt d'avoir mes tâches sous la forme d'un générateur. Dans mon cas chaque tâche implique la création d'un modèle numérique assez lourd (plusieurs centaines de mo en mémoire vive). Tel que réalisé mon code va commencer par créer N modèles avant de commencer les traitements. C'est pas l'idéal. Mais je ne comprends pas comment on peut créer une liste dynamique de taille fixe de futures, et traiter les résultats au fur et à mesure que les analyses sont terminées.

Je ne vois rien dans la doc de concurrent.futures. Peut-être est-ce réalisable avec asyncio?

Dernier point. Quand je lance la commande "pause", j'ai 2*N process que se terminent avec N le nombre max de workers. C'est parce que je check un booléen qu'après le processus sans doute. Mais comment réduire ce nombre?

J

[wiztricks]

Salut,

A mon avis, c'est loin d'être safe. Faut que je réfléchisse à comment m'assurer que les processus soient tués proprement si le thread est stoppé brutalement.

A priori, sauf à avoir dit daemon=True, c'est pas si facile de tuer un Thread.
Mais pour un premier jet, je ne comprends pas pourquoi vous compliquez déjà avec les threads.

e perds tout l'intérêt d'avoir mes tâches sous la forme d'un générateur. Dans mon cas chaque tâche implique la création d'un modèle numérique assez lourd (plusieurs centaines de mo en mémoire vive). Tel que réalisé mon code va commencer par créer N modèles avant de commencer les traitements.

La grosse différence entre Thread et process, est côté de la mémoire partagée (ou pas). C'est ce qui fait qu'avec les process, çà va recopier des objets (et planter lorsque c'est pas "pickable" puisqu'ils seront sérialisés avec cette mécanique là). Pourquoi ne pas construire le modèle au démarrage du process qui fera des calculs dessus?

Dernier point. Quand je lance la commande "pause", j'ai 2*N process que se terminent avec N le nombre max de workers. C'est parce que je check un booléen qu'après le processus sans doute. Mais comment réduire ce nombre?

Oula... vous me sortez des nombres bien trop précis pour des activités fictives qui attendent un nombre variable de pou-ièmes de secondes. Occuper un worker pendant une durée fixe plus ou moins longue devrait vous donner d'autres comportements.

- W

[Julien N]

Salut,

Si ce que je fais ne semble pas très clair, c'est que ça ne l'est effectivement pas pour moi.

A priori, sauf à avoir dit daemon=True, c'est pas si facile de tuer un Thread.

Initialement j'avais daemonisé mon thread. Mais je me suis rendu compte que si les calculs sont en pause l'utilisateur ne peut plus fermer proprement l'interface. Du moins sans autres modifications de ma part. Je n'ai pas encore creuser le sujet.

La grosse différence entre Thread et process, est côté de la mémoire partagée (ou pas). C'est ce qui fait qu'avec les process, çà va recopier des objets (et planter lorsque c'est pas "pickable" puisqu'ils seront sérialisés avec cette mécanique là). Pourquoi ne pas construire le modèle au démarrage du process qui fera des calculs dessus?

Mon principal problème est mon objet Solver. Dans le code (le vrai, pas le démonstrateur posté jusqu'à présent), le solver crée à l'initialisation un modèle numérique d'un logiciel tiers. Ce modèle est conservé en mémoire. Un peu comme si un Workbook Excel était créé, puis traité. Les modèles sont indépendants et ne doivent pas être partagés avec les processus.
Alors j'ai conscience que ce serait plus facile d'alléger l'iintialisation et de créer effectivement le modèle au lancement des analyses. Sauf qu'il faudrait que je modifie pas mal de chose d'un programme assez compliqué. Voilà pourquoi j'ai pensé aux générateurs.

Oula... vous me sortez des nombres bien trop précis pour des activités fictives qui attendent un nombre variable de pou-ièmes de secondes. Occuper un worker pendant une durée fixe plus ou moins longue devrait vous donner d'autres comportements.

Bien vu...

Je suis finalement parvenu à faire un truc qui me semble pas trop mal. En tout cas je peux traiter un générateur de tâches, stopper les calculs (enfin de pas lancer de nouveaux workers), et mettre en pause.

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
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
class TasksProcessor(threading.Thread):
 
    def __init__(self, target, tasks, nprocs=1, **kwgs):
 
        super(TasksProcessor, self).__init__()
 
        self.target = target
        self._generator_tasks = tasks
 
        self._max_workers = nprocs
 
        self._results = []
 
        # Flag to state if the processes shall be paused or stopped
        self._stopped = False
        self._paused = False
 
    def run(self):
 
        def add_future(fs):
            for idx, task in enumerate(self._generator_tasks):
                future = executor.submit(self.target, *task)
                futures[future] = idx
                yield
 
        with ProcessPoolExecutor(max_workers=self._max_workers) as executor:
 
            futures = {}
            generator_futures = add_future(futures)
 
            # Start by generating as many futures as workers
            for i in range(self._max_workers):
                next(generator_futures)
 
            while futures and not self._stopped:
 
                # Check for status of the futures which are working
                done, not_done = concurrent.futures.wait(
                    futures, return_when=concurrent.futures.FIRST_COMPLETED,
                    )
 
                # Process any compeleted futures
                for future in done:
                    _idx = futures[future]
                    try:
                        self._results.append((_idx, future.result()))
                    except:
                        pass
 
                    # Remove the completed task
                    del futures[future]
 
                    # Try to add a new worker
                    try:
                        next(generator_futures)
                    except:
                        pass
 
                while self._paused:
                    pass
 
    def pause(self):
        self._paused = True
 
    def resume(self):
        self._paused = False
 
    def stop(self):
        self._stopped = True
 
    @property
    def stopped(self):
        return self._stopped
 
    @property
    def results(self):
 
        _results = sorted(self._results, key=lambda res: res[0])
        results = [res[1] for res in _results]
 
        return results

Merci pour vos commentaires!

J