rt-poll: make small poll helper object

Remove the event loop code into a separate object so that we can share
the same loop for multiple nodes.
Some cleanups

pinos/client/stream.c

@@ -987,12 +987,23 @@ parse_control (PinosStream *stream,
       case SPA_CONTROL_CMD_SET_FORMAT:
       {
         SpaControlCmdSetFormat p;
+        SpaMemory *mem;
+        void *data;
+        size_t size;
 
-        if (spa_control_iter_parse_cmd (&it, &p) < 0)
+        data = spa_control_iter_get_data (&it, &size);
+        mem = spa_memory_alloc_size (SPA_MEMORY_POOL_LOCAL, data, size);
+        spa_control_iter_set_data (&it, spa_memory_ensure_ptr (mem), size);
+
+        if (spa_control_iter_parse_cmd (&it, &p) < 0) {
+          spa_memory_unref (&mem->mem);
           break;
+        }
 
         if (priv->format)
           spa_format_unref (priv->format);
+        if (p.format)
+          p.format->mem.mem = mem->mem;
         priv->format = p.format;
 
         spa_debug_format (p.format);

pinos/server/client-node.c

@@ -162,9 +162,19 @@ static void
 pinos_client_node_dispose (GObject * object)
 {
   PinosClientNode *this = PINOS_CLIENT_NODE (object);
+  PinosNode *node = PINOS_NODE (this);
+  SpaProps *props;
+  SpaPropValue value;
+  int fd = -1;
 
   g_debug ("client-node %p: dispose", this);
 
+  spa_node_get_props (node->node, &props);
+  value.value = &fd;
+  value.size = sizeof (int);
+  spa_props_set_prop (props, spa_props_index_for_name (props, "socket"), &value);
+  spa_node_set_props (node->node, props);
+
   G_OBJECT_CLASS (pinos_client_node_parent_class)->dispose (object);
 }
 

pinos/server/daemon.c

@@ -31,6 +31,7 @@
 #include "pinos/server/client-node.h"
 #include "pinos/server/client.h"
 #include "pinos/server/link.h"
+#include "pinos/server/rt-loop.h"
 
 #include "pinos/dbus/org-pinos.h"
 
@@ -47,6 +48,7 @@ struct _PinosDaemonPrivate
   GList *nodes;
 
   GHashTable *clients;
+  PinosRTLoop *loop;
 
   PinosProperties *properties;
 
@@ -326,9 +328,12 @@ static void
 on_node_added (PinosDaemon *daemon, PinosNode *node)
 {
   PinosNodeState state;
+  PinosDaemonPrivate *priv = daemon->priv;
 
   g_debug ("daemon %p: node %p added", daemon, node);
 
+  g_object_set (node, "rt-loop", priv->loop, NULL);
+
   g_signal_connect (node, "state-change", (GCallback) on_node_state_change, daemon);
 
   state = pinos_node_get_state (node);
@@ -763,6 +768,7 @@ pinos_daemon_finalize (GObject * object)
   g_debug ("daemon %p: finalize", object);
   g_clear_object (&priv->server_manager);
   g_clear_object (&priv->iface);
+  g_clear_object (&priv->loop);
   g_hash_table_unref (priv->clients);
   g_hash_table_unref (priv->node_factories);
 
@@ -820,6 +826,7 @@ pinos_daemon_init (PinosDaemon * daemon)
                                                 g_str_equal,
                                                 g_free,
                                                 g_object_unref);
+  priv->loop = pinos_rtloop_new();
 }
 
 /**

pinos/server/meson.build

@@ -7,6 +7,7 @@ pinoscore_headers = [
   'module.h',
   'node.h',
   'node-factory.h',
+  'rt-loop.h',
 ]
 
 pinoscore_sources = [
@@ -18,6 +19,7 @@ pinoscore_sources = [
   'module.c',
   'node.c',
   'node-factory.c',
+  'rt-loop.c',
 ]
 
 libpinoscore_c_args = [

pinos/server/node.c

@@ -19,17 +19,13 @@
 
 #include <string.h>
 #include <stdlib.h>
-#include <poll.h>
 #include <errno.h>
-#include <sys/eventfd.h>
-
-#include <gio/gio.h>
-#include <gio/gunixfdlist.h>
 
 #include "pinos/client/pinos.h"
 #include "pinos/client/enumtypes.h"
 
 #include "pinos/server/node.h"
+#include "pinos/server/rt-loop.h"
 #include "pinos/server/daemon.h"
 
 #include "pinos/dbus/org-pinos.h"
@@ -64,15 +60,7 @@ struct _PinosNodePrivate
 
   PinosProperties *properties;
 
-  unsigned int n_poll;
+  PinosRTLoop *loop;
-  SpaPollItem poll[16];
-
-  bool rebuild_fds;
-  SpaPollFd fds[16];
-  unsigned int n_fds;
-
-  gboolean running;
-  pthread_t thread;
 
   GArray *output_links;
   guint   n_used_output_links;
@@ -91,6 +79,7 @@ enum
   PROP_0,
   PROP_DAEMON,
   PROP_SENDER,
+  PROP_RTLOOP,
   PROP_OBJECT_PATH,
   PROP_NAME,
   PROP_PROPERTIES,
@@ -191,139 +180,6 @@ update_port_ids (PinosNode *node, gboolean create)
   }
 }
 
-static void *
-loop (void *user_data)
-{
-  PinosNode *this = user_data;
-  PinosNodePrivate *priv = this->priv;
-  unsigned int i, j;
-
-  g_debug ("node %p: enter thread", this);
-  while (priv->running) {
-    SpaPollNotifyData ndata;
-    unsigned int n_idle = 0;
-    int r;
-
-    /* prepare */
-    for (i = 0; i < priv->n_poll; i++) {
-      SpaPollItem *p = &priv->poll[i];
-
-      if (p->enabled && p->idle_cb) {
-        ndata.fds = NULL;
-        ndata.n_fds = 0;
-        ndata.user_data = p->user_data;
-        p->idle_cb (&ndata);
-        n_idle++;
-      }
-    }
-    if (n_idle > 0)
-      continue;
-
-    /* rebuild */
-    if (priv->rebuild_fds) {
-      g_debug ("node %p: rebuild fds", this);
-      priv->n_fds = 1;
-      for (i = 0; i < priv->n_poll; i++) {
-        SpaPollItem *p = &priv->poll[i];
-
-        if (!p->enabled)
-          continue;
-
-        for (j = 0; j < p->n_fds; j++)
-          priv->fds[priv->n_fds + j] = p->fds[j];
-        p->fds = &priv->fds[priv->n_fds];
-        priv->n_fds += p->n_fds;
-      }
-      priv->rebuild_fds = false;
-    }
-
-    /* before */
-    for (i = 0; i < priv->n_poll; i++) {
-      SpaPollItem *p = &priv->poll[i];
-
-      if (p->enabled && p->before_cb) {
-        ndata.fds = p->fds;
-        ndata.n_fds = p->n_fds;
-        ndata.user_data = p->user_data;
-        p->before_cb (&ndata);
-      }
-    }
-
-    r = poll ((struct pollfd *) priv->fds, priv->n_fds, -1);
-    if (r < 0) {
-      if (errno == EINTR)
-        continue;
-      break;
-    }
-    if (r == 0) {
-      g_debug ("node %p: select timeout", this);
-      break;
-    }
-
-    /* check wakeup */
-    if (priv->fds[0].revents & POLLIN) {
-      uint64_t u;
-      if (read (priv->fds[0].fd, &u, sizeof(uint64_t)) != sizeof(uint64_t))
-        g_warning ("node %p: failed to read fd", strerror (errno));
-      continue;
-    }
-
-    /* after */
-    for (i = 0; i < priv->n_poll; i++) {
-      SpaPollItem *p = &priv->poll[i];
-
-      if (p->enabled && p->after_cb) {
-        ndata.fds = p->fds;
-        ndata.n_fds = p->n_fds;
-        ndata.user_data = p->user_data;
-        p->after_cb (&ndata);
-      }
-    }
-  }
-  g_debug ("node %p: leave thread", this);
-
-  return NULL;
-}
-
-static void
-wakeup_thread (PinosNode *this)
-{
-  PinosNodePrivate *priv = this->priv;
-  uint64_t u = 1;
-
-  if (write (priv->fds[0].fd, &u, sizeof(uint64_t)) != sizeof(uint64_t))
-    g_warning ("node %p: failed to write fd", strerror (errno));
-}
-
-static void
-start_thread (PinosNode *this)
-{
-  PinosNodePrivate *priv = this->priv;
-  int err;
-
-  if (!priv->running) {
-    priv->running = true;
-    if ((err = pthread_create (&priv->thread, NULL, loop, this)) != 0) {
-      g_warning ("node %p: can't create thread", strerror (err));
-      priv->running = false;
-    }
-  }
-}
-
-static void
-stop_thread (PinosNode *this, gboolean in_thread)
-{
-  PinosNodePrivate *priv = this->priv;
-
-  if (priv->running) {
-    priv->running = false;
-    if (!in_thread) {
-      wakeup_thread (this);
-      pthread_join (priv->thread, NULL);
-    }
-  }
-}
-
 static SpaResult
 pause_node (PinosNode *this)
 {
@@ -508,7 +364,6 @@ on_node_event (SpaNode *node, SpaNodeEvent *event, void *user_data)
 {
   PinosNode *this = user_data;
   PinosNodePrivate *priv = this->priv;
-  gboolean in_thread = pthread_equal (priv->thread, pthread_self());
 
   switch (event->type) {
     case SPA_NODE_EVENT_TYPE_INVALID:
@@ -532,57 +387,20 @@ on_node_event (SpaNode *node, SpaNodeEvent *event, void *user_data)
 
     case SPA_NODE_EVENT_TYPE_ADD_POLL:
     {
-      SpaPollItem *poll = event->data;
+      SpaPollItem *item = event->data;
-
+      pinos_rtloop_add_poll (priv->loop, item);
-      g_debug ("node %p: add pollid %d, n_poll %d, n_fds %d", this, poll->id, priv->n_poll, poll->n_fds);
-      priv->poll[priv->n_poll] = *poll;
-      priv->n_poll++;
-      if (poll->n_fds)
-        priv->rebuild_fds = true;
-
-      if (!in_thread) {
-        wakeup_thread (this);
-        start_thread (this);
-      }
       break;
     }
     case SPA_NODE_EVENT_TYPE_UPDATE_POLL:
     {
-      unsigned int i;
+      SpaPollItem *item = event->data;
-      SpaPollItem *poll = event->data;
+      pinos_rtloop_update_poll (priv->loop, item);
-
-      for (i = 0; i < priv->n_poll; i++) {
-        if (priv->poll[i].id == poll->id)
-          priv->poll[i] = *poll;
-      }
-      if (poll->n_fds)
-        priv->rebuild_fds = true;
-
-      if (!in_thread)
-        wakeup_thread (this);
       break;
     }
     case SPA_NODE_EVENT_TYPE_REMOVE_POLL:
     {
-      SpaPollItem *poll = event->data;
+      SpaPollItem *item = event->data;
-      unsigned int i;
+      pinos_rtloop_remove_poll (priv->loop, item);
-
-      g_debug ("node %p: remove poll %d %d", this, poll->n_fds, priv->n_poll);
-      for (i = 0; i < priv->n_poll; i++) {
-        if (priv->poll[i].id == poll->id) {
-          priv->n_poll--;
-          for (; i < priv->n_poll; i++)
-            priv->poll[i] = priv->poll[i+1];
-          break;
-        }
-      }
-      if (priv->n_poll > 0) {
-        priv->rebuild_fds = true;
-        if (!in_thread)
-          wakeup_thread (this);
-      } else {
-        stop_thread (this, in_thread);
-      }
       break;
     }
     case SPA_NODE_EVENT_TYPE_NEED_INPUT:
@@ -671,10 +489,10 @@ handle_remove (PinosNode1             *interface,
                GDBusMethodInvocation  *invocation,
                gpointer                user_data)
 {
-  PinosNode *node = user_data;
+  PinosNode *this = user_data;
 
-  g_debug ("node %p: remove", node);
+  g_debug ("node %p: remove", this);
-  pinos_node_remove (node);
+  pinos_node_remove (this);
 
   g_dbus_method_invocation_return_value (invocation,
                                          g_variant_new ("()"));
@@ -687,8 +505,8 @@ pinos_node_get_property (GObject    *_object,
                          GValue     *value,
                          GParamSpec *pspec)
 {
-  PinosNode *node = PINOS_NODE (_object);
+  PinosNode *this = PINOS_NODE (_object);
-  PinosNodePrivate *priv = node->priv;
+  PinosNodePrivate *priv = this->priv;
 
   switch (prop_id) {
     case PROP_DAEMON:
@@ -699,6 +517,10 @@ pinos_node_get_property (GObject    *_object,
       g_value_set_string (value, priv->sender);
       break;
 
+    case PROP_RTLOOP:
+      g_value_set_object (value, priv->loop);
+      break;
+
     case PROP_OBJECT_PATH:
       g_value_set_string (value, priv->object_path);
       break;
@@ -712,11 +534,11 @@ pinos_node_get_property (GObject    *_object,
       break;
 
     case PROP_NODE:
-      g_value_set_pointer (value, node->node);
+      g_value_set_pointer (value, this->node);
       break;
 
     default:
-      G_OBJECT_WARN_INVALID_PROPERTY_ID (node, prop_id, pspec);
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (this, prop_id, pspec);
       break;
   }
 }
@@ -727,8 +549,8 @@ pinos_node_set_property (GObject      *_object,
                          const GValue *value,
                          GParamSpec   *pspec)
 {
-  PinosNode *node = PINOS_NODE (_object);
+  PinosNode *this = PINOS_NODE (_object);
-  PinosNodePrivate *priv = node->priv;
+  PinosNodePrivate *priv = this->priv;
 
   switch (prop_id) {
     case PROP_DAEMON:
@@ -739,6 +561,21 @@ pinos_node_set_property (GObject      *_object,
       priv->sender = g_value_dup_string (value);
       break;
 
+    case PROP_RTLOOP:
+    {
+      SpaResult res;
+
+      if (priv->loop)
+        g_object_unref (priv->loop);
+      priv->loop = g_value_dup_object (value);
+
+      if (priv->loop) {
+        if ((res = spa_node_set_event_callback (this->node, on_node_event, this)) < 0)
+         g_warning ("node %p: error setting callback", this);
+      }
+      break;
+    }
+
     case PROP_NAME:
       priv->name = g_value_dup_string (value);
       break;
@@ -752,21 +589,21 @@ pinos_node_set_property (GObject      *_object,
     case PROP_NODE:
     {
       void *iface;
-      node->node = g_value_get_pointer (value);
+      this->node = g_value_get_pointer (value);
-      if (node->node->handle->get_interface (node->node->handle, SPA_INTERFACE_ID_CLOCK, &iface) >= 0)
+      if (this->node->handle->get_interface (this->node->handle, SPA_INTERFACE_ID_CLOCK, &iface) >= 0)
-        node->clock = iface;
+        this->clock = iface;
       break;
     }
     default:
-      G_OBJECT_WARN_INVALID_PROPERTY_ID (node, prop_id, pspec);
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (this, prop_id, pspec);
       break;
   }
 }
 
 static void
-node_register_object (PinosNode *node)
+node_register_object (PinosNode *this)
 {
-  PinosNodePrivate *priv = node->priv;
+  PinosNodePrivate *priv = this->priv;
   PinosDaemon *daemon = priv->daemon;
   PinosObjectSkeleton *skel;
 
@@ -778,21 +615,21 @@ node_register_object (PinosNode *node)
   priv->object_path = pinos_daemon_export_uniquely (daemon, G_DBUS_OBJECT_SKELETON (skel));
   g_object_unref (skel);
 
-  g_debug ("node %p: register object %s", node, priv->object_path);
+  g_debug ("node %p: register object %s", this, priv->object_path);
-  pinos_daemon_add_node (daemon, node);
+  pinos_daemon_add_node (daemon, this);
 
   return;
 }
 
 static void
-node_unregister_object (PinosNode *node)
+node_unregister_object (PinosNode *this)
 {
-  PinosNodePrivate *priv = node->priv;
+  PinosNodePrivate *priv = this->priv;
 
-  g_debug ("node %p: unregister object %s", node, priv->object_path);
+  g_debug ("node %p: unregister object %s", this, priv->object_path);
   pinos_daemon_unexport (priv->daemon, priv->object_path);
   g_clear_pointer (&priv->object_path, g_free);
-  pinos_daemon_remove_node (priv->daemon, node);
+  pinos_daemon_remove_node (priv->daemon, this);
 }
 
 static void
@@ -800,17 +637,17 @@ on_property_notify (GObject    *obj,
                     GParamSpec *pspec,
                     gpointer    user_data)
 {
-  PinosNode *node = user_data;
+  PinosNode *this = user_data;
-  PinosNodePrivate *priv = node->priv;
+  PinosNodePrivate *priv = this->priv;
 
   if (pspec == NULL || strcmp (g_param_spec_get_name (pspec), "sender") == 0) {
     pinos_node1_set_owner (priv->iface, priv->sender);
   }
   if (pspec == NULL || strcmp (g_param_spec_get_name (pspec), "name") == 0) {
-    pinos_node1_set_name (priv->iface, pinos_node_get_name (node));
+    pinos_node1_set_name (priv->iface, pinos_node_get_name (this));
   }
   if (pspec == NULL || strcmp (g_param_spec_get_name (pspec), "properties") == 0) {
-    PinosProperties *props = pinos_node_get_properties (node);
+    PinosProperties *props = pinos_node_get_properties (this);
     pinos_node1_set_properties (priv->iface, props ? pinos_properties_to_variant (props) : NULL);
   }
 }
@@ -832,18 +669,12 @@ pinos_node_constructed (GObject * obj)
 {
   PinosNode *this = PINOS_NODE (obj);
   PinosNodePrivate *priv = this->priv;
-  SpaResult res;
 
   g_debug ("node %p: constructed", this);
 
   g_signal_connect (this, "notify", (GCallback) on_property_notify, this);
   G_OBJECT_CLASS (pinos_node_parent_class)->constructed (obj);
 
-  priv->fds[0].fd = eventfd (0, 0);
-  priv->fds[0].events = POLLIN | POLLPRI | POLLERR;
-  priv->fds[0].revents = 0;
-  priv->n_fds = 1;
-
   if (this->node->info) {
     unsigned int i;
 
@@ -856,9 +687,6 @@ pinos_node_constructed (GObject * obj)
                             this->node->info->items[i].value);
   }
 
-  if ((res = spa_node_set_event_callback (this->node, on_node_event, this)) < 0)
-    g_warning ("node %p: error setting callback", this);
-
   if (priv->sender == NULL)
     priv->sender = g_strdup (pinos_daemon_get_sender (priv->daemon));
 
@@ -878,7 +706,6 @@ pinos_node_dispose (GObject * obj)
 
   g_debug ("node %p: dispose", node);
   pinos_node_set_state (node, PINOS_NODE_STATE_SUSPENDED);
-  stop_thread (node, FALSE);
 
   node_unregister_object (node);
 
@@ -897,6 +724,7 @@ pinos_node_finalize (GObject * obj)
   g_debug ("node %p: finalize", node);
   g_clear_object (&priv->daemon);
   g_clear_object (&priv->iface);
+  g_clear_object (&priv->loop);
   g_free (priv->sender);
   g_free (priv->name);
   g_clear_error (&priv->error);
@@ -979,6 +807,15 @@ pinos_node_class_init (PinosNodeClass * klass)
                                                          G_PARAM_READWRITE |
                                                          G_PARAM_CONSTRUCT_ONLY |
                                                          G_PARAM_STATIC_STRINGS));
+  g_object_class_install_property (gobject_class,
+                                   PROP_RTLOOP,
+                                   g_param_spec_object ("rt-loop",
+                                                        "RTLoop",
+                                                        "The RTLoop",
+                                                        PINOS_TYPE_RTLOOP,
+                                                        G_PARAM_READWRITE |
+                                                        G_PARAM_CONSTRUCT |
+                                                        G_PARAM_STATIC_STRINGS));
 
   signals[SIGNAL_REMOVE] = g_signal_new ("remove",
                                          G_TYPE_FROM_CLASS (klass),
@@ -1539,17 +1376,19 @@ pinos_node_report_error (PinosNode *node,
                          GError    *error)
 {
   PinosNodePrivate *priv;
+  PinosNodeState old;
 
   g_return_if_fail (PINOS_IS_NODE (node));
   priv = node->priv;
 
+  old = priv->state;
   g_clear_error (&priv->error);
   remove_idle_timeout (node);
   priv->error = error;
   priv->state = PINOS_NODE_STATE_ERROR;
   g_debug ("node %p: got error state %s", node, error->message);
   pinos_node1_set_state (priv->iface, PINOS_NODE_STATE_ERROR);
-  g_object_notify (G_OBJECT (node), "state");
+  g_signal_emit (node, signals[SIGNAL_STATE_CHANGE], 0, old, priv->state);
 }
 
 static gboolean

pinos/server/rt-loop.c

@@ -0,0 +1,336 @@
+/* Pinos
+ * Copyright (C) 2016 Wim Taymans <wim.taymans@gmail.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#include <string.h>
+#include <stdlib.h>
+#include <poll.h>
+#include <errno.h>
+#include <sys/eventfd.h>
+
+#include <gio/gio.h>
+#include <gio/gunixfdlist.h>
+
+#include "pinos/server/rt-loop.h"
+
+#define PINOS_RTLOOP_GET_PRIVATE(loop)  \
+   (G_TYPE_INSTANCE_GET_PRIVATE ((loop), PINOS_TYPE_RTLOOP, PinosRTLoopPrivate))
+
+struct _PinosRTLoopPrivate
+{
+  unsigned int n_poll;
+  SpaPollItem poll[16];
+
+  bool rebuild_fds;
+  SpaPollFd fds[16];
+  unsigned int n_fds;
+
+  gboolean running;
+  pthread_t thread;
+};
+
+G_DEFINE_TYPE (PinosRTLoop, pinos_rtloop, G_TYPE_OBJECT);
+
+enum
+{
+  PROP_0,
+};
+
+enum
+{
+  LAST_SIGNAL
+};
+
+static void *
+loop (void *user_data)
+{
+  PinosRTLoop *this = user_data;
+  PinosRTLoopPrivate *priv = this->priv;
+  unsigned int i, j;
+
+  g_debug ("rt-loop %p: enter thread", this);
+  while (priv->running) {
+    SpaPollNotifyData ndata;
+    unsigned int n_idle = 0;
+    int r;
+
+    /* prepare */
+    for (i = 0; i < priv->n_poll; i++) {
+      SpaPollItem *p = &priv->poll[i];
+
+      if (p->enabled && p->idle_cb) {
+        ndata.fds = NULL;
+        ndata.n_fds = 0;
+        ndata.user_data = p->user_data;
+        p->idle_cb (&ndata);
+        n_idle++;
+      }
+    }
+    if (n_idle > 0)
+      continue;
+
+    /* rebuild */
+    if (priv->rebuild_fds) {
+      g_debug ("rt-loop %p: rebuild fds", this);
+      priv->n_fds = 1;
+      for (i = 0; i < priv->n_poll; i++) {
+        SpaPollItem *p = &priv->poll[i];
+
+        if (!p->enabled)
+          continue;
+
+        for (j = 0; j < p->n_fds; j++)
+          priv->fds[priv->n_fds + j] = p->fds[j];
+        p->fds = &priv->fds[priv->n_fds];
+        priv->n_fds += p->n_fds;
+      }
+      priv->rebuild_fds = false;
+    }
+
+    /* before */
+    for (i = 0; i < priv->n_poll; i++) {
+      SpaPollItem *p = &priv->poll[i];
+
+      if (p->enabled && p->before_cb) {
+        ndata.fds = p->fds;
+        ndata.n_fds = p->n_fds;
+        ndata.user_data = p->user_data;
+        p->before_cb (&ndata);
+      }
+    }
+
+    r = poll ((struct pollfd *) priv->fds, priv->n_fds, -1);
+    if (r < 0) {
+      if (errno == EINTR)
+        continue;
+      break;
+    }
+    if (r == 0) {
+      g_debug ("rt-loop %p: select timeout", this);
+      break;
+    }
+
+    /* check wakeup */
+    if (priv->fds[0].revents & POLLIN) {
+      uint64_t u;
+      if (read (priv->fds[0].fd, &u, sizeof(uint64_t)) != sizeof(uint64_t))
+        g_warning ("rt-loop %p: failed to read fd", strerror (errno));
+      continue;
+    }
+
+    /* after */
+    for (i = 0; i < priv->n_poll; i++) {
+      SpaPollItem *p = &priv->poll[i];
+
+      if (p->enabled && p->after_cb) {
+        ndata.fds = p->fds;
+        ndata.n_fds = p->n_fds;
+        ndata.user_data = p->user_data;
+        p->after_cb (&ndata);
+      }
+    }
+  }
+  g_debug ("rt-loop %p: leave thread", this);
+
+  return NULL;
+}
+
+static void
+wakeup_thread (PinosRTLoop *this)
+{
+  PinosRTLoopPrivate *priv = this->priv;
+  uint64_t u = 1;
+
+  if (write (priv->fds[0].fd, &u, sizeof(uint64_t)) != sizeof(uint64_t))
+    g_warning ("rt-loop %p: failed to write fd", strerror (errno));
+}
+
+static void
+start_thread (PinosRTLoop *this)
+{
+  PinosRTLoopPrivate *priv = this->priv;
+  int err;
+
+  if (!priv->running) {
+    priv->running = true;
+    if ((err = pthread_create (&priv->thread, NULL, loop, this)) != 0) {
+      g_warning ("rt-loop %p: can't create thread", strerror (err));
+      priv->running = false;
+    }
+  }
+}
+
+static void
+stop_thread (PinosRTLoop *this, gboolean in_thread)
+{
+  PinosRTLoopPrivate *priv = this->priv;
+
+  if (priv->running) {
+    priv->running = false;
+    if (!in_thread) {
+      wakeup_thread (this);
+      pthread_join (priv->thread, NULL);
+    }
+  }
+}
+
+gboolean
+pinos_rtloop_add_poll (PinosRTLoop *this, SpaPollItem *item)
+{
+  PinosRTLoopPrivate *priv = this->priv;
+  gboolean in_thread = pthread_equal (priv->thread, pthread_self());
+  unsigned int i;
+
+  g_debug ("rt-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)
+    priv->rebuild_fds = true;
+
+  if (!in_thread) {
+    wakeup_thread (this);
+    start_thread (this);
+  }
+  for (i = 0; i < priv->n_poll; i++) {
+    if (priv->poll[i].fds)
+      g_debug ("poll %d: %p %d", i, priv->poll[i].user_data, priv->poll[i].fds[0].fd);
+  }
+  return TRUE;
+}
+
+gboolean
+pinos_rtloop_update_poll (PinosRTLoop *this, SpaPollItem *item)
+{
+  PinosRTLoopPrivate *priv = this->priv;
+  gboolean in_thread = pthread_equal (priv->thread, pthread_self());
+  unsigned int i;
+
+  for (i = 0; i < priv->n_poll; i++) {
+    if (priv->poll[i].id == item->id && priv->poll[i].user_data == item->user_data)
+      priv->poll[i] = *item;
+  }
+  if (item->n_fds)
+    priv->rebuild_fds = true;
+
+  if (!in_thread)
+    wakeup_thread (this);
+
+  return TRUE;
+}
+
+
+gboolean
+pinos_rtloop_remove_poll (PinosRTLoop *this, SpaPollItem *item)
+{
+  PinosRTLoopPrivate *priv = this->priv;
+  gboolean in_thread = pthread_equal (priv->thread, pthread_self());
+  unsigned int i;
+
+  g_debug ("rt-loop %p: remove poll %d %d", this, item->n_fds, priv->n_poll);
+  for (i = 0; i < priv->n_poll; i++) {
+    if (priv->poll[i].id == item->id && priv->poll[i].user_data == item->user_data) {
+      priv->n_poll--;
+      for (; i < priv->n_poll; i++)
+        priv->poll[i] = priv->poll[i+1];
+      break;
+    }
+  }
+  if (item->n_fds) {
+    priv->rebuild_fds = true;
+    if (!in_thread)
+      wakeup_thread (this);
+  }
+  if (priv->n_poll == 0) {
+    stop_thread (this, in_thread);
+  }
+  for (i = 0; i < priv->n_poll; i++) {
+    if (priv->poll[i].fds)
+      g_debug ("poll %d: %p %d", i, priv->poll[i].user_data, priv->poll[i].fds[0].fd);
+  }
+  return TRUE;
+}
+
+static void
+pinos_rtloop_constructed (GObject * obj)
+{
+  PinosRTLoop *this = PINOS_RTLOOP (obj);
+  PinosRTLoopPrivate *priv = this->priv;
+
+  g_debug ("rt-loop %p: constructed", this);
+
+  G_OBJECT_CLASS (pinos_rtloop_parent_class)->constructed (obj);
+
+  priv->fds[0].fd = eventfd (0, 0);
+  priv->fds[0].events = POLLIN | POLLPRI | POLLERR;
+  priv->fds[0].revents = 0;
+  priv->n_fds = 1;
+}
+
+static void
+pinos_rtloop_dispose (GObject * obj)
+{
+  PinosRTLoop *this = PINOS_RTLOOP (obj);
+
+  g_debug ("rt-loop %p: dispose", this);
+  stop_thread (this, FALSE);
+
+  G_OBJECT_CLASS (pinos_rtloop_parent_class)->dispose (obj);
+}
+
+static void
+pinos_rtloop_finalize (GObject * obj)
+{
+  PinosRTLoop *this = PINOS_RTLOOP (obj);
+
+  g_debug ("rt-loop %p: finalize", this);
+
+  G_OBJECT_CLASS (pinos_rtloop_parent_class)->finalize (obj);
+}
+
+static void
+pinos_rtloop_class_init (PinosRTLoopClass * klass)
+{
+  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
+
+  g_type_class_add_private (klass, sizeof (PinosRTLoopPrivate));
+
+  gobject_class->constructed = pinos_rtloop_constructed;
+  gobject_class->dispose = pinos_rtloop_dispose;
+  gobject_class->finalize = pinos_rtloop_finalize;
+}
+
+static void
+pinos_rtloop_init (PinosRTLoop * this)
+{
+  this->priv = PINOS_RTLOOP_GET_PRIVATE (this);
+
+  g_debug ("rt-loop %p: new", this);
+}
+
+/**
+ * pinos_rtloop_new:
+ *
+ * Create a new #PinosRTLoop.
+ *
+ * Returns: a new #PinosRTLoop
+ */
+PinosRTLoop *
+pinos_rtloop_new (void)
+{
+  return g_object_new (PINOS_TYPE_RTLOOP, NULL);
+}

pinos/server/rt-loop.h

@@ -0,0 +1,76 @@
+/* Pinos
+ * Copyright (C) 2016 Wim Taymans <wim.taymans@gmail.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifndef __PINOS_RTLOOP_H__
+#define __PINOS_RTLOOP_H__
+
+#include <glib-object.h>
+
+G_BEGIN_DECLS
+
+#include <spa/include/spa/poll.h>
+
+typedef struct _PinosRTLoop PinosRTLoop;
+typedef struct _PinosRTLoopClass PinosRTLoopClass;
+typedef struct _PinosRTLoopPrivate PinosRTLoopPrivate;
+
+#define PINOS_TYPE_RTLOOP                 (pinos_rtloop_get_type ())
+#define PINOS_IS_RTLOOP(obj)              (G_TYPE_CHECK_INSTANCE_TYPE ((obj), PINOS_TYPE_RTLOOP))
+#define PINOS_IS_RTLOOP_CLASS(klass)      (G_TYPE_CHECK_CLASS_TYPE ((klass), PINOS_TYPE_RTLOOP))
+#define PINOS_RTLOOP_GET_CLASS(obj)       (G_TYPE_INSTANCE_GET_CLASS ((obj), PINOS_TYPE_RTLOOP, PinosRTLoopClass))
+#define PINOS_RTLOOP(obj)                 (G_TYPE_CHECK_INSTANCE_CAST ((obj), PINOS_TYPE_RTLOOP, PinosRTLoop))
+#define PINOS_RTLOOP_CLASS(klass)         (G_TYPE_CHECK_CLASS_CAST ((klass), PINOS_TYPE_RTLOOP, PinosRTLoopClass))
+#define PINOS_RTLOOP_CAST(obj)            ((PinosRTLoop*)(obj))
+#define PINOS_RTLOOP_CLASS_CAST(klass)    ((PinosRTLoopClass*)(klass))
+
+/**
+ * PinosRTLoop:
+ *
+ * Pinos rt-loop class.
+ */
+struct _PinosRTLoop {
+  GObject object;
+
+  PinosRTLoopPrivate *priv;
+};
+
+/**
+ * PinosRTLoopClass:
+ *
+ * Pinos rt-loop class.
+ */
+struct _PinosRTLoopClass {
+  GObjectClass parent_class;
+};
+
+/* normal GObject stuff */
+GType               pinos_rtloop_get_type                (void);
+
+PinosRTLoop *       pinos_rtloop_new                     (void);
+
+gboolean            pinos_rtloop_add_poll                (PinosRTLoop *loop,
+                                                          SpaPollItem *item);
+gboolean            pinos_rtloop_update_poll             (PinosRTLoop *loop,
+                                                          SpaPollItem *item);
+gboolean            pinos_rtloop_remove_poll             (PinosRTLoop *loop,
+                                                          SpaPollItem *item);
+
+G_END_DECLS
+
+#endif /* __PINOS_RTLOOP_H__ */

spa/include/spa/poll.h

@@ -43,8 +43,6 @@ typedef struct {
  * @user_data: user data
  * @fds: array of file descriptors
  * @n_fds: number of elements in @fds
- * @now: the current time
- * @timeout: the next desired wakeup time relative to @now
  *
  * Data passed to #SpaPollNotify.
  */
@@ -52,8 +50,6 @@ typedef struct {
   void *user_data;
   SpaPollFd *fds;
   unsigned int n_fds;
-  uint64_t now;
-  uint64_t timeout;
 } SpaPollNotifyData;
 
 typedef int (*SpaPollNotify) (SpaPollNotifyData *data);
@@ -67,7 +63,7 @@ typedef int (*SpaPollNotify) (SpaPollNotifyData *data);
  * @idle_cb: callback called when there is no other work
  * @before_cb: callback called before starting the poll
  * @after_cb: callback called after the poll loop
- * @user_data: user data pass to callbacks
+ * @user_data: user data passed to callbacks
  */
 typedef struct {
   uint32_t       id;

spa/plugins/alsa/alsa-sink.c

@@ -196,8 +196,6 @@ spa_alsa_sink_node_set_event_callback (SpaNode              *node,
   this->event_cb = event;
   this->user_data = user_data;
 
-  update_state (this, SPA_NODE_STATE_CONFIGURE);
-
   return SPA_RESULT_OK;
 }
 
@@ -656,6 +654,8 @@ alsa_sink_init (const SpaHandleFactory  *factory,
 
   this->status.flags = SPA_PORT_STATUS_FLAG_NEED_INPUT;
 
+  update_state (this, SPA_NODE_STATE_CONFIGURE);
+
   return SPA_RESULT_OK;
 }
 

spa/plugins/alsa/alsa-source.c

@@ -200,8 +200,6 @@ spa_alsa_source_node_set_event_callback (SpaNode              *node,
   this->event_cb = event;
   this->user_data = user_data;
 
-  update_state (this, SPA_NODE_STATE_CONFIGURE);
-
   return SPA_RESULT_OK;
 }
 
@@ -820,6 +818,9 @@ alsa_source_init (const SpaHandleFactory  *factory,
       this->props[1].props.unset_mask &= ~1;
     }
   }
+
+  update_state (this, SPA_NODE_STATE_CONFIGURE);
+
   return SPA_RESULT_OK;
 }
 

spa/plugins/v4l2/v4l2-source.c

@@ -286,8 +286,6 @@ spa_v4l2_source_node_set_event_callback (SpaNode              *node,
   this->event_cb = event;
   this->user_data = user_data;
 
-  update_state (this, SPA_NODE_STATE_CONFIGURE);
-
   return SPA_RESULT_OK;
 }
 
@@ -855,6 +853,8 @@ v4l2_source_init (const SpaHandleFactory  *factory,
     }
   }
 
+  update_state (this, SPA_NODE_STATE_CONFIGURE);
+
   return SPA_RESULT_OK;
 }