Streaming-Datenverarbeitung in C: Architektur und Beispiele

Streamdatenverarbeitung in C: Architektur und Implementierung

Streamverarbeitung erfordert minimale Latenz, vorhersehbare Ausführungszeiten und Flexibilität bei Konfigurationsänderungen. Im Gegensatz zur Batch-Verarbeitung, bei der Daten vor der Analyse akkumuliert werden und damit unvermeidbare Verzögerungen entstehen, reagieren Streaming-Systeme unmittelbar auf eingehende Eingaben. Das macht sie unverzichtbar für Videoanalysen, Netzwerkinspektion und andere Echtzeit-Anwendungen. C eignet sich ideal für diese Systeme, da es keinen Garbage Collection besitzt und vollständige Kontrolle über Speicher und Ausführungszeitpunkte bietet.

Architekturabstraktionen für Streamverarbeitung

Es ist praktisch, ein Streaming-System als Rechengraph zu modellieren, wobei Knoten Datenverarbeiter darstellen und Kanten die Datenflussrichtungen angeben. Jeder Knoten hat einen strikten Typ:

• INPUT_NODE — Datenquelle (Netzwerk, Festplatte, Sensor);
• PROCESSING_NODE — Transformer (Analyse, Filterung, Kodierung);
• OUTPUT_NODE — Konsument (Speicherung, Übertragung, Protokollierung).

Die Graph-Topologie bestimmt das Gesamtverhalten des Systems. Die einfachste Form ist eine einfach verkettete Liste für lineare Pipelines. Komplexere Aufgaben erfordern Baumstrukturen (Verzweigung nach Datentyp) oder beliebige Graphen (zyklische Abhängigkeiten, Verbindungszustände).

Die richtige Topologie für die Aufgabe wählen

Die Auswahl der passenden Graphstruktur wirkt sich direkt auf Leistung und Skalierbarkeit aus:

• Einfach verkettete Liste — ideal für homogene Eingabedaten und lineare Verarbeitung. Beispiel: JPEG-Dekodierung → Filterung → Speichern als BMP.
• Baum — für Routing basierend auf Eingabetyp. Zum Beispiel das Leiten von H.264- oder HEVC-Videoströmen zu verschiedenen Dekodern.
• Graph — erforderlich für zustandsbehaftete Operationen oder Feedback-Schleifen. TCP-Traffic-Verarbeitung benötigt Speicherung von Sitzungskontexten und erneutes Durchlaufen von Knoten.

Implementierung in C

Das System basiert auf dynamisch geladenen Modulen (.so unter Linux). Jedes Plugin exportiert eine node_t-Struktur, die sein Verhalten und seine Verbindungen definiert. Die Kerninfrastruktur befindet sich im Header 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; }

Jedes Plugin implementiert zwei Funktionen: init() für die Einrichtung von Ressourcen und processing() für die Kernlogik. Das Makro REGISTER_NODE erzeugt den Export automatisch.

Beispiel: Videoanalysen-Pipeline

Betrachten Sie ein synthetisches Szenario: Empfangen von JPEG-Frames über das Netzwerk, Tauschen der R- und G-Kanäle, dann Speichern als BMP und CSV.

Eingabeplugin (input_plugin.c) erfasst bis zu fünf Frames pro Iteration, dekodiert sie und leitet sie weiter:

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");

Verarbeitungsplugin (processing_plugin.c) tauscht Pixelkanäle:

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");

Ausgabeplugin (output_plugin.c) speichert Ergebnisse und gibt Speicher frei:

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);

Die Hauptschleife in main.c lädt Plugins, richtet die Ausführungsreihenfolge ein und führt die Verarbeitung in einer Endlosschleife durch, um Leerlaufzeiten zu minimieren.

Wichtige Erkenntnisse

• Streaming-Systeme auf C-Basis vermeiden unvorhersehbare Pausen, die in GC-Sprachen üblich sind.
• Wählen Sie die Graph-Topologie beim Design passend zum spezifischen Anwendungsfall.
• Dynamisches Plugin-Laden ermöglicht Erweiterbarkeit ohne Neukompilierung des Kerns.
• Jeder Knoten übernimmt genau eine Rolle: Eingabe, Verarbeitung oder Ausgabe.
• Das Speichermanagement liegt vollständig in der Verantwortung des Entwicklers – es erfordert Disziplin, bietet aber höchste Präzision.

— Editorial Team