Dans les applications temps-réel il est très important d’éviter les situations dites « d’inversion de priorité ». Il s’agit de cas dans lesquels une tâche de haute priorité est bloquée en attente de la terminaison d’une tâche de plus faible priorité alors qu’elles n’ont rien en commun.
Problème d’inversion de priorité
Examinons un cas classique, reposant sur trois tâches T1, T2 et T3 de priorités strictement croissantes ainsi qu’un mutex M.
Rappelons qu’un mutex est un élément logiciel permettant d’assurer l’exclusion mutuelle de tâches concurrentes. Une fois qu’une tâche a verrouillé le mutex, les autres ne pourront pas l’obtenir avant qu’elle le déverrouille. On utilise ceci pour synchroniser l’accès à des ressources partagées.
- Initialement, T2 et T3 sont endormies, en attente d’événements externes (par exemple des interruptions matérielles, ou l’expiration de timers). T1 s’exécute et va accéder à une ressource partagée avec T3 (par exemple une zone de mémoire commune). Aussi T1 verrouille-t-elle le mutex M.
- L’événement externe qu’attendait T3 se produit. La tâche T3 se réveille, et préempte T1 car elle est plus prioritaire. T3 est donc en exécution, T2 endormie et T1 préemptée.
- La tâche T3 va devoir accéder à son tour à la ressource partagée avec T1. Pour cela elle doit d’abord verrouiller le mutex M, toutefois ce dernier est déjà tenu par T1. T3 va donc s’endormir en attendant la libération de M, et T1 va reprendre son exécution. Jusqu’ici tout se déroule parfaitement normalement.
- A présent, l’événement attendu par T2 survient. Cette tâche va donc se réveiller. Comme elle est plus prioritaire que T1, cette dernière est préemptée et la situation est la suivante:
- T1 préemptée tient le mutex M
- T2 s’exécute
- T3 endormie attend le mutex M
La tâche T3 est bloquée par l’exécution d’une tâche T2 moins prioritaire, nous sommes dans une situation d’inversion de priorité.
Il est normal que T3 soit bloquée par l’exécution de T1, car ces deux tâches partagent un élément commun (une zone mémoire par exemple) et se synchronisent en utilisant un mutex. Mais il n’est pas normal que T3 soit bloquée alors qu’une tâche T2 avec laquelle elle ne partage rien s’exécute tranquillement.
Le cas survient généralement lorsque T1 et T2 sont approximativement de même priorité et T3 beaucoup plus élevée (par exemple T1 de priorité 10, T2 de priorité 12 et T3 de priorité 90).
Un exemple typique est le problème survenu au robot Pathfinder lors de sa mission sur Mars en 1997. Un descriptif détaillé est présenté dans un article de Glenn E. Reeves. En résumé T1 était une tâche de pilotage d’un outil nommé ASI/MET qui réalisait des analyses météorologiques (très basse priorité), T3 était la tâche principale de transfert des données entre les sous-systèmes et T2 était représenté par un ensemble de tâches de faibles priorités (mais plus élevées que T1). Enfin un sémaphore intégré dans l’implémentation de l’appel système select() – invoqué par T1 et par T3 – jouait le rôle de notre mutex.
Construisons un premier exemple afin de mettre en relief le comportement lors d’une inversion de priorité. Dans le programme suivant, nous allons créer trois threads avec des rôles spécifiques :
- T1, de faible priorité, démarre en premier et verrouille le mutex. Ensuite T1 va réaliser un nombre arbitraire de boucles actives (à ajuster en fonction de votre processeur afin qu’elles durent une dizaine de secondes) simulant un travail consommant du temps CPU, et libérera le mutex avant de se terminer..
- T2 de priorité moyenne va dormir deux secondes avant de démarrer sa série de boucles actives d’une dizaine de secondes.
- T3 de priorité élevée dort une seconde, verrouille le mutex (et devra donc patienter jusqu’à la fin de T1) et attaque une série de boucle active.
Pour que l’inversion se produise, il est nécessaire que les priorités soient fixées de manière précise, aussi utiliserons-nous un ordonnancement temps-réel FIFO. Notez que l’exécution du programme ne pourra réussir qu’avec les droits root. Je déconseille de réaliser une boucle active avec une priorité supérieure ou égale à 50, car c’est la priorité par défaut des gestionnaires d’interruption sur les noyaux Linux récents. Si un handler d’interruption est préempté trop longtemps (20 à 30 secondes), le driver peut cesser de fonctionner, c’est souvent le cas avec les cartes Wifi par exemple.
Le programme ci-dessous n’a d’intérêt que s’il y a une lutte pour le CPU entre T1 et T2, ce qui signifie qu’ils doivent s’exécuter sur le même processeur (même coeur).
test-inversion.c
#define _GNU_SOURCE // Pour sched_setaffinty
#include
#include
#include
#include
#include
#include
static void * T1 (void * unused);
static void * T2 (void * unused);
static void * T3 (void * unused);
static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
int main(void)
{
cpu_set_t cpu_set;
pthread_t t1, t2, t3;
// Fixer l'execution sur un seul coeur
CPU_ZERO(& cpu_set);
CPU_SET(0, & cpu_set); // Coeur 0
if (sched_setaffinity(0, sizeof(cpu_set), & cpu_set) != 0) {
perror("sched_setaffinity");
exit(EXIT_FAILURE);
}
// Creer les trois threads
if ((pthread_create(& t1, NULL, T1, NULL) != 0)
|| (pthread_create(& t2, NULL, T2, NULL) != 0)
|| (pthread_create(& t3, NULL, T3, NULL) != 0)) {
fprintf(stderr, "Erreur de creation des threads");
exit(EXIT_FAILURE);
}
// Terminer le thread main
pthread_exit(NULL);
}
#define BOUCLE_1 50000
#define BOUCLE_2 50000
static void * T1 (void * unused)
{
int i;
int j;
struct sched_param param;
fprintf(stderr, "T1 : Creation\n");
// passer en temps reel faible priorite
param.sched_priority = 5;
if (pthread_setschedparam(pthread_self(), SCHED_FIFO, & param) != 0) {
perror("pthread_setschedparam");
exit(EXIT_FAILURE);
}
// verrouiller le mutex
pthread_mutex_lock(& mutex);
// travail actif
fprintf(stderr, "T1 : Debut boucle\n");
for (i = 0; i < BOUCLE_1; i ++)
for (j = 0; j < BOUCLE_2; j ++)
;
fprintf(stderr, "T1 : Fin Boucle\n");
// liberer le mutex
pthread_mutex_unlock(& mutex);
return NULL;
}
static void * T2 (void * unused)
{
int i;
int j;
struct sched_param param;
fprintf(stderr, "T2 : Creation\n");
// passer en temps reel moyenne priorite
param.sched_priority = 15;
if (pthread_setschedparam(pthread_self(), SCHED_FIFO, & param) != 0) {
perror("pthread_setschedparam");
exit(EXIT_FAILURE);
}
// attendre deux secondes
sleep(2);
// travail actif
fprintf(stderr, "T2 : Debut boucle\n");
for (i = 0; i < BOUCLE_1; i ++)
for (j = 0; j < BOUCLE_2; j ++)
;
fprintf(stderr, "T2 : Fin boucle\n");
return NULL;
}
static void * T3 (void * unused)
{
int i;
int j;
struct sched_param param;
fprintf(stderr, "T3 : Creation\n");
// passer en temps reel haute priorite
param.sched_priority = 45;
if (pthread_setschedparam(pthread_self(), SCHED_FIFO, & param) != 0) {
perror("pthread_setschedparam");
exit(EXIT_FAILURE);
}
// attendre une seconde
sleep(1);
// prendre le mutex
pthread_mutex_lock(& mutex);
// travail actif
fprintf(stderr, "T3 : Debut boucle\n");
for (i = 0; i < BOUCLE_1; i ++)
for (j = 0; j < BOUCLE_2; j ++)
;
fprintf(stderr, "T3 : Fin boucle\n");
// liberer le mutex
pthread_mutex_unlock(&mutex);
return NULL;
}
Pensez à calibrer les valeurs de BOUCLE_1 et BOUCLE_2 pour que l’exécution du programme dure une dizaine de secondes. Voici un résultat d’exécution :
# cc test-inversion.c -o test-inversion -pthread -lrt -Wall # ./test-inversion T3 : Creation T2 : Creation T1 : Creation T1 : Debut boucle T2 : Debut boucle T2 : Fin boucle T1 : Fin Boucle T3 : Debut boucle T3 : Fin boucle #
Nous voyons bien que le thread T2 s’exécute intégralement avant que T3 puisse obtenir le mutex et commencer son travail. C’est un cas d’inversion de priorité
Un remède
Il existe un remède aux problèmes d’inversions de priorités : le système PIP. Toutefois précisons tout de suite qu’il ne fonctionne pas dans tous les cas. Il permit de sauver le robot Pathfinder mais il existe d’autres situations dans lesquelles le remède est pire que le mal et l’on risque de voir les performances du système s’effondrer.
PIP (priority inheritance protocol) est un mécanisme reposant sur deux règles et sur une valeur de priorité appliquée à chaque mutex (et éventuellement à toutes les ressources de synchronisation). Reprenons la situation décrite plus haut, T1 est actif, T2 et T3 dorment. T1 verrouille le mutex. La première règle de PIP s’applique alors :
- Première règle : un mutex hérite de la priorité la plus élevée parmi celles de tous les threads qui essayent de le verrouiller
Le thread T1 s’exécutant avec la priorité P1, le mutex hérite alors de cette valeur.
Ensuite T3 se réveille et demande à son tour le mutex. La priorité de celui-ci monte alors à la valeur P3. La seconde règle s’applique
- Seconde règle : un thread s’exécute avec la priorité la plus élevée parmi celles de tous les mutex qu’il a verrouilé.
Le thread T1 est alors propulsé à la priorité P3. Le thread T2 se réveille, mais sa priorité étant à présent inférieure à celle de T1 il ne peut le préempter, et T1 va s’exécuter intégralement, relâcher le mutex, redrescendre alors à la priorité P1. T3 pourra alors prendre le mutex et s’exécuter avant enfin de laisser passer T2.
Dans le programme suivant, j’ai ajouté quelques lignes au début de la fonction main(), qui activent le protocol PIP sur le mutex.
test-pip.c
[...]
int main(void)
{
cpu_set_t cpu_set;
pthread_t t1, t2, t3;
pthread_mutexattr_t attr;
pthread_mutexattr_init(& attr);
pthread_mutexattr_setprotocol(& attr, PTHREAD_PRIO_INHERIT);
pthread_mutex_init(& mutex, & attr);
// Fixer l'execution sur un seul coeur
[...]
Voyons l'exécution avec PIP : # cc test-pip.c -o test-pip -pthread -lrt -Wall # ./test-pip T3 : Creation T2 : Creation T1 : Creation T1 : Debut boucle T1 : Fin Boucle T3 : Debut boucle T3 : Fin boucle T2 : Debut boucle T2 : Fin boucle #
Cette fois, T3 a la possibilité de s’exécuter entièrement avant que T2 démarre. L’inversion de priorité est évitée.
Conclusion
La vraie solution au problème d’inversion de priorité réside dans la conception du système temps-réel. Il ne faut pas laisser de tâches de priorités intermédiaires entre deux tâches partageant des ressources communes. Naturellement ceci est très contraignant et ne peut pas toujours être respecté. PIP représente un correctif plus qu’une véritable solution, mais il a le mérite d’être facilement activable sur les mutex Posix.
