Traitement de Données en Streaming sur C : Architecture et Exemples

## Traitement des données en flux en C : Architecture et implémentation

Le traitement en flux exige une latence minimale, des temps d'exécution prévisibles et une flexibilité pour les changements de configuration. Contrairement au traitement par lots, où les données s'accumulent avant l'analyse avec des délais inhérents, les systèmes en flux répondent aux entrées immédiatement à leur arrivée. Cela les rend essentiels pour l'analyse vidéo, l'inspection réseau et d'autres applications en temps réel. C est idéal pour ces systèmes grâce à l'absence de ramasse-miettes et au contrôle total sur la mémoire et le timing d'exécution.

Abstractions architecturales pour le traitement en flux

Il est pratique de modéliser un système en flux comme un graphe de calcul, où les nœuds sont des processeurs de données et les arêtes représentent les directions du flux de données. Chaque nœud a un type strict :

• INPUT_NODE — source de données (réseau, disque, capteur) ;
• PROCESSING_NODE — transformateur (analyse, filtrage, encodage) ;
• OUTPUT_NODE — consommateur (stockage, transmission, journalisation).

La topologie du graphe dicte le comportement global du système. La plus simple est une liste chaînée simple pour les pipelines linéaires. Des tâches plus complexes nécessitent des structures arborescentes (branchement par type de données) ou des graphes arbitraires (dépendances cycliques, états de connexion).

Choisir la bonne topologie pour la tâche

La sélection de la structure de graphe appropriée impacte directement les performances et la scalabilité :

• Liste chaînée simple — idéale pour les données d'entrée homogènes et le traitement linéaire. Exemple : décodage JPEG → filtrage → sauvegarde en BMP.
• Arbre — pour le routage basé sur le type d'entrée. Par exemple, diriger les flux vidéo H.264 ou HEVC vers différents décodeurs.
• Graphe — requis pour les opérations stateful ou les boucles de rétroaction. Le traitement du trafic TCP nécessite le stockage du contexte de session et la reprise de traversée des nœuds.

Implémentation en C

Le système repose sur des modules chargés dynamiquement (.so sous Linux). Chaque plugin exporte une structure node_t définissant son comportement et ses connexions. L'infrastructure de base est dans l'en-tête infra.h :

typedef enum NODE_TYPE_C {
    INPUT_NODE,
    OUTPUT_NODE,
    PROCESSING_NODE
} NODE_TYPE_T;

typedef struct matrix_s {
  unsigned int width_;
  unsigned int height_;
  unsigned char *data_;
} matrix_t;

typedef struct data_s {
  matrix_t matrix_;
  char *metadata_;
} data_t;

typedef void (*init_function_t)(void);
typedef unsigned short (*processing_function_t)(data_t**, unsigned short);

typedef struct node_s {
  char *name_;
  NODE_TYPE_T type_;
  char *prev_;
  char *next_;
  init_function_t init;
  processing_function_t processing;
} node_t;

typedef node_t* (*get_node_structure)(void);

#define REGISTER_NODE(plugin_name, type, processing_function, init_function, prev, next) \
    static node_t node = {.name_ = plugin_name, .type_ = type, \
                          .init = init_function, .processing = processing_function, \
                          .prev_ = prev, .next_ = next}; \
    node_t* getnode_structure() { return &node; }

Chaque plugin implémente deux fonctions : init() pour la configuration des ressources et processing() pour la logique principale. La macro REGISTER_NODE génère automatiquement l'export.

Exemple : Pipeline d'analyse vidéo

Considérons un scénario synthétique : réception de trames JPEG sur le réseau, échange des canaux R et G, puis sauvegarde en BMP et CSV.

Plugin d'entrée (input_plugin.c) récupère jusqu'à cinq trames par itération, les décode et les transmet :

static unsigned short processing(data_t **data, unsigned short count) {
  if(data == NULL) return 0;
  unsigned short i = 0;
  for (; i < 5; ++i) {
    unsigned char *jpeg_image = get_image();
    if (jpeg_image == NULL) break;
    decode_image(data[i], jpeg_image);
    free(jpeg_image);
  }
  return i;
}
REGISTER_NODE("input-node", INPUT_NODE, processing, init, NULL, "processing-node");

Plugin de traitement (processing_plugin.c) échange les canaux des pixels :

static unsigned short processing(data_t **data, unsigned short count) {
  for (unsigned short i = 0; i < count; ++i) {
    unsigned int sz = data[i].width_ * data[i].height_;
    for (unsigned int j = 0; j < sz; j += 3) {
      unsigned char pix = data[i].data_[j];
      data[i].data_[j] = data[i].data_[j + 1];
      data[i].data_[j + 1] = pix;
    }
  }
  return count;
}
REGISTER_NODE("processing-node", PROCESSING_NODE, processing, init, "input-node", "output-node");

Plugin de sortie (output_plugin.c) sauvegarde les résultats et libère la mémoire :

static unsigned short processing(data_t **data, unsigned short count) {
  for (unsigned short i = 0; i < count; ++i) {
    save_data(data[i]);
    free(data[i].data_);
    free(data[i].metadata_);
  }
  free(data);
  return count;
}
REGISTER_NODE("output-node", OUTPUT_NODE, processing, init, "processing-node", NULL);

La boucle principale dans main.c charge les plugins, configure l'ordre d'exécution et exécute le traitement dans une boucle infinie pour minimiser le temps d'inactivité.

Points clés

• Les systèmes en flux basés sur C évitent les pauses imprévisibles courantes dans les langages à ramasse-miettes.
• Choisissez la topologie du graphe lors de la conception pour s'adapter au cas d'usage spécifique.
• Le chargement dynamique de plugins permet l'extensibilité sans recompiler le cœur.
• Chaque nœud gère un seul rôle : entrée, traitement ou sortie.
• La gestion de la mémoire est entièrement contrôlée par le développeur — exigeante en discipline mais offrant une précision ultime.

— Editorial Team