RidgeRun Developer Manual - Profiling Tools - Pipeline Latency
| RidgeRun Developer Manual |
|---|
|
| Coding Styles |
| Development Tools |
| Editors |
| Debugging Tools |
|
| Profiling Tools |
| Methodologies |
| Design Patterns |
| RidgeRun Developer Manual/Testing |
| RidgeRun Developer Manual/Build Systems |
| Contact Us |
Obtaining the latency of the element from a pipeline
The RidgeRun team has developed a Python Script that parses and summarizes latency measurements obtained from a Gstreamer pipeline. The idea behind it is to provide a tool capable of profiling pipelines' latency through each element individually. This statistics include mean, min/max, and percentiles to quantify the overall latency contribution across the pipeline being measured.
Find the Python Script here
#!/usr/bin/env python3
"""Parse GStreamer latency logs from tracers.
The logs are generated with the following options:
GST_DEBUG_NO_COLOR=1 GST_DEBUG=GST_TRACER:7 GST_TRACERS="latency(flags=element)"
Computes latency statistics per element:
count, avg, min, max, p50, p90, p95, p99
Additionally prints:
TOTAL_AVG_SUM_MS -> sum of per-element average latency (in ms)
TOTAL_AVG_SUM_NS -> same value in nanoseconds
Assumes `time=(guint64)N` is in nanoseconds (ns).
"""
# ruff: noqa: T201
from __future__ import annotations
import argparse
import re
import sys
from dataclasses import dataclass
from pathlib import Path
LINE_RE = re.compile(
r"element-latency,.*?element=\(string\)(?P<element>[^,]+),.*?time=\(guint64\)(?P<time>\d+)",
re.IGNORECASE,
)
@dataclass
class Stats:
"""Latency statistics."""
count: int = 0
total_ns: int = 0
min_ns: int | None = None
max_ns: int | None = None
samples_ns: list[int] | None = None
def _safe_int(value: str) -> int:
try:
return int(value)
except ValueError as exc:
msg = f"Invalid integer value: {value!r}"
raise ValueError(msg) from exc
def _add_sample(stats: Stats, value_ns: int, *, keep_samples: bool) -> None:
stats.count += 1
stats.total_ns += value_ns
if stats.min_ns is None or value_ns < stats.min_ns:
stats.min_ns = value_ns
if stats.max_ns is None or value_ns > stats.max_ns:
stats.max_ns = value_ns
if keep_samples:
if stats.samples_ns is None:
stats.samples_ns = []
stats.samples_ns.append(value_ns)
def _percentile(sorted_values: list[int], p: float) -> int:
if not sorted_values:
msg = "Cannot compute percentile of empty list"
raise ValueError(msg)
largest_percentile = 100
if p <= 0:
return sorted_values[0]
if p >= largest_percentile:
return sorted_values[-1]
n = len(sorted_values)
rank = int((p / largest_percentile) * n)
if rank <= 0:
return sorted_values[0]
if rank >= n:
return sorted_values[-1]
return sorted_values[rank - 1]
def _ns_to_ms(ns: int) -> float:
return ns / 1_000_000.0
def _parse_file(path: Path, *, keep_samples: bool) -> dict[str, Stats]:
per_element: dict[str, Stats] = {}
with path.open("r", encoding="utf-8", errors="replace") as f:
for line in f:
match = LINE_RE.search(line)
if not match:
continue
element = match.group("element").strip()
time_ns = _safe_int(match.group("time"))
stats = per_element.get(element)
if stats is None:
stats = Stats(samples_ns=[] if keep_samples else None)
per_element[element] = stats
_add_sample(stats, time_ns, keep_samples=keep_samples)
return per_element
def _average_ns(stats: Stats) -> int:
if stats.count == 0:
return 0
return stats.total_ns // stats.count
def _format_row(
element: str,
stats: Stats,
*,
with_percentiles: bool,
) -> str:
avg_ns = _average_ns(stats)
min_ns = stats.min_ns if stats.min_ns is not None else 0
max_ns = stats.max_ns if stats.max_ns is not None else 0
if with_percentiles:
values = sorted(stats.samples_ns or [])
p50 = _percentile(values, 50)
p90 = _percentile(values, 90)
p95 = _percentile(values, 95)
p99 = _percentile(values, 99)
return (
f"{element:<28} "
f"{stats.count:>8} "
f"{_ns_to_ms(avg_ns):>12.3f} "
f"{_ns_to_ms(min_ns):>12.3f} "
f"{_ns_to_ms(max_ns):>12.3f} "
f"{_ns_to_ms(p50):>12.3f} "
f"{_ns_to_ms(p90):>12.3f} "
f"{_ns_to_ms(p95):>12.3f} "
f"{_ns_to_ms(p99):>12.3f}"
)
return (
f"{element:<28} "
f"{stats.count:>8} "
f"{_ns_to_ms(avg_ns):>12.3f} "
f"{_ns_to_ms(min_ns):>12.3f} "
f"{_ns_to_ms(max_ns):>12.3f}"
)
def main(argv: list[str]) -> int:
"""Compute average element latency from GStreamer tracer logs."""
parser = argparse.ArgumentParser(
description="Compute average element latency from GStreamer tracer logs."
)
parser.add_argument("logfile", type=Path, help="Path to the tracer log file")
parser.add_argument(
"--no-percentiles",
action="store_true",
help="Disable percentile computation.",
)
parser.add_argument(
"--sort",
choices=["name", "avg", "count", "max"],
default="avg",
help="Sort output by this key (default: avg).",
)
args = parser.parse_args(argv)
with_percentiles = not args.no_percentiles
per_element = _parse_file(args.logfile, keep_samples=with_percentiles)
if not per_element:
print("No element-latency entries found.", file=sys.stderr)
return 2
def sort_key(item: tuple[str, Stats]) -> str | int:
element, stats = item
if args.sort == "name":
return element
if args.sort == "count":
return stats.count
if args.sort == "max":
return stats.max_ns or 0
return _average_ns(stats)
items = sorted(per_element.items(), key=sort_key, reverse=(args.sort != "name"))
if with_percentiles:
header = (
f"{'element':<28} {'count':>8} "
f"{'avg_ms':>12} {'min_ms':>12} {'max_ms':>12} "
f"{'p50_ms':>12} {'p90_ms':>12} {'p95_ms':>12} {'p99_ms':>12}"
)
else:
header = (
f"{'element':<28} {'count':>8} {'avg_ms':>12} {'min_ms':>12} {'max_ms':>12}"
)
print(header)
print("-" * len(header))
total_avg_sum_ns = 0
for element, stats in items:
avg_ns = _average_ns(stats)
total_avg_sum_ns += avg_ns
row = _format_row(element, stats, with_percentiles=with_percentiles)
print(row)
print("-" * len(header))
print(f"{'TOTAL_AVG_SUM_MS':<28} {'':>8} {_ns_to_ms(total_avg_sum_ns):>12.3f}")
print(f"{'TOTAL_AVG_SUM_NS':<28} {'':>8} {total_avg_sum_ns:>12}")
return 0
if __name__ == "__main__":
raise SystemExit(main(sys.argv[1:]))
How to Use the Tool
1. Define and Run the pipeline
Set up a Gstreamer pipeline with a fixed number of buffers to ensure controlled execution. The pipeline will run with a latency tracer enabled which will then be dumped into a log file named elements_latency.log that serves as the input for the Python script. Feel free to change the running pipeline as you need.
# Limit execution using num-buffers # This can also be replaced with a custom GStreamer application PIPELINE="gst-launch-1.0 videotestsrc is-live=1 num-buffers=300 ! videoconvert ! fakesink" # Run with latency tracer enabled GST_DEBUG_NO_COLOR=1 \ GST_DEBUG=GST_TRACER:7 \ GST_TRACERS="latency(flags=element)" \ eval "$PIPELINE" 2> elements_latency.log
2. Parse the Results
Run the parser script with the elements_latency.log as an input file to process and compute latency statistics:
python3 gst_tracer_latency_log_parser.py elements_latency.log
Expected Output
The script will print in terminal a table summarizing latency metrics for each pipeline element:
element count avg_ms min_ms max_ms p50_ms p90_ms p95_ms p99_ms -------------------------------------------------------------------------------------------------------------------------------- videoconvert0 300 0.045 0.016 0.278 0.044 0.064 0.073 0.092 -------------------------------------------------------------------------------------------------------------------------------- TOTAL_AVG_SUM_MS 0.045 TOTAL_AVG_SUM_NS 45346
- Per-element metrics: Include count, average latency, min/max latency, and percentile distributions
- TOTAL_AVG_SUM: Includes the sum of average latencies for each of the pipeline's elements, providing an estimate of the overall processing latency.