Aller au contenu

TP : Simulateur d'Ordonnanceur — Gestion des Processus

Thème : Processus, états, ordonnancement, interblocage

Contexte

Votre système d'exploitation préféré gère en permanence des dizaines de processus : navigateur web, lecteur de musique, mises à jour en arrière-plan... Comment fait-il pour donner l'illusion que tout s'exécute en même temps ?

Dans ce TP, vous allez créer un simulateur d'ordonnanceur qui gère une file de processus et les exécute tour à tour, comme le fait un vrai système d'exploitation.

Partie 1 : La classe Processus

Exercice 1 : Modéliser un processus

Créez une classe Processus représentant un processus avec ses caractéristiques.

Attributs : - pid : identifiant unique du processus (int) - nom : nom du processus (str) - duree_totale : durée d'exécution totale nécessaire (int, en unités de temps) - duree_restante : durée restant à exécuter (int) - etat : état actuel du processus (str) — "nouveau", "pret", "elu", "bloque", "termine" - priorite : niveau de priorité (int, optionnel, par défaut 0)

Méthodes : - __init__ : constructeur - executer(quantum) : exécute le processus pendant quantum unités de temps - est_termine() : renvoie True si le processus est terminé - __repr__ : affichage du processus

class Processus:
    # Compteur de PID (variable de classe)
    compteur_pid = 0

    def __init__(self, nom, duree_totale, priorite=0):
        """
        Constructeur de la classe Processus.

        :param nom: (str) Nom du processus
        :param duree_totale: (int) Durée totale d'exécution
        :param priorite: (int) Priorité (0 = normale)
        """
        Processus.compteur_pid += 1
        self.pid = Processus.compteur_pid
        # À compléter
        pass

    def executer(self, quantum):
        """
        Exécute le processus pendant quantum unités de temps.
        Met à jour la durée restante.

        :param quantum: (int) Temps d'exécution alloué
        :return: (int) Temps réellement utilisé
        """
        # À compléter
        pass

    def est_termine(self):
        """
        Vérifie si le processus est terminé.

        :return: (bool) True si terminé
        """
        # À compléter
        pass

    def __repr__(self):
        """Affichage du processus."""
        return f"[PID {self.pid}] {self.nom} ({self.etat}) - {self.duree_restante}/{self.duree_totale}"

Tests : 

>>> p = Processus("Firefox", 10)
>>> print(p)
[PID 1] Firefox (nouveau) - 10/10
>>> p.etat = "elu"
>>> p.executer(3)
3
>>> print(p)
[PID 1] Firefox (elu) - 7/10
>>> p.est_termine()
False

Partie 2 : L'ordonnanceur FIFO

Exercice 2 : Ordonnanceur simple (FIFO)

L'ordonnanceur FIFO (First In, First Out) exécute les processus dans l'ordre d'arrivée, chacun jusqu'à la fin.

class OrdonnanceurFIFO:
    def __init__(self):
        """Initialise l'ordonnanceur avec une file vide."""
        self.file_prets = []  # File des processus prêts
        self.historique = []  # Historique d'exécution
        self.temps = 0        # Temps courant

    def ajouter_processus(self, processus):
        """
        Ajoute un processus à la file des prêts.

        :param processus: (Processus) Processus à ajouter
        """
        processus.etat = "pret"
        self.file_prets.append(processus)
        print(f"[T={self.temps}] {processus.nom} ajouté à la file")

    def executer_suivant(self):
        """
        Exécute le prochain processus de la file jusqu'à la fin.
        """
        if len(self.file_prets) == 0:
            print("File vide, rien à exécuter")
            return

        # Récupérer le premier processus
        processus = self.file_prets.pop(0)
        processus.etat = "elu"

        print(f"\n[T={self.temps}] {processus.nom} commence son exécution")

        # Exécuter jusqu'à la fin
        while not processus.est_termine():
            temps_utilise = processus.executer(1)
            self.temps += temps_utilise
            self.historique.append((self.temps, processus.pid, processus.nom))

        processus.etat = "termine"
        print(f"[T={self.temps}] {processus.nom} terminé")

    def executer_tous(self):
        """Exécute tous les processus de la file."""
        print("\n" + "="*50)
        print("ORDONNANCEUR FIFO - Début de l'exécution")
        print("="*50)

        while len(self.file_prets) > 0:
            self.executer_suivant()

        print("\n" + "="*50)
        print(f"Exécution terminée - Temps total : {self.temps}")
        print("="*50)

    def afficher_historique(self):
        """Affiche l'historique d'exécution."""
        # À compléter
        pass

Tests : 

>>> ordonnanceur = OrdonnanceurFIFO()
>>> ordonnanceur.ajouter_processus(Processus("Firefox", 5))
>>> ordonnanceur.ajouter_processus(Processus("VSCode", 3))
>>> ordonnanceur.ajouter_processus(Processus("Spotify", 2))
>>> ordonnanceur.executer_tous()

Partie 3 : L'ordonnanceur Round Robin

Exercice 3 : Ordonnanceur à temps partagé

L'ordonnanceur Round Robin (tourniquet) alloue un quantum de temps à chaque processus, puis passe au suivant. Si le processus n'est pas terminé, il retourne en fin de file.

class OrdonnanceurRoundRobin:
    def __init__(self, quantum=2):
        """
        Initialise l'ordonnanceur Round Robin.

        :param quantum: (int) Temps alloué à chaque processus
        """
        self.file_prets = []
        self.quantum = quantum
        self.temps = 0
        self.historique = []

    def ajouter_processus(self, processus):
        """Ajoute un processus à la file."""
        processus.etat = "pret"
        self.file_prets.append(processus)

    def executer_cycle(self):
        """
        Exécute un cycle : le premier processus pendant quantum unités.
        S'il n'est pas terminé, il retourne en fin de file.

        :return: (bool) True si un processus a été exécuté
        """
        if len(self.file_prets) == 0:
            return False

        # Récupérer le premier processus
        processus = self.file_prets.pop(0)
        processus.etat = "elu"

        # Exécuter pendant le quantum
        temps_utilise = processus.executer(self.quantum)
        self.temps += temps_utilise

        # Enregistrer dans l'historique
        for _ in range(temps_utilise):
            self.historique.append(processus.nom[0])  # Première lettre

        if processus.est_termine():
            processus.etat = "termine"
            print(f"[T={self.temps}] {processus.nom} TERMINÉ")
        else:
            # Remettre en fin de file
            processus.etat = "pret"
            self.file_prets.append(processus)
            print(f"[T={self.temps}] {processus.nom} interrompu, retour en file ({processus.duree_restante} restant)")

        return True

    def executer_tous(self):
        """Exécute tous les processus jusqu'à la fin."""
        print(f"\n{'='*50}")
        print(f"ORDONNANCEUR ROUND ROBIN (quantum={self.quantum})")
        print(f"{'='*50}\n")

        while self.executer_cycle():
            pass

        print(f"\n{'='*50}")
        print(f"Temps total : {self.temps}")
        print(f"Historique : {''.join(self.historique)}")
        print(f"{'='*50}")

Tests : 

>>> Processus.compteur_pid = 0  # Reset des PID
>>> rr = OrdonnanceurRoundRobin(quantum=2)
>>> rr.ajouter_processus(Processus("Firefox", 6))
>>> rr.ajouter_processus(Processus("VSCode", 4))
>>> rr.ajouter_processus(Processus("Spotify", 2))
>>> rr.executer_tous()

Sortie attendue : 

==================================================
ORDONNANCEUR ROUND ROBIN (quantum=2)
==================================================

[T=2] Firefox interrompu, retour en file (4 restant)
[T=4] VSCode interrompu, retour en file (2 restant)
[T=6] Spotify TERMINÉ
[T=8] Firefox interrompu, retour en file (2 restant)
[T=10] VSCode TERMINÉ
[T=12] Firefox TERMINÉ

==================================================
Temps total : 12
Historique : FFVVSSFFVVFF
==================================================

Partie 4 : Simulation d'interblocage

Exercice 4 : Processus avec ressources

Modélisons des processus qui ont besoin de ressources partagées.

class Ressource:
    def __init__(self, nom):
        """
        :param nom: (str) Nom de la ressource
        """
        self.nom = nom
        self.proprietaire = None  # PID du processus qui détient la ressource

    def est_disponible(self):
        """Renvoie True si la ressource est libre."""
        return self.proprietaire is None

    def acquerir(self, pid):
        """
        Tente d'acquérir la ressource.

        :param pid: (int) PID du processus demandeur
        :return: (bool) True si acquisition réussie
        """
        if self.est_disponible():
            self.proprietaire = pid
            return True
        return False

    def liberer(self):
        """Libère la ressource."""
        self.proprietaire = None

    def __repr__(self):
        if self.est_disponible():
            return f"Ressource {self.nom} : LIBRE"
        else:
            return f"Ressource {self.nom} : occupée par PID {self.proprietaire}"

Exercice 5 : Détection d'interblocage

Créez une fonction qui détecte un interblocage potentiel.

class ProcessusAvecRessources(Processus):
    def __init__(self, nom, duree_totale, ressources_requises):
        """
        :param ressources_requises: (list) Liste des noms de ressources nécessaires
        """
        super().__init__(nom, duree_totale)
        self.ressources_requises = ressources_requises
        self.ressources_obtenues = []

    def demander_ressources(self, ressources_dispo):
        """
        Tente d'obtenir toutes les ressources nécessaires.

        :param ressources_dispo: (dict) {nom: Ressource}
        :return: (bool) True si toutes les ressources obtenues
        """
        for nom_res in self.ressources_requises:
            if nom_res not in self.ressources_obtenues:
                ressource = ressources_dispo.get(nom_res)
                if ressource and ressource.acquerir(self.pid):
                    self.ressources_obtenues.append(nom_res)
                    print(f"  {self.nom} obtient {nom_res}")
                else:
                    print(f"  {self.nom} BLOQUÉ sur {nom_res}")
                    self.etat = "bloque"
                    return False

        return True

    def liberer_ressources(self, ressources_dispo):
        """Libère toutes les ressources détenues."""
        for nom_res in self.ressources_obtenues:
            ressources_dispo[nom_res].liberer()
            print(f"  {self.nom} libère {nom_res}")
        self.ressources_obtenues = []


def detecter_interblocage(processus_list, ressources):
    """
    Détecte si un interblocage existe.

    :param processus_list: (list) Liste des processus
    :param ressources: (dict) Dictionnaire des ressources
    :return: (bool) True si interblocage détecté
    """
    # Un interblocage existe si tous les processus non terminés sont bloqués
    processus_actifs = [p for p in processus_list if p.etat != "termine"]

    if len(processus_actifs) == 0:
        return False

    tous_bloques = all(p.etat == "bloque" for p in processus_actifs)

    if tous_bloques:
        print("\n*** INTERBLOCAGE DÉTECTÉ ! ***")
        print("Processus bloqués :")
        for p in processus_actifs:
            print(f"  - {p.nom} : possède {p.ressources_obtenues}, attend {[r for r in p.ressources_requises if r not in p.ressources_obtenues]}")
        return True

    return False

Exemple de simulation d'interblocage : 

# Ressources
ressources = {
    "A": Ressource("A"),
    "B": Ressource("B")
}

# Processus (scénario classique d'interblocage)
p1 = ProcessusAvecRessources("Programme1", 5, ["A", "B"])
p2 = ProcessusAvecRessources("Programme2", 5, ["B", "A"])

# Simulation manuelle
print("=== Simulation d'interblocage ===\n")

print("Tour 1 : P1 demande A")
p1.demander_ressources(ressources)

print("\nTour 2 : P2 demande B")
p2.demander_ressources(ressources)

print("\nTour 3 : P1 demande B (déjà pris par P2)")
p1.demander_ressources(ressources)

print("\nTour 4 : P2 demande A (déjà pris par P1)")
p2.demander_ressources(ressources)

# Détection
detecter_interblocage([p1, p2], ressources)

Partie 5 : Statistiques et comparaison

Exercice 6 : Calcul des métriques

Ajoutez des méthodes pour calculer les statistiques d'ordonnancement :

def calculer_statistiques(self):
    """
    Calcule les statistiques d'ordonnancement.

    :return: (dict) Statistiques
    """
    # Temps de réponse moyen
    # Temps d'attente moyen
    # Temps de rotation moyen
    # À compléter
    pass

Métriques importantes : - Temps de rotation (turnaround) : temps total entre l'arrivée et la fin - Temps d'attente : temps passé en file d'attente - Temps de réponse : temps avant la première exécution

Exercice 7 : Comparaison FIFO vs Round Robin

Créez un programme qui compare les deux algorithmes sur le même jeu de processus.

def comparer_ordonnanceurs():
    """Compare FIFO et Round Robin."""

    processus_test = [
        ("Navigateur", 8),
        ("Editeur", 4),
        ("Terminal", 2),
        ("Musique", 6)
    ]

    print("="*60)
    print("COMPARAISON DES ORDONNANCEURS")
    print("="*60)

    # Test FIFO
    Processus.compteur_pid = 0
    fifo = OrdonnanceurFIFO()
    for nom, duree in processus_test:
        fifo.ajouter_processus(Processus(nom, duree))
    fifo.executer_tous()

    # Test Round Robin
    Processus.compteur_pid = 0
    rr = OrdonnanceurRoundRobin(quantum=2)
    for nom, duree in processus_test:
        rr.ajouter_processus(Processus(nom, duree))
    rr.executer_tous()

    # À compléter : afficher la comparaison des statistiques

Résumé des algorithmes

Algorithme Principe Avantages Inconvénients
FIFO Premier arrivé, premier servi Simple Temps d'attente long pour les petits processus
Round Robin Quantum de temps pour chacun Équitable, bon temps de réponse Overhead des changements de contexte
Priorité Processus prioritaire d'abord Urgent traité en premier Famine des processus peu prioritaires

Pour aller plus loin

  • Ordonnanceur à priorité : Implémenter un ordonnanceur qui choisit le processus de plus haute priorité
  • Ordonnanceur SJF (Shortest Job First) : Le processus le plus court d'abord
  • Vieillissement (Aging) : Augmenter la priorité des processus qui attendent longtemps
  • Interface graphique : Visualiser l'exécution avec une barre de Gantt

Auteur : Florian Mathieu

Licence CC BY NC

Licence Creative Commons
Ce cours est mis à disposition selon les termes de la Licence Creative Commons Attribution - Pas d'Utilisation Commerciale - Partage dans les Mêmes Conditions 4.0 International.