Multimedia Pipeline (mpipe)

Overview 

The Multimedia Pipeline subsystem (mpipe) builds a media stream out of self-contained processing components called elements. An application declares the elements it needs, links them into a graph, and drives that graph through a state machine. mpipe negotiates the data format between neighboring elements, settles the buffer configuration, and moves the buffers from one element to the next.

A pipeline is a graph of elements joined at their pads.

mpipe provides the pieces and the rules by which they fit together rather than finished solutions. The same graph runs on a different board by binding its elements to different devices, and a new requirement is usually one more element rather than a rewrite.

The elements themselves live in plugins, grouped by media domain. A plugin is added on its own, with its own directory, Kconfig and headers, without altering the framework.

mpipe allocates nothing dynamically. Elements and pipelines are objects the application owns, buffers come from pools sized when the pipeline starts, and a capability is a fixed-size structure passed by value. Negotiation runs on the thread that calls mpipe_element_set_state(), which therefore needs a stack sized for it.

mpipe is optional, and it is not always the right tool. An application driving a single device is better served by that device’s API directly. mpipe earns its place once several devices have to agree on a format and hand buffers to each other.

Building a pipeline 

An application includes <zephyr/mpipe/mpipe.h> for the pipeline API; each element it instantiates adds that element’s own header.

Elements are plain objects the application owns; mpipe allocates none of them. Each element type has its own init function taking that type and an id:

/* Concrete element types come from plugins; each has its own header. */
static struct mpipe pipe;
static struct my_src source;
static struct my_sink sink;

ret = mpipe_pipeline_init(&pipe, PIPE_ID);
ret = my_src_init(&source, SRC_ID);
ret = my_sink_init(&sink, SINK_ID);

Ids only have to be unique within the pipeline. Elements are configured through properties, which is how an element is set up without the caller knowing its concrete type:

ret = mpipe_object_set_properties((struct mpipe_object *)&source,
                                  MY_SRC_PROP_PATH, "/SD:/in.bin",
                                  MPIPE_PROP_LIST_END);

They are then added to the pipeline, in any order, and linked in stream order:

ret = mpipe_bin_add((struct mpipe_bin *)&pipe,
                    (struct mpipe_element *)&source,
                    (struct mpipe_element *)&sink, NULL);

ret = mpipe_element_link((struct mpipe_element *)&source,
                         (struct mpipe_element *)&sink, NULL);

The casts are well defined because every element embeds its base as its first member. Linking refuses a pair of elements whose capabilities cannot possibly intersect, so an impossible graph fails while it is being built rather than when it is started.

Setting the pipeline to MPIPE_STATE_PLAYING is not a jump to that state. The pipeline and its children step through the states in order, READY to PAUSED to PLAYING, which is why the format is negotiated and the pools are started along the way.

The application then observes the pipeline’s message channel, both to learn how the run ends and to act on failures. An error message names the element that raised it and the phase it was in, so an application can report that capability negotiation failed at element 3 rather than only that the run stopped. Returning the graph to MPIPE_STATE_READY tears it down:

/* Needs CONFIG_ZBUS_MSG_SUBSCRIBER=y */
ZBUS_MSG_SUBSCRIBER_DEFINE(main_sub);

struct zbus_channel *bus = mpipe_element_get_bus_chan((struct mpipe_element *)&pipe);

ret = zbus_chan_add_obs(bus, &main_sub, K_FOREVER);

if (mpipe_element_set_state((struct mpipe_element *)&pipe, MPIPE_STATE_PLAYING) != 0) {
        /* the element that refused is still in its previous state */
}

const struct zbus_channel *chan;
struct mpipe_message msg;

do {
        ret = zbus_sub_wait_msg(&main_sub, &chan, &msg, K_FOREVER);
} while ((msg.type & (MPIPE_MESSAGE_ERROR | MPIPE_MESSAGE_EOS)) == 0);

(void)mpipe_element_set_state((struct mpipe_element *)&pipe, MPIPE_STATE_READY);

Elements, pads and links 

Every mpipe element derives from mpipe_object by embedding it as its first member. mpipe_element adds the state machine and the pads. Each element base, source, sink, transform, parser and bin, adds what its kind of element needs, such as its pads and the hooks the negotiation calls, and a concrete element embeds one base and adds its own fields. The types that only carry data - a capability, a message, a dispatch - are plain structs outside that hierarchy:

Inheritance tree of the mpipe structures.

mpipe_src produces buffers and has source pads only. It also drives the negotiation, because it sits at the head of the graph.
mpipe_sink consumes them and has sink pads only.
mpipe_transform has one of each and turns input into output.
mpipe_parser cuts a formless byte stream into whole frames, which is a different job from transforming one format into another.
mpipe_bin holds other elements and forwards state changes to them in the order the transition requires.
mpipe is the top-level bin: it owns the streaming thread and the message channel.

A mpipe_pad is where two elements meet. It carries a direction (source or sink), the peer it is linked to, and the capability negotiated on it. It also carries the callbacks the framework dispatches to: process_fn receives a buffer, query_fn answers a query, and event_fn handles an event. Linking two elements links a source pad to a sink pad, and every hop of the stream is one such pair.

The state machine 

An element is in one of three states and moves between neighbors one step at a time:

Pipeline state machine.

READY means constructed and linked, holding no format and no buffers. The READY to PAUSED transition is where the work happens: the source drives capability negotiation across the whole graph, the buffer pool query settles who provides buffers and how many, and the pools start.

Which way a transition goes determines the order a bin works through its children:

Going up, children transition from the sink towards the source, so a downstream element is ready before anything is pushed into it.
Going down, children transition from the source towards the sink, so nothing keeps producing into an element that has already been torn down.

Nothing unwinds a failed transition. An element that refuses one stays where it was while the elements that already moved keep their new state, which is deliberate: the resulting graph shows exactly which element refused and in which transition.

Capability negotiation 

A capability describes the data crossing a link: a media type plus a set of fields, each an mpipe_caps_field identifier paired with an mpipe_value. It is a mpipe_structure, a fixed-size type held by value:

struct mpipe_structure s;

mpipe_structure_init_fields(&s, MPIPE_MEDIA_VIDEO,
    MPIPE_CAPS_PIXEL_FORMAT, MPIPE_TYPE_UINT, VIDEO_PIX_FMT_RGB565,
    MPIPE_CAPS_IMAGE_WIDTH, MPIPE_TYPE_UINT_RANGE, 16, 1280, 2,
    MPIPE_CAPS_IMAGE_HEIGHT, MPIPE_TYPE_UINT_RANGE, 16, 720, 2,
    MPIPE_CAPS_END);

A field holds either a single value or a range written [min, max, step], so the capability above covers every width from 16 to 1280 in steps of 2. Two capabilities intersect when they share a media type, have at least one field identifier in common, and every shared field has intersecting values. The result is the union of both: a shared field holds the intersected value, and a field only one side carries passes through unchanged. That is what lets a constraint travel down a chain of elements that do not themselves care about it. An ANY capability constrains nothing and intersects with anything; it is what a pad carries before it has negotiated. An empty one carries no field and intersects with nothing.

The source drives the negotiation on READY to PAUSED, in two passes:

        %%{init: {'themeVariables': {'fontSize': '18px'}, 'sequence': {'actorFontSize': 18, 'messageFontSize': 18, 'noteFontSize': 18}}}%%
sequenceDiagram
  autonumber
  participant so as src pad
  participant ti as sink pad
  participant to as src pad
  participant si as sink pad
  box rgba(79, 143, 214, 0.18) Source
  participant so
  end
  box rgba(148, 108, 196, 0.18) Transform
  participant ti
  participant to
  end
  box rgba(76, 168, 128, 0.18) Sink
  participant si
  end

  Note over so, si: Pass 1 - the caps query
  so ->> ti: can you take this format?
  ti ->> to: transform_caps()
  to ->> si: can you take this format?
  si -->> to: what I accept
  to ->> ti: transform_caps() back
  ti -->> so: what the chain accepts
  so ->> so: fixate to one format

  Note over so, si: Pass 2 - the caps event
  so ->> ti: this is the format
  ti ->> to: transform_caps(), narrowed
  to ->> si: this is the format
  si ->> si: set_caps()
  to ->> ti: set_caps() on both sides
  so ->> so: set_caps()

Capability negotiation.

A caps query travels downstream, each element narrowing it against what it supports before asking its own downstream, and the answer comes back narrowed to what the whole chain has in common. If an element refuses, the source simply offers the next format it supports rather than failing the negotiation.

A transform is the interesting case, because its two sides may speak different formats: a decoder takes one format in and produces another out. An element working in place rewrites the buffer where it lies, so the same format crosses it. Where the two sides differ, the transform_caps hook maps a capability from one side of the element to the other. The query crosses the element through it in both directions: out to ask the downstream peer, and back to express the answer in terms of the input side again.

The answer may still hold ranges, so the source fixates it - each range reduced to a single value - and announces the result downstream as a caps event. Every element applies its side through its set_caps hook, which is where hardware is actually configured.

Because a pad holds one capability rather than a set, an element that supports several formats is walked by index: the framework asks it for capability 0, then 1, and so on until it reports there are no more. An element whose formats come from a device answers each index by asking its driver.

Buffer pool negotiation 

Formats alone do not say how many buffers a graph needs, how large they must be, how they must be aligned, or whose pool provides them. A second query settles that, immediately after the format is fixed and in the same transition:

        %%{init: {'themeVariables': {'fontSize': '18px'}, 'sequence': {'actorFontSize': 18, 'messageFontSize': 18, 'noteFontSize': 18}}}%%
sequenceDiagram
  autonumber
  participant so as src pad
  participant ti as sink pad
  participant to as src pad
  participant si as sink pad
  box rgba(79, 143, 214, 0.18) Source
  participant so
  end
  box rgba(148, 108, 196, 0.18) Transform
  participant ti
  participant to
  end
  box rgba(76, 168, 128, 0.18) Sink
  participant si
  end

  so ->> ti: buffer pool query
  ti ->> to: forwarded downstream first
  to ->> si: buffer pool query
  si -->> to: propose_buffer_pool()
  to ->> to: decide_buffer_pool(), start out_pool
  ti -->> so: propose_buffer_pool()
  so ->> so: decide_buffer_pool(), start pool

Buffer pool negotiation.

The query travels down to the sink and proposals are written on the way back up, so an element always has its downstream’s proposal in hand before it decides anything.

A transform’s two sides are settled independently: decide_buffer_pool uses what its downstream proposed to settle the output pool, while propose_buffer_pool answers the upstream query with what the element needs on its input. The exception is passthrough, where one buffer crosses the element and there is only one pool to settle.

A proposal is either an entire pool or a bare config. Offering the pool lets the upstream element adopt it and avoid a copy; offering a config states requirements without handing anything over. Either way, a demand reaches a pool its proposer still owns only through mpipe_buffer_pool_set_config(), and the pool’s owner validates, clamps or refuses it.

Whoever starts a pool is who stops it, on PAUSED to READY. That symmetry is what makes a stop and replay behave like a first run.

Buffer flow 

Buffer flow is zero-copy. Each element that produces data owns a mpipe_buffer_pool, a vtable (configure, set_config, start, stop, acquire_buffer, release_buffer) over whatever backs it. That indirection lets a plugin hand out the buffer its driver already owns rather than a copy of it, so a frame captured by a camera can reach the display without a copy. What travels between elements is a reference, not the pixels.

Buffers are Zephyr net_buf allocations, with an mpipe_buffer_meta alongside carrying what the framework needs to know about one: the pool that owns it, how much of it is valid, a timestamp.

mpipe_push_buffer() walks a buffer downstream, calling the process_fn of each element in turn and following the buffer it produced. A NULL output means the buffer was consumed and the walk stops. The processing function owns the buffer it is given and releases it even when it fails.

Pipeline runtime 

The pipeline’s own thread drives the source: it acquires a buffer and pushes it downstream until the source reports the end of its data. It then sends an end-of-stream event downstream and pauses itself. A pipeline that needs two parts of its graph to run on separate threads places a queue element between them.

PLAYING to PAUSED only pauses the source thread. Whatever is queued is preserved, so resuming continues without loss.
PAUSED to READY tears the graph down: buffers still in flight are dropped, the children stop their pools, and the streaming thread is joined.

Messages travel to the application on the pipeline’s message channel, a zbus channel reachable with mpipe_element_get_bus_chan(). A message carries where something happened - the element that emitted it - and what happened: a type, plus, on a failure, a domain naming the phase that failed and an errno saying why. It carries no human-facing string; the sentence describing a failure belongs in the log at the site that detected it, while an application branches on the domain and the errno.

A pipeline with several sinks produces one end-of-stream message per sink. The pipeline counts them and passes on only the last, so the application is notified exactly once and never tears the graph down while a branch is still running.

Writing an element 

An element is written outside the framework and requires no change to it. It may embed one of the bases as its first member. It calls that base’s init with its own id and overrides only the hooks it needs:

struct my_sink {
        struct mpipe_sink sink;   /* must be first */
        const struct device *dev;
};

int my_sink_init(struct my_sink *self, uint8_t id)
{
        int ret = mpipe_sink_init(&self->sink, id);

        if (ret != 0) {
                return ret;
        }

        self->sink.sink_pad.enum_caps_fn = my_sink_enum_caps;
        self->sink.set_caps = my_sink_set_caps;

        return 0;
}

The hooks, in the order a pipeline exercises them:

Element configuration: set_property and get_property on the object, which the application reaches through mpipe_object_set_properties() and mpipe_object_get_properties(). A property configures this element only; nothing else has to agree on it.
Element capabilities: enum_caps_fn on a pad produces what the element supports, one capability per index, reporting -ENOENT past the last one. A capability known at build time is defined once with MPIPE_STRUCTURE_DEFINE and copied out; one only known at run time is built by querying the driver. set_caps applies the format the negotiation settled on, which is where hardware is configured. A transform element also implements transform_caps, mapping a capability on one side to the other side.
Buffer configuration: the downstream element proposes what it needs through propose_buffer_pool, and the upstream element decides through decide_buffer_pool with that proposal in hand.
State change: change_state runs what the element does on each transition. It must chain to its base, which performs the capability reset and the pool teardown every element is expected to do.
Processing function: process_fn on the sink pad is the heart of the element. It receives a buffer, owns it, and hands back the buffer it produced, or NULL when it consumed it. A source has no processing function: the pipeline thread acquires buffers from its pool.

Base elements 

The base plugin, CONFIG_MPIPE_BASE, holds the elements that belong to no media domain:

mpipe_caps_filter constrains the format negotiated on the link it sits on.
mpipe_tee pushes each buffer to every branch of a graph.
mpipe_queue runs the elements downstream of it on a thread of its own.
mpipe_app_src injects what the application pushes, as a source.
mpipe_app_sink hands what reaches it to the application, as a sink.

Debugging 

CONFIG_MPIPE_DUMP renders a pipeline as a Graphviz graph with mpipe_dump_bin(): every element with its state and pads, the peer of each pad, and the capability negotiated on it.

CONFIG_MPIPE_PLAYER adds a player that drives a pipeline through play, pause, stop and replay from a worker thread, with shell commands when the shell is enabled.

Configuration Options 

CONFIG_MPIPE enables the framework. Each plugin has its own option and contributes its own Kconfig file.
CONFIG_MPIPE_BASE builds the base elements, each behind its own option.
CONFIG_MPIPE_DUMP and CONFIG_MPIPE_PLAYER build the debugging utilities.
CONFIG_MPIPE_STRUCTURE_MAX_FIELDS sizes one capability. Size it for the largest union of fields that can meet in one intersection, not for the number of fields an element sets.
CONFIG_MPIPE_NET_BUF_POOL_COUNT sizes the shared net_buf pool: the sum of the in-flight buffers of every pipeline.
CONFIG_MPIPE_BIN_MAX_CHILDREN bounds the elements in one bin, which sizes the arrays used to order them during a state change.
CONFIG_MPIPE_THREADS_NUM, CONFIG_MPIPE_THREAD_STACK_SIZE and CONFIG_MPIPE_THREAD_DEFAULT_PRIORITY configure the pooled threads a pipeline draws from.
CONFIG_MPIPE_WORKQUEUE lets elements offload finite work items onto a shared priority work queue.
CONFIG_MPIPE_RPC builds the client-side elements that run their processing on another core.
CONFIG_MPIPE_FAKE_SRC provides a source of synthetic data, used to exercise a graph where a real source would need hardware.

API Reference 

Framework

Base Elements

Topology dump

Player