Streaming Data Processing on C: Architecture and Examples

# Stream Data Processing in C: Architecture and Implementation

Stream processing demands minimal latency, predictable execution times, and flexibility for configuration changes. Unlike batch processing, where data accumulates before analysis with inherent delays, streaming systems respond to inputs immediately upon arrival. This makes them essential for video analytics, network inspection, and other real-time applications. C is ideal for these systems thanks to no garbage collection and complete control over memory and execution timing.

Architectural Abstractions for Stream Processing

It's convenient to model a streaming system as a computation graph, where nodes are data processors and edges represent data flow directions. Each node has a strict type:

• INPUT_NODE — data source (network, disk, sensor);
• PROCESSING_NODE — transformer (analysis, filtering, encoding);
• OUTPUT_NODE — consumer (storage, transmission, logging).

The graph topology dictates the system's overall behavior. The simplest is a singly linked list for linear pipelines. More complex tasks need tree structures (branching by data type) or arbitrary graphs (cyclic dependencies, connection states).

Choosing the Right Topology for the Task

Selecting the proper graph structure directly impacts performance and scalability:

• Singly linked list — ideal for homogeneous input data and linear processing. Example: JPEG decoding → filtering → saving as BMP.
• Tree — for routing based on input type. For instance, directing H.264 or HEVC video streams to different decoders.
• Graph — required for stateful operations or feedback loops. TCP traffic processing needs session context storage and node re-traversal.

Implementation in C

The system revolves around dynamically loaded modules (.so on Linux). Each plugin exports a node_t structure defining its behavior and connections. Core infrastructure is in the infra.h header:

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

Each plugin implements two functions: init() for resource setup and processing() for core logic. The REGISTER_NODE macro auto-generates the export.

Example: Video Analytics Pipeline

Consider a synthetic scenario: receiving JPEG frames over the network, swapping R and G channels, then saving as BMP and CSV.

Input plugin (input_plugin.c) grabs up to five frames per iteration, decodes them, and passes them along:

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

Processing plugin (processing_plugin.c) swaps pixel channels:

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

Output plugin (output_plugin.c) saves results and frees memory:

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

The main loop in main.c loads plugins, sets up execution order, and runs processing in an infinite loop to minimize idle time.

Key Takeaways

• C-based streaming systems avoid unpredictable pauses common in GC languages.
• Choose graph topology during design to fit the specific use case.
• Dynamic plugin loading enables extensibility without core recompilation.
• Each node handles just one role: input, processing, or output.
• Memory management is fully developer-controlled — demanding discipline but offering ultimate precision.

— Editorial Team