Premiers pas avec Xenomai

Cette méthode, choisie lors de la configuration du noyau, permet de choisir le timer échu en O(1). Ce paragraphe décrit l'algorithme utilisé et ses limitations.

La seule manière d'obtenir un tel résultat est d'utiliser un tableau dont les indices représentent la date d'échéance du timer. Problème : il faudrait des ressources mémoire infinies ! Xenomai utilise une implémentation basée sur un tableau circulaire de timers (timing wheel). L'exemple ci-dessous décrit comment sont rangés trois timers.

Soit le tableau circulaire callwheel comprenant callwheelsize éléments. Les éléments sont de type liste de timers.

Un timer expirant à la date t sera rangé dans une liste du tableau de la manière suivante :
indice = t % callwheelsize
soit la case (backet) :
callwheel[t % callwheelsize]

On ajoute également au timer une information supplémentaire : c_time.
c_time est initialisé à la valeur :
c_time = (t-now)/callwheelsize
now étant la date courante.

À chaque interruption timer, on incrémente l'index. On décrémente le champ c_time de tous les timers de la liste pointée par callwheel[index]. Si des timers ont leur champ qui est inférieur ou égal à 0, alors ils sont expirés.

Dans l'exemple ci-dessous :

À now=1 et on stocke trois timers expirant aux dates : 3, 89 et 137.

Ci-dessous, la résolution de la roue : le paramètre Timer wheel step configurable dans le noyau :

Les deux derniers timers sont dans le même backet, ils seront donc parcourus lorsque l'index sera sur eux. C'est pourquoi, il est recommandé qu'il n'y ait pas plusieurs timers dans le même slot, on risque de retomber dans une complexité en O(N).
En interne, Nucleus utilise la notion de prochain timer expiré. Cette recherche implique le parcours de tous les slots jusqu'à ce qu'un timer soit rencontré. Dans le pire des cas, aucun timer n'est présent. C'est pourquoi il est recommandé d'avoir un timer de la période d'un slot (Timer wheel step) afin d'avoir un temps de recherche correct.

Cet exemple de code implémente l'affichage de messages sans basculer en secondary mode.

Les types utilisés :

typedef enum
{
  E_DEBUG,
  E_INFO,
  E_WARNING,
  E_ERREUR,
  E_FATAL,
  N_LEVELS
} e_log_level;

typedef struct
{
  char                   message[MESSAGE_SIZE];
  e_log_level            criticity;
  RTIME                 timestamp;
  char                   task_name[XNOBJECT_NAME_LEN];
} log_message_t;

Les variables globales :

RT_QUEUE log_MsgQueue;

unsigned int n_lost_messages = 0;

const char *criticity_label[N_LEVELS] = { "DEBUG  ", "INFO   ", "WARNING", "ERREUR  ", "FATAL  "};

e_log_level filter = E_INFO;

La fonction d'initialisation :

boolean log_init(void)
{
  if (rt_queue_create (&log_MsgQueue, "log_MsgQueue", sizeof (log_message_t) * N_MESSAGES, N_MESSAGES, Q_FIFO) != 0)
    return FALSE;
  return TRUE;
}

La fonction d'écriture :

void log_write (e_log_level n, const char *m)
{
  log_message_t lm;
  RT_TASK_INFO  task_info;
  RT_TIMER_INFO timer_info;

  memset (&task_info, 0x00, sizeof (task_info));
  if ( rt_task_inquire (NULL /* current task */ , &task_info) != 0 )
    assert(0); /* Should not happend */
  
  strncpy (lm.message, m, MESSAGE_SIZE);
  strncpy (lm.task_name, task_info.name, XNOBJECT_NAME_LEN);
  lm.criticity = n;
  
  /* Always returns 0: */
  (void)rt_timer_inquire( &timer_info ); 
  lm.timestamp = timer_info.date;

  if (rt_queue_write (&log_MsgQueue, (const void *) &lm, sizeof (lm), Q_NORMAL) < 0)
    ++n_lost_messages;
}

Quelques remarques :

• ce code est « primary mode safe », il ne provoquera pas de basculement en secondary mode ;
• la copie des chaînes de caractères est sécurisée, cela fait partie des bonnes pratiques afin de construire une application robuste ;
• le niveau de criticité est ignoré, par défaut tout est stocké. Cela signifie qu'une tâche importante du point de vue temporel, mise au point avec des traces E_DEBUG, inhibées par la suite, aura exactement les mêmes performances. Il est essentiel que les fonctions de validation ne modifient pas le comportement du logiciel si elles ne sont pas utilisées ;
• les messages non stockés sont comptabilisés.

La tâche de lecture :

void log_task (void *arg /* ignored */)
{
  log_message_t lm;

  while (flag /* externally set */ )
    {
      memset (&lm, 0x00, sizeof (lm));

      if (rt_queue_read (&log_MsgQueue, (void *) &lm, sizeof (lm), TM_INFINITE) != sizeof (lm))
        return; /* normal termination by queue destruction */

      if ( lm.criticity <= filter )
        {
          printf ("[%s] %llu (ns) - %s - %s\n",
            criticity_label[lm.criticity], lm.timestamp, lm.task_name, lm.message);
        }

      if (lm.niveau == E_FATAL)
        exit (EXIT_FAILURE);
    }
}

Le format de sortie est délimité afin de faciliter les post-traitements (awk, perl...).