Best Practices
Production-ready guidelines and optimization techniques for DeepStream applications.
Performance Optimization
1. Hardware Acceleration
Use Hardware Decoders
# Good: Hardware decoder
decoder = Gst.ElementFactory.make("nvv4l2decoder", "decoder")
# Avoid: Software decoder (slow)
# decoder = Gst.ElementFactory.make("avdec_h264", "decoder")
Enable Zero-Copy
# Use NVIDIA memory type
streammux.set_property('nvbuf-memory-type', 0) # 0=Default, 3=Unified
2. Batching Strategy
Dynamic Batching
streammux = Gst.ElementFactory.make("nvstreammux", "muxer")
streammux.set_property('batch-size', 8) # Process up to 8 streams
streammux.set_property('batched-push-timeout', 40000) # 40ms
streammux.set_property('live-source', 1) # For live/RTSP sources
Optimal Batch Size
# Test different batch sizes to find optimal
batch_sizes = [1, 2, 4, 8, 16]
for batch_size in batch_sizes:
# Measure FPS and latency
# Choose based on throughput vs latency requirements
3. Inference Optimization
Use Appropriate Precision
# FP32: Highest accuracy, slowest
network-mode=0
# FP16: Good balance (recommended)
network-mode=2
# INT8: Fastest, requires calibration
network-mode=1
int8-calib-file=calibration.table
Inference Interval
# Don't run inference on every frame
interval=4 # Run inference every 5th frame (0-based)
Multiple Model Instances
# For high-throughput scenarios
[property]
num-copies=2 # Run 2 parallel instances
4. Memory Management
Set Buffer Pool Sizes
streammux.set_property('buffer-pool-size', 4)
Configure GPU Memory
[property]
gpu-id=0
# Reserve GPU memory
workspace-size=1024 # MB
Monitor Memory Usage
# Check GPU memory
nvidia-smi --query-gpu=memory.used,memory.total --format=csv -l 1
# Jetson memory
tegrastats
5. Pipeline Design
Minimize Data Transfers
# Good: All GPU processing
filesrc → nvv4l2decoder → nvstreammux → nvinfer → nvvideoconvert → nvdsosd → sink
# Avoid: CPU-GPU transfers
# filesrc → avdec_h264 → videoconvert → nvstreammux → ...
Use Tiler for Multi-Stream Display
tiler = Gst.ElementFactory.make("nvmultistreamtiler", "tiler")
tiler.set_property('rows', 2)
tiler.set_property('columns', 2)
tiler.set_property('width', 1920)
tiler.set_property('height', 1080)
Scalability
Multi-GPU Setup
# Distribute streams across GPUs
def setup_multi_gpu(sources, num_gpus=2):
pipelines = []
streams_per_gpu = len(sources) // num_gpus
for gpu_id in range(num_gpus):
start_idx = gpu_id * streams_per_gpu
end_idx = start_idx + streams_per_gpu
gpu_sources = sources[start_idx:end_idx]
# Create pipeline for this GPU
pipeline = create_pipeline(gpu_sources, gpu_id)
pipelines.append(pipeline)
return pipelines
Container Deployment
Docker Compose Example
docker-compose.yml
version: '3.8'
services:
deepstream:
image: nvcr.io/nvidia/deepstream:6.4-gc-triton-devel
runtime: nvidia
environment:
- DISPLAY=${DISPLAY}
- NVIDIA_VISIBLE_DEVICES=all
- NVIDIA_DRIVER_CAPABILITIES=all
volumes:
- /tmp/.X11-unix:/tmp/.X11-unix
- ./models:/models
- ./configs:/configs
- ./streams:/streams
command: deepstream-app -c /configs/app_config.txt
deploy:
resources:
reservations:
devices:
- driver: nvidia
count: 1
capabilities: [gpu]
Kubernetes Deployment
deepstream-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: deepstream-app
spec:
replicas: 2
selector:
matchLabels:
app: deepstream
template:
metadata:
labels:
app: deepstream
spec:
containers:
- name: deepstream
image: nvcr.io/nvidia/deepstream:6.4-gc-triton-devel
resources:
limits:
nvidia.com/gpu: 1
volumeMounts:
- name: config
mountPath: /configs
- name: models
mountPath: /models
command: ["deepstream-app"]
args: ["-c", "/configs/app_config.txt"]
volumes:
- name: config
configMap:
name: deepstream-config
- name: models
persistentVolumeClaim:
claimName: models-pvc
Error Handling
Robust Pipeline
def create_robust_pipeline():
"""Pipeline with error handling"""
def bus_call(bus, message, loop):
t = message.type
if t == Gst.MessageType.EOS:
print("End-of-stream")
loop.quit()
elif t == Gst.MessageType.ERROR:
err, debug = message.parse_error()
print(f"Error: {err}: {debug}")
# Log error
with open('errors.log', 'a') as f:
f.write(f"{time.time()}: {err}\n")
# Attempt recovery
pipeline.set_state(Gst.State.NULL)
time.sleep(5)
# Restart pipeline
pipeline.set_state(Gst.State.PLAYING)
elif t == Gst.MessageType.WARNING:
warn, debug = message.parse_warning()
print(f"Warning: {warn}: {debug}")
return True
# Create pipeline with timeout
pipeline = Gst.Pipeline()
# ... setup pipeline ...
# Add bus watch
bus = pipeline.get_bus()
bus.add_signal_watch()
bus.connect("message", bus_call, loop)
return pipeline
Stream Reconnection
class RTSPReconnect:
def __init__(self, uri, max_retries=5):
self.uri = uri
self.max_retries = max_retries
self.retry_count = 0
self.reconnect_interval = 5 # seconds
def on_source_setup(self, element, source):
"""Handle source setup"""
source.set_property('location', self.uri)
source.set_property('latency', 100)
source.set_property('timeout', 10000000) # 10 seconds
source.set_property('drop-on-latency', True)
def handle_eos(self, pipeline):
"""Handle end-of-stream (connection loss)"""
if self.retry_count < self.max_retries:
print(f"Reconnecting... (attempt {self.retry_count + 1})")
time.sleep(self.reconnect_interval)
pipeline.set_state(Gst.State.NULL)
pipeline.set_state(Gst.State.PLAYING)
self.retry_count += 1
else:
print("Max retries reached. Giving up.")
return False
return True
Monitoring & Debugging
Performance Monitoring
class PerformanceMonitor:
def __init__(self):
self.frame_count = 0
self.start_time = time.time()
self.fps_samples = []
def update(self, frame_meta):
"""Update performance metrics"""
self.frame_count += 1
# Calculate FPS every second
elapsed = time.time() - self.start_time
if elapsed >= 1.0:
fps = self.frame_count / elapsed
self.fps_samples.append(fps)
print(f"FPS: {fps:.2f} | "
f"Avg: {sum(self.fps_samples)/len(self.fps_samples):.2f} | "
f"Frame: {frame_meta.frame_num}")
self.frame_count = 0
self.start_time = time.time()
def get_stats(self):
"""Get performance statistics"""
if not self.fps_samples:
return None
return {
'avg_fps': sum(self.fps_samples) / len(self.fps_samples),
'min_fps': min(self.fps_samples),
'max_fps': max(self.fps_samples),
'samples': len(self.fps_samples)
}
# Usage
perf_monitor = PerformanceMonitor()
def probe_callback(pad, info, u_data):
# ... process metadata ...
perf_monitor.update(frame_meta)
return Gst.PadProbeReturn.OK
Logging Best Practices
import logging
from datetime import datetime
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(f'deepstream_{datetime.now().strftime("%Y%m%d")}.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger('DeepStream')
def probe_with_logging(pad, info, u_data):
try:
# Process metadata
gst_buffer = info.get_buffer()
batch_meta = pyds.gst_buffer_get_nvds_batch_meta(hash(gst_buffer))
# ... processing ...
logger.debug(f"Processed frame {frame_meta.frame_num}")
except Exception as e:
logger.error(f"Error in probe: {e}", exc_info=True)
return Gst.PadProbeReturn.OK
Debug Environment Variables
# Enable GStreamer debug
export GST_DEBUG=3 # 0-9, higher = more verbose
export GST_DEBUG_FILE=gst_debug.log
# DeepStream debug
export NVDS_ENABLE_DEBUG=1
# TensorRT verbose logging
export TRT_LOGGER_VERBOSITY=4
Production Deployment
Configuration Management
import yaml
import json
class ConfigManager:
def __init__(self, config_file):
self.config = self.load_config(config_file)
def load_config(self, config_file):
"""Load configuration from file"""
with open(config_file, 'r') as f:
if config_file.endswith('.yaml'):
return yaml.safe_load(f)
elif config_file.endswith('.json'):
return json.load(f)
def get(self, key, default=None):
"""Get configuration value"""
keys = key.split('.')
value = self.config
for k in keys:
if isinstance(value, dict):
value = value.get(k)
if value is None:
return default
else:
return default
return value
# Usage
config = ConfigManager('config.yaml')
batch_size = config.get('inference.batch_size', 1)
Health Checks
class HealthCheck:
def __init__(self, pipeline):
self.pipeline = pipeline
self.last_frame_time = time.time()
self.timeout = 10 # seconds
def update(self):
"""Update last activity time"""
self.last_frame_time = time.time()
def is_healthy(self):
"""Check if pipeline is healthy"""
# Check pipeline state
state = self.pipeline.get_state(0)
if state[1] != Gst.State.PLAYING:
return False
# Check for frame timeout
if time.time() - self.last_frame_time > self.timeout:
return False
return True
def run_healthcheck_loop(self):
"""Periodic health check"""
while True:
time.sleep(5)
if not self.is_healthy():
print("Pipeline unhealthy! Taking action...")
# Restart pipeline or alert
self.recover()
def recover(self):
"""Attempt to recover pipeline"""
self.pipeline.set_state(Gst.State.NULL)
time.sleep(2)
self.pipeline.set_state(Gst.State.PLAYING)
Metrics Export
from prometheus_client import Counter, Gauge, Histogram, start_http_server
# Define metrics
frames_processed = Counter('deepstream_frames_processed_total',
'Total frames processed',
['source_id'])
objects_detected = Counter('deepstream_objects_detected_total',
'Total objects detected',
['class'])
inference_time = Histogram('deepstream_inference_seconds',
'Inference time in seconds')
current_fps = Gauge('deepstream_fps',
'Current FPS',
['source_id'])
def export_metrics(frame_meta):
"""Export metrics to Prometheus"""
source_id = str(frame_meta.source_id)
# Update counters
frames_processed.labels(source_id=source_id).inc()
# Count objects by class
l_obj = frame_meta.obj_meta_list
while l_obj:
obj_meta = pyds.NvDsObjectMeta.cast(l_obj.data)
objects_detected.labels(class=obj_meta.obj_label).inc()
l_obj = l_obj.next
# Start Prometheus server
start_http_server(8000)
Security Best Practices
Input Validation
def validate_video_source(uri):
"""Validate video source URI"""
import re
# Check RTSP URI format
if uri.startswith('rtsp://'):
pattern = r'^rtsp://[\w\-\.:]+(/[\w\-\.]+)*$'
if not re.match(pattern, uri):
raise ValueError(f"Invalid RTSP URI: {uri}")
# Check file path
elif uri.startswith('file://'):
path = uri[7:]
if not os.path.exists(path):
raise FileNotFoundError(f"Video file not found: {path}")
return True
Secure Configuration
import os
from cryptography.fernet import Fernet
class SecureConfig:
def __init__(self, key=None):
if key is None:
key = os.environ.get('CONFIG_ENCRYPTION_KEY')
if key:
self.cipher = Fernet(key.encode())
else:
self.cipher = None
def encrypt_value(self, value):
"""Encrypt sensitive value"""
if self.cipher:
return self.cipher.encrypt(value.encode()).decode()
return value
def decrypt_value(self, encrypted_value):
"""Decrypt sensitive value"""
if self.cipher:
return self.cipher.decrypt(encrypted_value.encode()).decode()
return encrypted_value
# Usage
secure_config = SecureConfig()
rtsp_password = secure_config.decrypt_value(config['rtsp_password'])
Testing
Unit Testing Metadata Processing
import unittest
class TestMetadataProcessing(unittest.TestCase):
def test_object_counting(self):
"""Test object counting logic"""
# Mock metadata
# Test counting logic
pass
def test_roi_filtering(self):
"""Test ROI filtering"""
# Create test ROI
# Test filtering logic
pass
if __name__ == '__main__':
unittest.main()
Integration Testing
#!/bin/bash
# integration_test.sh
# Test basic pipeline
python3 test_pipeline.py --video test.mp4 --config config.txt
# Verify output
if [ $? -eq 0 ]; then
echo "✓ Basic pipeline test passed"
else
echo "✗ Basic pipeline test failed"
exit 1
fi
# Test multi-stream
python3 test_multistream.py --streams 4
# Test RTSP reconnection
python3 test_rtsp_reconnect.py
echo "All tests passed!"
Common Pitfalls
❌ Avoid
# Don't block probe functions
def blocking_probe(pad, info, u_data):
time.sleep(1) # BAD: Blocks pipeline
heavy_computation() # BAD: Slows down pipeline
return Gst.PadProbeReturn.OK
# Don't create metadata objects directly
display_meta = pyds.NvDsDisplayMeta() # BAD: Memory leak
# Don't use software codecs
decoder = Gst.ElementFactory.make("avdec_h264") # BAD: Slow
✅ Do
# Use async processing
def async_probe(pad, info, u_data):
# Quick metadata extraction
data = extract_metadata(info)
# Offload processing to thread
thread_pool.submit(process_async, data)
return Gst.PadProbeReturn.OK
# Acquire from pool
display_meta = pyds.nvds_acquire_display_meta_from_pool(batch_meta)
# Use hardware acceleration
decoder = Gst.ElementFactory.make("nvv4l2decoder")
Resource Management
class ResourceManager:
def __init__(self):
self.pipelines = []
self.threads = []
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""Cleanup resources"""
# Stop all pipelines
for pipeline in self.pipelines:
pipeline.set_state(Gst.State.NULL)
# Join threads
for thread in self.threads:
thread.join(timeout=5)
print("Resources cleaned up")
# Usage
with ResourceManager() as rm:
pipeline = create_pipeline()
rm.pipelines.append(pipeline)
# ... run application ...
# Automatic cleanup on exit
Summary Checklist
Performance
- ✅ Use hardware acceleration
- ✅ Enable batching
- ✅ Use FP16/INT8 precision
- ✅ Set appropriate inference interval
- ✅ Monitor GPU memory
Reliability
- ✅ Implement error handling
- ✅ Add reconnection logic
- ✅ Configure timeouts
- ✅ Log errors and warnings
Scalability
- ✅ Use containerization
- ✅ Support multi-GPU
- ✅ Implement health checks
- ✅ Export metrics
Security
- ✅ Validate inputs
- ✅ Secure credentials
- ✅ Use authentication
- ✅ Monitor access
Maintenance
- ✅ Comprehensive logging
- ✅ Configuration management
- ✅ Automated testing
- ✅ Documentation