Streamové zpracování dat v C: architektura a příklady

# Zpracování datových proudů v C: architektura a implementace

Zpracování datových proudů vyžaduje minimální zpoždění, předvídatelný čas provádění a flexibilitu při změně konfigurace. Na rozdíl od dávkového zpracování, kde se data hromadí a analyzují se zpožděním, streamové systémy reagují na vstupní signály okamžitě – v okamžiku jejich příchodu. To je činí kriticky důležitými pro úlohy videoanalytiky, síťové inspekce a další scénáře v reálném čase. Jazyk C je pro takové systémy vhodný díky absenci sběrače odpadu a možnosti plné kontroly nad pamětí a časem provádění.

Architektonické abstrakce zpracování datových proudů

Streamový systém je pohodlné modelovat jako graf výpočtů, kde uzly představují zpracovatele dat a hrany směry přenosu informací. Každý uzel má přesně definovaný typ:

• INPUT_NODE — zdroj dat (síť, disk, senzor);
• PROCESSING_NODE — transformátor (analýza, filtrování, kódování);
• OUTPUT_NODE — spotřebitel (ukládání, přenos, logování).

Topologie grafu určuje chování celého systému. Nejjednodušší forma je jednosměrně vázaný seznam, vhodný pro lineární řady. Složitější úlohy vyžadují stromové struktury (větvení podle typu dat) nebo libovolné grafy (cyklické závislosti, stav spojení).

Volba topologie podle úlohy

Správná volba struktury grafu přímo ovlivňuje výkon a rozšiřitelnost:

• Jednosměrně vázaný seznam — vhodný, pokud jsou vstupní data homogenní a zpracování lineární. Příklad: dekódování JPEG → filtrování → uložení do BMP.
• Strom — používá se při potřebě směrování podle typu vstupu. Například videoproud H.264 nebo HEVC je směrován do různých dekodérů.
• Graf — nutný při přítomnosti stavu nebo zpětné vazby. Zpracování TCP provozu vyžaduje ukládání kontextu relace a opakovaný průchod uzly.

Implementace v jazyce C

Systém je postaven kolem dynamicky načítaných modulů (.so v Linuxu). Každý plugin exportuje strukturu node_t, která popisuje jeho chování a vazby. Obecná infrastruktura je definována v hlavičkovém souboru 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; }

Každý plugin implementuje dvě funkce: init() pro nastavení zdrojů a processing() pro hlavní logiku. Makro REGISTER_NODE automaticky vytváří exportní bod.

Příklad: řada videoanalytiky

Probereme syntetický scénář: příjem JPEG snímků po síti, výměna kanálů R a G, uložení do BMP a CSV.

Plugin vstupu (input_plugin.c) přijímá až pět snímků za iteraci, dekóduje je a předává dál:

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 zpracování (processing_plugin.c) provádí přesun pixelů:

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 výstupu (output_plugin.c) ukládá výsledky a uvolňuje paměť:

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

Hlavní smyčka v main.c načítá pluginy, sestavuje pořadí provádění a spouští zpracování v nekonečné smyčce, čímž minimalizuje prostoje.

Co je důležité

• Streamové systémy v C vylučují neurčité pauzy typické pro jazyky s GC.
• Topologie grafu musí být zvolena v fázi návrhu podle konkrétního scénáře.
• Dynamické načítání pluginů zajišťuje rozšiřitelnost bez překompilace jádra.
• Každý uzel odpovídá pouze za svou část: vstup, zpracování nebo výstup.
• Správa paměti leží plně na vývojáři – to vyžaduje disciplínu, ale poskytuje maximální kontrolu.

— Editorial Team