gst-mini 0.1.0

Very much simplified GStreamer-inspired learning framework for understanding buffer synchronization and timing

GstMini - Simplified GStreamer Learning Framework

A simplified implementation of GStreamer's core concepts for educational purposes. This framework demonstrates buffer synchronization, timing, and thread management in multimedia pipelines.

Overview

GstMini captures the essential mechanics of GStreamer without handling real multimedia content:

• Pipeline: Container managing elements and clock
• Elements: Processing units (Source, Queue, Segmenter, Sink)
• Pads: Connection points with chain functions
• Buffers: Data units carrying timestamps
• Synchronization: Clock-based timing at sinks
• Threading: Queue-based decoupling

Core Concepts Demonstrated

1. Buffer Flow via Chain Calls

Code: src/gst_mini/core/pad.py:push() → src/gst_mini/core/pad.py:_chain()

• Buffers flow synchronously through chain function calls
• Each element's chain function processes and forwards buffers
• Chain calls are blocking - upstream waits for downstream

GStreamer Reference: gstpad.c:4497-4586

2. Timestamps and Time Domains

Code: src/gst_mini/core/buffer.py:GstBuffer

Every buffer carries:

• pts: Presentation timestamp (when to display)
• duration: Buffer duration
• data: Payload

GStreamer Reference: gstbuffer.h:80-133

3. Segment Conversion

Code: src/gst_mini/core/segment.py:to_running_time()

Converts stream time to running time:

running_time = (pts - segment.start) / segment.rate + segment.base

GStreamer Reference: gstsegment.c:822-867

4. Queue Thread Decoupling

Code: src/gst_mini/elements/queue.py

Critical for live pipelines!

Two threads:

1. Chain function (upstream thread): Enqueues buffers
2. Loop function (downstream thread): Dequeues and pushes

This prevents live sources from blocking on sync waits.

GStreamer Reference: gstqueue.c:1271-1639

5. Synchronization at Sink

Code: src/gst_mini/elements/s3sink.py:_chain()

The synchronization formula:

running_time = segment.to_running_time(pts)
clock_time = running_time + base_time
clock.wait_until(clock_time)  # BLOCKS here!

GStreamer Reference:

• gstbasesink.c:2665-2828 (do_sync)
• gstbasesink.c:2333-2404 (wait_clock)

Installation

The project uses uv for package management. Install the package in development mode:

# Create virtual environment (first time only)
uv venv

Running Examples

Full Pipeline (Recommended Start)

uv run examples/full_pipeline.py

What to observe:

• LiveSource generates frames at steady 30fps
• Queue fills and empties
• Segments created every 6 seconds
• S3Sink waits on clock before "uploading"
• Frame generation never blocks!

Without Queue (Demonstrates Problem)

uv run examples/without_queue.py

What to observe:

• Frame generation becomes irregular
• LiveSource blocks during sync waits
• This shows WHY queues are essential!

Simple Synchronization

uv run examples/simple_sync.py

What to observe:

• Manual buffer pushing
• Clock wait mechanism
• Timing formula in action

Learning Path

Phase 1: Understanding Data Flow

1. Read src/gst_mini/core/buffer.py - See how timestamps are stored
2. Read src/gst_mini/core/pad.py - Understand chain calls
3. Run uv run examples/simple_sync.py - See synchronization in action

Key Insight: Chain calls are synchronous and blocking.

Phase 2: Understanding Timing

1. Read src/gst_mini/core/segment.py - Time domain conversion
2. Read src/gst_mini/core/clock.py - Clock wait mechanism
3. Read src/gst_mini/elements/s3sink.py:_chain() - See synchronization formula

Key Insight: Sync happens at sinks, not during chain calls.

Phase 3: Understanding Threading

1. Read src/gst_mini/elements/livesource.py - Live frame generation
2. Run uv run examples/without_queue.py - See the blocking problem
3. Read src/gst_mini/elements/queue.py - Thread decoupling solution
4. Run uv run examples/full_pipeline.py - See smooth operation

Key Insight: Queues separate fast producers from slow consumers.

Phase 4: Understanding Buffering

1. Read src/gst_mini/elements/hlssegmenter.py - Accumulation strategy
2. Modify target_duration in examples
3. Modify max_size and leaky mode in queue

Key Insight: Buffering enables batching and handles rate mismatches.

Experiments to Try

1. Disable Queue

In full_pipeline.py, remove the queue and link source directly to segmenter:

# pipeline.link(source, queue)
# pipeline.link(queue, segmenter)
pipeline.link(source, segmenter)  # Direct link

Expected: Irregular frame generation, blocking.

2. Disable Synchronization

Set sync=False in S3Sink:

sink = S3Sink("s3sink", bucket="live-streams", sync=False)

Expected: Segments "upload" immediately, no clock waits.

3. Change Queue Size

Try different queue sizes:

queue = Queue("queue", max_size=3, leaky="upstream")   # Small queue
queue = Queue("queue", max_size=100, leaky=None)       # Large queue, blocking

Expected:

• Small queue with leaky: drops frames when full
• Large queue without leaky: never drops, but high latency

4. Change Leaky Mode

queue = Queue("queue", max_size=10, leaky="downstream")  # Drop old buffers

Expected: Maintains low latency by dropping old data.

5. Vary FPS and Segment Duration

source = LiveSource("camera", fps=60)  # Higher frame rate
segmenter = HLSSegmenter("segmenter", target_duration=2.0)  # Shorter segments

Expected: More frequent segment creation.

Code Mapping to Real GStreamer

GstMini Component	Real GStreamer	File:Line
GstBuffer.pts	GST_BUFFER_PTS	gstbuffer.h:89
GstSegment.to_running_time()	gst_segment_to_running_time()	gstsegment.c:822
GstPad.push()	gst_pad_push()	gstpad.c:4795
GstPad._chain()	gst_pad_chain_data_unchecked()	gstpad.c:4497
Queue._chain()	gst_queue_chain_buffer_or_list()	gstqueue.c:1271
Queue._loop()	gst_queue_loop()	gstqueue.c:1590
S3Sink._chain()	gst_base_sink_do_sync()	gstbasesink.c:2665
GstClock.wait_until()	gst_base_sink_wait_clock()	gstbasesink.c:2333

Key Takeaways

1. Buffers carry timestamps - Set once by source, preserved through pipeline
2. Segments define time domains - Enable seeking, rate changes
3. Chain calls are synchronous - Blocking data flow
4. Synchronization happens at sinks - Via clock waits
5. Queues decouple threads - Essential for live sources
6. The sync formula: clock_time = running_time + base_time
7. Thread safety: Mutexes protect queue, pad stream locks serialize data

Differences from Real GStreamer

Simplified for learning:

• No caps negotiation
• No events (except implicit segment)
• No queries
• No buffer pools or memory management
• No state change ordering
• No preroll mechanism
• Single clock (no clock providers/selection)
• No activation modes (push/pull)
• No scheduling/chain optimization
• Simplified error handling

These simplifications let you focus on the core synchronization concepts without the full GStreamer's functionality.

Next Steps

After understanding GstMini:

1. Explore real GStreamer code using the file:line references
2. Build a real pipeline with GStreamer applying these concepts
3. Extend GstMini - Add features like:
   • Multiple sinks (audio + video sync)
   • Seeking support
   • QoS events
   • Buffer pools
   • More element types

Questions to Explore

1. How does the queue decide when to drop buffers?
2. How would you implement rate changes (slow motion)?
3. What happens if clock time goes backwards?

Happy learning!