Tutoriel pour apprendre la programmation parallèle en Python

Pour cela, j'ai préparé un ensemble d'URL pour les images à télécharger, ceci permettant des tests plus stables pour ce tutoriel (si nous ne tenons pas compte de la mise en cache du routeur Web). Cependant, je vais vous montrer les sources vous permettant de manipuler le code.

Regardons maintenant le code :

__author__ = 'GHajba'

from urllib.request import urlopen, Request
import json
import os


def get_image_urls(client_id):
    headers = {'Authorization': 'Client-ID {0}'.format(client_id)}
    with urlopen(Request('https://api.imgur.com/3/g/memes/', headers=headers)) as response:
        data = json.loads(response.read().decode('utf-8'))
    return map(lambda image: image['link'], data['data'])


def download_images(target_dir, url):
    path = target_dir + '/' + os.path.basename(url)

    with open(path, 'wb') as file:
        file.write(urlopen(url).read())

Ce code télécharge les derniers éléments de la page « memes » de imgur.com dont les liens pointent vers des fichiers .png ou jpg. Je vais éviter d'entrer dans les détails sur la façon d'obtenir votre «  IMGUR_CLIENT_ID », mais cela doit être spécifié dans l'environnement, car l'API imgur requiert une authentification qui est fournie à travers un id client.

Néanmoins, ce code télécharge les images une à une.

Sur mon Mac, l'exécution du code donne les résultats suivants :

52 images téléchargées en 13.790853023529053 secondes avec Python 3
52 images téléchargées en 15.189190864562988 secondes avec Python 3
52 images téléchargées en 13.965453863143921 secondes avec Python 3
52 images téléchargées en 13.087532997131348 secondes avec Python 3
52 images téléchargées en in 14.43852710723877 secondes avec Python 3

Comme vous pouvez le voir, cela prend en moyenne 14 secondes pour télécharger 52 images depuis imgur en utilisant seulement un des huit cœurs dont je dispose.

Dans cet exemple, je vais utiliser une version pas vraiment optimisée de décomposition en facteurs premiers. Ce calcul est très intensif au niveau CPU. Cela signifie qu'il requiert plus de puissance de calcul de la part de l'ordinateur que l'exemple précédent, où nous passions pas mal de temps à attendre du réseau toutes les données requises.

Le code de calcul des facteurs premiers ressemble à celui-ci :

def factors(result, n):
    if n <= 1:
        return result
    for i in range(2, n + 1):
        if n % i == 0:
            result.append(i)
            return factors(result, n // i)

Comme vous pouvez le voir, nous recevons comme arguments de la fonction un nombre n et une liste partielle de résultats nommée « result ».

Pour chaque nombre compris entre 2 et n, nous testons si n est divisible par ce nombre. Si oui, nous ajoutons ce nombre à la liste de résultat et appelons récursivement la fonction factors en lui passant en paramètres la nouvelle liste partielle des résultats et le résultat de la division entière de n (n//i).

Ce n'est pas optimisé, car nous parcourons chaque nombre entre 2 et n fois, et calculons le reste de la division n de chacun de ces nombres, au lieu d'utiliser une liste de nombres premiers (ou un générateur pour le faire plus efficacement). Mais c'est parfait pour cet exemple, car ces tâches nécessitent beaucoup de calculs de la part du processeur.

Lançons maintenant ce code avec une liste de 50 000 nombres et voyons ce qu'il se passe :

La factorisation de 50000 nombres a pris  25,89749503135681 secondes avec l'approche série récursive et Python 3.5.0
La factorisation de 50000 nombres a pris  26,277130842208862 secondes avec l'approche série récursive et Python 3.5.0
La factorisation de 50000 nombres a pris  26,53605008125305 secondes avec l'approche série récursive et Python 3.5.0
La factorisation de 50000 nombres a pris  25,725329160690308 secondes avec l'approche série récursive et Python 3.5.0
La factorisation de 50000 nombres a pris  25,732399940490723 secondes avec l'approche série récursive et Python 3.5.0

Cela paraît pas mal, mais si vous augmentez le nombre de calculs, nous aurons des temps de calcul plus longs :

La factorisation de 55000 nombres a pris  33.56754684448242 secondes avec l'approche série récursive et Python 3.5.0
La factorisation de 60000 nombres a pris  37.03928780555725 secondes avec l'approche série récursive et Python 3.5.0
La factorisation de 65000 nombres a pris  41.349984884262085 secondes avec l'approche série récursive et Python 3.5.0
La factorisation de 75000 nombres a pris  56.48437809944153 secondes avec l'approche série récursive et Python 3.5.0
La factorisation de 100000 nombres a pris  100.21058487892151 secondes avec l'approche série récursive et Python 3.5.0

À la fin, nous atteignons la moyenne de 1 seconde pour 1000 nombres. Et ce n'est pas bon si nous voulons traiter plus de nombres.

Ceci est toujours un sujet intéressant. Si vous travaillez avec le parallélisme indépendamment du langage, la question se pose ; comment pouvez-vous partager des données entre les threads sans avoir un mauvais résultat ou une exception ?

Un mauvais résultat peut arriver si chaque thread télécharge les mêmes images (ou si juste deux threads téléchargent la même image en parallèle).

Nous pouvons utiliser une classe Queue du module queue de Python.

Un objet Queue (file d'attente) est une implémentation FIFO (first-in first-out - premier entré, premier sorti), ce qui signifie que si vous placez un élément A avant l'élément B dans la queue, l'élément A sera sorti avant l'élément B.

J'ai précisé que l'application se termine quand le thread principal se termine lui-même. Cependant, si nous faisons tout le travail dans des threads séparés, nous pourrions terminer avant les autres threads, ce qui signifie que cela peut être super rapide, mais nous n'aurons pas atteint notre but.

Voyons ce que nous pouvons accomplir quand nous ajoutons des threads pour exécuter les téléchargements en parallèle. Tout d'abord, nous définissons une classe fille de Thread que nous appelons Downloader .

Voici le code :

__author__ = 'GHajba'

from threading import Thread
from queue import Queue


class Downloader(Thread):
    def __init__(self, queue, folder):
        Thread.__init__(self)
        self.queue = queue
        self.folder = folder

    def run(self):
        while True:
            url = self.queue.get()
            download_images(self.folder, url)
            self.queue.task_done()

Comme vous pouvez le voir, je crée une classe fille de la classe Thread et initialise le nouvel objet avec la référence de la queue partageant les données dans le dossier dans lequel nous exportons les fichiers. Ces deux éléments ne devraient pas changer pendant la durée de vie du thread.

La méthode run a une boucle infinie, car les threads démons ne savent pas habituellement combien de travail ils ont à faire. À chaque itération, nous obtenons l'URL suivante depuis la queue et appelons la fonction de téléchargement des images avec la nouvelle URL et le dossier où nous voulons que le résultat soit sauvegardé. Une fois le téléchargement fini, nous pouvons informer la queue que nous avons fini avec l'élément que nous avons traité, afin qu'elle puisse être assurée que nous avons effectué notre partie de la charge de travail, et n'attende pas sans fin dans le thread principal (là où nous appelons queue.join()).

Voyons maintenant comment initialiser les threads et comment remplir la queue ;

thread_count = 4

queue = Queue()

for i in range(thread_count):
    downloader = Downloader(queue, 'images')
    downloader.daemon = True
    downloader.start()

for url in image_url_list:
    queue.put(url)

queue.join()

Le code ci-dessus démarre l'application avec 4 threads parallèles puis alimente la queue. Comme je l'ai mentionné, nous créons les threads en tant que threads démons et les démarrons. Une fois tous les threads démarrés, nous pouvons remplir la queue avec les données des images pour télécharger les URL.

Finalement, nous bloquons notre thread principal avec queue.join() jusqu'à ce que tous les éléments de la queue aient été traités par les threads.

J'ai utilisé les URL des images comme données en entrée pour la queue, car elles ont une liste définie de multiples éléments (dans le cas de mon test : 52 fichiers .jpg ou .png). Nous pourrions également faire l'extraction en parallèle, mais cela ne donnerait pas beaucoup de gain de performances pour un site. Si nous extrayions un paquet de sites de la galerie memes (pas seulement les 60 premières images du premier site), nous pourrions le paralléliser.

Quand je lance cet exemple (et ajoute un peu de mesure de temps) sur mon Mac, j'obtiens le résultat suivant :

52 images téléchargées en  5.1321189403533936 secondes avec 4 threads et Python 3.5.0
52 images téléchargées en  4.801907062530518 secondes avec 4 threads et Python 3.5.0
52 images téléchargées en  5.145358085632324 secondes avec 4 threads et Python 3.5.0
52 images téléchargées en  4.81237006187439  secondes avec 4 threads et Python 3.5.0
52 images téléchargées en  5.1538519859313965  secondes avec 4 threads et Python 3.5.0

Nettement plus rapide ! C'est ce que nous espérions quand nous avons ajouté plusieurs threads au code. Maintenant, voyons ce qui se passe quand nous utilisons 8 threads :

52 images téléchargées en  3.7929270267486572  secondes avec 8 threads et Python 3.5.0
52 images téléchargées en  3.6049129962921143  secondes avec 8 threads et Python 3.5.0
52 images téléchargées en  3.4051239490509033  secondes avec 8 threads et Python 3.5.0
52 images téléchargées en  3.8651368618011475  secondes avec 8 threads et Python 3.5.0
52 images téléchargées en  3.204490900039673  secondes avec 8 threads et Python 3.5.0

Un peu plus rapide, mais comme vous pouvez le voir, il n'est pas utile d'augmenter encore le nombre de threads. Naturellement, si le script doit télécharger de plus gros fichiers ou si vous n'avez pas une connexion réseau rapide, alors vous pouvez ajouter plus de threads pour utiliser les temps d'attente d'entrées/sorties.

Pour ce calcul, nous créons encore une sous-classe Thread et l'appelons Factorizer :

__author__ = 'GHajba'

from threading import Thread
from queue import Queue

class Factorizer(Thread):
    def __init__(self, queue):
        Thread.__init__(self)
        self.queue = queue

    def run(self):
        while True:
            n = self.queue.get()
            result = factors([], n)
            self.queue.task_done()

L'enchaînement des opérations est le même : nous fournissons dans le constructeur la queue des données partagées et dans la méthode run, nous y prenons un élément, créons ses facteurs et informons la queue de ce que nous en avons fait (et nous n'utilisons pas le résultat, mais ça n'a ici pas d'importance).

Pour remplir la queue et initialiser les threads, j'utilise la même approche que précédemment :

thread_count = 4

queue = Queue()

for i in range(thread_count):
    factorizer = Factorizer(queue)
    factorizer.daemon = True
    factorizer.start()

for n in numbers:
    queue.put(n)

queue.join()

Les données partagées entre les threads sont les nombres que nous voulons factoriser.

Si nous lançons le script avec 4 threads, nous obtenons le résultat suivant :

Factoriser 50000 nombres a pris 30.301374912261963 avec 4 threads et Python 3.5.0 
Factoriser 50000 nombres a pris 30.535978078842163 avec 4 threads et Python 3.5.0 
Factoriser 50000 nombres a pris 30.37640905380249 avec 4 threads et Python 3.5.0 
Factoriser 50000 nombres a pris 30.411335945129395 avec 4 threads et Python 3.5.0 
Factoriser 50000 nombres a pris 31.054304122924805 avec 4 threads et Python 3.5.0

Quasiment les mêmes durées d'exécution que précédemment. Mais que s'est-il passé ? Ajoutons plus de threads et voyons si nous pouvons faire mieux :

Factoriser 50000 nombres a pris 27.70519995689392 secondes avec 8 threads et Python 3.5.0
Factoriser 50000 nombres a pris 27.566843032836914 secondes avec 8 threads et Python 3.5.0
Factoriser 50000 nombres a pris 27.92934489250183 secondes avec 8 threads et Python 3.5.0
Factoriser 50000 nombres a pris 28.050817012786865 secondes avec 8 threads et Python 3.5.0
Factoriser 50000 nombres a pris 27.751455068588257 secondes avec 8 threads et Python 3.5.0

Ah non, toujours pas, mais pourquoi ? Et pourquoi cela prend-il plus de temps avec des threads qu'en utilisant une simple boucle for ?

Créer la queue et la remplir avec les nombres, puis créer les threads rend le déroulement du processus un peu plus lent, mais ce n'est pas la raison principale : nous avons la même durée d'exécution à la fin, car la tâche est intensive en temps CPU et l'implémentation CPython de Python (celle que vous pouvez télécharger par défaut sur le site officiel de Python) a un mécanisme nommé Global Interpreter Lock (GIL) qui garantit qu'un seul thread peut exécuter du code Python à un moment donné.

Quand nous téléchargeons des images, nous attendons la fin d'opérations réseau. C'est pourquoi il est possible d'effectuer des opérations en parallèle beaucoup plus vite. Il n'y a pas d'exécution de code quand on est en attente du réseau.

Je vous ai parlé de deux moyens de créer des threads : à travers la création d'une sous-classe ou avec des expressions lambda. J'ai montré la première version, car je pense qu'elle est plus claire et plus flexible. Je vais cependant maintenant vous montrer la seconde version. L'exemple est la même factorisation vue précédemment, juste à la place de la classe Factorizer, nous créons une classe « anonyme » :

queue = Queue() 

for i in range(thread_count):
    t = Thread(target = lambda: calculate(queue))
    t.daemon = True
    t.start()

for n in numbers:
    queue.put(n)

queue.join()

Comme vous pouvez le voir, nous fournissons une expression lambda au constructeur du thread. Cette expression se réfère à une fonction nommée calculate qui prend une queue en paramètre. Et cette queue est la queue des nombres que nous voulons factoriser :

def calculate(numbers_queue):
    while True:
        number = numbers_queue.get()
        result = factors([], number)
        numbers_queue.task_done()

Comme vous pouvez le voir, le corps de la méthode est le même pour le thread sous-classe. La performance est aussi la même. Il n'y a pas d'amélioration en changeant les solutions.

À vous de choisir la version que vous préférez et de l'appliquer.