실험 : RTSP 기반 스트리밍 파이프라인 레이턴시 및 QoE 분석

Media Server ----- (RTSP Stream) ---> RTSP ingest -> Decoder -> Processing -> Encoder -> File Output

 

Measurement Layer
- timestamp
- latency
- jitter
- frame drop
- buffering
- CPU / GPU
- recovery time
- QoE

Queue 배치 전략
source, decode, process, encode, sink 각 모듈 구현체에 queue 엘리먼트를 포함시키지 않고, 컨트롤러에서 각 모듈의 bin을 조합하여 파이프를 생성할 때, 중간 중간에 queue를 끼워넣는 구조로 변경

그리고 GstPadProbe를 해당 queue를 타겟으로 작성
callback을 통해 각 단계의 정확한 latency와 버퍼의 pts라던지 여러 정보 추출 가능

GST_TRACE를 통해 E2E 트레이싱 수행 시 문제점은 내장 latency tracer는 버퍼가 생성된 순간 내부 추적 ID를 통해 sink까지의 시간을 재지만 이번 파이프라인처럼 디코딩 후 재인코딩을 수행하는 경우에는 새로운 압축 버퍼가 생성되므로 E2E 추적 링크가 끊김
또한 모든 엘리먼트의 패드마다 텍스트 로그를 stderr로 쏟아내므로 I/O 병목으로인한 실시간성 왜곡 발생

또한 내부 트레이서 모듈만으로 지원되지 않는 케이스 존재 (재복구 시간, 드롭율 계산 등)

1. Pad 등록

// src/apps/operator/core/src/Controller.cpp (initialize 함수 내부)

for (size_t i = 0; i + 1 < stages.size(); ++i) {
    const auto& cur_stage = stages[i];
    const auto& next_stage = stages[i + 1];

    // 1. 모듈 간 분리 및 버퍼링을 위한 queue 생성
    std::string q_name = "q_" + cur_stage.name + "_to_" + next_stage.name;
    GstElement* queue = gst_element_factory_make("queue", q_name.c_str());
    
    g_object_set(queue,
                 "max-size-buffers", 200u,
                 "max-size-time", static_cast<guint64>(200) * GST_MSECOND,
                 "max-size-bytes", 0u,
                 "leaky", 2, // GST_QUEUE_LEAK_DOWNSTREAM (실시간성 유지)
                 nullptr);

    gst_bin_add(GST_BIN(pipeline), queue);
    inter_module_queues_.push_back(queue);

    // 2. 모듈과 queue 연결: [cur_stage] -> [queue] -> [next_stage]
    gst_element_link(cur_stage.element, queue);
    gst_element_link(queue, next_stage.element);

    // =========================================================================
    // 3. [앞단 모듈 출력 프로브] queue의 sink 패드에 프로브 등록
    //    -> 버퍼가 cur_stage(Source, Decoder, Processor, Encoder)를 막 통과한 시점
    // =========================================================================
    GstPad* q_sink_pad = gst_element_get_static_pad(queue, "sink");
    if (q_sink_pad != nullptr) {
        auto ctx = std::make_unique<ProbeContext>();
        ctx->controller = this;
        ctx->stage_type = static_cast<int>(cur_stage.kind); // 0: Source, 1: Decoder, 2: Processor, 3: Encoder
        
        gst_pad_add_probe(q_sink_pad, GST_PAD_PROBE_TYPE_BUFFER, on_queue_pad_probe, ctx.get(), nullptr);
        gst_object_unref(q_sink_pad);
        probe_contexts_.push_back(std::move(ctx));
    }

    // =========================================================================
    // 4. [최종 Sink 입력 프로브] 다음 단계가 SINK인 경우 마지막 queue의 src 패드에 등록
    //    -> 버퍼가 큐를 빠져나와 최종 VideoSink로 인입되는 시점
    // =========================================================================
    if (next_stage.kind == StageKind::SINK) {
        GstPad* q_src_pad = gst_element_get_static_pad(queue, "src");
        if (q_src_pad != nullptr) {
            auto ctx = std::make_unique<ProbeContext>();
            ctx->controller = this;
            ctx->stage_type = static_cast<int>(StageKind::SINK); // 4: Sink
            
            gst_pad_add_probe(q_src_pad, GST_PAD_PROBE_TYPE_BUFFER, on_queue_pad_probe, ctx.get(), nullptr);
            gst_object_unref(q_src_pad);
            probe_contexts_.push_back(std::move(ctx));
        }
    }
}

 

2. 프로브 디스패처 및 callback

// src/apps/operator/core/src/Controller.cpp

// [정적 프로브 콜백] GStreamer 이벤트 발생 시 Controller의 해당 단계 콜백으로 분기
GstPadProbeReturn Controller::on_queue_pad_probe(GstPad*, GstPadProbeInfo* info, gpointer user_data) {
    if (info->type & GST_PAD_PROBE_TYPE_BUFFER) {
        GstBuffer* buf = GST_PAD_PROBE_INFO_BUFFER(info);
        auto* ctx = static_cast<ProbeContext*>(user_data);
        if (ctx != nullptr && ctx->controller != nullptr && buf != nullptr) {
            switch (ctx->stage_type) {
                case 0: ctx->controller->on_source_frame(buf);  break;
                case 1: ctx->controller->on_decode_frame(buf);  break;
                case 2: ctx->controller->on_process_frame(buf); break;
                case 3: ctx->controller->on_encode_frame(buf);  break;
                case 4: ctx->controller->on_sink_frame(buf);    break;
            }
        }
    }
    return GST_PAD_PROBE_OK;
}

// 1. Source 완료 콜백 (RTSP/WebRTC 수신 직후)
void Controller::on_source_frame(GstBuffer* buf) {
    GstClockTime pts = GST_BUFFER_PTS(buf);
    auto now_us = std::chrono::duration_cast<std::chrono::microseconds>(
        std::chrono::steady_clock::now().time_since_epoch()).count();

    std::lock_guard<std::mutex> lock(metrics_mutex_);
    frame_timings_[pts].pts = pts;
    frame_timings_[pts].t_source_us = now_us;

    static uint64_t source_count = 0;
    if (++source_count % 300 == 0) {
        BOOST_LOG(info) << "[Controller::on_source_frame] " << source_count << " frames ingested (PTS: " << pts << ")";
    }
}

// 2. Decode 완료 콜백 (H.264/H.265 디코딩 후)
void Controller::on_decode_frame(GstBuffer* buf) {
    GstClockTime pts = GST_BUFFER_PTS(buf);
    auto now_us = std::chrono::duration_cast<std::chrono::microseconds>(
        std::chrono::steady_clock::now().time_since_epoch()).count();

    std::lock_guard<std::mutex> lock(metrics_mutex_);
    auto it = frame_timings_.find(pts);
    if (it != frame_timings_.end()) {
        it->second.t_decode_us = now_us;
    }
}

// 3. Process 완료 콜백 (오버레이/필터 처리 후)
void Controller::on_process_frame(GstBuffer* buf) {
    GstClockTime pts = GST_BUFFER_PTS(buf);
    auto now_us = std::chrono::duration_cast<std::chrono::microseconds>(
        std::chrono::steady_clock::now().time_since_epoch()).count();

    std::lock_guard<std::mutex> lock(metrics_mutex_);
    auto it = frame_timings_.find(pts);
    if (it != frame_timings_.end()) {
        it->second.t_process_us = now_us;
    }
}

// 4. Encode 완료 콜백 (x264 재인코딩 후)
void Controller::on_encode_frame(GstBuffer* buf) {
    GstClockTime pts = GST_BUFFER_PTS(buf);
    auto now_us = std::chrono::duration_cast<std::chrono::microseconds>(
        std::chrono::steady_clock::now().time_since_epoch()).count();

    std::lock_guard<std::mutex> lock(metrics_mutex_);
    auto it = frame_timings_.find(pts);
    if (it != frame_timings_.end()) {
        it->second.t_encode_us = now_us;
    }
}

// 5. Sink 진입 콜백 (최종 파일 저장 또는 SDL 화면 표시 직전)
void Controller::on_sink_frame(GstBuffer* buf) {
    GstClockTime pts = GST_BUFFER_PTS(buf);
    auto now_us = std::chrono::duration_cast<std::chrono::microseconds>(
        std::chrono::steady_clock::now().time_since_epoch()).count();

    std::lock_guard<std::mutex> lock(metrics_mutex_);
    auto it = frame_timings_.find(pts);
    if (it != frame_timings_.end()) {
        it->second.t_sink_us = now_us;
        double total_lat_ms = (it->second.t_sink_us - it->second.t_source_us) / 1000.0;

        static uint64_t sink_count = 0;
        if (++sink_count % 300 == 0) {
            BOOST_LOG(info) << "[Controller::on_sink_frame] " << sink_count 
                            << " frames outputted (Total Latency: " << total_lat_ms << " ms)";
        }
        frame_timings_.erase(it); // 완료된 프레임 제거 (메모리 정리)
    }

    // 중간 드롭된 프레임 누적으로 인한 맵 비대화 방지
    if (frame_timings_.size() > 500) {
        frame_timings_.erase(frame_timings_.begin());
    }
}

 

3. 헤더 클래스 선언부

// src/apps/operator/core/inc/Controller.h

public:
    // Queue Pad 프로브에서 호출되는 5대 스테이지 콜백
    void on_source_frame(GstBuffer* buf);
    void on_decode_frame(GstBuffer* buf);
    void on_process_frame(GstBuffer* buf);
    void on_encode_frame(GstBuffer* buf);
    void on_sink_frame(GstBuffer* buf);

private:
    struct ProbeContext {
        Controller* controller = nullptr;
        int stage_type = 0; // 0: source, 1: decode, 2: process, 3: encode, 4: sink
    };

    static GstPadProbeReturn on_queue_pad_probe(GstPad* pad, GstPadProbeInfo* info, gpointer user_data);

    // 동적으로 생성된 inter-module queue 및 프로브 컨텍스트 라이프사이클 관리
    std::vector<GstElement*> inter_module_queues_;
    std::vector<std::unique_ptr<ProbeContext>> probe_contexts_;

    // 프레임별 타임스탬프 추적 장부
    struct FrameTiming {
        uint64_t pts = 0;
        int64_t t_source_us = 0;
        int64_t t_decode_us = 0;
        int64_t t_process_us = 0;
        int64_t t_encode_us = 0;
        int64_t t_sink_us = 0;
    };
    std::mutex metrics_mutex_;
    std::unordered_map<uint64_t, FrameTiming> frame_timings_;