Discussion :

[Arrakis]

Bonjour,

C'est plus une question générale concernant les threads mais je bosse en python donc je poste ici,

Comment faire pour qu'un programme qui ne comporte qu'une fonction récursive puisse utiliser les différents processeurs du pc sur lequel il est éxécuté ?

Le seul moyen je pense est de lancer un nouveau thread à chaque appel de la fonction, "acquérir" un verrou au debut de la fonction et le libérer avant l'appel récursif, est-ce bien la méthode appropriée ?

Je précise que la fonction ne retourne rien, ne modifie pas ses arguments et enregistre les résultats dans une variable globale, voila le schéma général :

Code :

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

1
2
3
4
5
6
7
8
9
 
def solveur(<arguments>):
    if <condition arret> :
        <modification de la variable globale>
        return
    boucle :
        <traitement>
        if <condition>:
            solveur(<arguments>)

Merci pour votre aide

edit :
un peu de precisions:

ce que je voudrai faire c'est :
- lancement des n threads de profondeur 1 (ceux créés lors du premier appel) et les mettre en attente tant que le premier appel n'est pas terminé
- execution des n threads et création des m threads de profondeur 2, attente des m threads que les n soient terminés
- execution des m threads...
etc...

Je ne sais pas si je suis clair

[wiztricks]

Bonsoir,

Il n'est peut être pas raisonnable d'avoir un nombre de threads très supérieur par rapport au nombre de CPU disponibles.

CPython utilise un Global InterLock (GIL) faisant que les threads sont mono-cpu par défaut.

Note: Si vous appelez une bibliothèque C elle pourra rentre le GIL et permettre une exécution multi-cpu - cas de nombreux drivers de bases de données -.

Avec une version de Python >= 2.6, vous avez une bibliothèque 'multiprocessing' qui passe à côté du GIL en créant des "process" plutôt que des threads et vous offre une API qui a de nombreuses similitudes avec celle de threads.

Pour le reste, sans trop réfléchir à votre problème, je dirais que la récursivité suppose une sauvegarde du contexte de l'appelant dans la pile d'exécution. Les threads ayant leur propre contexte d'exécution, il me semble difficile de paralléliser sans dé-récurser (i.e. rendre visible tout ou partie du contexte caché dans la pile).
Comme la fonction est tail-recursive, le contexte à sauver est plus "simple".

Un autre axe pourrait être de partitionner le problème.

J'espère que d'autres auront de meilleures idées.
- W

[Arrakis]

Bonjour,

Avec une version de Python >= 2.6, vous avez une bibliothèque 'multiprocessing' qui passe à côté du GIL en créant des "process" plutôt que des threads et vous offre une API qui a de nombreuses similitudes avec celle de threads.

Je suis justement en train de regarder de ce coté, il y a un mecanisme de Queue que je n'arrive pas bien a comprendre mais c'est surement la solution, si quelqu'un avait des précisions sur cette librairie...

Pour le reste, sans trop réfléchir à votre problème, je dirais que la récursivité suppose une sauvegarde du contexte de l'appelant dans la pile d'exécution. Les threads ayant leur propre contexte d'exécution, il me semble difficile de paralléliser sans dé-récurser (i.e. rendre visible tout ou partie du contexte caché dans la pile).
Comme la fonction est tail-recursive, le contexte à sauver est plus "simple".

Justement j'ai précisé que la fonction ne renvoyait rien, l'appelé n'a pas besoin de connaître le contexte de l'appelant, tout ce dont il doit savoir se trouve dans ses arguments, il me semble que cela change quelquechose, ou alors je n'ai pas compris ce que vous avez dit ici

edit : le contexte de l'appelant ne serait pas justement l'état des arguments lors de l'appel ?

[Arrakis]

J'avance petit à petit avec le module multiprocessing, j'arrive a créer plusieurs processus et a partager de la mémoire entre les processus pour enregistrer les solutions (grace aux Queue justement), voila maintenant a quoi ressemble le script :

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
 
from multiprocessing import Process,Queue
 
def solveur(<arguments>,q):
    if <condition arret> :
        q.put(solution)
        return
    boucle :
        <traitement>
        if <condition>:
            p=Process(target=solveur,args=(<arguments>,q))
            p.start()
            p.join()
 
q=Queue()
solveur(<arguments>,q)
 
print q.qsize(),"solutions"

Cela marche comme je veux, les résultats sont les mêmes qu'avec la version précédente.
Plusieurs problèmes restent :
- Mes deux processeur sont bien utilisés, mais a environ 50% chacun
- C'est beaucoup plus lent avec cette version, environ 10 fois !! la cause doit être liée au 1er problème

La cause de la lenteur provient peut-être de la méthode join(), elle force l'appelant à attendre que le processus appelé soit terminé pour continuer, a priori cela ne me pose pas de problème qu'il attende ou pas, mais lorsque j'enleve la ligne p.join(), les deux processeurs montent en charge directement et le pc plante presque immédiatement...

Quelqu'un aurait-il une idée ?
Merci pour vos réponses

[DelphiManiac]

C'est vraiment fait à l'arrache, vu que je ne connaissais pas les mécanismes auparavant.

C'est un quicksort adapté au traitement en parallèle.

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
from random import randint
from multiprocessing import Process, Queue, current_process, freeze_support
from time import clock
from sys import argv
 
def worker(input, output):
    for args in iter(input.get, 'STOP'):
        result = quicksort(args)
        output.put(result)
 
def quicksort(q):
    pos = q[0]
    q = q[1]
 
    for x in xrange(100000):
        pass
 
    lesser=[]
    greater=[]
    if len(q) <= 1:
        return q
    for i in q[1:]:
        if i < q[0]:
            lesser.append(i)
        else:
            greater.append(i)
    return [pos, [lesser, q[0:1], greater]]
 
def test(NUMBER_OF_PROCESSES):
    #NUMBER_OF_PROCESSES = 4
    #NUMBER_OF_PROCESSES = 1
 
    task_queue = Queue()
    done_queue = Queue()
 
    for i in range(NUMBER_OF_PROCESSES):
        Process(target=worker, args=(task_queue, done_queue)).start()
 
    q = []
    for x in xrange(10000):
        q.append (randint(0, 100000))
 
 
    print "Nombre de nombre : ", len(q)
#    q = [500, 250, 251, 551, 552]
    q_ori = q[:]
    q = [q]
#    print q
 
    end = False
 
    while not end:
        res = {}
        tasks = []
 
        pos = 1
        nbr_task = 0
        for v in q:
            if len(v) > 1:
                res[pos] = []
                task_queue.put([pos, v])
                nbr_task += 1
            else: 
                res[pos] = [v]
 
            pos += 1
 
        end = True
        print nbr_task
        for i in range(nbr_task):
            end = False
            quick_sort_res = done_queue.get()
            item_index = quick_sort_res[0]
            res[item_index] = [value for value in quick_sort_res[1] if value]
 
#        print res
        q = []
        for value in res.itervalues():
            for value1 in value:
                q += [value1]
#        print q
            #res += [value for value in quick_sort_res if value]
            #print res
 
 
    for i in range(NUMBER_OF_PROCESSES):
        task_queue.put('STOP')
 
    q = [value[0] for value in q]
    q_sorted = sorted(q_ori)
 
    print q_sorted == q
#    print [value[0] for value in q]
 
if __name__ == '__main__':
    if len(argv) < 2:
        print "Passer en argument le nombre de process."
    else:
        NUMBER_OF_PROCESSES = int(argv[1])
        freeze_support()
        t0 = clock()
        test(NUMBER_OF_PROCESSES)
        print clock() - t0

Suivant que tu lances le projet avec 1 ou 2 process (paramètre à passer en ligne de commande), le temps est a peu près divisé par 2.

Je ralenti volontairement le traitement dans la méthode quicksort (avec une boucle for), sinon le processeur passe plus de temps à gérer l'ordonancement des taches que de traiter les tâches elles mêmes.

Moralité ce genre de traitement n'est intéressant que si les tâches à lancer en parallèle prennent un peu de temps, sinon l'overhead est important. D'une autre coté mon code étant loin d'être optimal, il génère lui même beaucoup d'overhead dans le thread principal (dictionnaire, copie de dictionnaire, de liste, etc :/).


Concernant le fait que le proc ne monte pas à plus de 50% c'est normal, enfin j'ai toujours vu ça. 100% = la totalité des processeurs, donc sur un système à 2 proc = 50% chacun; à 4, 25% chacun, etc...

Je rappelle que c'est vraiment très moche, pas optimisé et écrit à l'arrache ^^

[Arrakis]

Mon code précédent générait beaucoup trop de processus, voila pourquoi il etait si lent (enfin je pense).

J'ai finalement réussi avec le module pp en m'inspirant de l'exemple situé ici qui implémente lui aussi un quicksort, ca pourrait être intérréssant de comparer les performances.

Concernant le fait que le proc ne monte pas à plus de 50% c'est normal, enfin j'ai toujours vu ça

Avec pp, les deux proc montent bien a 100% (enfin a partir du moment que plus d'1 processus est créé), et on peut également régler le nombre de processeur utilisés et le nombre de processus maximal à créer, c'est assez pratique.
Par contre le temps de calcul n'est pas divisé par deux, j'arrive a un quotient de 1.6 au mieux, peut être que multiprocessing fait mieux...

Je vais essayer de décrypter ton code pour tester le module multiprocessing afin de comparer les deux.

merci.