C에서의 스트리밍 데이터 처리: 아키텍처 및 예제

C에서의 스트림 데이터 처리: 아키텍처와 구현

스트림 처리는 최소 지연 시간, 예측 가능한 실행 시간, 그리고 구성 변경에 대한 유연성을 요구합니다. 배치 처리와 달리 데이터가 분석 전에 축적되어 내재된 지연이 발생하는 반면, 스트리밍 시스템은 입력이 도착하는 즉시 응답합니다. 이는 비디오 분석, 네트워크 검사, 기타 실시간 애플리케이션에 필수적입니다. C는 가비지 컬렉션이 없고 메모리와 실행 타이밍에 대한 완전한 제어를 제공하기 때문에 이러한 시스템에 이상적입니다.

스트림 처리의 아키텍처 추상화

스트리밍 시스템을 계산 그래프로 모델링하는 것이 편리합니다. 여기서 노드는 데이터 프로세서이고, 에지는 데이터 흐름 방향을 나타냅니다. 각 노드는 엄격한 타입을 가집니다:

• INPUT_NODE — 데이터 소스 (네트워크, 디스크, 센서);
• PROCESSING_NODE — 변환기 (분석, 필터링, 인코딩);
• OUTPUT_NODE — 소비자 (저장, 전송, 로깅).

그래프 토폴로지가 시스템의 전체 동작을 결정합니다. 가장 단순한 것은 선형 파이프라인을 위한 단일 연결 리스트입니다. 더 복잡한 작업은 데이터 타입에 따른 분기인 트리 구조나 순환 종속성, 연결 상태를 가진 임의 그래프를 필요로 합니다.

작업에 맞는 적절한 토폴로지 선택

적절한 그래프 구조를 선택하는 것은 성능과 확장성에 직접적인 영향을 미칩니다:

• Singly linked list — 동질 입력 데이터와 선형 처리에 이상적입니다. 예: JPEG 디코딩 → 필터링 → BMP로 저장.
• Tree — 입력 타입에 기반한 라우팅을 위해. 예를 들어, H.264 또는 HEVC 비디오 스트림을 서로 다른 디코더로 지시.
• Graph — 상태 유지 작업이나 피드백 루프에 필요합니다. TCP 트래픽 처리는 세션 컨텍스트 저장과 노드 재방문을 필요로 합니다.

C에서의 구현

시스템은 동적으로 로드되는 모듈(Linux에서 .so)을 중심으로 합니다. 각 플러그인은 동작과 연결을 정의하는 node_t 구조체를 export합니다. 핵심 인프라는 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; }

각 플러그인은 리소스 설정을 위한 init()과 핵심 로직을 위한 processing() 두 함수를 구현합니다. REGISTER_NODE 매크로는 export를 자동 생성합니다.

예제: 비디오 분석 파이프라인

합성 시나리오를 고려해보세요: 네트워크를 통해 JPEG 프레임을 수신하고, R과 G 채널을 교환한 후 BMP와 CSV로 저장합니다.

입력 플러그인 (input_plugin.c) 각 반복당 최대 5개의 프레임을 가져와 디코딩한 후 전달합니다:

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.c) 픽셀 채널을 교환합니다:

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.c) 결과를 저장하고 메모리를 해제합니다:

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

main.c의 메인 루프는 플러그인을 로드하고 실행 순서를 설정한 후, 유휴 시간을 최소화하기 위해 무한 루프에서 처리를 실행합니다.

주요 요점

• C 기반 스트리밍 시스템은 GC 언어에서 흔한 예측 불가능한 일시 정지를 피합니다.
• 특정 사용 사례에 맞게 설계 시 그래프 토폴로지를 선택하세요.
• 동적 플러그인 로딩은 핵심 재컴파일 없이 확장성을 가능하게 합니다.
• 각 노드는 입력, 처리, 또는 출력 중 하나의 역할만 처리합니다.
• 메모리 관리는 개발자가 완전히 제어합니다 — 규율을 요구하지만 최고의 정밀도를 제공합니다.

— Editorial Team