Introduction aux systèmes distribués

import time


def prepare_seq(ingredients):
    for fruit, t in ingredients:
        print(f"1 {fruit} en préparation ({t}s)...")
        time.sleep(t)
    print("\nLa salade est prête ! Bonne dégustation !")


if __name__ == "__main__":
    # Une liste d'ingrédients. Chaque ingrédient est un couple
    # (nom, temps de préparation en secondes). Les temps inscrits ici ne
    # sont pas réalistes.
    INGREDIENTS = [
        ("banane", 2),
        ("pêche", 3),
        ("banane", 2),
        ("cerise", 1),
        ("cerise", 1),
        ("poire", 2),
        ("pêche", 3),
        ("pomme", 3),
        ("poire", 2),
        ("pastèque", 4),
        ("pomme", 3)
    ]
    start_time = time.time()
    prepare_seq(INGREDIENTS)
    end_time = time.time()
    print(f"Temps de préparation : {end_time - start_time:.1f}s")

import rpyc
from rpyc.utils.server import ThreadedServer

class MyService(rpyc.Service):
    def exposed_f(self):
        # Cette méthode sera accessible sur le réseau du fait de son préfix "exposed_"
        return 42

    def g(self):
        # Cette méthode ne sera pas accessible sur le réseau
        return 43

def start():
    t = ThreadedServer(MyService, port=18861)
    t.start()

if __name__ == "__main__":
    start()

>>> import rpyc
>>> conn = rpyc.connect("adresse_ip_du_serveur", 18861)
>>> conn.root.exposed_f()
42
>>> conn.root.f()  # Peut aussi être appelée sans le "exposed_"
42
>>> # Par contre, on n'a pas accès à g
>>> conn.root.g()
...
AttributeError: cannot access 'g'

import os
import sys
import time

import rpyc


PREPARATION_TIMES = {  # En secondes
    "pomme": 3,
    "poire": 2,
    "banane": 2,
    "cerise": 1,
    "pêche": 3,
    "pastèque": 4,
}


def log(agent, msg, task, out=None):
    now = dt.datetime.now().timestamp()
    if out is None:
        direction = ""
    elif out:
        direction = "OUT"
    else:
        direction = "IN"
    # On utilise os.getpid() pour récupérer l'identifiant du processus en cours.
    # Nous l'utilisons comme identifiant de l'esclave.
    # Génère un message de log du genre :
    # [1542562396.20902][E-19264][T-10][IN] 1 banane à préparer reçue
    print(f"[{now}][E-{os.getpid()}][T-{task:02d}][{direction}] {msg}")


def prepare_fruit(id_, fruit):
    t = PREPARATION_TIMES[fruit]
    log(f"1 {fruit} en préparation ({t}s)...", id_)
    time.sleep(t)
    return f"1 {fruit} préparée"


def send_result(conn, task, result):
    conn.root.receive_result(task, result)


def ask_task(conn):
    return conn.root.give_task()


def run(conn):
    task = ask_task(conn)

    while task is not None:
        id_, fruit = task

        log(f"1 {fruit} à préparer reçue", id_, out=False)
        prepared_fruit = prepare_fruit(id_, fruit)

        log(f"1 {fruit} prête envoyée", id_, out=True)
        send_result(conn, task, prepared_fruit)

        # Quand il n'y a plus de tâche, le maître retourne None
        # donc on sort de la boucle et de la fonction
        task = ask_task(conn)


if __name__ == "__main__":
    # sys.argv contient les arguments passés en ligne de commande
    # sys.argv[0] est le nom du script Python exécuté. L'indice
    # du premier argument est donc 1.
    # Un esclave s'exécute de cette manière : `python3 slave.py master_ip master_service_port`
    # Par exemple : `python3 slave.py 192.168.168.1 18861`
    master_addr = sys.argv[1]
    master_port = int(sys.argv[2])
    conn = rpyc.connect(master_addr, master_port)
    run(conn)
    
    # Le processus Python se termine dès que run() se termine,
    # c'est-à-dire dès le moment où l'esclave a demandé une tâche
    # et que le maître ne lui en a pas donnée

import sys
from threading import Lock
import time

import rpyc
from rpyc.utils.server import ThreadedServer


def log(agent, msg, task, out=None):
    now = dt.datetime.now().timestamp()
    if out is None:
        direction = ""
    elif out:
        direction = "OUT"
    else:
        direction = "IN"
    # Génère un message de log du genre :
    # [1542562403.242837][MAITRE][T-00][OUT] 1 poire envoyée à la préparation
    print(f"[{now}][MAITRE][T-{task:02d}][{direction}] {msg}")


def make_service(fruits):
    """Retourne un service RPC appelé par les esclaves pour demander du travail
    et notifier des résultats.
    """

    # On attribue un identifiant à chaque tâche, c'est-à-dire à chaque fruit,
    # pour pouvoir déterminer lesquels ont été préparés
    # Ici, on prend simplement l'indice dans le tableau
    tasks_to_do = [(i, fruit) for i, fruit in enumerate(fruits)]
    tasks_being_done = []
    tasks_done = []

    # On crée un verrou commun pour tous les threads attribués
    # aux clients.
    # Nous expliquons plus loin de quoi il est question.
    # Pour les connaisseurs : je fais comme s'il n'y avait pas de GIL
    # pour les besoins du tutoriel
    lock = Lock()

    start_time = None

    class MasterService(rpyc.Service):
        def exposed_give_task(self):
            nonlocal start_time
            
            if start_time is None:
                # On commence le chrono à la sollicitation du premier client
                start_time = time.time()

            with lock:
                if not tasks_to_do:
                    return None
                task = tasks_to_do.pop()
                tasks_being_done.append(task)
            id_, fruit = task
            log(f"1 {fruit} envoyée à la préparation", id_, out=True)
            return task
            
        def exposed_receive_result(self, task, result):
            log(f"{result} reçue", task[0], out=False)

            with lock:
                tasks_being_done.remove(task)
                tasks_done.append((task, result))

            if not tasks_to_do and not tasks_being_done:
                end_time = time.time()
                print("\nLa salade est prête ! Bonne dégustation !")
                print(f"Temps de préparation : {end_time - start_time:.1f}s")
                # server est définie en variable globale plus bas
                server.close()

    return MasterService


if __name__ == "__main__":
    service = make_service([
        "pomme",
        "pomme",
        "poire",
        "poire",
        "banane",
        "banane",
        "cerise",
        "cerise",
        "pêche",
        "pêche",
        "pastèque",
    ])

    port = int(sys.argv[1])
    server = ThreadedServer(service, port=port)
    print(f"Le maître est accessible à {server.host}:{server.port}.", end="\n\n")
    # Cette méthode est bloquante : tant que personne appelle server.close()
    # l'interpréteur Python est bloqué à ce niveau
    server.start()
    # On atteint ici quand on a appelé server.close() dans le service (cf. ci-dessus),
    # et le processus Python s'arrête également

N = 1000000
i = 0


def incr():
    # Cette fonction incrémente `N fois` la variable globale `i`.
    global i
    for _ in range(N):
        i += 1


if __name__ == "__main__":
    print("Un seul thread :")
    incr()
    print("    N - i =", N-i)

import time
from threading import Thread

N = 1000000
i = 0


def incr():
    # Cette fonction incrémente `N` fois la variable globale `i`.
    global i
    for _ in range(N):
        i += 1


def run(t1, t2):
    start = time.time()
    # Les méthodes `t1.run()` et `t2.run()` sont exécutées en parallèle.
    # Elles vont chacune incrémenter `i` en parallèle.
    t1.start()
    t2.start()
    # On attend que `t1` et `t2` terminent, c'est-à-dire qu'ils aient chacun
    # incrémenté `i` `N` fois.
    t1.join()
    t2.join()
    end = time.time()
    # Comme `i` a en théorie été incrémentée `N` fois par chacun des threads,
    # on s'attend à ce qu'elle soit égale à `2N`.
    print(f"    {end - start:.2f} secondes")
    print(f"    2N - i = {2*N-i}")


if __name__ == "__main__":
    print("Un seul thread :")
    incr()
    print("    N - i =", N-i)

    print("\nDeux threads (sans exclusion mutuelle) :")
    i = 0
    run(Thread(target=incr), Thread(target=incr))

with lock:
    if not tasks_to_do:
        return None
    task = tasks_to_do.pop()
    tasks_being_done.append(task)

import time
from threading import Thread, Lock

N = 1000000
i = 0


def incr():
    # Cette fonction incrémente `N` fois la variable globale `i`.
    global i
    for _ in range(N):
        i += 1


def incr_with_lock(lock):
    global i
    for _ in range(N):
        # On utilise le verrou le moins longtemps possible pour ne pas bloquer
        # excessivement les autres threads. C'est pourquoi la ressource est
        # réservée dans la boucle et non pas à l'extérieur.
        with lock:
            i += 1


def run(t1, t2):
    start = time.time()
    # Les méthodes `t1.run()` et `t2.run()` sont exécutées en parallèle.
    # Elles vont chacune incrémenter `i` en parallèle.
    t1.start()
    t2.start()
    # On attend que `t1` et `t2` terminent, c'est-à-dire qu'ils aient chacun
    # incrémenté `i` `N` fois.
    t1.join()
    t2.join()
    end = time.time()
    # Comme `i` a en théorie été incrémentée `N` fois par chacun des threads,
    # on s'attend à ce qu'elle soit égale à `2N`.
    print(f"    {end - start:.2f} secondes")
    print(f"    2N - i = {2*N-i}")


if __name__ == "__main__":
    print("Un seul thread :")
    incr()
    print("    N - i =", N-i)

    print("\nDeux threads (sans exclusion mutuelle) :")
    i = 0
    run(Thread(target=incr), Thread(target=incr))

    print("\nDeux threads (avec exclusion mutuelle) :")
    # Nous définissons un seul verrou que nous partageons entre les threads.
    lock = Lock()
    i = 0
    run(
        Thread(target=incr_with_lock, args=(lock,)),
        Thread(target=incr_with_lock, args=(lock,)),
    )

with lock:
    if not tasks_to_do:
        return None
    task = tasks_to_do.pop()
    tasks_being_done.append(task)

try:
    task = tasks_to_do.pop()
except IndexError:
    return None
else:
    tasks_being_done.append(task)

# Je n'affiche que le code modifié de l'esclave

import random


def read_crash_prob():
    try:
        return float(sys.argv[3])
    except IndexError:
        return 0
        
        
def run(conn):
    task = ask_task(conn)
    crash_prob = read_crash_prob()

    while task is not None:
        id_, fruit = task

        log(f"1 {fruit} à préparer reçue", id_, out=False)
        prepared_fruit = prepare_fruit(id_, fruit)
        
        if crash_prob and random.random() < crash_prob:
            log(f"alerte, une panne ! 1 {fruit} en préparation", id_)
            break

        log(f"1 {fruit} prête envoyée", id_, out=True)
        send_result(conn, task, prepared_fruit)

        # Quand il n'y a plus de tâche, le maître retourne None
        # donc on sort de la boucle et de la fonction
        task = ask_task(conn)

# ...

def make_service(fruits):
    
    # ...

    class MasterService(rpyc.Service):
        def exposed_receive_result(self, task, result):
            with lock:
                tasks_being_done.remove(task)
                tasks_done.append((task, result))
                
            tasks_being_done_formatted = [
                f"{task[1]} (T-{task[0]})"
                for task in tasks_being_done
            ].join(", ")
            log(
              f"{result} reçue. En cours : {tasks_being_done_formatted}",
              task[0],
              out=False
            )

            if not tasks_to_do and not tasks_being_done:
                end_time = time.time()
                print("\nLa salade est prête ! Bonne dégustation !")
                print(f"Temps de préparation : {end_time - start_time:.1f}s")
                # server est définie en variable globale plus bas
                server.close()

        # ...

WAITING_DELAY = 3

def run(conn):
    task = ask_task(conn)
    crash_prob = read_crash_prob()

    # On ajoute une 3è valeur possible à `task` : un tuple vide
    # Cela signifie que la maître n'a pas de tâche à donner
    # mais que toutes ne sont pas encore finies donc qu'il
    # en aura plus tard
    # Quand `task` vaut `None` c'est que toutes les tâches
    # ont été réalisées
    while task is not None:
        if task == tuple():
            # Pour éviter de surcharger le serveur avec des demandes, on attend
            # un peu avant de redemander une tâche.
            time.sleep(WAITING_DELAY)
            task = ask_task(conn)
            continue
    
        id_, fruit = task

        log(f"1 {fruit} à préparer reçue", id_, out=False)
        prepared_fruit = prepare_fruit(id_, fruit)
        
        if crash_prob and random.random() < crash_prob:
            log(f"alerte, une panne ! 1 {fruit} en préparation", id_)
            break

        log(f"1 {fruit} prête envoyée", id_, out=True)
        send_result(conn, task, prepared_fruit)

        # Quand il n'y a plus de tâche, le maître retourne None
        # donc on sort de la boucle et de la fonction
        task = ask_task(conn)

from threading import Lock, Timer

TIMEOUT = 5

# ...

def check_task_being_done(task, tasks_being_done, tasks_to_do, lock):
    """Déplace une tâche en cours d'exécution dans la pile des tâches à exécuter,
    si elle y est toujours.
    """
    try:
        # En Python, les listes ne sont pas copiées quand elles sont passées en
        # paramètre, donc l'instruction ci-dessous modifie bien la liste lue par
        # le maître.
        with lock:
            tasks_being_done.remove(task)
    except ValueError:
        # La tâche n'est plus dans la liste, on ne fait rien.
        pass
    else:
        with lock:
            tasks_to_do.append(task)
        log(
            f"remet la {task[1]} dans le panier à préparer",
            task[0],
        )


def make_service(fruits):
    # ...

    class MasterService(rpyc.Service):
        def exposed_give_task(self):
            nonlocal start_time
            
            if start_time is None:
                # On commence le chrono à la sollicitation du premier client
                start_time = time.time()

            with lock:
                if not tasks_to_do:
                    # On introduit la différence entre None (toutes les tâches
                    # réalisées) et tuple() (plus de tâche disponible mais
                    # certaines encore en réalisation donc potentiellement
                    # à récupérer plus tard).
                    return None if not tasks_being_done else tuple()
                task = tasks_to_do.pop()
                tasks_being_done.append(task)
                # On lance un décompte en parallèle. La fonction `check_task_being_done`
                # sera appelée avec les arguments `args` dans `timeout` secondes.
                # Si l'esclave a retourné le résultat d'ici là, la tâche aura déjà été
                # enlevée de la liste par `exposed_receive_result` et l'appel
                # à la fonction n'aura aucun effet. Notons qu'elle sera tout de même
                # appelée car le timer s'exécutera toujours. On pourrait compléter
                # le code en tenant à jour la liste des timers associés aux tâches
                # en cours et en arrêtant le timer quand une tâche est réalisée.
                Timer(
                    TIMEOUT,
                    check_task_being_done,
                    args=(task, tasks_being_done, tasks_to_do, lock)
                ).start()
            id_, fruit = task
            log(f"1 {fruit} envoyée à la préparation", id_, out=True)
            return task
    
        def exposed_receive_result(self, task, result):
            with lock:
                try:
                    tasks_being_done.remove(task)
                except ValueError:
                    # On rentre ici si le délai a été atteint alors que
                    # l'esclave est toujours en vie et répond en retard. Dans ce
                    # cas, sa tâche a été réalisée par quelqu'un d'autre donc on
                    # ignore son message.
                    return
                tasks_done.append((task, result))
                
            tasks_being_done_formatted = [
                f"{task[1]} (T-{task[0]})"
                for task in tasks_being_done
            ].join(", ")
            log(
              f"{result} reçue. En cours : {tasks_being_done_formatted}",
              task[0],
              out=False
            )

            if not tasks_to_do and not tasks_being_done:
                end_time = time.time()
                print("\nLa salade est prête ! Bonne dégustation !")
                print(f"Temps de préparation : {end_time - start_time:.1f}s")
                # server est définie en variable globale plus bas
                server.close()

    return MasterService

# ...