TNSI - TD Processus en Python : Le module multiprocessing

Création d’un Processus avec Process

La classe Process du module multiprocessing permet de créer un processus fils du processus courant.

1. Etudier le code incomplet ci-dessous..

   # creation.py (A sauvegarder sous ce nom-là)
   from multiprocessing import Process
   import os
   import time
   
   def f(nom):
       # A Compléter
       time.sleep(0.1)
   
   if __name__ == '__main__':
       p = Process(target=f, args=('bob',))  # Crée un processus fils
       p.start()
   

   Puis, compléter la partie A Compléter du code ci-dessus (en commentaire), de sorte à créer un processus fils et obtenir l’affichage suivant dans le Terminal :
   

   hello bob
   PID : 11481
   PID du père : 11476
   

2. Visualiser les processus dans la console (augmenter si nécessaire la durée du time.sleep().

Ordonnancement & Pool de Processus

La classe Pool du module multiprocessing permet de créer un pool (comprendre un groupe) de plusieurs processus fils, en parallèle.
La méthode map du module multiprocessing est parallélisée pour exécuter en parallèle, chaque processus p du pool. On utilise le gestionnaire de contexte with .. as .. de Python (comme pour les fichiers. Rappels ici)

On se donne le programme suivant, à copier-coller et sauvegarder sous le nom pool1.py :

# pool1.py
import time
from multiprocessing import Pool
import os

def f(x):
    time.sleep(0.5)
    print("\n",os.getpid())
    return x*x

if __name__ == '__main__':
    with Pool(10) as p:
        print(p.map(f, range(30)))
        # res=p.apply_async(f, (20,))
        #print(res.get(timeout=1))

Création d’un Pool de processus

1. Exécuter pool1.py, observer le nombre de fois consécutives qu’un même processus est utilisé. Que peut-on en conclure ?
2. Décocher le time.sleep() de la fonction f(), observer le nombre de fois consécutives qu’un même processus est utilisé. Que peut-on en conclure ?

Pile des Processus

1. Ouvrez 2 consoles, dans la 1ère, tenez-vous prêt à exécuter pool1.py. Dans la seconde, tapez $ vmstat -w 1 et exécutez. Le champ à observer est r (le premier) qui montre les processus en attente d’exécution. Puis exécutez pool1.py. Que constatez-vous concernant r ?

2. Recommencer l’opération en utilisant le programme suivant nommé pool2.py (au lieu du précédent pool1.py). Que constatez-vous concernant r ?

   # pool2.py
   import time
   from multiprocessing import Pool
   import os
   
   def f(x):
       # time.sleep(0.01)
       print("\n",os.getpid())
       return x*x
   
   if __name__ == '__main__':
       with Pool(10) as p:
           print(p.map(f, range(1000000)))
           # res=p.apply_async(f, (20,))
           # print(res.get(timeout=1))
   

Verrous

La Classe Lock du module multiprocessing permet de créer des verrous synchronisés, et dispose des méthodes .acquire() (pour verrouiller) et .release() (pour déverrouiller) le verrou.

1. Exécuter le programme suivant, nommé verrou.py. Que fait ce programme ? Compléter le chronogramme ci-dessous.

   # verrou.py
   from multiprocessing import Process, Lock
   import time
   
   def f(l, i):
       #l.acquire()
       print (l)
       time.sleep(0.02)
       print('hello world1 ', i)
       time.sleep(0.02)
       print('hello world2 ', i)
   
       #l.release()
   
       time.sleep(0.3)
       print('hello world3 ', i)
   
   if __name__ == '__main__':
       lock = Lock()
   
       for num in range(10):
           Process(target=f, args=(lock, num)).start()
   

   

2. Décocher les lignes l.acquire() et l.release(). Que constatez-vous ?

3. Mettre l.release() après print('hello world3 ', i). Que constatez-vous ?

Sémaphores

La classe Semaphore(n:int) du module multiprocessing permet d'instancier un sémaphore avec un nombre n de jetons (d'accès simultanés).

1. Un jeton
   Exécuter le programme suivant, nommé semaphore01.py, et mettre des numéros pour indiquer l’ordre d’exécution des tâches.

   # semaphore01.py (Sauvegarder sous ce nom)
   from datetime import datetime
   import time
   import multiprocessing
   from multiprocessing import Semaphore, Process
   
   def proc(sem, interval):
       print("Le processus: %r, demande un jeton, le nombre de jetons disponible est: %r"
               % (multiprocessing.current_process().name, sem.get_value()))
       #time.sleep(1)
       sem.acquire()
       try:
           print("Le processus: %r, a pris un jeton et utilise la ressource, le nombre de jetons disponible est %r."
               % (multiprocessing.current_process().name, sem.get_value()))
   
           time.sleep(interval*10)
       finally:
           print("Le processus: %r, libére un jeton et n'utilise plus la ressource."
               % (multiprocessing.current_process().name))
   
           sem.release()
       # print("Le nombre de jetons disponible est %r."
               # % (sem.get_value()))
   
   if __name__ == '__main__':
       sem = Semaphore(1)      # 1 seul jeton (accès simultané)
       print("le nombre de jetons disponible dans le sémaphore est: %r" % sem.get_value())
       for i in range(3):
           Process(target = proc, args = (sem, i)).start()
   

2. Plusieurs jetons
   Modifier le programme précédent, en le nommant semaphore2.py, de sorte à mettre 2 jetons dans le sémaphore. Expliquez comment est utilisée la ressource.

Inter blocage

Dans cette partie, 2 processus utilisent 2 ressources

Cas 1 (Tout se passe bien)

1. Exécuter le programme suivant, en le nommant semaphore03.py.

   # semaphore03.py
   from datetime import datetime
   import time
   import multiprocessing
   from multiprocessing import Semaphore, Process
   
   def proc1(sem1, sem2):
   
       sem1.acquire()
       try:
           print("Le processus: %r, a pris un jeton dans sem et utilise la ressource , le nombre de jetons disponible dans sem est %r."
               % (multiprocessing.current_process().name, sem1.get_value()))
           time.sleep(2)
       finally:
           print("Le processus: %r, libére un jeton dans sem et n'utilise plus la ressource ."
               % (multiprocessing.current_process().name))
   
           sem1.release()
       print("Le processus: %r, demande un jeton dans sem pour accèder à la ressource , le nombre de jetons disponible dans sem est: %r"
               % (multiprocessing.current_process().name, sem2.get_value()))
       sem2.acquire()
       try:
   
           print("Le processus: %r, a pris un jeton dans sem et utilise la ressource , le nombre de jetons disponible dans sem est %r."
               % (multiprocessing.current_process().name, sem2.get_value()))
   
           time.sleep(3)
       finally:
           print("Le processus: %r, libére un jeton dans sem et n'utilise plus la ressource ."
               % (multiprocessing.current_process().name))
   
           sem2.release()
   
   def proc2(sem1,sem2):
       sem2.acquire()
       try:
           print("Le processus: %r, a pris un jeton dans sem et utilise la ressource , le nombre de jetons disponible dans sem est %r."
               % (multiprocessing.current_process().name, sem2.get_value()))
           time.sleep(5)
       finally:
           print("Le processus: %r, libére un jeton dans sem et n'utilise plus la ressource ."
               % (multiprocessing.current_process().name))
   
           sem2.release()
       print("Le processus: %r, demande un jeton pour accèder à la ressource , le nombre de jetons dans sem disponible est: %r"
               % (multiprocessing.current_process().name, sem1.get_value()))
   
       sem1.acquire()
       try:
           print("Le processus: %r, a pris un jeton dans sem et utilise la ressource , le nombre de jetons disponible dans sem est %r."
               % (multiprocessing.current_process().name, sem1.get_value()))
   
           time.sleep(5)
       finally:
           print("Le processus: %r, libére un jeton dans sem et n'utilise plus la ressource ."
               % (multiprocessing.current_process().name))
   
           sem1.release()
   
   if __name__ == '__main__':
       sem1 = Semaphore(1)
       sem2 = Semaphore(1)
       print("le nombre de jetons disponible dans le sémaphore 1 est: %r" % sem1.get_value())
       print("le nombre de jetons disponible dans le sémaphore 2 est: %r" % sem2.get_value())
   
       Process(target = proc1, args = (sem1, sem2)).start()
       Process(target = proc2, args = (sem1, sem2)).start()
   

2. Modifier le programme précédent, de sorte que l'affichage respecte la logique suivante (vous n’aurez peut-être pas exactement le même ordre), avec les bons numéros de sémaphores, et de ressources,

   le nombre de jetons disponible dans le sémaphore 1 est: 1
   le nombre de jetons disponible dans le sémaphore 2 est: 1
   Le processus: 'Process-1', a pris un jeton dans sem1 et utilise la ressource 1, le nombre de jetons disponible dans sem1 est 0.
   Le processus: 'Process-2', a pris un jeton dans sem2 et utilise la ressource 2, le nombre de jetons disponible dans sem2 est 0.
   Le processus: 'Process-1', libére un jeton dans sem1 et n'utilise plus la ressource 1.
   Le processus: 'Process-1', demande un jeton dans sem2 pour accèder à la ressource 2, le nombre de jetons disponible dans sem2 est: 0
   Le processus: 'Process-2', libére un jeton dans sem2 et n'utilise plus la ressource 2.
   Le processus: 'Process-2', demande un jeton pour accèder à la ressource 1, le nombre de jetons dans sem1 disponible est: 1
   Le processus: 'Process-1', a pris un jeton dans sem2 et utilise la ressource 2, le nombre de jetons disponible dans sem2 est 0.
   Le processus: 'Process-2', a pris un jeton dans sem1 et utilise la ressource 1, le nombre de jetons disponible dans sem1 est 0.
   Le processus: 'Process-1', libére un jeton dans sem2 et n'utilise plus la ressource 2.
   Le processus: 'Process-2', libére un jeton dans sem1 et n'utilise plus la ressource 1.
   

Cas 2

1. Exécuter le programme suivant, sous le nom interblocage.py. Que se passe t-il ? Expliquez pourquoi.

   # interblocage.py
   from datetime import datetime
   import time
   import multiprocessing
   from multiprocessing import Semaphore, Process
   
   def proc1(sem1, sem2):
       #print("Le processus: %r, demande un jeton pour accèder à la ressource 1, le nombre de jetons disponible est: %r"
       #         % (multiprocessing.current_process().name, sem1.get_value()))
       #time.sleep(1)
       sem1.acquire()
       try:
   
           print("Le processus: %r, a pris un jeton dans sem1 et utilise la ressource 1, le nombre de jetons disponible dans sem1 est %r."
               % (multiprocessing.current_process().name, sem1.get_value()))
           time.sleep(2)
   
   
   
           print("Le processus: %r, demande un jeton pour accèder à la ressource 2, le nombre de jetons disponible disponible dans sem2 est: %r"
                   % (multiprocessing.current_process().name, sem2.get_value()))
           #time.sleep(1)
           sem2.acquire()
           try:
   
               print("Le processus: %r, a pris un jeton dans sem2 et utilise la ressource 2, le nombre de jetons disponible est %r."
                   % (multiprocessing.current_process().name, sem2.get_value()))
   
               time.sleep(3)
           finally:
               print("Le processus: %r, libére un jeton et n'utilise plus la ressource 2."
                   % (multiprocessing.current_process().name))
   
               sem2.release()
       finally:
           print("Le processus: %r, utilise la ressource 1 et à également besoin de la ressource 2."
               % (multiprocessing.current_process().name))
           sem1.release()
   
   def proc2(sem1,sem2):
   
       sem2.acquire()
       try:
   
           print("Le processus: %r, a pris un jeton dans sem2 et utilise la ressource 2 et à également besoin de la ressource 1, le nombre de jetons disponible disponible dans sem1 est %r."
               % (multiprocessing.current_process().name, sem1.get_value()))
           time.sleep(5)
   
           sem1.acquire()
           try:
   
               print("Le processus: %r, a pris un jeton et utilise la ressource 1, le nombre de jetons disponible est %r."
                   % (multiprocessing.current_process().name, sem1.get_value()))
   
               time.sleep(5)
           finally:
               print("Le processus: %r, libére un jeton et n'utilise plus la ressource 1."
                   % (multiprocessing.current_process().name))
   
               sem1.release()
       finally:
           print("Le processus: %r, libére un jeton et n'utilise plus la ressource 2."
               % (multiprocessing.current_process().name))
           sem2.release()
   
   if __name__ == '__main__':
       sem1 = Semaphore(1)
       sem2 = Semaphore(1)
       print("le nombre de jetons disponible dans le sémaphore 1 est: %r" % sem1.get_value())
       print("le nombre de jetons disponible dans le sémaphore 2 est: %r" % sem2.get_value())
       Process(target = proc1, args = (sem1, sem2)).start()
       Process(target = proc2, args = (sem1, sem2)).start()
   

Le join de Python

1. Exécuter le programme suivant sous le nom join.py. Que constatez-vous ?

   # join.py
   from multiprocessing import Process
   import time
   
   def f(i):
   
       time.sleep(0.02)
       print('hello world1 ', i)
       time.sleep(0.02)
       print('hello world2 ', i)
   
       time.sleep(0.3)
       print('hello world3 ', i)
   
   if __name__ == '__main__':
       for num in range(10):
           p = Process(target=f, args=(num,))
           p.start()
           #p.join()
   

2. Décommenter la ligne p.join() et réexécuter. Que se passe t-il ? A quoi sert join() ?