Procesamiento de Datos en Streaming en C: Arquitectura y Ejemplos

Procesamiento de Datos en Streaming en C: Arquitectura e Implementación

El procesamiento en streaming exige latencia mínima, tiempos de ejecución predecibles y flexibilidad para cambios de configuración. A diferencia del procesamiento por lotes, donde los datos se acumulan antes del análisis con retrasos inherentes, los sistemas de streaming responden a las entradas inmediatamente al llegar. Esto los hace esenciales para análisis de video, inspección de redes y otras aplicaciones en tiempo real. C es ideal para estos sistemas gracias a la ausencia de recolección de basura y el control total sobre la memoria y el tiempo de ejecución.

Abstracciones Arquitectónicas para el Procesamiento en Streaming

Es conveniente modelar un sistema de streaming como un grafo de cómputo, donde los nodos son procesadores de datos y las aristas representan las direcciones del flujo de datos. Cada nodo tiene un tipo estricto:

• INPUT_NODE — fuente de datos (red, disco, sensor);
• PROCESSING_NODE — transformador (análisis, filtrado, codificación);
• OUTPUT_NODE — consumidor (almacenamiento, transmisión, registro).

La topología del grafo dicta el comportamiento general del sistema. La más simple es una lista enlazada simple para tuberías lineales. Tareas más complejas necesitan estructuras de árbol (ramificación por tipo de dato) o grafos arbitrarios (dependencias cíclicas, estados de conexión).

Elegir la Topología Adecuada para la Tarea

Seleccionar la estructura de grafo adecuada impacta directamente en el rendimiento y la escalabilidad:

• Lista enlazada simple — ideal para datos de entrada homogéneos y procesamiento lineal. Ejemplo: decodificación JPEG → filtrado → guardado como BMP.
• Árbol — para enrutamiento basado en el tipo de entrada. Por ejemplo, dirigir flujos de video H.264 o HEVC a decodificadores diferentes.
• Grafo — necesario para operaciones con estado o bucles de retroalimentación. El procesamiento de tráfico TCP necesita almacenamiento de contexto de sesión y revisita de nodos.

Implementación en C

El sistema gira en torno a módulos cargados dinámicamente (.so en Linux). Cada plugin exporta una estructura node_t que define su comportamiento y conexiones. La infraestructura principal está en el encabezado 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; }

Cada plugin implementa dos funciones: init() para la configuración de recursos y processing() para la lógica principal. La macro REGISTER_NODE genera automáticamente la exportación.

Ejemplo: Pipeline de Análisis de Video

Considera un escenario sintético: recepción de fotogramas JPEG por la red, intercambio de canales R y G, luego guardado como BMP y CSV.

Plugin de entrada (input_plugin.c) captura hasta cinco fotogramas por iteración, los decodifica y los pasa adelante:

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 procesamiento (processing_plugin.c) intercambia canales de píxeles:

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 salida (output_plugin.c) guarda resultados y libera memoria:

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

El bucle principal en main.c carga los plugins, configura el orden de ejecución y ejecuta el procesamiento en un bucle infinito para minimizar el tiempo inactivo.

Puntos Clave

• Los sistemas de streaming basados en C evitan pausas impredecibles comunes en lenguajes con recolección de basura.
• Elige la topología del grafo durante el diseño para adaptarse al caso de uso específico.
• La carga dinámica de plugins permite extensibilidad sin recompilación del núcleo.
• Cada nodo maneja solo un rol: entrada, procesamiento o salida.
• La gestión de memoria está completamente controlada por el desarrollador — exige disciplina pero ofrece precisión máxima.

— Editorial Team