# 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
```python
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 :**
```python
>>> 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.
```python
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 :**
```python
>>> 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.
```python
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 :**
```python
>>> 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.
```python
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.
```python
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 :**
```python
# 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 :
```python
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.
```python
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
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.