The new version of Polypaudio includes a threading abstraction that allows application of a more synchronous nature to use the API more easily. Using this, the complexity of the Polypaudio plug-in is greatly reduced and also removes the risk of stalling the communications layer. Signed-off-by: Pierre Ossman <ossman@xxxxxxxxx>
? polyp-09.patch
? polyp-update.patch
? mix/pcm_upmix.loT
Index: configure.in
===================================================================
RCS file: /cvsroot/alsa/alsa-plugins/configure.in,v
retrieving revision 1.21
diff -u -r1.21 configure.in
--- configure.in 19 Apr 2006 07:41:49 -0000 1.21
+++ configure.in 28 May 2006 14:41:32 -0000
@@ -16,7 +16,7 @@
PKG_CHECK_MODULES(JACK, jack >= 0.98, [HAVE_JACK=yes], [HAVE_JACK=no])
AM_CONDITIONAL(HAVE_JACK, test x$HAVE_JACK = xyes)
-PKG_CHECK_MODULES(polypaudio, [polyplib], [HAVE_POLYP=yes], [HAVE_POLYP=no])
+PKG_CHECK_MODULES(polypaudio, [polyplib >= 0.9.0], [HAVE_POLYP=yes], [HAVE_POLYP=no])
AM_CONDITIONAL(HAVE_POLYP, test x$HAVE_POLYP = xyes)
PKG_CHECK_MODULES(samplerate, [samplerate], [HAVE_SAMPLERATE=yes], [HAVE_SAMPLERATE=no])
Index: polyp/ctl_polyp.c
===================================================================
RCS file: /cvsroot/alsa/alsa-plugins/polyp/ctl_polyp.c,v
retrieving revision 1.4
diff -u -r1.4 ctl_polyp.c
--- polyp/ctl_polyp.c 8 Mar 2006 12:23:13 -0000 1.4
+++ polyp/ctl_polyp.c 28 May 2006 14:41:32 -0000
@@ -20,8 +20,6 @@
#include <sys/poll.h>
-#include <pthread.h>
-
#include <alsa/asoundlib.h>
#include <alsa/control_external.h>
@@ -43,8 +41,6 @@
int subscribed;
int updated;
-
- pthread_mutex_t mutex;
} snd_ctl_polyp_t;
#define SOURCE_VOL_NAME "Capture Volume"
@@ -62,14 +58,19 @@
snd_ctl_polyp_t *ctl = (snd_ctl_polyp_t*)userdata;
int chan;
- if (is_last)
+ assert(ctl);
+
+ if (is_last) {
+ pa_threaded_mainloop_signal(ctl->p->mainloop, 0);
return;
+ }
- assert(ctl && i);
+ assert(i);
if (!!ctl->sink_muted != !!i->mute) {
ctl->sink_muted = i->mute;
ctl->updated |= UPDATE_SINK_MUTE;
+ polyp_poll_activate(ctl->p);
}
if (ctl->sink_volume.channels == i->volume.channels) {
@@ -81,6 +82,7 @@
return;
ctl->updated |= UPDATE_SINK_VOL;
+ polyp_poll_activate(ctl->p);
}
memcpy(&ctl->sink_volume, &i->volume, sizeof(pa_cvolume));
@@ -91,14 +93,19 @@
snd_ctl_polyp_t *ctl = (snd_ctl_polyp_t*)userdata;
int chan;
- if (is_last)
+ assert(ctl);
+
+ if (is_last) {
+ pa_threaded_mainloop_signal(ctl->p->mainloop, 0);
return;
+ }
- assert(ctl && i);
+ assert(i);
if (!!ctl->source_muted != !!i->mute) {
ctl->source_muted = i->mute;
ctl->updated |= UPDATE_SOURCE_MUTE;
+ polyp_poll_activate(ctl->p);
}
if (ctl->source_volume.channels == i->volume.channels) {
@@ -110,6 +117,7 @@
return;
ctl->updated |= UPDATE_SOURCE_VOL;
+ polyp_poll_activate(ctl->p);
}
memcpy(&ctl->source_volume, &i->volume, sizeof(pa_cvolume));
@@ -163,14 +171,14 @@
assert(ctl);
- pthread_mutex_lock(&ctl->mutex);
+ pa_threaded_mainloop_lock(ctl->p->mainloop);
if (ctl->source)
count += 2;
if (ctl->sink)
count += 2;
- pthread_mutex_unlock(&ctl->mutex);
+ pa_threaded_mainloop_unlock(ctl->p->mainloop);
return count;
}
@@ -184,7 +192,7 @@
snd_ctl_elem_id_set_interface(id, SND_CTL_ELEM_IFACE_MIXER);
- pthread_mutex_lock(&ctl->mutex);
+ pa_threaded_mainloop_lock(ctl->p->mainloop);
if (ctl->source) {
if (offset == 0)
@@ -194,7 +202,7 @@
} else
offset += 2;
- pthread_mutex_unlock(&ctl->mutex);
+ pa_threaded_mainloop_unlock(ctl->p->mainloop);
if (offset == 2)
snd_ctl_elem_id_set_name(id, SINK_VOL_NAME);
@@ -229,18 +237,13 @@
snd_ctl_polyp_t *ctl = ext->private_data;
int err = 0;
- assert(ctl);
-
if (key > 3)
return -EINVAL;
- pthread_mutex_lock(&ctl->mutex);
-
+ assert(ctl);
assert(ctl->p);
- err = polyp_finish_poll(ctl->p);
- if (err < 0)
- goto finish;
+ pa_threaded_mainloop_lock(ctl->p->mainloop);
err = polyp_check_connection(ctl->p);
if (err < 0)
@@ -265,7 +268,7 @@
*count = 1;
finish:
- pthread_mutex_unlock(&ctl->mutex);
+ pa_threaded_mainloop_unlock(ctl->p->mainloop);
return err;
}
@@ -288,14 +291,9 @@
pa_cvolume *vol = NULL;
assert(ctl);
-
- pthread_mutex_lock(&ctl->mutex);
-
assert(ctl->p);
- err = polyp_finish_poll(ctl->p);
- if (err < 0)
- goto finish;
+ pa_threaded_mainloop_lock(ctl->p->mainloop);
err = polyp_check_connection(ctl->p);
if (err < 0)
@@ -329,7 +327,7 @@
}
finish:
- pthread_mutex_unlock(&ctl->mutex);
+ pa_threaded_mainloop_unlock(ctl->p->mainloop);
return err;
}
@@ -343,14 +341,9 @@
pa_cvolume *vol = NULL;
assert(ctl);
-
- pthread_mutex_lock(&ctl->mutex);
-
assert(ctl->p && ctl->p->context);
- err = polyp_finish_poll(ctl->p);
- if (err < 0)
- goto finish;
+ pa_threaded_mainloop_lock(ctl->p->mainloop);
err = polyp_check_connection(ctl->p);
if (err < 0)
@@ -395,17 +388,17 @@
if (key == 0)
o = pa_context_set_source_volume_by_name(ctl->p->context,
- ctl->source, vol, NULL, NULL);
+ ctl->source, vol, polyp_context_success_cb, ctl->p);
else
o = pa_context_set_sink_volume_by_name(ctl->p->context,
- ctl->sink, vol, NULL, NULL);
+ ctl->sink, vol, polyp_context_success_cb, ctl->p);
} else {
if (key == 1)
o = pa_context_set_source_mute_by_name(ctl->p->context,
- ctl->source, ctl->source_muted, NULL, NULL);
+ ctl->source, ctl->source_muted, polyp_context_success_cb, ctl->p);
else
o = pa_context_set_sink_mute_by_name(ctl->p->context,
- ctl->sink, ctl->sink_muted, NULL, NULL);
+ ctl->sink, ctl->sink_muted, polyp_context_success_cb, ctl->p);
}
err = polyp_wait_operation(ctl->p, o);
@@ -416,7 +409,7 @@
err = 1;
finish:
- pthread_mutex_unlock(&ctl->mutex);
+ pa_threaded_mainloop_unlock(ctl->p->mainloop);
return err;
}
@@ -427,11 +420,11 @@
assert(ctl);
- pthread_mutex_lock(&ctl->mutex);
+ pa_threaded_mainloop_lock(ctl->p->mainloop);
ctl->subscribed = !!(subscribe & SND_CTL_EVENT_MASK_VALUE);
- pthread_mutex_unlock(&ctl->mutex);
+ pa_threaded_mainloop_unlock(ctl->p->mainloop);
}
static int polyp_read_event(snd_ctl_ext_t *ext, snd_ctl_elem_id_t *id,
@@ -443,7 +436,7 @@
assert(ctl);
- pthread_mutex_lock(&ctl->mutex);
+ pa_threaded_mainloop_lock(ctl->p->mainloop);
if (!ctl->updated || !ctl->subscribed)
goto finish;
@@ -469,10 +462,13 @@
*event_mask = SND_CTL_EVENT_MASK_VALUE;
- err = 0;
+ if (!ctl->updated)
+ polyp_poll_deactivate(ctl->p);
+
+ err = 1;
finish:
- pthread_mutex_unlock(&ctl->mutex);
+ pa_threaded_mainloop_unlock(ctl->p->mainloop);
return err;
}
@@ -483,14 +479,13 @@
int count;
assert(ctl);
-
- pthread_mutex_lock(&ctl->mutex);
-
assert(ctl->p);
+ pa_threaded_mainloop_lock(ctl->p->mainloop);
+
count = polyp_poll_descriptors_count(ctl->p);
- pthread_mutex_unlock(&ctl->mutex);
+ pa_threaded_mainloop_unlock(ctl->p->mainloop);
return count;
}
@@ -502,20 +497,16 @@
snd_ctl_polyp_t *ctl = ext->private_data;
assert(ctl);
-
- pthread_mutex_lock(&ctl->mutex);
-
assert(ctl->p);
+ pa_threaded_mainloop_lock(ctl->p->mainloop);
+
num = polyp_poll_descriptors(ctl->p, pfd, space);
if (num < 0)
goto finish;
- if (ctl->updated)
- pa_mainloop_wakeup(ctl->p->mainloop);
-
finish:
- pthread_mutex_unlock(&ctl->mutex);
+ pa_threaded_mainloop_unlock(ctl->p->mainloop);
return num;
}
@@ -526,11 +517,10 @@
int err = 0;
assert(ctl);
-
- pthread_mutex_lock(&ctl->mutex);
-
assert(ctl->p);
+ pa_threaded_mainloop_lock(ctl->p->mainloop);
+
err = polyp_poll_revents(ctl->p, pfd, nfds, revents);
if (err < 0)
goto finish;
@@ -541,7 +531,7 @@
*revents |= POLLIN;
finish:
- pthread_mutex_unlock(&ctl->mutex);
+ pa_threaded_mainloop_unlock(ctl->p->mainloop);
return err;
}
@@ -560,8 +550,6 @@
if (ctl->sink)
free(ctl->sink);
- pthread_mutex_destroy(&ctl->mutex);
-
free(ctl);
}
@@ -591,6 +579,8 @@
ctl->source = strdup(i->default_source_name);
if (i->default_sink_name && !ctl->sink)
ctl->sink = strdup(i->default_sink_name);
+
+ pa_threaded_mainloop_signal(ctl->p->mainloop, 0);
}
SND_CTL_PLUGIN_DEFINE_FUNC(polyp)
@@ -603,7 +593,6 @@
int err;
snd_ctl_polyp_t *ctl;
pa_operation *o;
- pthread_mutexattr_t mutexattr;
snd_config_for_each(i, next, conf) {
snd_config_t *n = snd_config_iterator_entry(i);
@@ -647,21 +636,15 @@
ctl = calloc(1, sizeof(*ctl));
ctl->p = polyp_new();
- assert(ctl->p);
-
- err = polyp_connect(ctl->p, server);
- if (err < 0)
+ if (!ctl->p) {
+ err = -EIO;
goto error;
+ }
- err = polyp_start_thread(ctl->p);
+ err = polyp_connect(ctl->p, server);
if (err < 0)
goto error;
- pthread_mutexattr_init(&mutexattr);
- pthread_mutexattr_settype(&mutexattr, PTHREAD_MUTEX_RECURSIVE);
- pthread_mutex_init(&ctl->mutex, &mutexattr);
- pthread_mutexattr_destroy(&mutexattr);
-
if (source)
ctl->source = strdup(source);
else if (device)
@@ -673,19 +656,33 @@
ctl->sink = strdup(device);
if (!ctl->source || !ctl->sink) {
+ pa_threaded_mainloop_lock(ctl->p->mainloop);
+
o = pa_context_get_server_info(ctl->p->context, server_info_cb, ctl);
err = polyp_wait_operation(ctl->p, o);
+
pa_operation_unref(o);
+
+ pa_threaded_mainloop_unlock(ctl->p->mainloop);
+
if (err < 0)
goto error;
}
+ pa_threaded_mainloop_lock(ctl->p->mainloop);
+
pa_context_set_subscribe_callback(ctl->p->context, event_cb, ctl);
o = pa_context_subscribe(ctl->p->context,
- PA_SUBSCRIPTION_MASK_SINK | PA_SUBSCRIPTION_MASK_SOURCE, NULL, NULL);
+ PA_SUBSCRIPTION_MASK_SINK | PA_SUBSCRIPTION_MASK_SOURCE,
+ polyp_context_success_cb, ctl->p);
+
err = polyp_wait_operation(ctl->p, o);
+
pa_operation_unref(o);
+
+ pa_threaded_mainloop_unlock(ctl->p->mainloop);
+
if (err < 0)
goto error;
Index: polyp/pcm_polyp.c
===================================================================
RCS file: /cvsroot/alsa/alsa-plugins/polyp/pcm_polyp.c,v
retrieving revision 1.5
diff -u -r1.5 pcm_polyp.c
--- polyp/pcm_polyp.c 8 Mar 2006 12:23:13 -0000 1.5
+++ polyp/pcm_polyp.c 28 May 2006 14:41:33 -0000
@@ -21,8 +21,6 @@
#include <stdio.h>
#include <sys/poll.h>
-#include <pthread.h>
-
#include <alsa/asoundlib.h>
#include <alsa/pcm_external.h>
@@ -46,8 +44,6 @@
pa_sample_spec ss;
unsigned int frame_size;
pa_buffer_attr buffer_attr;
-
- pthread_mutex_t mutex;
} snd_pcm_polyp_t;
static void update_ptr(snd_pcm_polyp_t *pcm)
@@ -74,20 +70,17 @@
int err = 0;
assert(pcm);
+ assert(pcm->p);
- pthread_mutex_lock(&pcm->mutex);
-
- assert(pcm->p && pcm->stream);
+ pa_threaded_mainloop_lock(pcm->p->mainloop);
- err = polyp_finish_poll(pcm->p);
- if (err < 0)
- goto finish;
+ assert(pcm->stream);
err = polyp_check_connection(pcm->p);
if (err < 0)
goto finish;
- o = pa_stream_cork(pcm->stream, 0, NULL, NULL);
+ o = pa_stream_cork(pcm->stream, 0, polyp_stream_success_cb, pcm->p);
assert(o);
err = polyp_wait_operation(pcm->p, o);
@@ -100,7 +93,7 @@
}
finish:
- pthread_mutex_unlock(&pcm->mutex);
+ pa_threaded_mainloop_unlock(pcm->p->mainloop);
return err;
}
@@ -112,20 +105,17 @@
int err = 0;
assert(pcm);
+ assert(pcm->p);
- pthread_mutex_lock(&pcm->mutex);
-
- assert(pcm->p && pcm->stream);
+ pa_threaded_mainloop_lock(pcm->p->mainloop);
- err = polyp_finish_poll(pcm->p);
- if (err < 0)
- goto finish;
+ assert(pcm->stream);
err = polyp_check_connection(pcm->p);
if (err < 0)
goto finish;
- o = pa_stream_flush(pcm->stream, NULL, NULL);
+ o = pa_stream_flush(pcm->stream, polyp_stream_success_cb, pcm->p);
assert(o);
err = polyp_wait_operation(pcm->p, o);
@@ -137,7 +127,7 @@
goto finish;
}
- o = pa_stream_cork(pcm->stream, 1, NULL, NULL);
+ o = pa_stream_cork(pcm->stream, 1, polyp_stream_success_cb, pcm->p);
assert(o);
err = polyp_wait_operation(pcm->p, o);
@@ -150,7 +140,7 @@
}
finish:
- pthread_mutex_unlock(&pcm->mutex);
+ pa_threaded_mainloop_unlock(pcm->p->mainloop);
return err;
}
@@ -162,20 +152,17 @@
int err = 0;
assert(pcm);
+ assert(pcm->p);
- pthread_mutex_lock(&pcm->mutex);
-
- assert(pcm->p && pcm->stream);
+ pa_threaded_mainloop_lock(pcm->p->mainloop);
- err = polyp_finish_poll(pcm->p);
- if (err < 0)
- goto finish;
+ assert(pcm->stream);
err = polyp_check_connection(pcm->p);
if (err < 0)
goto finish;
- o = pa_stream_drain(pcm->stream, NULL, NULL);
+ o = pa_stream_drain(pcm->stream, polyp_stream_success_cb, pcm->p);
assert(o);
err = polyp_wait_operation(pcm->p, o);
@@ -188,7 +175,7 @@
}
finish:
- pthread_mutex_unlock(&pcm->mutex);
+ pa_threaded_mainloop_unlock(pcm->p->mainloop);
return err;
}
@@ -199,14 +186,11 @@
int err = 0;
assert(pcm);
+ assert(pcm->p);
- pthread_mutex_lock(&pcm->mutex);
-
- assert(pcm->p && pcm->stream);
+ pa_threaded_mainloop_lock(pcm->p->mainloop);
- err = polyp_finish_poll(pcm->p);
- if (err < 0)
- goto finish;
+ assert(pcm->stream);
err = polyp_check_connection(pcm->p);
if (err < 0)
@@ -214,14 +198,41 @@
update_ptr(pcm);
- err = polyp_start_poll(pcm->p);
+ err = snd_pcm_bytes_to_frames(io->pcm, pcm->ptr);
+
+finish:
+ pa_threaded_mainloop_unlock(pcm->p->mainloop);
+
+ return err;
+}
+
+static int polyp_delay(snd_pcm_ioplug_t *io,
+ snd_pcm_sframes_t *delayp)
+{
+ snd_pcm_polyp_t *pcm = io->private_data;
+ int err = 0;
+ pa_usec_t lat;
+
+ assert(pcm);
+ assert(pcm->p);
+
+ pa_threaded_mainloop_lock(pcm->p->mainloop);
+
+ assert(pcm->stream);
+
+ err = polyp_check_connection(pcm->p);
if (err < 0)
goto finish;
- err = snd_pcm_bytes_to_frames(io->pcm, pcm->ptr);
+ if (pa_stream_get_latency(pcm->stream, &lat, NULL)) {
+ err = -EIO;
+ goto finish;
+ }
+
+ *delayp = snd_pcm_bytes_to_frames(io->pcm, pa_usec_to_bytes(lat, &pcm->ss));
finish:
- pthread_mutex_unlock(&pcm->mutex);
+ pa_threaded_mainloop_unlock(pcm->p->mainloop);
return err;
}
@@ -236,14 +247,11 @@
int err = 0;
assert(pcm);
+ assert(pcm->p);
- pthread_mutex_lock(&pcm->mutex);
-
- assert(pcm->p && pcm->stream);
+ pa_threaded_mainloop_lock(pcm->p->mainloop);
- err = polyp_finish_poll(pcm->p);
- if (err < 0)
- goto finish;
+ assert(pcm->stream);
err = polyp_check_connection(pcm->p);
if (err < 0)
@@ -261,10 +269,13 @@
/* Make sure the buffer pointer is in sync */
update_ptr(pcm);
+ if (pcm->last_size < pcm->buffer_attr.minreq)
+ polyp_poll_deactivate(pcm->p);
+
err = size;
finish:
- pthread_mutex_unlock(&pcm->mutex);
+ pa_threaded_mainloop_unlock(pcm->p->mainloop);
return err;
}
@@ -280,14 +291,11 @@
int err = 0;
assert(pcm);
+ assert(pcm->p);
- pthread_mutex_lock(&pcm->mutex);
-
- assert(pcm->p && pcm->stream);
+ pa_threaded_mainloop_lock(pcm->p->mainloop);
- err = polyp_finish_poll(pcm->p);
- if (err < 0)
- goto finish;
+ assert(pcm->stream);
err = polyp_check_connection(pcm->p);
if (err < 0)
@@ -325,28 +333,40 @@
/* Make sure the buffer pointer is in sync */
update_ptr(pcm);
+ if (pcm->last_size < pcm->buffer_attr.minreq)
+ polyp_poll_deactivate(pcm->p);
+
err = size - (remain_size / pcm->frame_size);
finish:
- pthread_mutex_unlock(&pcm->mutex);
+ pa_threaded_mainloop_unlock(pcm->p->mainloop);
return err;
}
+static void stream_request_cb(pa_stream *p, size_t length, void *userdata)
+{
+ snd_pcm_polyp_t *pcm = userdata;
+
+ assert(pcm);
+ assert(pcm->p);
+
+ polyp_poll_activate(pcm->p);
+}
+
static int polyp_pcm_poll_descriptors_count(snd_pcm_ioplug_t *io)
{
snd_pcm_polyp_t *pcm = io->private_data;
int count;
assert(pcm);
-
- pthread_mutex_lock(&pcm->mutex);
-
assert(pcm->p);
+ pa_threaded_mainloop_lock(pcm->p->mainloop);
+
count = polyp_poll_descriptors_count(pcm->p);
- pthread_mutex_unlock(&pcm->mutex);
+ pa_threaded_mainloop_unlock(pcm->p->mainloop);
return count;
}
@@ -357,14 +377,13 @@
int err;
assert(pcm);
-
- pthread_mutex_lock(&pcm->mutex);
-
assert(pcm->p);
+ pa_threaded_mainloop_lock(pcm->p->mainloop);
+
err = polyp_poll_descriptors(pcm->p, pfd, space);
- pthread_mutex_unlock(&pcm->mutex);
+ pa_threaded_mainloop_unlock(pcm->p->mainloop);
return err;
}
@@ -375,11 +394,10 @@
int err = 0;
assert(pcm);
-
- pthread_mutex_lock(&pcm->mutex);
-
assert(pcm->p);
+ pa_threaded_mainloop_lock(pcm->p->mainloop);
+
err = polyp_poll_revents(pcm->p, pfd, nfds, revents);
if (err < 0)
goto finish;
@@ -402,25 +420,21 @@
}
finish:
- pthread_mutex_unlock(&pcm->mutex);
+ pa_threaded_mainloop_unlock(pcm->p->mainloop);
return err;
}
static int polyp_prepare(snd_pcm_ioplug_t *io)
{
+ pa_channel_map map;
snd_pcm_polyp_t *pcm = io->private_data;
int err = 0;
assert(pcm);
-
- pthread_mutex_lock(&pcm->mutex);
-
assert(pcm->p);
- err = polyp_finish_poll(pcm->p);
- if (err < 0)
- goto finish;
+ pa_threaded_mainloop_lock(pcm->p->mainloop);
if (pcm->stream) {
pa_stream_disconnect(pcm->stream);
@@ -436,15 +450,24 @@
assert(pcm->stream == NULL);
if (io->stream == SND_PCM_STREAM_PLAYBACK)
- pcm->stream = pa_stream_new(pcm->p->context, "ALSA Playback", &pcm->ss, NULL);
+ pcm->stream = pa_stream_new(pcm->p->context, "ALSA Playback", &pcm->ss,
+ pa_channel_map_init_auto(&map, pcm->ss.channels, PA_CHANNEL_MAP_ALSA));
else
- pcm->stream = pa_stream_new(pcm->p->context, "ALSA Capture", &pcm->ss, NULL);
+ pcm->stream = pa_stream_new(pcm->p->context, "ALSA Capture", &pcm->ss,
+ pa_channel_map_init_auto(&map, pcm->ss.channels, PA_CHANNEL_MAP_ALSA));
assert(pcm->stream);
- if (io->stream == SND_PCM_STREAM_PLAYBACK)
- pa_stream_connect_playback(pcm->stream, pcm->device, &pcm->buffer_attr, 0, NULL, NULL);
- else
- pa_stream_connect_record(pcm->stream, pcm->device, &pcm->buffer_attr, 0);
+ pa_stream_set_state_callback(pcm->stream, polyp_stream_state_cb, pcm->p);
+
+ if (io->stream == SND_PCM_STREAM_PLAYBACK) {
+ pa_stream_set_write_callback(pcm->stream, stream_request_cb, pcm);
+ pa_stream_connect_playback(pcm->stream, pcm->device, &pcm->buffer_attr,
+ PA_STREAM_AUTO_TIMING_UPDATE | PA_STREAM_INTERPOLATE_TIMING, NULL, NULL);
+ } else {
+ pa_stream_set_read_callback(pcm->stream, stream_request_cb, pcm);
+ pa_stream_connect_record(pcm->stream, pcm->device, &pcm->buffer_attr,
+ PA_STREAM_AUTO_TIMING_UPDATE | PA_STREAM_INTERPOLATE_TIMING);
+ }
err = polyp_wait_stream_state(pcm->p, pcm->stream, PA_STREAM_READY);
if (err < 0) {
@@ -459,7 +482,7 @@
pcm->offset = 0;
finish:
- pthread_mutex_unlock(&pcm->mutex);
+ pa_threaded_mainloop_unlock(pcm->p->mainloop);
return err;
}
@@ -470,10 +493,11 @@
int err = 0;
assert(pcm);
+ assert(pcm->p);
- pthread_mutex_lock(&pcm->mutex);
+ pa_threaded_mainloop_lock(pcm->p->mainloop);
- assert(pcm->p && !pcm->stream);
+ assert(!pcm->stream);
pcm->frame_size = (snd_pcm_format_physical_width(io->format) * io->channels) / 8;
@@ -516,7 +540,7 @@
pcm->buffer_attr.fragsize = io->period_size * pcm->frame_size;
finish:
- pthread_mutex_unlock(&pcm->mutex);
+ pa_threaded_mainloop_unlock(pcm->p->mainloop);
return err;
}
@@ -527,51 +551,55 @@
assert(pcm);
+ pa_threaded_mainloop_lock(pcm->p->mainloop);
+
if (pcm->stream) {
pa_stream_disconnect(pcm->stream);
polyp_wait_stream_state(pcm->p, pcm->stream, PA_STREAM_TERMINATED);
pa_stream_unref(pcm->stream);
}
+ pa_threaded_mainloop_unlock(pcm->p->mainloop);
+
if (pcm->p)
polyp_free(pcm->p);
if (pcm->device)
free(pcm->device);
- pthread_mutex_destroy(&pcm->mutex);
-
free(pcm);
return 0;
}
static snd_pcm_ioplug_callback_t polyp_playback_callback = {
- .start = polyp_start,
- .stop = polyp_stop,
+ .start = polyp_start,
+ .stop = polyp_stop,
.drain = polyp_drain,
- .pointer = polyp_pointer,
- .transfer = polyp_write,
+ .pointer = polyp_pointer,
+ .transfer = polyp_write,
+ .delay = polyp_delay,
.poll_descriptors_count = polyp_pcm_poll_descriptors_count,
.poll_descriptors = polyp_pcm_poll_descriptors,
.poll_revents = polyp_pcm_poll_revents,
- .prepare = polyp_prepare,
- .hw_params = polyp_hw_params,
- .close = polyp_close,
+ .prepare = polyp_prepare,
+ .hw_params = polyp_hw_params,
+ .close = polyp_close,
};
static snd_pcm_ioplug_callback_t polyp_capture_callback = {
- .start = polyp_start,
- .stop = polyp_stop,
- .pointer = polyp_pointer,
- .transfer = polyp_read,
+ .start = polyp_start,
+ .stop = polyp_stop,
+ .pointer = polyp_pointer,
+ .transfer = polyp_read,
+ .delay = polyp_delay,
.poll_descriptors_count = polyp_pcm_poll_descriptors_count,
.poll_descriptors = polyp_pcm_poll_descriptors,
.poll_revents = polyp_pcm_poll_revents,
- .prepare = polyp_prepare,
- .hw_params = polyp_hw_params,
- .close = polyp_close,
+ .prepare = polyp_prepare,
+ .hw_params = polyp_hw_params,
+ .close = polyp_close,
};
@@ -645,7 +673,6 @@
const char *device = NULL;
int err;
snd_pcm_polyp_t *pcm;
- pthread_mutexattr_t mutexattr;
snd_config_for_each(i, next, conf) {
snd_config_t *n = snd_config_iterator_entry(i);
@@ -678,21 +705,15 @@
pcm->device = strdup(device);
pcm->p = polyp_new();
- assert(pcm->p);
-
- err = polyp_connect(pcm->p, server);
- if (err < 0)
+ if (!pcm->p) {
+ err = -EIO;
goto error;
+ }
- err = polyp_start_thread(pcm->p);
+ err = polyp_connect(pcm->p, server);
if (err < 0)
goto error;
- pthread_mutexattr_init(&mutexattr);
- pthread_mutexattr_settype(&mutexattr, PTHREAD_MUTEX_RECURSIVE);
- pthread_mutex_init(&pcm->mutex, &mutexattr);
- pthread_mutexattr_destroy(&mutexattr);
-
pcm->io.version = SND_PCM_IOPLUG_VERSION;
pcm->io.name = "ALSA <-> Polypaudio PCM I/O Plugin";
pcm->io.poll_fd = -1;
Index: polyp/polyp.c
===================================================================
RCS file: /cvsroot/alsa/alsa-plugins/polyp/polyp.c,v
retrieving revision 1.4
diff -u -r1.4 polyp.c
--- polyp/polyp.c 8 Mar 2006 12:23:13 -0000 1.4
+++ polyp/polyp.c 28 May 2006 14:41:33 -0000
@@ -19,177 +19,71 @@
*/
#include <stdio.h>
+#include <unistd.h>
#include <signal.h>
-#include <sys/types.h>
-#include <sys/socket.h>
-#include <sys/un.h>
-
-#include <pthread.h>
+#include <sys/poll.h>
#include "polyp.h"
-enum {
- COMMAND_POLL = 'p',
- COMMAND_QUIT = 'q',
- COMMAND_POLL_DONE = 'P',
- COMMAND_POLL_FAILED = 'F',
-};
-
-static int write_command(snd_polyp_t *p, char command)
-{
- if (write(p->main_fd, &command, 1) != 1)
- return -errno;
- return 0;
-}
-
-static int write_reply(snd_polyp_t *p, char reply)
-{
- if (write(p->thread_fd, &reply, 1) != 1)
- return -errno;
- return 0;
-}
-
-static int read_command(snd_polyp_t *p)
-{
- char command;
-
- if (read(p->thread_fd, &command, 1) != 1)
- return -errno;
-
- return command;
-}
-
-static int read_reply(snd_polyp_t *p)
-{
- char reply;
-
- if (read(p->main_fd, &reply, 1) != 1)
- return -errno;
-
- return reply;
-}
-
-static void* thread_func(void *data)
+int polyp_check_connection(snd_polyp_t *p)
{
- snd_polyp_t *p = (snd_polyp_t*)data;
- sigset_t mask;
- char command;
- int ret;
+ pa_context_state_t state;
- sigfillset(&mask);
- pthread_sigmask(SIG_BLOCK, &mask, NULL);
+ assert(p && p->context && p->mainloop);
- do {
- command = read_command(p);
- if (command < 0)
- break;
-
- switch (command) {
- case COMMAND_POLL:
- do {
- ret = pa_mainloop_poll(p->mainloop);
- } while ((ret < 0) && (errno == EINTR));
-
- ret = write_reply(p, (ret < 0) ? COMMAND_POLL_FAILED : COMMAND_POLL_DONE);
- if (ret < 0)
- return NULL;
+ state = pa_context_get_state(p->context);
- break;
- }
- } while (command != COMMAND_QUIT);
+ if (state != PA_CONTEXT_READY)
+ return -EIO;
- return NULL;
+ return 0;
}
-int polyp_start_poll(snd_polyp_t *p)
+void polyp_stream_state_cb(pa_stream *s, void * userdata)
{
- int err;
+ snd_polyp_t *p = userdata;
+ assert(s);
assert(p);
- if (p->state == POLYP_STATE_POLLING)
- return 0;
-
- assert(p->state == POLYP_STATE_READY);
-
- err = pa_mainloop_prepare(p->mainloop, -1);
- if (err < 0)
- return err;
-
- err = write_command(p, COMMAND_POLL);
- if (err < 0)
- return err;
-
- p->state = POLYP_STATE_POLLING;
-
- return 0;
+ pa_threaded_mainloop_signal(p->mainloop, 0);
}
-int polyp_finish_poll(snd_polyp_t *p)
+void polyp_stream_success_cb(pa_stream *s, int success, void *userdata)
{
- char reply;
- int err;
+ snd_polyp_t *p = userdata;
+ assert(s);
assert(p);
- if (p->state == POLYP_STATE_READY)
- return 0;
-
- assert(p->state == POLYP_STATE_POLLING);
-
- p->state = POLYP_STATE_READY;
-
- pa_mainloop_wakeup(p->mainloop);
-
- reply = read_reply(p);
-
- if (reply == COMMAND_POLL_DONE) {
- err = pa_mainloop_dispatch(p->mainloop);
- if (err < 0)
- return err;
- } else
- return -EIO;
-
- return 0;
+ pa_threaded_mainloop_signal(p->mainloop, 0);
}
-int polyp_check_connection(snd_polyp_t *p)
+void polyp_context_success_cb(pa_context *c, int success, void *userdata)
{
- pa_context_state_t state;
+ snd_polyp_t *p = userdata;
- assert(p && p->context);
-
- state = pa_context_get_state(p->context);
-
- if (state != PA_CONTEXT_READY)
- return -EIO;
+ assert(c);
+ assert(p);
- return 0;
+ pa_threaded_mainloop_signal(p->mainloop, 0);
}
int polyp_wait_operation(snd_polyp_t *p, pa_operation *o)
{
- int err;
+ assert(p && o && (p->state == POLYP_STATE_READY) && p->mainloop);
- assert(p && o && (p->state == POLYP_STATE_READY));
-
- while (pa_operation_get_state(o) == PA_OPERATION_RUNNING) {
- p->state = POLYP_STATE_POLLING;
- err = pa_mainloop_iterate(p->mainloop, 1, NULL);
- p->state = POLYP_STATE_READY;
- if (err < 0)
- return err;
- }
+ while (pa_operation_get_state(o) == PA_OPERATION_RUNNING)
+ pa_threaded_mainloop_wait(p->mainloop);
return 0;
}
int polyp_wait_stream_state(snd_polyp_t *p, pa_stream *stream, pa_stream_state_t target)
{
- int err;
pa_stream_state_t state;
- assert(p && stream && (p->state == POLYP_STATE_READY));
+ assert(p && stream && (p->state == POLYP_STATE_READY) && p->mainloop);
while (1) {
state = pa_stream_get_state(stream);
@@ -200,34 +94,72 @@
if (state == target)
break;
- p->state = POLYP_STATE_POLLING;
- err = pa_mainloop_iterate(p->mainloop, 1, NULL);
- p->state = POLYP_STATE_READY;
- if (err < 0)
- return -EIO;
+ pa_threaded_mainloop_wait(p->mainloop);
}
return 0;
}
+static void context_state_cb(pa_context *c, void *userdata) {
+ snd_polyp_t *p = userdata;
+ assert(c);
+
+ switch (pa_context_get_state(c)) {
+ case PA_CONTEXT_READY:
+ case PA_CONTEXT_TERMINATED:
+ case PA_CONTEXT_FAILED:
+ pa_threaded_mainloop_signal(p->mainloop, 0);
+ break;
+
+ case PA_CONTEXT_UNCONNECTED:
+ case PA_CONTEXT_CONNECTING:
+ case PA_CONTEXT_AUTHORIZING:
+ case PA_CONTEXT_SETTING_NAME:
+ break;
+ }
+}
+
snd_polyp_t *polyp_new()
{
snd_polyp_t *p;
+ int fd[2] = { -1, -1 };
+ char proc[PATH_MAX], buf[PATH_MAX + 20];
p = calloc(1, sizeof(snd_polyp_t));
assert(p);
p->state = POLYP_STATE_INIT;
- p->main_fd = -1;
- p->thread_fd = -1;
- p->thread_running = 0;
+ if (pipe(fd)) {
+ free(p);
+ return NULL;
+ }
+
+ p->main_fd = fd[0];
+ p->thread_fd = fd[1];
+
+ fcntl(fd[0], F_SETFL, O_NONBLOCK);
+ fcntl(fd[1], F_SETFL, O_NONBLOCK);
+
+ signal(SIGPIPE, SIG_IGN); /* Yes, ugly as hell */
- p->mainloop = pa_mainloop_new();
+ p->mainloop = pa_threaded_mainloop_new();
assert(p->mainloop);
- p->context = pa_context_new(pa_mainloop_get_api(p->mainloop),
- "ALSA Plugin");
+ if (pa_threaded_mainloop_start(p->mainloop) < 0) {
+ pa_threaded_mainloop_free(p->mainloop);
+ close(fd[0]);
+ close(fd[1]);
+ free(p);
+ return NULL;
+ }
+
+ if (pa_get_binary_name(proc, sizeof(proc)))
+ snprintf(buf, sizeof(buf), "ALSA plug-in [%s]", pa_path_get_filename(proc));
+ else
+ snprintf(buf, sizeof(buf), "ALSA plug-in");
+
+ p->context = pa_context_new(pa_threaded_mainloop_get_api(p->mainloop), buf);
assert(p->context);
return p;
@@ -235,22 +167,13 @@
void polyp_free(snd_polyp_t *p)
{
- if (p->thread_running) {
- assert(p->mainloop && p->thread);
- write_command(p, COMMAND_QUIT);
- pa_mainloop_wakeup(p->mainloop);
- pthread_join(p->thread, NULL);
- }
+ pa_threaded_mainloop_stop(p->mainloop);
- if (p->context)
- pa_context_unref(p->context);
- if (p->mainloop)
- pa_mainloop_free(p->mainloop);
+ pa_context_unref(p->context);
+ pa_threaded_mainloop_free(p->mainloop);
- if (p->thread_fd >= 0)
- close(p->thread_fd);
- if (p->main_fd >= 0)
- close(p->main_fd);
+ close(p->thread_fd);
+ close(p->main_fd);
free(p);
}
@@ -258,70 +181,52 @@
int polyp_connect(snd_polyp_t *p, const char *server)
{
int err;
- pa_context_state_t state;
assert(p && p->context && p->mainloop && (p->state == POLYP_STATE_INIT));
+ pa_threaded_mainloop_lock(p->mainloop);
+
err = pa_context_connect(p->context, server, 0, NULL);
if (err < 0)
goto error;
- while (1) {
- state = pa_context_get_state(p->context);
+ pa_context_set_state_callback(p->context, context_state_cb, p);
- if (state == PA_CONTEXT_FAILED)
- goto error;
+ pa_threaded_mainloop_wait(p->mainloop);
- if (state == PA_CONTEXT_READY)
- break;
+ if (pa_context_get_state(p->context) != PA_CONTEXT_READY)
+ goto error;
- err = pa_mainloop_iterate(p->mainloop, 1, NULL);
- if (err < 0)
- return -EIO;
- }
+ pa_threaded_mainloop_unlock(p->mainloop);
- p->state = POLYP_STATE_CONNECTED;
+ p->state = POLYP_STATE_READY;
return 0;
error:
fprintf(stderr, "*** POLYPAUDIO: Unable to connect: %s\n",
pa_strerror(pa_context_errno(p->context)));
+
+ pa_threaded_mainloop_unlock(p->mainloop);
+
return -ECONNREFUSED;
}
-int polyp_start_thread(snd_polyp_t *p)
+void polyp_poll_activate(snd_polyp_t *p)
{
- int err;
- int fd[2] = { -1, -1 };
-
- assert(p && (p->state == POLYP_STATE_CONNECTED));
-
- if (socketpair(AF_UNIX, SOCK_STREAM, 0, fd) < 0) {
- perror("socketpair()");
- return -errno;
- }
-
- p->thread_fd = fd[0];
- p->main_fd = fd[1];
-
- p->thread_running = 0;
+ assert(p);
- err = pthread_create(&p->thread, NULL, thread_func, p);
- if (err) {
- SNDERR("pthread_create(): %s", strerror(err));
- close(fd[0]);
- close(fd[1]);
- p->main_fd = -1;
- p->thread_fd = -1;
- return -err;
- }
+ write(p->thread_fd, "a", 1);
+}
- p->thread_running = 1;
+void polyp_poll_deactivate(snd_polyp_t *p)
+{
+ char buf[10];
- p->state = POLYP_STATE_READY;
+ assert(p);
- return 0;
+ /* Drain the pipe */
+ while (read(p->main_fd, buf, sizeof(buf)) > 0);
}
int polyp_poll_descriptors_count(snd_polyp_t *p)
@@ -336,22 +241,12 @@
int polyp_poll_descriptors(snd_polyp_t *p, struct pollfd *pfd, unsigned int space)
{
- int err;
-
assert(p);
- err = polyp_finish_poll(p);
- if (err < 0)
- return err;
-
- err = polyp_start_poll(p);
- if (err < 0)
- return err;
-
assert(space >= 1);
pfd[0].fd = p->main_fd;
- pfd[0].events = POLL_IN;
+ pfd[0].events = POLLIN;
pfd[0].revents = 0;
return 1;
@@ -359,20 +254,7 @@
int polyp_poll_revents(snd_polyp_t *p, struct pollfd *pfd, unsigned int nfds, unsigned short *revents)
{
- int err;
-
assert(p);
- err = polyp_finish_poll(p);
- if (err < 0)
- return err;
-
- err = polyp_check_connection(p);
- if (err < 0)
- return err;
-
- /*
- * The application might redo the poll immediatly.
- */
- return polyp_poll_descriptors(p, pfd, nfds);
+ return 1;
}
Index: polyp/polyp.h
===================================================================
RCS file: /cvsroot/alsa/alsa-plugins/polyp/polyp.h,v
retrieving revision 1.2
diff -u -r1.2 polyp.h
--- polyp/polyp.h 8 Mar 2006 12:16:51 -0000 1.2
+++ polyp/polyp.h 28 May 2006 14:41:33 -0000
@@ -21,32 +21,27 @@
#include <alsa/asoundlib.h>
#include <polyp/polypaudio.h>
-#include <polyp/mainloop.h>
#define ARRAY_SIZE(a) (sizeof(a)/sizeof((a)[0]))
typedef struct snd_polyp {
- pa_mainloop *mainloop;
+ pa_threaded_mainloop *mainloop;
pa_context *context;
int thread_fd, main_fd;
- pthread_t thread;
- int thread_running;
-
enum {
POLYP_STATE_INIT,
- POLYP_STATE_CONNECTED,
POLYP_STATE_READY,
- POLYP_STATE_POLLING,
} state;
} snd_polyp_t;
-int polyp_start_poll(snd_polyp_t *p);
-int polyp_finish_poll(snd_polyp_t *p);
-
int polyp_check_connection(snd_polyp_t *p);
+void polyp_stream_state_cb(pa_stream *s, void * userdata);
+void polyp_stream_success_cb(pa_stream *s, int success, void *userdata);
+void polyp_context_success_cb(pa_context *c, int success, void *userdata);
+
int polyp_wait_operation(snd_polyp_t *p, pa_operation *o);
int polyp_wait_stream_state(snd_polyp_t *p, pa_stream *stream, pa_stream_state_t target);
@@ -54,8 +49,9 @@
void polyp_free(snd_polyp_t *p);
int polyp_connect(snd_polyp_t *p, const char *server);
-int polyp_start_thread(snd_polyp_t *p);
+void polyp_poll_activate(snd_polyp_t *p);
+void polyp_poll_deactivate(snd_polyp_t *p);
int polyp_poll_descriptors_count(snd_polyp_t *p);
int polyp_poll_descriptors(snd_polyp_t *p, struct pollfd *pfd, unsigned int space);
int polyp_poll_revents(snd_polyp_t *p, struct pollfd *pfd, unsigned int nfds, unsigned short *revents);
Attachment:
signature.asc
Description: OpenPGP digital signature
_______________________________________________ Alsa-devel mailing list Alsa-devel@xxxxxxxxxxxxxxxxxxxxx https://lists.sourceforge.net/lists/listinfo/alsa-devel
