From 4d514abfd60dcdcbadc4b748e56c1c4da4e051d9 Mon Sep 17 00:00:00 2001 From: Johan Sternerup Date: Thu, 22 Apr 2021 10:43:55 +0200 Subject: [PATCH] webrtcbin: Fix deadlock when receiving new sctp stream When receiving an sctp message for a stream that not yet has an sctpdec pad associated with it means we end up in _on_sctpdec_pad_added. At this point we're holding the sctpassocation lock. Then it's not possible to take the pc_lock because then code executing under the pc_lock (which means anything in the webrtc thread) may not take the sctpassociation lock. For example, running the data channel close procedure from the webrtc thread means we eventually end up sending a SCTP_RESET_STREAMS packet which needs to grab the sctpassociation lock. This means _on_sctpdec_pad_added simply cannot take the pc_lock and also it is not possible to postpone the channel creation as we need to link the pads right there. The solution is to introduce a more granular dc_lock that protects only the things that needs to be done to create the datachannel. Part-of: --- ext/webrtc/gstwebrtcbin.c | 76 +++++++++++++++++++++++---------------- ext/webrtc/gstwebrtcbin.h | 3 ++ 2 files changed, 49 insertions(+), 30 deletions(-) diff --git a/ext/webrtc/gstwebrtcbin.c b/ext/webrtc/gstwebrtcbin.c index 9ef835c48a..9ba1850b88 100644 --- a/ext/webrtc/gstwebrtcbin.c +++ b/ext/webrtc/gstwebrtcbin.c @@ -55,6 +55,9 @@ #define ICE_LOCK(w) (g_mutex_lock (ICE_GET_LOCK(w))) #define ICE_UNLOCK(w) (g_mutex_unlock (ICE_GET_LOCK(w))) +#define DC_GET_LOCK(w) (&w->priv->dc_lock) +#define DC_LOCK(w) (g_mutex_lock (DC_GET_LOCK(w))) +#define DC_UNLOCK(w) (g_mutex_unlock (DC_GET_LOCK(w))) /* The extra time for the rtpstorage compared to the RTP jitterbuffer (in ms) */ #define RTPSTORAGE_EXTRA_TIME (50) @@ -579,6 +582,7 @@ data_channel_match_for_id (WebRTCDataChannel * channel, gint * id) return channel->parent.id == *id; } +/* always called with dc_lock held */ static WebRTCDataChannel * _find_data_channel_for_id (GstWebRTCBin * webrtc, gint id) { @@ -1801,12 +1805,14 @@ gst_webrtc_bin_update_sctp_priority (GstWebRTCBin * webrtc) if (!webrtc->priv->sctp_transport) return; + DC_LOCK (webrtc); for (i = 0; i < webrtc->priv->data_channels->len; i++) { GstWebRTCDataChannel *channel = g_ptr_array_index (webrtc->priv->data_channels, i); sctp_priority = MAX (sctp_priority, channel->priority); } + DC_UNLOCK (webrtc); /* Default priority is low means DSCP field is left as 0 */ if (sctp_priority == 0) @@ -1961,13 +1967,16 @@ _on_data_channel_ready_state (WebRTCDataChannel * channel, if (ready_state == GST_WEBRTC_DATA_CHANNEL_STATE_OPEN) { gboolean found; + DC_LOCK (webrtc); found = g_ptr_array_remove (webrtc->priv->pending_data_channels, channel); if (found == FALSE) { GST_FIXME_OBJECT (webrtc, "Received open for unknown data channel"); + DC_UNLOCK (webrtc); return; } g_ptr_array_add (webrtc->priv->data_channels, gst_object_ref (channel)); + DC_UNLOCK (webrtc); gst_webrtc_bin_update_sctp_priority (webrtc); @@ -1976,12 +1985,14 @@ _on_data_channel_ready_state (WebRTCDataChannel * channel, } else if (ready_state == GST_WEBRTC_DATA_CHANNEL_STATE_CLOSED) { gboolean found; + DC_LOCK (webrtc); found = g_ptr_array_remove (webrtc->priv->pending_data_channels, channel) || g_ptr_array_remove (webrtc->priv->data_channels, channel); if (found == FALSE) { GST_FIXME_OBJECT (webrtc, "Received close for unknown data channel"); } + DC_UNLOCK (webrtc); } } @@ -1996,7 +2007,7 @@ _on_sctpdec_pad_added (GstElement * sctpdec, GstPad * pad, if (sscanf (GST_PAD_NAME (pad), "src_%u", &stream_id) != 1) return; - PC_LOCK (webrtc); + DC_LOCK (webrtc); channel = _find_data_channel_for_id (webrtc, stream_id); if (!channel) { channel = g_object_new (WEBRTC_TYPE_DATA_CHANNEL, NULL); @@ -2013,6 +2024,7 @@ _on_sctpdec_pad_added (GstElement * sctpdec, GstPad * pad, g_ptr_array_add (webrtc->priv->pending_data_channels, channel); } + DC_UNLOCK (webrtc); g_signal_connect (channel, "notify::ready-state", G_CALLBACK (_on_data_channel_ready_state), webrtc); @@ -2022,7 +2034,6 @@ _on_sctpdec_pad_added (GstElement * sctpdec, GstPad * pad, GST_WARNING_OBJECT (channel, "Failed to link sctp pad %s with channel %" GST_PTR_FORMAT, GST_PAD_NAME (pad), channel); gst_object_unref (sink_pad); - PC_UNLOCK (webrtc); } static void @@ -2036,9 +2047,9 @@ _on_sctp_state_notify (GstWebRTCSCTPTransport * sctp, GParamSpec * pspec, if (state == GST_WEBRTC_SCTP_TRANSPORT_STATE_CONNECTED) { int i; - PC_LOCK (webrtc); GST_DEBUG_OBJECT (webrtc, "SCTP association established"); + DC_LOCK (webrtc); for (i = 0; i < webrtc->priv->data_channels->len; i++) { WebRTCDataChannel *channel; @@ -2049,7 +2060,7 @@ _on_sctp_state_notify (GstWebRTCSCTPTransport * sctp, GParamSpec * pspec, if (!channel->parent.negotiated && !channel->opened) webrtc_data_channel_start_negotiation (channel); } - PC_UNLOCK (webrtc); + DC_UNLOCK (webrtc); } } @@ -2152,7 +2163,6 @@ _get_or_create_data_channel_transports (GstWebRTCBin * webrtc, guint session_id) if (!webrtc->priv->data_channel_transport) { TransportStream *stream; GstWebRTCSCTPTransport *sctp_transport; - int i; stream = _find_transport_for_session (webrtc, session_id); @@ -2201,14 +2211,6 @@ _get_or_create_data_channel_transports (GstWebRTCBin * webrtc, guint session_id) GST_ELEMENT (stream->send_bin), "data_sink")) g_warn_if_reached (); - for (i = 0; i < webrtc->priv->data_channels->len; i++) { - WebRTCDataChannel *channel; - - channel = g_ptr_array_index (webrtc->priv->data_channels, i); - - webrtc_data_channel_link_to_sctp (channel, webrtc->priv->sctp_transport); - } - gst_element_sync_state_with_parent (GST_ELEMENT (stream->send_bin)); gst_element_sync_state_with_parent (GST_ELEMENT (stream->receive_bin)); @@ -2226,6 +2228,7 @@ _get_or_create_data_channel_transports (GstWebRTCBin * webrtc, guint session_id) } webrtc->priv->sctp_transport = sctp_transport; + gst_webrtc_bin_update_sctp_priority (webrtc); } @@ -4502,6 +4505,7 @@ _update_data_channel_from_sdp_media (GstWebRTCBin * webrtc, remote_port, NULL); } + DC_LOCK (webrtc); for (i = 0; i < webrtc->priv->data_channels->len; i++) { WebRTCDataChannel *channel; @@ -4519,6 +4523,7 @@ _update_data_channel_from_sdp_media (GstWebRTCBin * webrtc, webrtc_data_channel_start_negotiation (channel); } } + DC_UNLOCK (webrtc); stream->active = TRUE; @@ -5650,6 +5655,7 @@ gst_webrtc_bin_create_data_channel (GstWebRTCBin * webrtc, const gchar * label, return NULL; PC_LOCK (webrtc); + DC_LOCK (webrtc); /* check if the id has been used already */ if (id != -1) { WebRTCDataChannel *channel = _find_data_channel_for_id (webrtc, id); @@ -5657,6 +5663,7 @@ gst_webrtc_bin_create_data_channel (GstWebRTCBin * webrtc, const gchar * label, GST_ELEMENT_WARNING (webrtc, LIBRARY, SETTINGS, ("Attempting to add a data channel with a duplicate ID: %i", id), NULL); + DC_UNLOCK (webrtc); PC_UNLOCK (webrtc); return NULL; } @@ -5669,6 +5676,7 @@ gst_webrtc_bin_create_data_channel (GstWebRTCBin * webrtc, const gchar * label, if (id == -1) { GST_ELEMENT_WARNING (webrtc, RESOURCE, NOT_FOUND, ("%s", "Failed to generate an identifier for a data channel"), NULL); + DC_UNLOCK (webrtc); PC_UNLOCK (webrtc); return NULL; } @@ -5679,25 +5687,31 @@ gst_webrtc_bin_create_data_channel (GstWebRTCBin * webrtc, const gchar * label, "max-retransmits", max_retransmits, "protocol", protocol, "negotiated", negotiated, "id", id, "priority", priority, NULL); - if (ret) { - gst_bin_add (GST_BIN (webrtc), ret->appsrc); - gst_bin_add (GST_BIN (webrtc), ret->appsink); + if (!ret) { + DC_UNLOCK (webrtc); + PC_UNLOCK (webrtc); + return ret; + } - gst_element_sync_state_with_parent (ret->appsrc); - gst_element_sync_state_with_parent (ret->appsink); + gst_bin_add (GST_BIN (webrtc), ret->appsrc); + gst_bin_add (GST_BIN (webrtc), ret->appsink); - ret = gst_object_ref (ret); - ret->webrtcbin = webrtc; - g_ptr_array_add (webrtc->priv->data_channels, ret); - gst_webrtc_bin_update_sctp_priority (webrtc); - webrtc_data_channel_link_to_sctp (ret, webrtc->priv->sctp_transport); - if (webrtc->priv->sctp_transport && - webrtc->priv->sctp_transport->association_established - && !ret->parent.negotiated) { - webrtc_data_channel_start_negotiation (ret); - } else { - _update_need_negotiation (webrtc); - } + gst_element_sync_state_with_parent (ret->appsrc); + gst_element_sync_state_with_parent (ret->appsink); + + ret = gst_object_ref (ret); + ret->webrtcbin = webrtc; + g_ptr_array_add (webrtc->priv->data_channels, ret); + DC_UNLOCK (webrtc); + + gst_webrtc_bin_update_sctp_priority (webrtc); + webrtc_data_channel_link_to_sctp (ret, webrtc->priv->sctp_transport); + if (webrtc->priv->sctp_transport && + webrtc->priv->sctp_transport->association_established + && !ret->parent.negotiated) { + webrtc_data_channel_start_negotiation (ret); + } else { + _update_need_negotiation (webrtc); } PC_UNLOCK (webrtc); @@ -6756,6 +6770,7 @@ gst_webrtc_bin_finalize (GObject * object) gst_webrtc_session_description_free (webrtc->priv->last_generated_offer); webrtc->priv->last_generated_offer = NULL; + g_mutex_clear (DC_GET_LOCK (webrtc)); g_mutex_clear (ICE_GET_LOCK (webrtc)); g_mutex_clear (PC_GET_LOCK (webrtc)); g_cond_clear (PC_GET_COND (webrtc)); @@ -7219,6 +7234,7 @@ gst_webrtc_bin_init (GstWebRTCBin * webrtc) g_cond_init (PC_GET_COND (webrtc)); g_mutex_init (ICE_GET_LOCK (webrtc)); + g_mutex_init (DC_GET_LOCK (webrtc)); webrtc->rtpbin = _create_rtpbin (webrtc); gst_bin_add (GST_BIN (webrtc), webrtc->rtpbin); diff --git a/ext/webrtc/gstwebrtcbin.h b/ext/webrtc/gstwebrtcbin.h index 808bdfa791..b180aa62f9 100644 --- a/ext/webrtc/gstwebrtcbin.h +++ b/ext/webrtc/gstwebrtcbin.h @@ -100,6 +100,9 @@ struct _GstWebRTCBinPrivate /* list of data channels we've received a sctp stream for but no data * channel protocol for */ GPtrArray *pending_data_channels; + /* dc_lock protects data_channels and pending_data_channels */ + /* lock ordering is pc_lock first, then dc_lock */ + GMutex dc_lock; guint jb_latency;