From d55315e75fe9d68e866e68538d64c7521ec9c335 Mon Sep 17 00:00:00 2001 From: Goffi Date: Thu, 6 Nov 2025 22:10:13 +0100 Subject: [PATCH] tools (webrtc): improved pipeline: - Adjust parameters and elements to better handle the buffering and jitter. - Add logging, better error handling. --- libervia/frontends/tools/webrtc.py | 160 ++++++++++++++++++----------- 1 file changed, 99 insertions(+), 61 deletions(-) diff --git a/libervia/frontends/tools/webrtc.py b/libervia/frontends/tools/webrtc.py index 8140ba60..2ce30f5c 100644 --- a/libervia/frontends/tools/webrtc.py +++ b/libervia/frontends/tools/webrtc.py @@ -537,7 +537,7 @@ class WebRTC: case SinksApp(): local_video_sink_elt = ( "appsink name=local_video_sink emit-signals=true drop=true " - "max-buffers=1 sync=True" + "max-buffers=2 sync=false" ) case SinksAuto(): if isinstance(self.sources_data, SourcesNone): @@ -552,7 +552,7 @@ class WebRTC: ) gst_pipe_elements = [ - "webrtcbin latency=30 name=sendrecv bundle-policy=max-bundle" + "webrtcbin latency=100 name=sendrecv bundle-policy=max-bundle" ] if self.merge_pip and local_video_sink_elt: @@ -572,12 +572,11 @@ class WebRTC: gst_pipe_elements.append( f""" input-selector name=video_selector - ! videorate drop-only=1 max-rate=30 ! video/x-raw,framerate=30/1 ! tee name=t {video_source_elt} name=video_src - ! queue max-size-buffers=1 max-size-time=0 max-size-bytes=0 leaky=downstream + ! queue max-size-buffers=3 leaky=downstream ! video_selector. videotestsrc name=muted_src is-live=true pattern=black @@ -585,10 +584,10 @@ class WebRTC: ! video_selector. t. - ! queue max-size-buffers=1 max-size-time=0 max-size-bytes=0 leaky=downstream + ! queue max-size-buffers=3 leaky=downstream ! videoscale ! videoconvert - ! vp8enc deadline=1 keyframe-max-dist=30 + ! vp8enc deadline=1 keyframe-max-dist=30 cpu-used=5 target-bitrate=1000000 ! rtpvp8pay picture-id-mode=15-bit ! application/x-rtp,media=video,encoding-name=VP8,payload={video_pt} ! sendrecv. @@ -600,7 +599,7 @@ class WebRTC: gst_pipe_elements.append( f""" t. - ! queue max-size-buffers=1 max-size-time=0 max-size-bytes=0 leaky=downstream + ! queue max-size-buffers=3 leaky=downstream ! videoconvert ! {local_video_sink_elt} """ @@ -766,11 +765,16 @@ class WebRTC: if ret == Gst.StateChangeReturn.FAILURE: log.error(f"Failed to change pipeline state to {state}") return False + elif ret == Gst.StateChangeReturn.ASYNC: + log.debug(f"Waiting for async state change to {state.value_nick.upper()}...") + ret, _, _ = self.pipeline.get_state(timeout=5 * Gst.SECOND) + if ret != Gst.StateChangeReturn.SUCCESS: + log.warning( + f"Async state change to {state.value_nick.upper()} timed out or failed!" + ) + return False - __, pending_state = self.pipeline.get_state(Gst.CLOCK_TIME_NONE)[:2] - if pending_state != Gst.State.VOID_PENDING: - log.warning(f"State change to {state} still pending") - + log.debug(f"Pipeline state successfully changed to {state.value_nick.upper()}") return True def monitor_connection_state(self): @@ -852,7 +856,9 @@ class WebRTC: ) return False elif isinstance(dest, Gst.Element): - return source.link(dest) + if not source.link(dest): + log.error(f"Failed to link {source.get_name()} to {dest.get_name()}") + return False else: log.error(f"Unexpected type for dest: {type(dest)}") return False @@ -879,29 +885,27 @@ class WebRTC: new_height = int(max_width / aspect_ratio) return max_width, new_height - def on_remote_decodebin_stream(self, _, pad: Gst.Pad) -> None: + def on_remote_decodebin_stream(self, decodebin, pad: Gst.Pad) -> None: """Handle the stream from the remote decodebin. This method processes the incoming stream from the remote decodebin, determining whether it's video or audio. It then sets up the appropriate GStreamer elements for video/audio processing and adds them to the pipeline. + @param decodebin: signal source. @param pad: The Gst.Pad from the remote decodebin producing the stream. """ assert self.pipeline is not None - if not pad.has_current_caps(): - log.error(f"{pad} has no caps, ignoring") - return - caps = pad.get_current_caps() - assert len(caps) - s = caps[0] - name = s.get_name() - log.debug(f"====> NAME START: {name}") + if not caps: + caps = pad.query_caps(None) - q = Gst.ElementFactory.make("queue") + name = caps.get_structure(0).get_name() + log.debug(f"Handling decoded pad of type {name}") - if name.startswith("video"): + queue = Gst.ElementFactory.make("queue") + + if name.startswith('video/'): log.debug("===> VIDEO OK") self._remote_video_pad = pad @@ -921,7 +925,9 @@ class WebRTC: height = (height + 3) // 4 * 4 log.info(f"Adjusted video size: {width}x{height}") - conv = Gst.ElementFactory.make("videoconvert") + + convert = Gst.ElementFactory.make("videoconvert") + if self.merge_pip: # with ``merge_pip`` set, we plug the remote stream to the composer compositor = self.pipeline.get_by_name("compositor") @@ -949,31 +955,32 @@ class WebRTC: remote_video_sink.set_property("width", width) remote_video_sink.set_property("height", height) remote_video_sink.set_property("sizing-policy", 1) + elif isinstance(self.sinks_data, SinksApp): # ``app`` sink without ``self.merge_pip`` set, be create the sink and # connect it to the ``remote_video_cb``. - remote_video_sink = Gst.ElementFactory.make("appsink") - - remote_video_caps = Gst.Caps.from_string("video/x-raw,format=RGB") - remote_video_sink.set_property("caps", remote_video_caps) - + remote_video_sink = Gst.ElementFactory.make("appsink", "remote_video_sink") + remote_video_sink.set_property("caps", Gst.Caps.from_string("video/x-raw,format=RGB")) remote_video_sink.set_property("emit-signals", True) remote_video_sink.set_property("drop", True) - remote_video_sink.set_property("max-buffers", 1) - remote_video_sink.set_property("sync", True) - remote_video_sink.connect("new-sample", self.sinks_data.remote_video_cb) - self.pipeline.add(remote_video_sink) + remote_video_sink.set_property("max-buffers", 2) + remote_video_sink.set_property("sync", False) + if self.sinks_data.remote_video_cb: + remote_video_sink.connect("new-sample", self.sinks_data.remote_video_cb) + else: + log.error("sinks_data.remote_video_cb is not available! Cannot display remote video.") + return elif isinstance(self.sinks_data, SinksAuto): # if ``self.merge_pip`` is not set, we create a dedicated # ``autovideosink`` for remote stream. remote_video_sink = Gst.ElementFactory.make("autovideosink") - self.pipeline.add(remote_video_sink) else: raise exceptions.InternalError( f'Unhandled "sinks_data" value: {self.sinks_data!r}' ) if adjust_resolution: + log.debug("Applying adjust resolution pipeline.") videoscale = Gst.ElementFactory.make("videoscale") adjusted_caps = Gst.Caps.from_string( f"video/x-raw,width={width},height={height}" @@ -981,20 +988,20 @@ class WebRTC: capsfilter = Gst.ElementFactory.make("capsfilter") capsfilter.set_property("caps", adjusted_caps) - self.pipeline.add(q, conv, videoscale, capsfilter) + self.pipeline.add(queue, convert, videoscale, capsfilter) self.pipeline.sync_children_states() - ret = pad.link(q.get_static_pad("sink")) + ret = pad.link(queue.get_static_pad("sink")) if ret != Gst.PadLinkReturn.OK: log.error(f"Error linking pad to queue: {ret}") return - if not q.link(conv): + if not queue.link(convert): log.error("Failed to link queue to videoconvert") return - if not conv.link(videoscale): + if not convert.link(videoscale): log.error("Failed to link videoconvert to videoscale") return @@ -1007,50 +1014,81 @@ class WebRTC: return else: - self.pipeline.add(q, conv) + self.pipeline.add(queue, convert, remote_video_sink) - self.pipeline.sync_children_states() - ret = pad.link(q.get_static_pad("sink")) - if ret != Gst.PadLinkReturn.OK: - log.error(f"Error linking pad: {ret}") - q.link(conv) - self.link_element_or_pad(conv, remote_video_sink) + pad.link(queue.get_static_pad("sink")) + Gst.Element.link_many(queue, convert, remote_video_sink) - elif name.startswith("audio"): + queue.sync_state_with_parent() + convert.sync_state_with_parent() + remote_video_sink.sync_state_with_parent() + + log.info("Successfully linked remote video stream.") + + elif name.startswith('audio/'): log.debug("===> Audio OK") - conv = Gst.ElementFactory.make("audioconvert") + # Create the audio sink chain: queue -> convert -> resample -> sink + convert = Gst.ElementFactory.make("audioconvert") resample = Gst.ElementFactory.make("audioresample") - remote_audio_sink = Gst.ElementFactory.make("autoaudiosink") - self.pipeline.add(q, conv, resample, remote_audio_sink) - self.pipeline.sync_children_states() - ret = pad.link(q.get_static_pad("sink")) - if ret != Gst.PadLinkReturn.OK: - log.error(f"Error linking pad: {ret}") - q.link(conv) - conv.link(resample) - resample.link(remote_audio_sink) + remote_video_sink = Gst.ElementFactory.make("autoaudiosink") + + self.pipeline.add(queue, convert, resample, remote_video_sink) + + pad.link(queue.get_static_pad("sink")) + Gst.Element.link_many(queue, convert, resample, remote_video_sink) + + # Bring all new elements to the PLAYING state + queue.sync_state_with_parent() + convert.sync_state_with_parent() + resample.sync_state_with_parent() + remote_video_sink.sync_state_with_parent() + log.info("Successfully linked remote audio stream.") else: - log.warning(f"unmanaged name: {name!r}") + log.warning(f"Ignoring unknown pad type from decodebin: {name}") - def on_pad_added(self, __, pad: Gst.Pad) -> None: + def on_pad_added(self, webrtcbin, pad: Gst.Pad) -> None: """Handle the addition of a new pad to the element. When a new source pad is added to the element, this method creates a decodebin, connects it to handle the stream, and links the pad to the decodebin. - @param __: Placeholder for the signal source. Not used in this method. + @param webrtcbin: Signal source. @param pad: The newly added pad. """ - log.debug("on_pad_added") + assert self.pipeline is not None + log.debug(f"Attempting to handle new source pad {pad.get_name()} from {webrtcbin.get_name()}") if pad.direction != Gst.PadDirection.SRC: return + # Create the potential downstream element. decodebin = Gst.ElementFactory.make("decodebin") + if not decodebin: + log.error("Critical: Failed to create decodebin element") + return + decodebin.connect("pad-added", self.on_remote_decodebin_stream) + self.pipeline.add(decodebin) decodebin.sync_state_with_parent() - pad.link(decodebin.get_static_pad("sink")) + + link_ret = pad.link(decodebin.get_static_pad("sink")) + + if link_ret == Gst.PadLinkReturn.OK: + log.info(f"Successfully linked pad '{pad.get_name()}' to a new decodebin.") + elif link_ret == Gst.PadLinkReturn.WAS_LINKED: + log.debug(f"Pad '{pad.get_name()}' was already linked. Cleaning up redundant decodebin.") + # The pad is already handled. We must clean up the decodebin we just created. + decodebin.set_state(Gst.State.NULL) + self.pipeline.remove(decodebin) + else: + log.error( + f"Failed to link pad '{pad.get_name()}' for an unexpected reason: {link_ret}. " + "Cleaning up." + ) + decodebin.set_state(Gst.State.NULL) + self.pipeline.remove(decodebin) + async def _call_start(self, callee: jid.JID, call_data: dict, profile: str) -> str: return await self.bridge.call_start(