module-rtp: Add RTP jitter buffer

This new data structure is useful for reordering incoming packets if they
arrive out-of-order. Many audio codecs require frames and/or packets to be
processed in sequence order due to inter-frame dependencies, so reordering
is critical for such encoded data. It also detects lost packets and reports
those in sequence with received packets (crucial for proper PLC), and
detects and drops late and duplicate packets.

src/modules/meson.build

@@ -595,7 +595,8 @@ summary({'zeroconf-discover': build_module_zeroconf_discover}, bool_yn: true, se
 # (by avoiding build script code duplication), create a static library
 # that contains that common code.
 pipewire_module_rtp_common_lib = static_library('pipewire-module-rtp-common-lib',
-  [ 'module-rtp/stream.c' ],
+  [ 'module-rtp/stream.c',
+    'module-rtp/jitter-buffer.c' ],
   include_directories : [configinc],
   install : false,
   dependencies : [mathlib, dl_lib, rt_lib, pipewire_dep, opus_dep],

src/modules/module-rtp/jitter-buffer.h

@@ -0,0 +1,317 @@
+/* PipeWire */
+/* SPDX-FileCopyrightText: Copyright © 2026 Carlos Rafael Giani */
+/* SPDX-License-Identifier: MIT */
+
+#ifndef PIPEWIRE_RTP_JITTER_BUFFER_H
+#define PIPEWIRE_RTP_JITTER_BUFFER_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <stdint.h>
+#include <spa/utils/ringbuffer.h>
+#include <pipewire/loop.h>
+
+struct rtp_jitter_buffer_params {
+	/* How many slots the jitter buffer shall have. This defines
+	 * the capacity of the jitter buffer. Must be at least 1.
+	 * For a more detailed explanation what slots are, see
+	 * further below. */
+	size_t num_slots;
+	/* Maximum size of the packets, in bytes. This must include
+	 * the RTP header bytes. Must be at least 1. */
+	size_t max_packet_size;
+	/* Duration of each packet, in nanoseconds.
+	 * Must be at least 1. */
+	uint64_t packet_duration;
+	/* PipeWire loop, used for setting up a timeout timer.
+	 * This needs to be a loop that runs in the same thread
+	 * as this jitter buffer.
+	 * This must be set to a valid pointer. */
+	struct pw_loop *loop;
+	/* User-defined context that is passed to the function pointers. */
+	void *context;
+
+	/* Function pointer called when the jitter buffer decides to output
+	 * an RTP packet. In regular mode, this is immediately invoked when
+	 * rtp_jitter_buffer_insert_packet() is called. No data is copied
+	 * then; this function pointer is directly passed the packet_data and
+	 * packet_size argument values that rtp_jitter_buffer_insert_packet()
+	 * was called with. In hold-back mode, this is called when the jitter
+	 * buffer is able to output previously held back packets in order
+	 * (for example, because a preceding packet just now arrived), or
+	 * when the jitter buffer gets drained. packet_data points to the
+	 * bytes of the RTP packet that is output, and packet_size contains
+	 * the size of the packet in bytes.
+	 *
+	 * Should this function experience an error, it returns a negated
+	 * errno, and 0 if it succeeds. In case of an error, if the overall
+	 * logic can continue, it is recommended to fully reset the jitter
+	 * buffer by calling rtp_jitter_buffer_flush().
+	 *
+	 * This must be set to a valid pointer. */
+	int (*output_rtp_packet)(void *context, const uint8_t *packet_data, size_t packet_size,
+				 size_t header_size, uint32_t timestamp, uint16_t seqnum);
+	/* Function pointer called when the jitter buffer detected one or
+	 * more lost packets in a row. For example, if the last received packet
+	 * had sequence number 16, and then, the jitter buffer received packet
+	 * with sequence number 26, then 27, 28, etc., it eventually will
+	 * detect that packets 17 through 25 are lost, and call this, setting
+	 * seqnum_of_first_lost_packet to 17, and gap_length to 8. The  last
+	 * argument is set to false if the packet loss is detected mid stream
+	 * when the gap is small enough that measures like PLC can reasonably
+	 * cover all lost packets. If the detected gap is too large, open_ended
+	 * will be set to true. Implementations must then interpret the value of
+	 * gap_length as a maximum; whatever PLC measures the implementation
+	 * uses must not produce output larger than gap_length. This is a
+	 * safety measure to avoid cases where PLC would have to produce an
+	 * excessive amount of data.
+	 *
+	 * When open_ended is true, implementations should also apply a fade
+	 * out at the end of the produced PLC content to avoid a potential
+	 * hard cutoff at the end.
+	 *
+	 * Should this function experience an error, it returns a negated
+	 * errno, and 0 if it succeeds. In case of an error, if the overall
+	 * logic can continue, it is recommended to fully reset the jitter
+	 * buffer by calling rtp_jitter_buffer_flush().
+	 *
+	 * This must be set to a valid pointer. */
+	int (*signal_lost_packets)(void *context, uint16_t seqnum_of_first_lost_packet,
+		size_t gap_length, bool open_ended);
+};
+
+enum rtp_jitter_buffer_element_type {
+	RTP_JITTER_BUFFER_ELEMENT_TYPE_PACKET,
+	RTP_JITTER_BUFFER_ELEMENT_TYPE_GAP
+};
+
+/* Slot structure. A slot stores an element (a packer or
+ * a gap). Slots outside of the valid seqnum window have
+ * undefined contents. See the "slots" array documentation
+ * in the rtp_jitter_buffer structure below. */
+struct rtp_jitter_buffer_slot {
+	enum rtp_jitter_buffer_element_type element_type;
+	/* This is only used when the element_type is
+	 * RTP_JITTER_BUFFER_ELEMENT_TYPE_PACKET. See
+	 * the packet_buffer documentation below for more. */
+	size_t packet_size;
+	/* Size of the RTP header. This must be less than
+	 * or equal to packet_size. The packet data that
+	 * is stored in packet_buffer includes this header. */
+	size_t header_size;
+	/* RTP timestamp. */
+	uint32_t timestamp;
+};
+
+struct rtp_jitter_buffer {
+	/* Copy of the params passed to rtp_jitter_buffer_init(). */
+	struct rtp_jitter_buffer_params params;
+
+	/* Set to true by rtp_jitter_buffer_init(), and set to false
+	 * by rtp_jitter_buffer_shutdown(). */
+	bool initialized;
+
+	/* Array of slots. The size of this array equals num_slots.
+	 * Only slots that are within the valid seqnum window have
+	 * valid values in this array. For example, if the slot at
+	 * array index 3 corresponds to sequence number 400, but
+	 * the valid seqnum window starts at 402, then the value
+	 * in this array at index 3 has no meaning. Not used when
+	 * regular mode is active. */
+	struct rtp_jitter_buffer_slot *slots;
+
+	/* Buffer for storing packets in hold-back mode. The size
+	 * of this buffer is max_packet_size*num_slots. To access
+	 * a packet, do it this way:
+	 *
+	 * packet_data = &(packet_buffer[max_packet_size * index]);
+	 *
+	 * Doing this is technically less space efficient if the actual
+	 * packet size is smaller than max_packet_size, but it makes
+	 * buffer management and packet addressing much easier, and
+	 * avoids costly reallocations. The packet_size field in the
+	 * slot structure specifies how many bytes at that location
+	 * actually make up the packet. FOr example, if the slot at
+	 * index 5 in the slots array has its element_type value set
+	 * to RTP_JITTER_BUFFER_ELEMENT_TYPE_PACKET, then at location
+	 * &(packet_buffer[max_packet_size * 5]), valid packet data
+	 * can be found. And, starting from that location, the amount
+	 * of bytes the packet there is made of equals the packet_size
+	 * value of the slot at index 5.
+	 *
+	 * Only the locations that correspond to valid slots with
+	 * a packet as element contain valid data. For example, if the
+	 * slot at index 3 corresponds to sequence number 400, but the
+	 * valid seqnum window starts at 402, then the data at location
+	 * &(packet_buffer[max_packet_size * 3]) has no meaning. Also,
+	 * if the slot does lie within the valid window, but the type
+	 * of the contained element is RTP_JITTER_BUFFER_ELEMENT_TYPE_GAP,
+	 * then the associated location in this packert buffer is
+	 * meaningless. Not used when regular mode is active. */
+	uint8_t *packet_buffer;
+
+	/* If true, hold-back mode is active. */
+	bool hold_back_mode;
+	/* Sequence number of the last packet observed while in
+	 * regular mode. If no last packet is known (at startup for
+	 * example), this is set to -1.
+	 * Not used when hold-back mode is active. */
+	int32_t last_seqnum;
+	/* Timer that, when it times out, drains all slots and switches
+	 * back to regular mode. This is armed when hold-back mode is
+	 * activated, and when it is reset.
+	 * Not used when regular mode is active. */
+	struct spa_source *timeout_timer;
+	/* These define the valid seqnum window. Packet sequence numbers
+	 * in the [start_seqnum .. (start_seqnum + window_length - 1)]
+	 * range (both start and end of it inclusive) are within this
+	 * window, and are considered valid. A sequence number below the
+	 * window's start_seqnum is considered stale. A sequence number
+	 * beyond the end of the window extends the window; the window
+	 * length is increased such that this new sequence number is at
+	 * the very end of the window, that is, the length is extended
+	 * such that new_seqnum == (start_seqnum + window_length - 1)
+	 *
+	 * However, if this overextends the window (meaning that the
+	 * window length would be greater than the number of slots),
+	 * the logic in rtp_jitter_buffer_insert_packet() will adjust
+	 * the window's start_seqnum and length.
+	 *
+	 * If hold-back mode is off, these two values have no meaning.
+	 *
+	 * This window is necessary for the following:
+	 *
+	 * 1. It allows for detecting and dropping old, stale packets
+	 *    by looking at their sequence numbers.
+	 * 2. It detects when currently held-back packets need to be
+	 *    drained, that is, the gap that precedes them is not getting
+	 *    filled in time. This happens when packets are lost.
+	 * 3. It defines what slots are valid. Only those slots that are
+	 *    associated with sequence numbers that fit inside this
+	 *    window are valid. Slots outside of the window have no
+	 *    defined nor meaningful content (that is, packet size,
+	 *    packet buffer data, or slot element type).
+	 *
+	 * Note that even though valid_seqnum_window_start_seqnum is a
+	 * sequence number, it is a uint32_t instead of a uint16_t, for
+	 * easier arithmetics. */
+	uint32_t valid_seqnum_window_start_seqnum;
+	uint32_t valid_seqnum_window_length;
+};
+
+/* Initializes an uninitialized rtp_jitter_buffer instance with the given params.
+ *
+ * See rtp_jitter_buffer_params for details about the input parameters.
+ *
+ * params must be a valid pointer. An internal copy of the rtp_jitter_buffer_params
+ * instance that pointer refers to will be made, so the caller does not have to
+ * keep that instance around for the duration of the jitter buffer's lifetime.
+ *
+ * The jitter buffer must not be already initialized.
+ *
+ * Returns 0 if initialization was successful, nonzero in case of an error.
+ * If an error happens, this automatically calls rtp_jitter_buffer_shutdown(). */
+int rtp_jitter_buffer_init(struct rtp_jitter_buffer *jitter_buffer, struct rtp_jitter_buffer_params *params);
+
+/* Shuts down a previously initialized jitter buffer.
+ *
+ * jitter_buffer must be a valid pointer.
+ *
+ * Calling this on an uninitialized jitter buffer results in a no-op. */
+void rtp_jitter_buffer_shutdown(struct rtp_jitter_buffer *jitter_buffer);
+
+/* Returns true if the jitter buffer is initialized. */
+static inline bool rtp_jitter_buffer_is_initialized(struct rtp_jitter_buffer *jitter_buffer)
+{
+	return jitter_buffer->initialized;
+}
+
+/* Inserts an RTP packet into the jitter buffer.
+ *
+ * Depending on the sequence number of the RTP packet, the jitter buffer
+ * will immediately output that packet, or hold it back inside a slot
+ * for purposes of reordering and packet loss detection. (The packet's data
+ * will be copied into the packet buffer.) The packet order is defined by
+ * the RTP sequence number. (16-bit unsigned int RTP sequence wrap-around
+ * is handled properly.)
+ *
+ * If for example packets with sequence numbers 1, 4, 3, 5 arrived (in this
+ * incorrect order), packet 1 will have been output immediately, but packets
+ * 4, 3, 5 will have been held back (due to the gap at 2 and the incorrect
+ * order of packets). If later, packet 2 arrives, the gap at 2 is filled,
+ * and the jitter buffer will then output all packets from 2 to 5, in the
+ * correct order. If however packet 2 never arrives, but packets past
+ * packet 5 keep arriving, then eventually, packet 2 will be considered
+ * lost (causing signal_lost_packets() to be called), and the held-packets
+ * 4, 3, 5 will be drained (and output via output_rtp_packet(), in correct
+ * order 3, 4, 5).
+ *
+ * jitter_buffer and packet_data must be valid pointers. packet_size must
+ * be the size of the memory block pointed to by packet_data .
+ *
+ * header_size is the size of the RTP header, and must be less than or equa
+ * to packet_size. timestamp is the RTP timestamp.
+ *
+ * Note that while the values of header_size, timestamp, seqnum could be
+ * parsed from the packet header, this is not done by this function for
+ * efficiency reasons. Parsing those may require byte swapping to handle
+ * endianness, and computing the header size may depend on whether the
+ * header has extensions or not. These computations and proceses are done
+ * once, by the caller, and passed to this call via the arguments.
+ *
+ * Returns 0 if the call was successful, nonzero in case of an error.
+ * If an error happens, consider the jitter buffer as no longer usable
+ * (it can only be shut down then). */
+int rtp_jitter_buffer_insert_packet(struct rtp_jitter_buffer *jitter_buffer,
+				    const uint8_t *packet_data, size_t packet_size,
+				    size_t header_size, uint32_t timestamp, uint16_t seqnum);
+
+/* Drains all held-back packets and reports packet loss based on those packets.
+ *
+ * If the jitter buffer is currently holding back packets due to some packets
+ * having been added out of order previously, or because gaps were detected, this
+ * drains them, causing the output_rtp_packet() and signal_lost_packets() function
+ * pointers from the params passed to rtp_jitter_buffer_init() to be called. If
+ * for example packets with sequence numbers 1, 4, 3, 5 arrived (in this incorrect
+ * order), packet 1 will have been output immediately, but packets 4, 3, 5 will
+ * have been held back (due to the gap at 2 and the incorrect order of packets).
+ * If then, this function is called, the jitter buffer will (in this order) call
+ * signal_lost_packets() to signal the loss of packet 2, and then call
+ * output_rtp_packet() to output packets 3, 4, 5 (in that corrected order).
+ *
+ * After draining, the jitter buffer will be back in regular mode.
+ *
+ * If no packets were added, or if the jitter buffer is in regular mode,
+ * this does nothing.
+ *
+ * The jitter buffer also has an internal timer that is activated as soon
+ * as gaps and out of order packets are detected. If that timer expires,
+ * the jitter buffer will automatically drain itself. If however all gaps
+ * are filled in time, the timer is deactivated, and no automatic drain occurs.
+ *
+ * jitter_buffer must be a valid pointer.
+ *
+ * Returns 0 if the call was successful, nonzero in case of an error.
+ * If an error happens, consider the jitter buffer as no longer usable
+ * (it can only be shut down then). */
+int rtp_jitter_buffer_drain(struct rtp_jitter_buffer *jitter_buffer);
+
+/* Flushes all held-back packets.
+ *
+ * Unlike draining, this does not call any function pointers. Instead, it
+ * flushes any and all held-back packets (meaning, these packets are all
+ * gone, and _not_ reported as lost via signal_lost_packets()), and
+ * resets the jitter buffer back to the regular mode.
+ *
+ * This is useful for performing a hard reset on the jitter buffer.
+ *
+ * jitter_buffer must be a valid pointer. */
+void rtp_jitter_buffer_flush(struct rtp_jitter_buffer *jitter_buffer);
+
+#ifdef __cplusplus
+}  /* extern "C" */
+#endif
+
+#endif /* PIPEWIRE_RTP_JITTER_BUFFER_H */

test/meson.build

@@ -124,6 +124,17 @@ test('test-spa',
                link_with: pwtest_lib)
 )
 
+test('test-module-rtp-common-lib',
+    executable('test-module-rtp-common-lib',
+               'modules/module-rtp/test-jitter-buffer.c',
+               include_directories : [
+                 pwtest_inc,
+                 include_directories('../src/modules'),
+                 ],
+               dependencies: [ spa_dep, pipewire_module_rtp_common_dep ],
+               link_with: [pwtest_lib])
+)
+
 openal_info = find_program('openal-info', required: false)
 if openal_info.found()
     cdata.set_quoted('OPENAL_INFO_PATH', openal_info.full_path())