mirrors/gstreamer
mirrors/gstreamer
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gstreamer.git synced 2024-10-20 01:13:47 +00:00
Code Issues 1 Projects Releases Wiki Activity

pulsesink: rework context sharing

Browse source 
We also need to share the main-loop threads as this owns the context. Thus have
a class wide main-loop thread. From this we create a context per client-name.
Instead of always looking up the context, we keep this with the instance. The
reverse mapping is only needed in pulse singal handlers. This saves a lot of
locking. Also one signal handler becomes simpler as ther eis only one mainloop
to notify.

Now valgind happy - no leaks, no bad reads/writes.

This reverts major parts of commit 69a397c32f.

Fixes #628996
This commit is contained in:
Stefan Kost 2010-09-13 16:24:26 +03:00
parent d541f5e24d
commit f62dc6976b
2 changed files with 126 additions and 140 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

260
ext/pulse/pulsesink.c
View file

@ -115,6 +115,8 @@ struct _GstPulseContext
* (strings) and values should be GstPulseContext pointers. */ * (strings) and values should be GstPulseContext pointers. */
static GHashTable *gst_pulse_shared_contexts = NULL; static GHashTable *gst_pulse_shared_contexts = NULL;
static GMutex *pa_shared_ressource_mutex = NULL;
/* We keep a custom ringbuffer that is backed up by data allocated by /* We keep a custom ringbuffer that is backed up by data allocated by
* pulseaudio. We must also overide the commit function to write into * pulseaudio. We must also overide the commit function to write into
* pulseaudio memory instead. */ * pulseaudio memory instead. */
@ -125,6 +127,7 @@ struct _GstPulseRingBuffer
gchar *context_name; gchar *context_name;
gchar *stream_name; gchar *stream_name;
pa_context *context;
pa_stream *stream; pa_stream *stream;
pa_sample_spec sample_spec; pa_sample_spec sample_spec;
@ -166,13 +169,13 @@ static guint gst_pulseringbuffer_commit (GstRingBuffer * buf,
G_DEFINE_TYPE (GstPulseRingBuffer, gst_pulseringbuffer, GST_TYPE_RING_BUFFER); G_DEFINE_TYPE (GstPulseRingBuffer, gst_pulseringbuffer, GST_TYPE_RING_BUFFER);
static GMutex *pa_ring_buffer_mutex = NULL;
static void static void
gst_pulseringbuffer_init_contexts (void) gst_pulsesink_init_contexts (void)
{ {
g_assert (pa_ring_buffer_mutex == NULL); g_assert (pa_shared_ressource_mutex == NULL);
pa_ring_buffer_mutex = g_mutex_new (); pa_shared_ressource_mutex = g_mutex_new ();
gst_pulse_shared_contexts = g_hash_table_new (g_str_hash, g_str_equal); gst_pulse_shared_contexts = g_hash_table_new_full (g_str_hash, g_str_equal,
g_free, NULL);
} }
static void static void
@ -210,6 +213,7 @@ static void
gst_pulseringbuffer_init (GstPulseRingBuffer * pbuf) gst_pulseringbuffer_init (GstPulseRingBuffer * pbuf)
{ {
pbuf->stream_name = NULL; pbuf->stream_name = NULL;
pbuf->context = NULL;
pbuf->stream = NULL; pbuf->stream = NULL;
#ifdef HAVE_PULSE_0_9_13 #ifdef HAVE_PULSE_0_9_13
@ -268,38 +272,35 @@ gst_pulsering_destroy_stream (GstPulseRingBuffer * pbuf)
pbuf->stream_name = NULL; pbuf->stream_name = NULL;
} }
static GstPulseContext *
gst_pulsering_get_context (GstPulseRingBuffer * pbuf)
{
GstPulseContext *pctx;
g_mutex_lock (pa_ring_buffer_mutex);
pctx = g_hash_table_lookup (gst_pulse_shared_contexts, pbuf->context_name);
g_mutex_unlock (pa_ring_buffer_mutex);
return pctx;
}
static void static void
gst_pulsering_destroy_context (GstPulseRingBuffer * pbuf) gst_pulsering_destroy_context (GstPulseRingBuffer * pbuf)
{ {
GstPulseSink *psink; g_mutex_lock (pa_shared_ressource_mutex);
g_mutex_lock (pa_ring_buffer_mutex); GST_DEBUG_OBJECT (pbuf, "destroying ringbuffer %p", pbuf);
psink = GST_PULSESINK_CAST (GST_OBJECT_PARENT (pbuf));
gst_pulsering_destroy_stream (pbuf);
if (pbuf->context) {
pa_context_unref (pbuf->context);
pbuf->context = NULL;
}
if (pbuf->context_name) { if (pbuf->context_name) {
GstPulseContext *pctx; GstPulseContext *pctx;
GST_DEBUG_OBJECT (psink, "destroying context for pbuf=%p '%s'",
pbuf, pbuf->context_name);
pctx = g_hash_table_lookup (gst_pulse_shared_contexts, pbuf->context_name); pctx = g_hash_table_lookup (gst_pulse_shared_contexts, pbuf->context_name);
gst_pulsering_destroy_stream (pbuf); GST_DEBUG_OBJECT (pbuf, "releasing context with name %s, pbuf=%p, pctx=%p",
pbuf->context_name, pbuf, pctx);
if (pctx) { if (pctx) {
pctx->ring_buffers = g_slist_remove (pctx->ring_buffers, pbuf); pctx->ring_buffers = g_slist_remove (pctx->ring_buffers, pbuf);
if (pctx->ring_buffers == NULL) { if (pctx->ring_buffers == NULL) {
GST_DEBUG_OBJECT (pbuf,
"destroying final context with name %s, pbuf=%p, pctx=%p",
pbuf->context_name, pbuf, pctx);
pa_context_disconnect (pctx->context); pa_context_disconnect (pctx->context);
/* Make sure we don't get any further callbacks */ /* Make sure we don't get any further callbacks */
@ -309,15 +310,15 @@ gst_pulsering_destroy_context (GstPulseRingBuffer * pbuf)
#endif #endif
g_hash_table_remove (gst_pulse_shared_contexts, pbuf->context_name); g_hash_table_remove (gst_pulse_shared_contexts, pbuf->context_name);
g_free (pbuf->context_name);
pbuf->context_name = NULL;
pa_context_unref (pctx->context); pa_context_unref (pctx->context);
g_slice_free (GstPulseContext, pctx); g_slice_free (GstPulseContext, pctx);
} }
} }
g_free (pbuf->context_name);
pbuf->context_name = NULL;
} }
g_mutex_unlock (pa_ring_buffer_mutex); g_mutex_unlock (pa_shared_ressource_mutex);
} }
static void static void
@ -328,26 +329,18 @@ gst_pulseringbuffer_finalize (GObject * object)
ringbuffer = GST_PULSERING_BUFFER_CAST (object); ringbuffer = GST_PULSERING_BUFFER_CAST (object);
gst_pulsering_destroy_context (ringbuffer); gst_pulsering_destroy_context (ringbuffer);
G_OBJECT_CLASS (ring_parent_class)->finalize (object); G_OBJECT_CLASS (ring_parent_class)->finalize (object);
} }
static gboolean static gboolean
gst_pulsering_is_dead (GstPulseSink * psink, GstPulseRingBuffer * pbuf) gst_pulsering_is_dead (GstPulseSink * psink, GstPulseRingBuffer * pbuf)
{ {
GstPulseContext *pctx = gst_pulsering_get_context (pbuf); if (!pbuf->context
|| !PA_CONTEXT_IS_GOOD (pa_context_get_state (pbuf->context))
if (!pctx) {
GST_ELEMENT_ERROR (psink, RESOURCE, FAILED, ("Disconnected"), (NULL));
return TRUE;
}
if (!pctx->context
|| !PA_CONTEXT_IS_GOOD (pa_context_get_state (pctx->context))
|| !pbuf->stream || !pbuf->stream
|| !PA_STREAM_IS_GOOD (pa_stream_get_state (pbuf->stream))) { || !PA_STREAM_IS_GOOD (pa_stream_get_state (pbuf->stream))) {
const gchar *err_str = const gchar *err_str =
pctx->context ? pa_strerror (pa_context_errno (pctx->context)) : NULL; pbuf->context ? pa_strerror (pa_context_errno (pbuf->context)) : NULL;
GST_ELEMENT_ERROR (psink, RESOURCE, FAILED, ("Disconnected: %s", GST_ELEMENT_ERROR (psink, RESOURCE, FAILED, ("Disconnected: %s",
err_str), (NULL)); err_str), (NULL));
return TRUE; return TRUE;
@ -358,38 +351,26 @@ gst_pulsering_is_dead (GstPulseSink * psink, GstPulseRingBuffer * pbuf)
static void static void
gst_pulsering_context_state_cb (pa_context * c, void *userdata) gst_pulsering_context_state_cb (pa_context * c, void *userdata)
{ {
GstPulseSink *psink;
pa_context_state_t state; pa_context_state_t state;
pa_threaded_mainloop *mainloop = (pa_threaded_mainloop *) userdata;
GstPulseContext *pctx = (GstPulseContext *) userdata;
GSList *walk;
state = pa_context_get_state (c); state = pa_context_get_state (c);
for (walk = pctx->ring_buffers; walk; walk = g_slist_next (walk)) { GST_LOG ("got new context state %d", state);
GstPulseRingBuffer *pbuf = (GstPulseRingBuffer *) walk->data;
psink = GST_PULSESINK_CAST (GST_OBJECT_PARENT (pbuf));
GST_LOG_OBJECT (psink, "got new context state %d", state);
/* psink can be null when we are shutting down and the ringbuffer is already switch (state) {
* unparented */ case PA_CONTEXT_READY:
if (psink == NULL) case PA_CONTEXT_TERMINATED:
continue; case PA_CONTEXT_FAILED:
GST_LOG ("signaling");
pa_threaded_mainloop_signal (mainloop, 0);
break;
switch (state) { case PA_CONTEXT_UNCONNECTED:
case PA_CONTEXT_READY: case PA_CONTEXT_CONNECTING:
case PA_CONTEXT_TERMINATED: case PA_CONTEXT_AUTHORIZING:
case PA_CONTEXT_FAILED: case PA_CONTEXT_SETTING_NAME:
GST_LOG_OBJECT (psink, "signaling"); break;
pa_threaded_mainloop_signal (psink->mainloop, 0);
break;
case PA_CONTEXT_UNCONNECTED:
case PA_CONTEXT_CONNECTING:
case PA_CONTEXT_AUTHORIZING:
case PA_CONTEXT_SETTING_NAME:
break;
}
} }
} }
@ -443,7 +424,6 @@ gst_pulseringbuffer_open_device (GstRingBuffer * buf)
pbuf = GST_PULSERING_BUFFER_CAST (buf); pbuf = GST_PULSERING_BUFFER_CAST (buf);
g_assert (!pbuf->stream); g_assert (!pbuf->stream);
g_assert (psink->client_name); g_assert (psink->client_name);
if (psink->server) if (psink->server)
@ -453,25 +433,25 @@ gst_pulseringbuffer_open_device (GstRingBuffer * buf)
pbuf->context_name = g_strdup (psink->client_name); pbuf->context_name = g_strdup (psink->client_name);
pa_threaded_mainloop_lock (psink->mainloop); pa_threaded_mainloop_lock (psink->mainloop);
g_mutex_lock (pa_ring_buffer_mutex); g_mutex_lock (pa_shared_ressource_mutex);
pctx = g_hash_table_lookup (gst_pulse_shared_contexts, pbuf->context_name); pctx = g_hash_table_lookup (gst_pulse_shared_contexts, pbuf->context_name);
if (pctx == NULL) { if (pctx == NULL) {
pctx = g_slice_new0 (GstPulseContext); pctx = g_slice_new0 (GstPulseContext);
/* get the mainloop api and create a context */ /* get the mainloop api and create a context */
GST_LOG_OBJECT (psink, "new context with name %s", GST_INFO_OBJECT (psink, "new context with name %s, pbuf=%p, pctx=%p",
GST_STR_NULL (pbuf->context_name)); pbuf->context_name, pbuf, pctx);
api = pa_threaded_mainloop_get_api (psink->mainloop); api = pa_threaded_mainloop_get_api (psink->mainloop);
if (!(pctx->context = pa_context_new (api, pbuf->context_name))) if (!(pctx->context = pa_context_new (api, pbuf->context_name)))
goto create_failed; goto create_failed;
pctx->ring_buffers = g_slist_prepend (pctx->ring_buffers, pbuf); pctx->ring_buffers = g_slist_prepend (pctx->ring_buffers, pbuf);
g_hash_table_insert (gst_pulse_shared_contexts, pbuf->context_name, g_hash_table_insert (gst_pulse_shared_contexts,
(gpointer) pctx); g_strdup (pbuf->context_name), (gpointer) pctx);
/* register some essential callbacks */ /* register some essential callbacks */
pa_context_set_state_callback (pctx->context, pa_context_set_state_callback (pctx->context,
gst_pulsering_context_state_cb, pctx); gst_pulsering_context_state_cb, psink->mainloop);
#ifdef HAVE_PULSE_0_9_12 #ifdef HAVE_PULSE_0_9_12
pa_context_set_subscribe_callback (pctx->context, pa_context_set_subscribe_callback (pctx->context,
gst_pulsering_context_subscribe_cb, pctx); gst_pulsering_context_subscribe_cb, pctx);
@ -485,15 +465,19 @@ gst_pulseringbuffer_open_device (GstRingBuffer * buf)
PA_CONTEXT_NOAUTOSPAWN, NULL) < 0) PA_CONTEXT_NOAUTOSPAWN, NULL) < 0)
goto connect_failed; goto connect_failed;
} else { } else {
GST_LOG_OBJECT (psink, "reusing shared pulseaudio context with name %s", GST_INFO_OBJECT (psink,
GST_STR_NULL (pbuf->context_name)); "reusing shared context with name %s, pbuf=%p, pctx=%p",
pbuf->context_name, pbuf, pctx);
pctx->ring_buffers = g_slist_prepend (pctx->ring_buffers, pbuf); pctx->ring_buffers = g_slist_prepend (pctx->ring_buffers, pbuf);
} }
/* context created or shared okay */
pbuf->context = pa_context_ref (pctx->context);
for (;;) { for (;;) {
pa_context_state_t state; pa_context_state_t state;
state = pa_context_get_state (pctx->context); state = pa_context_get_state (pbuf->context);
GST_LOG_OBJECT (psink, "context state is now %d", state); GST_LOG_OBJECT (psink, "context state is now %d", state);
@ -510,7 +494,7 @@ gst_pulseringbuffer_open_device (GstRingBuffer * buf)
GST_LOG_OBJECT (psink, "opened the device"); GST_LOG_OBJECT (psink, "opened the device");
g_mutex_unlock (pa_ring_buffer_mutex); g_mutex_unlock (pa_shared_ressource_mutex);
pa_threaded_mainloop_unlock (psink->mainloop); pa_threaded_mainloop_unlock (psink->mainloop);
return TRUE; return TRUE;
@ -518,7 +502,7 @@ gst_pulseringbuffer_open_device (GstRingBuffer * buf)
/* ERRORS */ /* ERRORS */
unlock_and_fail: unlock_and_fail:
{ {
g_mutex_unlock (pa_ring_buffer_mutex); g_mutex_unlock (pa_shared_ressource_mutex);
gst_pulsering_destroy_context (pbuf); gst_pulsering_destroy_context (pbuf);
pa_threaded_mainloop_unlock (psink->mainloop); pa_threaded_mainloop_unlock (psink->mainloop);
@ -528,6 +512,7 @@ create_failed:
{ {
GST_ELEMENT_ERROR (psink, RESOURCE, FAILED, GST_ELEMENT_ERROR (psink, RESOURCE, FAILED,
("Failed to create context"), (NULL)); ("Failed to create context"), (NULL));
g_slice_free (GstPulseContext, pctx);
goto unlock_and_fail; goto unlock_and_fail;
} }
connect_failed: connect_failed:
@ -725,7 +710,6 @@ gst_pulseringbuffer_acquire (GstRingBuffer * buf, GstRingBufferSpec * spec)
{ {
GstPulseSink *psink; GstPulseSink *psink;
GstPulseRingBuffer *pbuf; GstPulseRingBuffer *pbuf;
GstPulseContext *pctx;
pa_buffer_attr wanted; pa_buffer_attr wanted;
const pa_buffer_attr *actual; const pa_buffer_attr *actual;
pa_channel_map channel_map; pa_channel_map channel_map;
@ -749,12 +733,12 @@ gst_pulseringbuffer_acquire (GstRingBuffer * buf, GstRingBufferSpec * spec)
pa_threaded_mainloop_lock (psink->mainloop); pa_threaded_mainloop_lock (psink->mainloop);
/* we need a context and a no stream */ /* we need a context and a no stream */
pctx = gst_pulsering_get_context (pbuf); g_assert (pbuf->context);
g_assert (!pbuf->stream); g_assert (!pbuf->stream);
/* enable event notifications */ /* enable event notifications */
GST_LOG_OBJECT (psink, "subscribing to context events"); GST_LOG_OBJECT (psink, "subscribing to context events");
if (!(o = pa_context_subscribe (pctx->context, if (!(o = pa_context_subscribe (pbuf->context,
PA_SUBSCRIPTION_MASK_SINK_INPUT, NULL, NULL))) PA_SUBSCRIPTION_MASK_SINK_INPUT, NULL, NULL)))
goto subscribe_failed; goto subscribe_failed;
@ -772,10 +756,10 @@ gst_pulseringbuffer_acquire (GstRingBuffer * buf, GstRingBufferSpec * spec)
/* create a stream */ /* create a stream */
GST_LOG_OBJECT (psink, "creating stream with name %s", name); GST_LOG_OBJECT (psink, "creating stream with name %s", name);
if (psink->proplist) { if (psink->proplist) {
if (!(pbuf->stream = pa_stream_new_with_proplist (pctx->context, if (!(pbuf->stream = pa_stream_new_with_proplist (pbuf->context,
name, &pbuf->sample_spec, &channel_map, psink->proplist))) name, &pbuf->sample_spec, &channel_map, psink->proplist)))
goto stream_failed; goto stream_failed;
} else if (!(pbuf->stream = pa_stream_new (pctx->context, } else if (!(pbuf->stream = pa_stream_new (pbuf->context,
name, &pbuf->sample_spec, &channel_map))) name, &pbuf->sample_spec, &channel_map)))
goto stream_failed; goto stream_failed;
@ -910,21 +894,21 @@ subscribe_failed:
{ {
GST_ELEMENT_ERROR (psink, RESOURCE, FAILED, GST_ELEMENT_ERROR (psink, RESOURCE, FAILED,
("pa_context_subscribe() failed: %s", ("pa_context_subscribe() failed: %s",
pa_strerror (pa_context_errno (pctx->context))), (NULL)); pa_strerror (pa_context_errno (pbuf->context))), (NULL));
goto unlock_and_fail; goto unlock_and_fail;
} }
stream_failed: stream_failed:
{ {
GST_ELEMENT_ERROR (psink, RESOURCE, FAILED, GST_ELEMENT_ERROR (psink, RESOURCE, FAILED,
("Failed to create stream: %s", ("Failed to create stream: %s",
pa_strerror (pa_context_errno (pctx->context))), (NULL)); pa_strerror (pa_context_errno (pbuf->context))), (NULL));
goto unlock_and_fail; goto unlock_and_fail;
} }
connect_failed: connect_failed:
{ {
GST_ELEMENT_ERROR (psink, RESOURCE, FAILED, GST_ELEMENT_ERROR (psink, RESOURCE, FAILED,
("Failed to connect stream: %s", ("Failed to connect stream: %s",
pa_strerror (pa_context_errno (pctx->context))), (NULL)); pa_strerror (pa_context_errno (pbuf->context))), (NULL));
goto unlock_and_fail; goto unlock_and_fail;
} }
} }
@ -966,7 +950,6 @@ gst_pulsering_set_corked (GstPulseRingBuffer * pbuf, gboolean corked,
{ {
pa_operation *o = NULL; pa_operation *o = NULL;
GstPulseSink *psink; GstPulseSink *psink;
GstPulseContext *pctx = NULL;
gboolean res = FALSE; gboolean res = FALSE;
psink = GST_PULSESINK_CAST (GST_OBJECT_PARENT (pbuf)); psink = GST_PULSESINK_CAST (GST_OBJECT_PARENT (pbuf));
@ -1002,10 +985,9 @@ server_dead:
} }
cork_failed: cork_failed:
{ {
pctx = gst_pulsering_get_context (pbuf);
GST_ELEMENT_ERROR (psink, RESOURCE, FAILED, GST_ELEMENT_ERROR (psink, RESOURCE, FAILED,
("pa_stream_cork() failed: %s", ("pa_stream_cork() failed: %s",
pa_strerror (pa_context_errno (pctx->context))), (NULL)); pa_strerror (pa_context_errno (pbuf->context))), (NULL));
goto cleanup; goto cleanup;
} }
} }
@ -1264,7 +1246,6 @@ gst_pulseringbuffer_commit (GstRingBuffer * buf, guint64 * sample,
{ {
GstPulseSink *psink; GstPulseSink *psink;
GstPulseRingBuffer *pbuf; GstPulseRingBuffer *pbuf;
GstPulseContext *pctx;
guint result; guint result;
guint8 *data_end; guint8 *data_end;
gboolean reverse; gboolean reverse;
@ -1636,20 +1617,16 @@ was_paused:
} }
writable_size_failed: writable_size_failed:
{ {
pctx = gst_pulsering_get_context (pbuf);
GST_ELEMENT_ERROR (psink, RESOURCE, FAILED, GST_ELEMENT_ERROR (psink, RESOURCE, FAILED,
("pa_stream_writable_size() failed: %s", ("pa_stream_writable_size() failed: %s",
pa_strerror (pa_context_errno (pctx->context))), (NULL)); pa_strerror (pa_context_errno (pbuf->context))), (NULL));
goto unlock_and_fail; goto unlock_and_fail;
} }
write_failed: write_failed:
{ {
pctx = gst_pulsering_get_context (pbuf);
GST_ELEMENT_ERROR (psink, RESOURCE, FAILED, GST_ELEMENT_ERROR (psink, RESOURCE, FAILED,
("pa_stream_write() failed: %s", ("pa_stream_write() failed: %s",
pa_strerror (pa_context_errno (pctx->context))), (NULL)); pa_strerror (pa_context_errno (pbuf->context))), (NULL));
goto unlock_and_fail; goto unlock_and_fail;
} }
} }
@ -1676,7 +1653,7 @@ static void gst_pulsesink_init_interfaces (GType type);
GST_IMPLEMENT_PULSEPROBE_METHODS (GstPulseSink, gst_pulsesink); GST_IMPLEMENT_PULSEPROBE_METHODS (GstPulseSink, gst_pulsesink);
#define _do_init(type) \ #define _do_init(type) \
gst_pulseringbuffer_init_contexts (); \ gst_pulsesink_init_contexts (); \
gst_pulsesink_init_interfaces (type); gst_pulsesink_init_interfaces (type);
GST_BOILERPLATE_FULL (GstPulseSink, gst_pulsesink, GstBaseAudioSink, GST_BOILERPLATE_FULL (GstPulseSink, gst_pulsesink, GstBaseAudioSink,
@ -2007,7 +1984,6 @@ gst_pulsesink_set_volume (GstPulseSink * psink, gdouble volume)
pa_cvolume v; pa_cvolume v;
pa_operation *o = NULL; pa_operation *o = NULL;
GstPulseRingBuffer *pbuf; GstPulseRingBuffer *pbuf;
GstPulseContext *pctx;
uint32_t idx; uint32_t idx;
if (!psink->mainloop) if (!psink->mainloop)
@ -2026,9 +2002,7 @@ gst_pulsesink_set_volume (GstPulseSink * psink, gdouble volume)
gst_pulse_cvolume_from_linear (&v, pbuf->sample_spec.channels, volume); gst_pulse_cvolume_from_linear (&v, pbuf->sample_spec.channels, volume);
pctx = gst_pulsering_get_context (pbuf); if (!(o = pa_context_set_sink_input_volume (pbuf->context, idx,
if (!(o = pa_context_set_sink_input_volume (pctx->context, idx,
&v, NULL, NULL))) &v, NULL, NULL)))
goto volume_failed; goto volume_failed;
@ -2068,7 +2042,7 @@ volume_failed:
{ {
GST_ELEMENT_ERROR (psink, RESOURCE, FAILED, GST_ELEMENT_ERROR (psink, RESOURCE, FAILED,
("pa_stream_set_sink_input_volume() failed: %s", ("pa_stream_set_sink_input_volume() failed: %s",
pa_strerror (pa_context_errno (pctx->context))), (NULL)); pa_strerror (pa_context_errno (pbuf->context))), (NULL));
goto unlock; goto unlock;
} }
} }
@ -2078,7 +2052,6 @@ gst_pulsesink_set_mute (GstPulseSink * psink, gboolean mute)
{ {
pa_operation *o = NULL; pa_operation *o = NULL;
GstPulseRingBuffer *pbuf; GstPulseRingBuffer *pbuf;
GstPulseContext *pctx;
uint32_t idx; uint32_t idx;
if (!psink->mainloop) if (!psink->mainloop)
@ -2095,9 +2068,7 @@ gst_pulsesink_set_mute (GstPulseSink * psink, gboolean mute)
if ((idx = pa_stream_get_index (pbuf->stream)) == PA_INVALID_INDEX) if ((idx = pa_stream_get_index (pbuf->stream)) == PA_INVALID_INDEX)
goto no_index; goto no_index;
pctx = gst_pulsering_get_context (pbuf); if (!(o = pa_context_set_sink_input_mute (pbuf->context, idx,
if (!(o = pa_context_set_sink_input_mute (pctx->context, idx,
mute, NULL, NULL))) mute, NULL, NULL)))
goto mute_failed; goto mute_failed;
@ -2137,7 +2108,7 @@ mute_failed:
{ {
GST_ELEMENT_ERROR (psink, RESOURCE, FAILED, GST_ELEMENT_ERROR (psink, RESOURCE, FAILED,
("pa_stream_set_sink_input_mute() failed: %s", ("pa_stream_set_sink_input_mute() failed: %s",
pa_strerror (pa_context_errno (pctx->context))), (NULL)); pa_strerror (pa_context_errno (pbuf->context))), (NULL));
goto unlock; goto unlock;
} }
} }
@ -2174,7 +2145,6 @@ static gdouble
gst_pulsesink_get_volume (GstPulseSink * psink) gst_pulsesink_get_volume (GstPulseSink * psink)
{ {
GstPulseRingBuffer *pbuf; GstPulseRingBuffer *pbuf;
GstPulseContext *pctx;
pa_operation *o = NULL; pa_operation *o = NULL;
gdouble v = DEFAULT_VOLUME; gdouble v = DEFAULT_VOLUME;
uint32_t idx; uint32_t idx;
@ -2191,9 +2161,7 @@ gst_pulsesink_get_volume (GstPulseSink * psink)
if ((idx = pa_stream_get_index (pbuf->stream)) == PA_INVALID_INDEX) if ((idx = pa_stream_get_index (pbuf->stream)) == PA_INVALID_INDEX)
goto no_index; goto no_index;
pctx = gst_pulsering_get_context (pbuf); if (!(o = pa_context_get_sink_input_info (pbuf->context, idx,
if (!(o = pa_context_get_sink_input_info (pctx->context, idx,
gst_pulsesink_sink_input_info_cb, pbuf))) gst_pulsesink_sink_input_info_cb, pbuf)))
goto info_failed; goto info_failed;
@ -2239,7 +2207,7 @@ info_failed:
{ {
GST_ELEMENT_ERROR (psink, RESOURCE, FAILED, GST_ELEMENT_ERROR (psink, RESOURCE, FAILED,
("pa_context_get_sink_input_info() failed: %s", ("pa_context_get_sink_input_info() failed: %s",
pa_strerror (pa_context_errno (pctx->context))), (NULL)); pa_strerror (pa_context_errno (pbuf->context))), (NULL));
goto unlock; goto unlock;
} }
} }
@ -2248,7 +2216,6 @@ static gboolean
gst_pulsesink_get_mute (GstPulseSink * psink) gst_pulsesink_get_mute (GstPulseSink * psink)
{ {
GstPulseRingBuffer *pbuf; GstPulseRingBuffer *pbuf;
GstPulseContext *pctx;
pa_operation *o = NULL; pa_operation *o = NULL;
uint32_t idx; uint32_t idx;
gboolean mute = FALSE; gboolean mute = FALSE;
@ -2266,9 +2233,7 @@ gst_pulsesink_get_mute (GstPulseSink * psink)
if ((idx = pa_stream_get_index (pbuf->stream)) == PA_INVALID_INDEX) if ((idx = pa_stream_get_index (pbuf->stream)) == PA_INVALID_INDEX)
goto no_index; goto no_index;
pctx = gst_pulsering_get_context (pbuf); if (!(o = pa_context_get_sink_input_info (pbuf->context, idx,
if (!(o = pa_context_get_sink_input_info (pctx->context, idx,
gst_pulsesink_sink_input_info_cb, pbuf))) gst_pulsesink_sink_input_info_cb, pbuf)))
goto info_failed; goto info_failed;
@ -2279,7 +2244,6 @@ gst_pulsesink_get_mute (GstPulseSink * psink)
} }
unlock: unlock:
if (o) if (o)
pa_operation_unref (o); pa_operation_unref (o);
@ -2308,7 +2272,7 @@ info_failed:
{ {
GST_ELEMENT_ERROR (psink, RESOURCE, FAILED, GST_ELEMENT_ERROR (psink, RESOURCE, FAILED,
("pa_context_get_sink_input_info() failed: %s", ("pa_context_get_sink_input_info() failed: %s",
pa_strerror (pa_context_errno (pctx->context))), (NULL)); pa_strerror (pa_context_errno (pbuf->context))), (NULL));
goto unlock; goto unlock;
} }
} }
@ -2343,7 +2307,6 @@ static gchar *
gst_pulsesink_device_description (GstPulseSink * psink) gst_pulsesink_device_description (GstPulseSink * psink)
{ {
GstPulseRingBuffer *pbuf; GstPulseRingBuffer *pbuf;
GstPulseContext *pctx;
pa_operation *o = NULL; pa_operation *o = NULL;
gchar *t; gchar *t;
@ -2355,9 +2318,7 @@ gst_pulsesink_device_description (GstPulseSink * psink)
if (pbuf == NULL || pbuf->stream == NULL) if (pbuf == NULL || pbuf->stream == NULL)
goto no_buffer; goto no_buffer;
pctx = gst_pulsering_get_context (pbuf); if (!(o = pa_context_get_sink_info_by_index (pbuf->context,
if (!(o = pa_context_get_sink_info_by_index (pctx->context,
pa_stream_get_device_index (pbuf->stream), pa_stream_get_device_index (pbuf->stream),
gst_pulsesink_sink_info_cb, pbuf))) gst_pulsesink_sink_info_cb, pbuf)))
goto info_failed; goto info_failed;
@ -2369,7 +2330,6 @@ gst_pulsesink_device_description (GstPulseSink * psink)
} }
unlock: unlock:
if (o) if (o)
pa_operation_unref (o); pa_operation_unref (o);
@ -2393,7 +2353,7 @@ info_failed:
{ {
GST_ELEMENT_ERROR (psink, RESOURCE, FAILED, GST_ELEMENT_ERROR (psink, RESOURCE, FAILED,
("pa_context_get_sink_info_by_index() failed: %s", ("pa_context_get_sink_info_by_index() failed: %s",
pa_strerror (pa_context_errno (pctx->context))), (NULL)); pa_strerror (pa_context_errno (pbuf->context))), (NULL));
goto unlock; goto unlock;
} }
} }
@ -2489,7 +2449,6 @@ gst_pulsesink_change_title (GstPulseSink * psink, const gchar * t)
{ {
pa_operation *o = NULL; pa_operation *o = NULL;
GstPulseRingBuffer *pbuf; GstPulseRingBuffer *pbuf;
GstPulseContext *pctx;
pa_threaded_mainloop_lock (psink->mainloop); pa_threaded_mainloop_lock (psink->mainloop);
@ -2501,8 +2460,6 @@ gst_pulsesink_change_title (GstPulseSink * psink, const gchar * t)
g_free (pbuf->stream_name); g_free (pbuf->stream_name);
pbuf->stream_name = g_strdup (t); pbuf->stream_name = g_strdup (t);
pctx = gst_pulsering_get_context (pbuf);
if (!(o = pa_stream_set_name (pbuf->stream, pbuf->stream_name, NULL, NULL))) if (!(o = pa_stream_set_name (pbuf->stream, pbuf->stream_name, NULL, NULL)))
goto name_failed; goto name_failed;
@ -2525,7 +2482,7 @@ name_failed:
{ {
GST_ELEMENT_ERROR (psink, RESOURCE, FAILED, GST_ELEMENT_ERROR (psink, RESOURCE, FAILED,
("pa_stream_set_name() failed: %s", ("pa_stream_set_name() failed: %s",
pa_strerror (pa_context_errno (pctx->context))), (NULL)); pa_strerror (pa_context_errno (pbuf->context))), (NULL));
goto unlock; goto unlock;
} }
} }
@ -2551,7 +2508,6 @@ gst_pulsesink_change_props (GstPulseSink * psink, GstTagList * l)
gboolean empty = TRUE; gboolean empty = TRUE;
pa_operation *o = NULL; pa_operation *o = NULL;
GstPulseRingBuffer *pbuf; GstPulseRingBuffer *pbuf;
GstPulseContext *pctx;
pl = pa_proplist_new (); pl = pa_proplist_new ();
@ -2572,13 +2528,10 @@ gst_pulsesink_change_props (GstPulseSink * psink, GstTagList * l)
goto finish; goto finish;
pa_threaded_mainloop_lock (psink->mainloop); pa_threaded_mainloop_lock (psink->mainloop);
pbuf = GST_PULSERING_BUFFER_CAST (GST_BASE_AUDIO_SINK (psink)->ringbuffer); pbuf = GST_PULSERING_BUFFER_CAST (GST_BASE_AUDIO_SINK (psink)->ringbuffer);
if (pbuf == NULL || pbuf->stream == NULL) if (pbuf == NULL || pbuf->stream == NULL)
goto no_buffer; goto no_buffer;
pctx = gst_pulsering_get_context (pbuf);
if (!(o = pa_stream_proplist_update (pbuf->stream, PA_UPDATE_REPLACE, if (!(o = pa_stream_proplist_update (pbuf->stream, PA_UPDATE_REPLACE,
pl, NULL, NULL))) pl, NULL, NULL)))
goto update_failed; goto update_failed;
@ -2608,7 +2561,7 @@ update_failed:
{ {
GST_ELEMENT_ERROR (psink, RESOURCE, FAILED, GST_ELEMENT_ERROR (psink, RESOURCE, FAILED,
("pa_stream_proplist_update() failed: %s", ("pa_stream_proplist_update() failed: %s",
pa_strerror (pa_context_errno (pctx->context))), (NULL)); pa_strerror (pa_context_errno (pbuf->context))), (NULL));
goto unlock; goto unlock;
} }
} }
@ -2672,18 +2625,30 @@ static GstStateChangeReturn
gst_pulsesink_change_state (GstElement * element, GstStateChange transition) gst_pulsesink_change_state (GstElement * element, GstStateChange transition)
{ {
GstPulseSink *pulsesink = GST_PULSESINK (element); GstPulseSink *pulsesink = GST_PULSESINK (element);
GstPulseSinkClass *klass = GST_PULSESINK_GET_CLASS (pulsesink);
GstStateChangeReturn ret; GstStateChangeReturn ret;
guint res; guint res;
switch (transition) { switch (transition) {
case GST_STATE_CHANGE_NULL_TO_READY: case GST_STATE_CHANGE_NULL_TO_READY:
g_assert (pulsesink->mainloop == NULL); g_mutex_lock (pa_shared_ressource_mutex);
pulsesink->mainloop = pa_threaded_mainloop_new (); if (!klass->main_loop_ref_ct) {
if (!pulsesink->mainloop) GST_INFO_OBJECT (element, "new pa main loop thread");
goto mainloop_failed; klass->mainloop = pa_threaded_mainloop_new ();
res = pa_threaded_mainloop_start (pulsesink->mainloop); if (!klass->mainloop)
g_assert (res == 0); goto mainloop_failed;
klass->main_loop_ref_ct = 1;
res = pa_threaded_mainloop_start (klass->mainloop);
g_assert (res == 0);
g_mutex_unlock (pa_shared_ressource_mutex);
pulsesink->mainloop = klass->mainloop;
} else {
GST_INFO_OBJECT (element, "reusing pa main loop thread");
pulsesink->mainloop = klass->mainloop;
klass->main_loop_ref_ct++;
g_mutex_unlock (pa_shared_ressource_mutex);
}
break; break;
case GST_STATE_CHANGE_READY_TO_PAUSED: case GST_STATE_CHANGE_READY_TO_PAUSED:
gst_element_post_message (element, gst_element_post_message (element,
@ -2706,9 +2671,16 @@ gst_pulsesink_change_state (GstElement * element, GstStateChange transition)
break; break;
case GST_STATE_CHANGE_READY_TO_NULL: case GST_STATE_CHANGE_READY_TO_NULL:
if (pulsesink->mainloop) { if (pulsesink->mainloop) {
pa_threaded_mainloop_stop (pulsesink->mainloop); g_mutex_lock (pa_shared_ressource_mutex);
pa_threaded_mainloop_free (pulsesink->mainloop); klass->main_loop_ref_ct--;
pulsesink->mainloop = NULL; pulsesink->mainloop = NULL;
if (!klass->main_loop_ref_ct) {
GST_INFO_OBJECT (element, "terminating pa main loop thread");
pa_threaded_mainloop_stop (klass->mainloop);
pa_threaded_mainloop_free (klass->mainloop);
klass->mainloop = NULL;
}
g_mutex_unlock (pa_shared_ressource_mutex);
} }
break; break;
default: default:
@ -2720,6 +2692,7 @@ gst_pulsesink_change_state (GstElement * element, GstStateChange transition)
/* ERRORS */ /* ERRORS */
mainloop_failed: mainloop_failed:
{ {
g_mutex_unlock (pa_shared_ressource_mutex);
GST_ELEMENT_ERROR (pulsesink, RESOURCE, FAILED, GST_ELEMENT_ERROR (pulsesink, RESOURCE, FAILED,
("pa_threaded_mainloop_new() failed"), (NULL)); ("pa_threaded_mainloop_new() failed"), (NULL));
return GST_STATE_CHANGE_FAILURE; return GST_STATE_CHANGE_FAILURE;
@ -2729,9 +2702,16 @@ state_failure:
if (transition == GST_STATE_CHANGE_NULL_TO_READY) { if (transition == GST_STATE_CHANGE_NULL_TO_READY) {
/* Clear the PA mainloop if baseaudiosink failed to open the ring_buffer */ /* Clear the PA mainloop if baseaudiosink failed to open the ring_buffer */
g_assert (pulsesink->mainloop); g_assert (pulsesink->mainloop);
pa_threaded_mainloop_stop (pulsesink->mainloop); g_mutex_lock (pa_shared_ressource_mutex);
pa_threaded_mainloop_free (pulsesink->mainloop); klass->main_loop_ref_ct--;
pulsesink->mainloop = NULL; pulsesink->mainloop = NULL;
if (!klass->main_loop_ref_ct) {
GST_INFO_OBJECT (element, "terminating pa main loop thread");
pa_threaded_mainloop_stop (klass->mainloop);
pa_threaded_mainloop_free (klass->mainloop);
klass->mainloop = NULL;
}
g_mutex_unlock (pa_shared_ressource_mutex);
} }
return ret; return ret;
} }

6
ext/pulse/pulsesink.h
View file

@ -46,6 +46,9 @@ G_BEGIN_DECLS
(G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_PULSESINK)) (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_PULSESINK))
#define GST_PULSESINK_CAST(obj) \ #define GST_PULSESINK_CAST(obj) \
((GstPulseSink *)(obj)) ((GstPulseSink *)(obj))
#define GST_PULSESINK_GET_CLASS(obj) \
(G_TYPE_INSTANCE_GET_CLASS((obj),GST_TYPE_PULSESINK,GstPulseSinkClass))
typedef struct _GstPulseSink GstPulseSink; typedef struct _GstPulseSink GstPulseSink;
typedef struct _GstPulseSinkClass GstPulseSinkClass; typedef struct _GstPulseSinkClass GstPulseSinkClass;
@ -79,6 +82,9 @@ struct _GstPulseSink
struct _GstPulseSinkClass struct _GstPulseSinkClass
{ {
GstBaseAudioSinkClass parent_class; GstBaseAudioSinkClass parent_class;
pa_threaded_mainloop *mainloop;
guint main_loop_ref_ct;
}; };
GType gst_pulsesink_get_type (void); GType gst_pulsesink_get_type (void);