mirrors/pipewire
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/pipewire/pipewire.git synced 2025-10-29 05:40:27 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

cleanups:

Browse source 
Keep lock while doing recycle_buffer
Pass context to main loop
Recycle WorkItems in a free list
Push and pull from the node data loop
Do async start/pause in alsa
Add alsa fds right after open.
This commit is contained in:
Wim Taymans 2016-10-25 11:15:15 +02:00
parent 73e6272488
commit 4148e0ff78
12 changed files with 312 additions and 87 deletions
Download patch file Download diff file Expand all files Collapse all files

2
pinos/client/connection.c
View file

@ -156,8 +156,8 @@ connection_ensure_size (PinosConnection *conn, ConnectionBuffer *buf, size_t siz
{
if (buf->buffer_size + size > buf->buffer_maxsize) {
buf->buffer_maxsize = buf->buffer_size + MAX_BUFFER_SIZE * ((size + MAX_BUFFER_SIZE-1) / MAX_BUFFER_SIZE);
buf->buffer_data = realloc (buf->buffer_data, buf->buffer_maxsize);
g_debug ("connection %p: resize buffer to %zd", conn, buf->buffer_maxsize);
buf->buffer_data = realloc (buf->buffer_data, buf->buffer_maxsize);
}
return (uint8_t *) buf->buffer_data + buf->buffer_size;
}

2
pinos/gst/gstpinossrc.c
View file

@ -354,7 +354,9 @@ buffer_recycle (GstMiniObject *obj)
src = data->src;
GST_LOG_OBJECT (obj, "recycle buffer");
pinos_thread_main_loop_lock (src->loop);
pinos_stream_recycle_buffer (src->stream, data->id);
pinos_thread_main_loop_unlock (src->loop);
return FALSE;
}

3
pinos/server/daemon.c
View file

@ -990,7 +990,8 @@ pinos_daemon_init (PinosDaemon * daemon)
g_str_equal,
g_free,
g_object_unref);
daemon->main_loop = pinos_main_loop_new();
daemon->main_loop = pinos_main_loop_new (g_main_context_get_thread_default ());
priv->data_loop = pinos_data_loop_new();

20
pinos/server/data-loop.c
View file

@ -106,7 +106,6 @@ loop (void *user_data)
/* rebuild */
if (priv->rebuild_fds) {
g_debug ("data-loop %p: rebuild fds", this);
priv->n_fds = 1;
for (i = 0; i < priv->n_poll; i++) {
SpaPollItem *p = &priv->poll[i];
@ -142,7 +141,7 @@ loop (void *user_data)
break;
}
if (r == 0) {
g_debug ("data-loop %p: select timeout should not happen", this);
g_warning ("data-loop %p: select timeout should not happen", this);
continue;
}
@ -226,10 +225,8 @@ do_add_item (SpaPoll *poll,
PinosDataLoop *this = SPA_CONTAINER_OF (poll, PinosDataLoop, poll);
PinosDataLoopPrivate *priv = this->priv;
gboolean in_thread = pthread_equal (priv->thread, pthread_self());
unsigned int i;
item->id = ++priv->counter;
g_debug ("data-loop %p: %d: add pollid %d, n_poll %d, n_fds %d", this, in_thread, item->id, priv->n_poll, item->n_fds);
priv->poll[priv->n_poll] = *item;
priv->n_poll++;
if (item->n_fds)
@ -239,10 +236,6 @@ do_add_item (SpaPoll *poll,
wakeup_thread (this);
start_thread (this);
}
for (i = 0; i < priv->n_poll; i++) {
if (priv->poll[i].n_fds > 0)
g_debug ("poll %d: %d %d", i, priv->poll[i].id, priv->poll[i].fds[0].fd);
}
return SPA_RESULT_OK;
}
@ -278,7 +271,6 @@ do_remove_item (SpaPoll *poll,
gboolean in_thread = pthread_equal (priv->thread, pthread_self());
unsigned int i;
g_debug ("data-loop %p: %d: remove poll %d %d", this, item->id, item->n_fds, priv->n_poll);
for (i = 0; i < priv->n_poll; i++) {
if (priv->poll[i].id == item->id) {
priv->n_poll--;
@ -295,10 +287,6 @@ do_remove_item (SpaPoll *poll,
if (priv->n_poll == 0) {
stop_thread (this, in_thread);
}
for (i = 0; i < priv->n_poll; i++) {
if (priv->poll[i].n_fds > 0)
g_debug ("poll %d: %d %d", i, priv->poll[i].id, priv->poll[i].fds[0].fd);
}
return SPA_RESULT_OK;
}
@ -432,3 +420,9 @@ pinos_data_loop_new (void)
{
return g_object_new (PINOS_TYPE_DATA_LOOP, NULL);
}
gboolean
pinos_data_loop_in_thread (PinosDataLoop *loop)
{
return pthread_equal (loop->priv->thread, pthread_self());
}

2
pinos/server/data-loop.h
View file

@ -67,6 +67,8 @@ GType pinos_data_loop_get_type (void);
PinosDataLoop * pinos_data_loop_new (void);
gboolean pinos_data_loop_in_thread (PinosDataLoop *loop);
G_END_DECLS
#endif /* __PINOS_DATA_LOOP_H__ */

105
pinos/server/main-loop.c
View file

@ -58,6 +58,9 @@ struct _WorkItem {
struct _PinosMainLoopPrivate
{
GMainContext *context;
GMainLoop *loop;
gulong counter;
SpaRingbuffer buffer;
@ -77,6 +80,7 @@ G_DEFINE_TYPE (PinosMainLoop, pinos_main_loop, G_TYPE_OBJECT);
enum
{
PROP_0,
PROP_MAIN_CONTEXT,
};
enum
@ -225,6 +229,47 @@ do_invoke (SpaPoll *poll,
return res;
}
static void
pinos_main_loop_get_property (GObject *_object,
guint prop_id,
GValue *value,
GParamSpec *pspec)
{
PinosMainLoop *loop = PINOS_MAIN_LOOP (_object);
PinosMainLoopPrivate *priv = loop->priv;
switch (prop_id) {
case PROP_MAIN_CONTEXT:
g_value_set_boxed (value, priv->context);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (loop, prop_id, pspec);
break;
}
}
static void
pinos_main_loop_set_property (GObject *_object,
guint prop_id,
const GValue *value,
GParamSpec *pspec)
{
PinosMainLoop *loop = PINOS_MAIN_LOOP (_object);
PinosMainLoopPrivate *priv = loop->priv;
switch (prop_id) {
case PROP_MAIN_CONTEXT:
priv->context = g_value_dup_boxed (value);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (loop, prop_id, pspec);
break;
}
}
static void
pinos_main_loop_constructed (GObject * obj)
{
@ -232,6 +277,7 @@ pinos_main_loop_constructed (GObject * obj)
PinosMainLoopPrivate *priv = this->priv;
g_debug ("main-loop %p: constructed", this);
priv->loop = g_main_loop_new (priv->context, FALSE);
priv->fds[0].fd = eventfd (0, 0);
priv->fds[0].events = POLLIN | POLLPRI | POLLERR;
@ -264,9 +310,12 @@ static void
pinos_main_loop_finalize (GObject * obj)
{
PinosMainLoop *this = PINOS_MAIN_LOOP (obj);
PinosMainLoopPrivate *priv = this->priv;
g_debug ("main-loop %p: finalize", this);
g_slice_free_chain (WorkItem, priv->free_list, next);
G_OBJECT_CLASS (pinos_main_loop_parent_class)->finalize (obj);
}
@ -280,6 +329,19 @@ pinos_main_loop_class_init (PinosMainLoopClass * klass)
gobject_class->constructed = pinos_main_loop_constructed;
gobject_class->dispose = pinos_main_loop_dispose;
gobject_class->finalize = pinos_main_loop_finalize;
gobject_class->set_property = pinos_main_loop_set_property;
gobject_class->get_property = pinos_main_loop_get_property;
g_object_class_install_property (gobject_class,
PROP_MAIN_CONTEXT,
g_param_spec_boxed ("main-context",
"Main Context",
"The main context to use",
G_TYPE_MAIN_CONTEXT,
G_PARAM_READWRITE |
G_PARAM_CONSTRUCT_ONLY |
G_PARAM_STATIC_STRINGS));
}
static void
@ -302,15 +364,18 @@ pinos_main_loop_init (PinosMainLoop * this)
/**
* pinos_main_loop_new:
* @context: a #GMainContext or %NULL to use the default context
*
* Create a new #PinosMainLoop.
*
* Returns: a new #PinosMainLoop
*/
PinosMainLoop *
pinos_main_loop_new (void)
pinos_main_loop_new (GMainContext *context)
{
return g_object_new (PINOS_TYPE_MAIN_LOOP, NULL);
return g_object_new (PINOS_TYPE_MAIN_LOOP,
"main-context", context,
NULL);
}
static gboolean
@ -340,7 +405,10 @@ process_work_queue (PinosMainLoop *this)
item->func (item->obj, item->data, item->res, item->id);
if (item->notify)
item->notify (item->data);
g_slice_free (WorkItem, item);
item->next = priv->free_list;
priv->free_list = item;
item = prev;
}
return FALSE;
@ -361,7 +429,12 @@ pinos_main_loop_defer (PinosMainLoop *loop,
g_return_val_if_fail (PINOS_IS_MAIN_LOOP (loop), 0);
priv = loop->priv;
item = g_slice_new (WorkItem);
if (priv->free_list) {
item = priv->free_list;
priv->free_list = item->next;
} else {
item = g_slice_new (WorkItem);
}
item->id = ++priv->counter;
item->obj = obj;
item->func = func;
@ -440,3 +513,27 @@ pinos_main_loop_defer_complete (PinosMainLoop *loop,
if (priv->work_id == 0 && have_work)
priv->work_id = g_idle_add ((GSourceFunc) process_work_queue, loop);
}
GMainLoop *
pinos_main_loop_get_impl (PinosMainLoop *loop)
{
g_return_val_if_fail (PINOS_IS_MAIN_LOOP (loop), NULL);
return loop->priv->loop;
}
void
pinos_main_loop_quit (PinosMainLoop *loop)
{
g_return_if_fail (PINOS_IS_MAIN_LOOP (loop));
g_main_loop_quit (loop->priv->loop);
}
void
pinos_main_loop_run (PinosMainLoop *loop)
{
g_return_if_fail (PINOS_IS_MAIN_LOOP (loop));
g_main_loop_run (loop->priv->loop);
}

5
pinos/server/main-loop.h
View file

@ -70,8 +70,11 @@ typedef void (*PinosDeferFunc) (gpointer obj,
/* normal GObject stuff */
GType pinos_main_loop_get_type (void);
PinosMainLoop * pinos_main_loop_new (void);
PinosMainLoop * pinos_main_loop_new (GMainContext *context);
GMainLoop * pinos_main_loop_get_impl (PinosMainLoop *loop);
void pinos_main_loop_run (PinosMainLoop *loop);
void pinos_main_loop_quit (PinosMainLoop *loop);
gulong pinos_main_loop_defer (PinosMainLoop *loop,
gpointer obj,

51
pinos/server/node.c
View file

@ -327,37 +327,40 @@ send_clock_update (PinosNode *this)
g_debug ("got error %d", res);
}
static gboolean
do_read_link (PinosNode *this, PinosLink *link)
static SpaResult
do_read_link (SpaPoll *poll,
bool async,
uint32_t seq,
size_t size,
void *data,
void *user_data)
{
SpaRingbufferArea areas[2];
PinosNode *this = user_data;
PinosLink *link = ((PinosLink**)data)[0];
size_t offset;
SpaResult res;
gboolean pushed = FALSE;
spa_ringbuffer_get_read_areas (&link->ringbuffer, areas);
if (areas[0].len > 0) {
if (spa_ringbuffer_get_read_offset (&link->ringbuffer, &offset) > 0) {
SpaPortInputInfo iinfo[1];
if (link->in_ready <= 0 || link->input == NULL)
return FALSE;
return SPA_RESULT_OK;
link->in_ready--;
iinfo[0].port_id = link->input->port;
iinfo[0].buffer_id = link->queue[areas[0].offset];
iinfo[0].buffer_id = link->queue[offset];
iinfo[0].flags = SPA_PORT_INPUT_FLAG_NONE;
if ((res = spa_node_port_push_input (link->input->node->node, 1, iinfo)) < 0)
g_warning ("node %p: error pushing buffer: %d, %d", this, res, iinfo[0].status);
else
pushed = TRUE;
spa_ringbuffer_read_advance (&link->ringbuffer, 1);
}
return pushed;
return SPA_RESULT_OK;
}
static void
on_node_event (SpaNode *node, SpaNodeEvent *event, void *user_data)
{
@ -393,7 +396,12 @@ on_node_event (SpaNode *node, SpaNodeEvent *event, void *user_data)
continue;
link->in_ready++;
do_read_link (this, link);
spa_poll_invoke (&link->input->node->priv->data_loop->poll,
do_read_link,
SPA_ID_INVALID,
sizeof (PinosLink *),
&link,
link->input->node);
}
break;
}
@ -414,17 +422,22 @@ on_node_event (SpaNode *node, SpaNodeEvent *event, void *user_data)
for (i = 0; i < priv->rt.links->len; i++) {
PinosLink *link = g_ptr_array_index (priv->rt.links, i);
SpaRingbufferArea areas[2];
size_t offset;
if (link->output == NULL || link->output->port != ho->port_id)
continue;
spa_ringbuffer_get_write_areas (&link->ringbuffer, areas);
if (areas[0].len > 0) {
link->queue[areas[0].offset] = oinfo[0].buffer_id;
if (spa_ringbuffer_get_write_offset (&link->ringbuffer, &offset) > 0) {
link->queue[offset] = oinfo[0].buffer_id;
spa_ringbuffer_write_advance (&link->ringbuffer, 1);
pushed = do_read_link (this, link);
spa_poll_invoke (&link->input->node->priv->data_loop->poll,
do_read_link,
SPA_ID_INVALID,
sizeof (PinosLink *),
&link,
link->input->node);
pushed = TRUE;
}
}
if (!pushed) {
@ -443,7 +456,7 @@ on_node_event (SpaNode *node, SpaNodeEvent *event, void *user_data)
for (i = 0; i < priv->rt.links->len; i++) {
PinosLink *link = g_ptr_array_index (priv->rt.links, i);
if (link->output == NULL || link->output->port != rb->port_id)
if (link->input == NULL || link->input->port != rb->port_id)
continue;
if ((res = spa_node_port_reuse_buffer (link->output->node->node,

2
spa/plugins/alsa/alsa-sink.c
View file

@ -170,7 +170,7 @@ spa_alsa_sink_node_send_command (SpaNode *node,
update_state (this, SPA_NODE_STATE_STREAMING);
break;
case SPA_NODE_COMMAND_PAUSE:
spa_alsa_stop (this);
spa_alsa_pause (this);
update_state (this, SPA_NODE_STATE_PAUSED);
break;

124
spa/plugins/alsa/alsa-source.c
View file

@ -38,8 +38,8 @@ update_state (SpaALSASource *this, SpaNodeState state)
}
static const char default_device[] = "hw:0";
static const uint32_t default_buffer_time = 100000;
static const uint32_t default_period_time = 10000;
static const uint32_t default_buffer_time = 1000;
static const uint32_t default_period_time = 100;
static const bool default_period_event = 0;
static void
@ -151,9 +151,86 @@ spa_alsa_source_node_set_props (SpaNode *node,
return res;
}
static SpaResult
do_send_event (SpaPoll *poll,
bool async,
uint32_t seq,
size_t size,
void *data,
void *user_data)
{
SpaALSASource *this = user_data;
this->event_cb (&this->node, data, this->user_data);
return SPA_RESULT_OK;
}
static SpaResult
do_start (SpaPoll *poll,
bool async,
uint32_t seq,
size_t size,
void *data,
void *user_data)
{
SpaALSASource *this = user_data;
SpaResult res;
SpaNodeEventAsyncComplete ac;
if (SPA_RESULT_IS_OK (res = spa_alsa_start (this))) {
update_state (this, SPA_NODE_STATE_STREAMING);
}
if (async) {
ac.event.type = SPA_NODE_EVENT_TYPE_ASYNC_COMPLETE;
ac.event.size = sizeof (SpaNodeEventAsyncComplete);
ac.seq = seq;
ac.res = res;
spa_poll_invoke (this->main_loop,
do_send_event,
SPA_ID_INVALID,
sizeof (ac),
&ac,
this);
}
return res;
}
static SpaResult
do_pause (SpaPoll *poll,
bool async,
uint32_t seq,
size_t size,
void *data,
void *user_data)
{
SpaALSASource *this = user_data;
SpaResult res;
SpaNodeEventAsyncComplete ac;
if (SPA_RESULT_IS_OK (res = spa_alsa_pause (this))) {
update_state (this, SPA_NODE_STATE_PAUSED);
}
if (async) {
ac.event.type = SPA_NODE_EVENT_TYPE_ASYNC_COMPLETE;
ac.event.size = sizeof (SpaNodeEventAsyncComplete);
ac.seq = seq;
ac.res = res;
spa_poll_invoke (this->main_loop,
do_send_event,
SPA_ID_INVALID,
sizeof (ac),
&ac,
this);
}
return res;
}
static SpaResult
spa_alsa_source_node_send_command (SpaNode *node,
SpaNodeCommand *command)
SpaNodeCommand *command)
{
SpaALSASource *this;
@ -167,17 +244,36 @@ spa_alsa_source_node_send_command (SpaNode *node,
return SPA_RESULT_INVALID_COMMAND;
case SPA_NODE_COMMAND_START:
if (spa_alsa_start (this) < 0)
return SPA_RESULT_ERROR;
{
if (!this->have_format)
return SPA_RESULT_NO_FORMAT;
update_state (this, SPA_NODE_STATE_STREAMING);
break;
if (this->n_buffers == 0)
return SPA_RESULT_NO_BUFFERS;
return spa_poll_invoke (this->data_loop,
do_start,
++this->seq,
0,
NULL,
this);
}
case SPA_NODE_COMMAND_PAUSE:
if (spa_alsa_stop (this) < 0)
return SPA_RESULT_ERROR;
{
if (!this->have_format)
return SPA_RESULT_NO_FORMAT;
update_state (this, SPA_NODE_STATE_PAUSED);
break;
if (this->n_buffers == 0)
return SPA_RESULT_NO_BUFFERS;
return spa_poll_invoke (this->data_loop,
do_pause,
++this->seq,
0,
NULL,
this);
}
case SPA_NODE_COMMAND_FLUSH:
case SPA_NODE_COMMAND_DRAIN:
case SPA_NODE_COMMAND_MARKER:
@ -345,8 +441,10 @@ spa_alsa_source_node_port_set_format (SpaNode *node,
return SPA_RESULT_INVALID_PORT;
if (format == NULL) {
this->have_format = false;
spa_alsa_pause (this);
spa_alsa_clear_buffers (this);
spa_alsa_close (this);
this->have_format = false;
update_state (this, SPA_NODE_STATE_CONFIGURE);
return SPA_RESULT_OK;
}
@ -768,6 +866,8 @@ alsa_source_init (const SpaHandleFactory *factory,
this->log = support[i].data;
else if (strcmp (support[i].uri, SPA_POLL__DataLoop) == 0)
this->data_loop = support[i].data;
else if (strcmp (support[i].uri, SPA_POLL__MainLoop) == 0)
this->main_loop = support[i].data;
}
if (this->map == NULL) {
spa_log_error (this->log, "an id-map is needed");

73
spa/plugins/alsa/alsa-utils.c
View file

@ -11,6 +11,8 @@
#define CHECK(s,msg) if ((err = (s)) < 0) { spa_log_error (state->log, msg ": %s\n", snd_strerror(err)); return err; }
static int alsa_on_fd_events (SpaPollNotifyData *data);
static int
spa_alsa_open (SpaALSAState *state)
{
@ -31,12 +33,30 @@ spa_alsa_open (SpaALSAState *state)
SND_PCM_NO_AUTO_CHANNELS |
SND_PCM_NO_AUTO_FORMAT), "open failed");
if ((state->poll.n_fds = snd_pcm_poll_descriptors_count (state->hndl)) <= 0) {
spa_log_error (state->log, "Invalid poll descriptors count %d\n", state->poll.n_fds);
return SPA_RESULT_ERROR;
}
if ((err = snd_pcm_poll_descriptors (state->hndl, (struct pollfd *)state->fds, state->poll.n_fds)) < 0) {
spa_log_error (state->log, "Unable to obtain poll descriptors for playback: %s\n", snd_strerror(err));
return SPA_RESULT_ERROR;
}
state->poll.id = 0;
state->poll.enabled = false;
state->poll.fds = state->fds;
state->poll.idle_cb = NULL;
state->poll.before_cb = NULL;
state->poll.after_cb = alsa_on_fd_events;
state->poll.user_data = state;
spa_poll_add_item (state->data_loop, &state->poll);
state->opened = true;
return 0;
}
static int
int
spa_alsa_close (SpaALSAState *state)
{
int err = 0;
@ -44,7 +64,9 @@ spa_alsa_close (SpaALSAState *state)
if (!state->opened)
return 0;
spa_log_info (state->log, "Playback device closing\n");
spa_poll_remove_item (state->data_loop, &state->poll);
spa_log_info (state->log, "Device closing\n");
CHECK (snd_pcm_close (state->hndl), "close failed");
state->opened = false;
@ -426,54 +448,41 @@ alsa_on_fd_events (SpaPollNotifyData *data)
return 0;
}
int
SpaResult
spa_alsa_start (SpaALSAState *state)
{
int err;
if (spa_alsa_open (state) < 0)
return -1;
if (state->n_buffers == 0)
return -1;
if (state->started)
return SPA_RESULT_OK;
CHECK (set_swparams (state), "swparams");
snd_pcm_dump (state->hndl, state->output);
if ((state->poll.n_fds = snd_pcm_poll_descriptors_count (state->hndl)) <= 0) {
spa_log_error (state->log, "Invalid poll descriptors count\n");
return state->poll.n_fds;
}
if ((err = snd_pcm_poll_descriptors (state->hndl, (struct pollfd *)state->fds, state->poll.n_fds)) < 0) {
spa_log_error (state->log, "Unable to obtain poll descriptors for playback: %s\n", snd_strerror(err));
return err;
if (state->stream == SND_PCM_STREAM_PLAYBACK) {
mmap_write (state);
}
state->poll.id = 0;
state->poll.enabled = true;
state->poll.fds = state->fds;
state->poll.idle_cb = NULL;
state->poll.before_cb = NULL;
state->poll.after_cb = alsa_on_fd_events;
state->poll.user_data = state;
spa_poll_add_item (state->data_loop, &state->poll);
spa_poll_update_item (state->data_loop, &state->poll);
mmap_write (state);
err = snd_pcm_start (state->hndl);
state->started = true;
return err;
return SPA_RESULT_OK;
}
int
spa_alsa_stop (SpaALSAState *state)
SpaResult
spa_alsa_pause (SpaALSAState *state)
{
if (!state->opened)
return 0;
if (!state->started)
return SPA_RESULT_OK;
state->poll.enabled = false;
spa_poll_update_item (state->data_loop, &state->poll);
spa_poll_remove_item (state->data_loop, &state->poll);
snd_pcm_drop (state->hndl);
spa_alsa_close (state);
state->started = false;
return 0;
return SPA_RESULT_OK;
}

10
spa/plugins/alsa/alsa-utils.h
View file

@ -66,9 +66,12 @@ struct _SpaALSAState {
SpaNode node;
SpaClock clock;
uint32_t seq;
URI uri;
SpaIDMap *map;
SpaLog *log;
SpaPoll *main_loop;
SpaPoll *data_loop;
snd_pcm_stream_t stream;
@ -104,7 +107,7 @@ struct _SpaALSAState {
SpaQueue free;
SpaQueue ready;
bool running;
bool started;
SpaPollFd fds[16];
SpaPollItem poll;
@ -117,8 +120,9 @@ int spa_alsa_set_format (SpaALSAState *state,
SpaFormatAudio *fmt,
SpaPortFormatFlags flags);
int spa_alsa_start (SpaALSAState *state);
int spa_alsa_stop (SpaALSAState *state);
SpaResult spa_alsa_start (SpaALSAState *state);
SpaResult spa_alsa_pause (SpaALSAState *state);
SpaResult spa_alsa_close (SpaALSAState *state);
#ifdef __cplusplus
} /* extern "C" */