Przetwarzanie strumieniowe danych w C: architektura i przykłady

Przetwarzanie strumieniowe danych w C: architektura i implementacja

Przetwarzanie strumieniowe danych wymaga minimalnych opóźnień, przewidywalnego czasu wykonania i elastyczności przy zmianie konfiguracji. W przeciwieństwie do przetwarzania wsadowego, gdzie dane są gromadzone i analizowane z opóźnieniem, systemy strumieniowe reagują na sygnały wejściowe natychmiast — w momencie ich nadejścia. To czyni je krytycznie ważnymi dla zadań analizy wideo, inspekcji sieciowej i innych scenariuszy czasu rzeczywistego. Język C nadaje się do takich systemów dzięki braku garbage collectora i możliwości pełnej kontroli nad pamięcią i czasem wykonania.

Architektoniczne abstrakcje przetwarzania strumieniowego

System strumieniowy wygodnie modelować jako graf obliczeń, gdzie węzły reprezentują przetwarzać danych, a krawędzie — kierunki transmisji informacji. Każdy węzeł ma ściśle określony typ:

• INPUT_NODE — źródło danych (sieć, dysk, sensor);
• PROCESSING_NODE — transformator (analiza, filtrowanie, kodowanie);
• OUTPUT_NODE — konsument (zapis, transmisja, logowanie).

Topologia grafu określa zachowanie całego systemu. Najprostsza forma — lista jednokierunkowa, odpowiednia dla liniowych konwejerów. Bardziej złożone zadania wymagają struktur drzewiastych (rozgałęzianie według typu danych) lub dowolnych grafów (cykliczne zależności, stan połączeń).

Wybór topologii pod zadanie

Właściwy wybór struktury grafu bezpośrednio wpływa na wydajność i skalowalność:

• Lista jednokierunkowa — nadaje się, jeśli dane wejściowe są jednorodne i przetwarzanie liniowe. Przykład: dekodowanie JPEG → filtrowanie → zapis w BMP.
• Drzewo — używane przy potrzebie routingu według typu wejścia. Na przykład, strumień wideo H.264 lub HEVC kierowany do różnych dekoderów.
• Graf — niezbędny przy nalichii stanu lub sprzężenia zwrotnego. Przetwarzanie ruchu TCP wymaga przechowywania kontekstu sesji i ponownego przejścia przez węzły.

Implementacja w języku C

System jest budowany wokół dynamicznie ładowanych modułów (.so w Linux). Każdy plugin eksportuje strukturę node_t, opisującą jego zachowanie i połączenia. Ogólna infrastruktura jest zdefiniowana w pliku nagłówkowym 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żdy plugin implementuje dwie funkcje: init() do konfiguracji zasobów i processing() do głównej logiki. Makro REGISTER_NODE automatycznie tworzy punkt eksportu.

Przykład: konwejer analizy wideo

Rozważmy syntetyczny scenariusz: odbiór klatek JPEG przez sieć, zamiana kanałów R i G, zapis w BMP i CSV.

Plugin wejściowy (input_plugin.c) odbiera do pięciu klatek na iterację, dekoduje je i przekazuje dalej:

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 przetwarzający (processing_plugin.c) wykonuje permutację pikseli:

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 wyjściowy (output_plugin.c) zapisuje wyniki i zwalnia pamięć:

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

Główna pętla w main.c ładuje pluginy, buduje kolejność wykonania i uruchamia przetwarzanie w nieskończonej pętli, minimalizując przestoje.

Co ważne

• Systemy strumieniowe w C eliminują nieokreślone pauzy charakterystyczne dla języków GC.
• Topologia grafu powinna być wybierana na etapie projektowania pod konkretny scenariusz.
• Dynamiczne ładowanie pluginów zapewnia skalowalność bez rekompilacji jądra.
• Każdy węzeł odpowiada tylko za swoją część: wejście, przetwarzanie lub wyjście.
• Zarządzanie pamięcią spoczywa w pełni na programiście — wymaga to dyscypliny, ale daje maksymalną kontrolę.

— Editorial Team