Add invoke method to schedule executing in a main loop

pinos/server/client-node.c

@@ -217,9 +217,11 @@ spa_proxy_node_send_command (SpaNode        *node,
       cnc.command = command;
       pinos_connection_add_cmd (this->conn, PINOS_CONTROL_CMD_NODE_COMMAND, &cnc);
 
-      if (!pinos_connection_flush (this->conn))
+      if (!pinos_connection_flush (this->conn)) {
         spa_log_error (this->log, "proxy %p: error writing connection\n", this);
-
+        res = SPA_RESULT_ERROR;
+      }
+      else
         res = SPA_RESULT_RETURN_ASYNC (cnc.seq);
       break;
     }
@@ -232,8 +234,10 @@ spa_proxy_node_send_command (SpaNode        *node,
       cnc.command = command;
       pinos_connection_add_cmd (this->conn, PINOS_CONTROL_CMD_NODE_COMMAND, &cnc);
 
-      if (!pinos_connection_flush (this->conn))
+      if (!pinos_connection_flush (this->conn)) {
         spa_log_error (this->log, "proxy %p: error writing connection\n", this);
+        res = SPA_RESULT_ERROR;
+      }
 
       break;
     }
@@ -1287,7 +1291,7 @@ proxy_init (SpaProxy         *this,
   this->rtpoll.after_cb = proxy_on_rtfd_events;
   this->rtpoll.user_data = this;
 
-  return SPA_RESULT_OK;
+  return SPA_RESULT_RETURN_ASYNC (this->seq++);
 }
 
 static SpaResult

pinos/server/data-loop.c

@@ -24,15 +24,29 @@
 #include <sys/eventfd.h>
 
 #include <gio/gio.h>
-#include <gio/gunixfdlist.h>
 
+#include "spa/include/spa/ringbuffer.h"
 #include "pinos/server/data-loop.h"
 
 #define PINOS_DATA_LOOP_GET_PRIVATE(loop)  \
    (G_TYPE_INSTANCE_GET_PRIVATE ((loop), PINOS_TYPE_DATA_LOOP, PinosDataLoopPrivate))
 
+#define DATAS_SIZE (4096 * 8)
+
+typedef struct {
+  size_t             item_size;
+  SpaPollInvokeFunc  func;
+  uint32_t           seq;
+  size_t             size;
+  void              *data;
+  void              *user_data;
+} InvokeItem;
+
 struct _PinosDataLoopPrivate
 {
+  SpaRingbuffer buffer;
+  uint8_t       buffer_data[DATAS_SIZE];
+
   unsigned int n_poll;
   SpaPollItem poll[16];
   int idx[16];
@@ -42,6 +56,7 @@ struct _PinosDataLoopPrivate
   unsigned int n_fds;
 
   uint32_t counter;
+  uint32_t seq;
 
   gboolean running;
   pthread_t thread;
@@ -64,6 +79,7 @@ loop (void *user_data)
 {
   PinosDataLoop *this = user_data;
   PinosDataLoopPrivate *priv = this->priv;
+  SpaPoll *p = &this->poll;
   unsigned int i, j;
 
   g_debug ("data-loop %p: enter thread", this);
@@ -80,7 +96,7 @@ loop (void *user_data)
         ndata.fds = NULL;
         ndata.n_fds = 0;
         ndata.user_data = p->user_data;
-        if (p->idle_cb (&ndata) < 0)
+        if (SPA_RESULT_IS_ERROR (p->idle_cb (&ndata)))
           p->enabled = false;
         n_idle++;
       }
@@ -114,7 +130,8 @@ loop (void *user_data)
         ndata.fds = &priv->fds[priv->idx[i]];
         ndata.n_fds = p->n_fds;
         ndata.user_data = p->user_data;
-        p->before_cb (&ndata);
+        if (SPA_RESULT_IS_ERROR (p->before_cb (&ndata)))
+          p->enabled = false;
       }
     }
 
@@ -125,15 +142,24 @@ loop (void *user_data)
       break;
     }
     if (r == 0) {
-      g_debug ("data-loop %p: select timeout", this);
+      g_debug ("data-loop %p: select timeout should not happen", this);
-      break;
+      continue;
     }
 
     /* check wakeup */
     if (priv->fds[0].revents & POLLIN) {
       uint64_t u;
+      size_t offset;
+
       if (read (priv->fds[0].fd, &u, sizeof(uint64_t)) != sizeof(uint64_t))
-        g_warning ("data-loop %p: failed to read fd", strerror (errno));
+        g_warning ("data-loop %p: failed to read fd: %s", this, strerror (errno));
+
+      while (spa_ringbuffer_get_read_offset (&priv->buffer, &offset) > 0) {
+        InvokeItem *item = SPA_MEMBER (priv->buffer_data, offset, InvokeItem);
+        g_debug ("data-loop %p: invoke %d", this, item->seq);
+        item->func (p, item->seq, item->size, item->data, item->user_data);
+        spa_ringbuffer_read_advance (&priv->buffer, item->item_size);
+      }
       continue;
     }
 
@@ -145,7 +171,8 @@ loop (void *user_data)
         ndata.fds = &priv->fds[priv->idx[i]];
         ndata.n_fds = p->n_fds;
         ndata.user_data = p->user_data;
-        p->after_cb (&ndata);
+        if (SPA_RESULT_IS_ERROR (p->after_cb (&ndata)))
+          p->enabled = false;
       }
     }
   }
@@ -161,7 +188,7 @@ wakeup_thread (PinosDataLoop *this)
   uint64_t u = 1;
 
   if (write (priv->fds[0].fd, &u, sizeof(uint64_t)) != sizeof(uint64_t))
-    g_warning ("data-loop %p: failed to write fd", strerror (errno));
+    g_warning ("data-loop %p: failed to write fd: %s", this, strerror (errno));
 }
 
 static void
@@ -173,7 +200,7 @@ start_thread (PinosDataLoop *this)
   if (!priv->running) {
     priv->running = true;
     if ((err = pthread_create (&priv->thread, NULL, loop, this)) != 0) {
-      g_warning ("data-loop %p: can't create thread", strerror (err));
+      g_warning ("data-loop %p: can't create thread: %s", this, strerror (err));
       priv->running = false;
     }
   }
@@ -276,6 +303,48 @@ do_remove_item (SpaPoll         *poll,
   return SPA_RESULT_OK;
 }
 
+static SpaResult
+do_invoke (SpaPoll           *poll,
+           SpaPollInvokeFunc  func,
+           uint32_t           seq,
+           size_t             size,
+           void              *data,
+           void              *user_data)
+{
+  PinosDataLoop *this = SPA_CONTAINER_OF (poll, PinosDataLoop, poll);
+  PinosDataLoopPrivate *priv = this->priv;
+  SpaRingbufferArea areas[2];
+  InvokeItem *item;
+
+  spa_ringbuffer_get_write_areas (&priv->buffer, areas);
+  if (areas[0].len < sizeof (InvokeItem)) {
+    g_warning ("queue full");
+    return SPA_RESULT_ERROR;
+  }
+  item = SPA_MEMBER (priv->buffer_data, areas[0].offset, InvokeItem);
+  item->seq = seq;
+  item->func = func;
+  item->user_data = user_data;
+  item->size = size;
+
+  if (areas[0].len > sizeof (InvokeItem) + size) {
+    item->data = SPA_MEMBER (item, sizeof (InvokeItem), void);
+    item->item_size = sizeof (InvokeItem) + size;
+    if (areas[0].len < sizeof (InvokeItem) + item->item_size)
+      item->item_size = areas[0].len;
+  } else {
+    item->data = SPA_MEMBER (priv->buffer_data, areas[1].offset, void);
+    item->item_size = areas[0].len + 1 + size;
+  }
+  memcpy (item->data, data, size);
+
+  spa_ringbuffer_write_advance (&priv->buffer, item->item_size);
+
+  wakeup_thread (this);
+
+  return SPA_RESULT_RETURN_ASYNC (seq);
+}
+
 static void
 pinos_data_loop_constructed (GObject * obj)
 {
@@ -328,7 +397,7 @@ pinos_data_loop_class_init (PinosDataLoopClass * klass)
 static void
 pinos_data_loop_init (PinosDataLoop * this)
 {
-  this->priv = PINOS_DATA_LOOP_GET_PRIVATE (this);
+  PinosDataLoopPrivate *priv = this->priv = PINOS_DATA_LOOP_GET_PRIVATE (this);
 
   g_debug ("data-loop %p: new", this);
 
@@ -337,6 +406,9 @@ pinos_data_loop_init (PinosDataLoop * this)
   this->poll.add_item = do_add_item;
   this->poll.update_item = do_update_item;
   this->poll.remove_item = do_remove_item;
+  this->poll.invoke = do_invoke;
+
+  spa_ringbuffer_init (&priv->buffer, DATAS_SIZE);
 }
 
 /**

pinos/server/data-loop.h

@@ -25,6 +25,7 @@
 G_BEGIN_DECLS
 
 #include <spa/include/spa/poll.h>
+#include <spa/include/spa/node-command.h>
 
 typedef struct _PinosDataLoop PinosDataLoop;
 typedef struct _PinosDataLoopClass PinosDataLoopClass;
@@ -61,6 +62,10 @@ struct _PinosDataLoopClass {
   GObjectClass parent_class;
 };
 
+typedef void (*PinosCommandFunc) (SpaNodeCommand *command,
+                                  uint32_t        seq,
+                                  void           *user_data);
+
 /* normal GObject stuff */
 GType               pinos_data_loop_get_type         (void);
 

pinos/server/main-loop.c

@@ -19,19 +19,56 @@
 
 #include <string.h>
 #include <stdlib.h>
+#include <errno.h>
+#include <poll.h>
+#include <sys/eventfd.h>
 
 #include <gio/gio.h>
 
+#include "spa/include/spa/queue.h"
+#include "spa/include/spa/ringbuffer.h"
 #include "pinos/server/main-loop.h"
 
 #define PINOS_MAIN_LOOP_GET_PRIVATE(loop)  \
    (G_TYPE_INSTANCE_GET_PRIVATE ((loop), PINOS_TYPE_MAIN_LOOP, PinosMainLoopPrivate))
 
+#define DATAS_SIZE (4096 * 8)
+
+typedef struct {
+  size_t             item_size;
+  SpaPollInvokeFunc  func;
+  uint32_t           seq;
+  size_t             size;
+  void              *data;
+  void              *user_data;
+} InvokeItem;
+
+typedef struct _WorkItem WorkItem;
+
+struct _WorkItem {
+  gulong          id;
+  gpointer        obj;
+  uint32_t        seq;
+  SpaResult       res;
+  PinosDeferFunc  func;
+  gpointer       *data;
+  GDestroyNotify  notify;
+  WorkItem       *next;
+};
+
 struct _PinosMainLoopPrivate
 {
   gulong counter;
 
-  GQueue work;
+  SpaRingbuffer buffer;
+  uint8_t       buffer_data[DATAS_SIZE];
+
+  SpaPollFd fds[1];
+  SpaPollItem wakeup;
+
+  SpaQueue  work;
+  WorkItem *free_list;
+
   gulong work_id;
 };
 
@@ -115,13 +152,94 @@ do_remove_item (SpaPoll     *poll,
   return SPA_RESULT_OK;
 }
 
+static int
+main_loop_dispatch (SpaPollNotifyData *data)
+{
+  PinosMainLoop *this = data->user_data;
+  PinosMainLoopPrivate *priv = this->priv;
+  SpaPoll *p = &this->poll;
+  uint64_t u;
+  size_t offset;
+  InvokeItem *item;
+
+  if (read (priv->fds[0].fd, &u, sizeof(uint64_t)) != sizeof(uint64_t))
+    g_warning ("main-loop %p: failed to read fd", strerror (errno));
+
+  while (spa_ringbuffer_get_read_offset (&priv->buffer, &offset) > 0) {
+    item = SPA_MEMBER (priv->buffer_data, offset, InvokeItem);
+    item->func (p, item->seq, item->size, item->data, item->user_data);
+    spa_ringbuffer_read_advance (&priv->buffer, item->item_size);
+  }
+
+  return 0;
+}
+
+static SpaResult
+do_invoke (SpaPoll           *poll,
+           SpaPollInvokeFunc  func,
+           uint32_t           seq,
+           size_t             size,
+           void              *data,
+           void              *user_data)
+{
+  PinosMainLoop *this = SPA_CONTAINER_OF (poll, PinosMainLoop, poll);
+  PinosMainLoopPrivate *priv = this->priv;
+  SpaRingbufferArea areas[2];
+  InvokeItem *item;
+  uint64_t u = 1;
+
+  spa_ringbuffer_get_write_areas (&priv->buffer, areas);
+  if (areas[0].len < sizeof (InvokeItem)) {
+    g_warning ("queue full");
+    return SPA_RESULT_ERROR;
+  }
+  item = SPA_MEMBER (priv->buffer_data, areas[0].offset, InvokeItem);
+  item->seq = seq;
+  item->func = func;
+  item->user_data = user_data;
+  item->size = size;
+
+  if (areas[0].len > sizeof (InvokeItem) + size) {
+    item->data = SPA_MEMBER (item, sizeof (InvokeItem), void);
+    item->item_size = sizeof (InvokeItem) + size;
+    if (areas[0].len < sizeof (InvokeItem) + item->item_size)
+      item->item_size = areas[0].len;
+  } else {
+    item->data = SPA_MEMBER (priv->buffer_data, areas[1].offset, void);
+    item->item_size = areas[0].len + 1 + size;
+  }
+  memcpy (item->data, data, size);
+
+  spa_ringbuffer_write_advance (&priv->buffer, item->item_size);
+
+  if (write (priv->fds[0].fd, &u, sizeof(uint64_t)) != sizeof(uint64_t))
+    g_warning ("data-loop %p: failed to write fd", strerror (errno));
+
+  return SPA_RESULT_RETURN_ASYNC (seq);
+}
+
 static void
 pinos_main_loop_constructed (GObject * obj)
 {
   PinosMainLoop *this = PINOS_MAIN_LOOP (obj);
+  PinosMainLoopPrivate *priv = this->priv;
 
   g_debug ("main-loop %p: constructed", this);
 
+  priv->fds[0].fd = eventfd (0, 0);
+  priv->fds[0].events = POLLIN | POLLPRI | POLLERR;
+  priv->fds[0].revents = 0;
+
+  priv->wakeup.id = SPA_ID_INVALID;
+  priv->wakeup.enabled = false;
+  priv->wakeup.fds = priv->fds;
+  priv->wakeup.n_fds = 1;
+  priv->wakeup.idle_cb = NULL;
+  priv->wakeup.before_cb = NULL;
+  priv->wakeup.after_cb = main_loop_dispatch;
+  priv->wakeup.user_data = this;
+  do_add_item (&this->poll, &priv->wakeup);
+
   G_OBJECT_CLASS (pinos_main_loop_parent_class)->constructed (obj);
 }
 
@@ -169,8 +287,10 @@ pinos_main_loop_init (PinosMainLoop * this)
   this->poll.add_item = do_add_item;
   this->poll.update_item = do_update_item;
   this->poll.remove_item = do_remove_item;
+  this->poll.invoke = do_invoke;
 
-  g_queue_init (&priv->work);
+  SPA_QUEUE_INIT (&priv->work);
+  spa_ringbuffer_init (&priv->buffer, DATAS_SIZE);
 }
 
 /**
@@ -186,42 +306,36 @@ pinos_main_loop_new (void)
   return g_object_new (PINOS_TYPE_MAIN_LOOP, NULL);
 }
 
-typedef struct {
-  gulong          id;
-  gpointer        obj;
-  uint32_t        seq;
-  SpaResult       res;
-  PinosDeferFunc  func;
-  gpointer       *data;
-  GDestroyNotify  notify;
-} WorkItem;
-
 static gboolean
 process_work_queue (PinosMainLoop *this)
 {
   PinosMainLoopPrivate *priv = this->priv;
-  GList *walk, *next;
+  WorkItem *prev, *item, *next;
 
-  for (walk = priv->work.head; walk; walk = next) {
+  priv->work_id = 0;
-    WorkItem *item = walk->data;
 
-    next = g_list_next (walk);
+  for (item = priv->work.head, prev = NULL; item; prev = item, item = next) {
+    next = item->next;
 
-    g_debug ("main-loop %p: peek work queue item %p seq %d", this, item, item ? item->seq : -1);
+    g_debug ("main-loop %p: peek work queue item %p seq %d, prev %p next %p", this, item->obj, item->seq, prev, next);
     if (item->seq != SPA_ID_INVALID)
       continue;
 
-    g_debug ("main-loop %p: process work item %p", this, item);
+    g_debug ("main-loop %p: process work item %p", this, item->obj);
+    if (priv->work.tail == item)
+      priv->work.tail = prev;
+    if (prev == NULL)
+      priv->work.head = next;
+    else
+      prev->next = next;
+
     if (item->func)
-      item->func (item->data, item->res, item->id);
+      item->func (item->obj, item->data, item->res, item->id);
     if (item->notify)
       item->notify (item->data);
-
-    g_queue_delete_link (&priv->work, walk);
     g_slice_free (WorkItem, item);
+    item = prev;
   }
-
-  priv->work_id = 0;
   return FALSE;
 }
 
@@ -246,6 +360,7 @@ pinos_main_loop_defer (PinosMainLoop  *loop,
   item->func = func;
   item->data = data;
   item->notify = notify;
+  item->next = NULL;
 
   if (SPA_RESULT_IS_ASYNC (res)) {
     item->seq = SPA_RESULT_ASYNC_SEQ (res);
@@ -257,7 +372,7 @@ pinos_main_loop_defer (PinosMainLoop  *loop,
     have_work = TRUE;
     g_debug ("main-loop %p: defer object %p", loop, obj);
   }
-  g_queue_push_tail (&priv->work, item);
+  SPA_QUEUE_PUSH_TAIL (&priv->work, WorkItem, next, item);
 
   if (priv->work_id == 0 && have_work)
     priv->work_id = g_idle_add ((GSourceFunc) process_work_queue, loop);
@@ -270,19 +385,18 @@ pinos_main_loop_defer_cancel (PinosMainLoop  *loop,
                               gpointer        obj,
                               gulong          id)
 {
-  GList *walk;
   PinosMainLoopPrivate *priv;
   gboolean have_work = FALSE;
+  WorkItem *item;
 
   g_return_if_fail (PINOS_IS_MAIN_LOOP (loop));
   priv = loop->priv;
 
-  for (walk = priv->work.head; walk; walk = g_list_next (walk)) {
+  for (item = priv->work.head; item; item = item->next) {
-    WorkItem *i = walk->data;
+    if ((id == 0 || item->id == id) && (obj == NULL || item->obj == obj)) {
-    if ((id == 0 || i->id == id) && (obj == NULL || i->obj == obj)) {
+      g_debug ("main-loop %p: cancel defer %d for object %p", loop, item->seq, item->obj);
-      g_debug ("main-loop %p: cancel defer %d for object %p", loop, i->seq, i->obj);
+      item->seq = SPA_ID_INVALID;
-      i->seq = SPA_ID_INVALID;
+      item->func = NULL;
-      i->func = NULL;
       have_work = TRUE;
     }
   }
@@ -296,7 +410,7 @@ pinos_main_loop_defer_complete (PinosMainLoop  *loop,
                                 uint32_t        seq,
                                 SpaResult       res)
 {
-  GList *walk;
+  WorkItem *item;
   PinosMainLoopPrivate *priv;
   gboolean have_work = FALSE;
 
@@ -305,16 +419,17 @@ pinos_main_loop_defer_complete (PinosMainLoop  *loop,
 
   g_debug ("main-loop %p: async complete %d %d for object %p", loop, seq, res, obj);
 
-  for (walk = priv->work.head; walk; walk = g_list_next (walk)) {
+  for (item = priv->work.head; item; item = item->next) {
-    WorkItem *i = walk->data;
+    if (item->obj == obj && item->seq == seq) {
-
-    if (i->obj == obj && i->seq == seq) {
       g_debug ("main-loop %p: found defered %d for object %p", loop, seq, obj);
-      i->seq = SPA_ID_INVALID;
+      item->seq = SPA_ID_INVALID;
-      i->res = res;
+      item->res = res;
       have_work = TRUE;
     }
   }
+  if (!have_work)
+    g_debug ("main-loop %p: no defered %d found for object %p", loop, seq, obj);
+
   if (priv->work_id == 0 && have_work)
     priv->work_id = g_idle_add ((GSourceFunc) process_work_queue, loop);
 }

pinos/server/main-loop.h

@@ -25,6 +25,7 @@
 G_BEGIN_DECLS
 
 #include <spa/include/spa/poll.h>
+#include <spa/include/spa/node-event.h>
 
 typedef struct _PinosMainLoop PinosMainLoop;
 typedef struct _PinosMainLoopClass PinosMainLoopClass;
@@ -61,7 +62,11 @@ struct _PinosMainLoopClass {
   GObjectClass parent_class;
 };
 
-typedef void (*PinosDeferFunc) (gpointer       data,
+typedef void (*PinosEventFunc) (SpaNodeEvent *event,
+                                void         *user_data);
+
+typedef void (*PinosDeferFunc) (gpointer       obj,
+                                gpointer       data,
                                 SpaResult      res,
                                 gulong         id);
 

pinos/server/node.c

@@ -687,7 +687,7 @@ pinos_node_constructed (GObject * obj)
                            this,
                            SPA_RESULT_RETURN_ASYNC (0),
                            (PinosDeferFunc) init_complete,
-                           this,
+                           NULL,
                            NULL);
   }
   node_register_object (this);
@@ -1222,15 +1222,22 @@ remove_idle_timeout (PinosNode *node)
   }
 }
 
-typedef struct {
-  PinosNode *node;
-  PinosNodeState state;
-} StateData;
-
 static void
-on_state_complete (StateData *data)
+on_state_complete (PinosNode *node,
+                   gpointer   data,
+                   SpaResult  res)
 {
-  pinos_node_update_state (data->node, data->state);
+  PinosNodeState state = GPOINTER_TO_INT (data);
+
+  if (SPA_RESULT_IS_ERROR (res)) {
+    GError *error = NULL;
+    g_set_error (&error,
+                 PINOS_ERROR,
+                 PINOS_ERROR_NODE_STATE,
+                 "error changing node state: %d", res);
+    pinos_node_report_error (node, error);
+  } else
+    pinos_node_update_state (node, state);
 }
 
 /**
@@ -1248,7 +1255,6 @@ pinos_node_set_state (PinosNode      *node,
 {
   PinosNodePrivate *priv;
   SpaResult res = SPA_RESULT_OK;
-  StateData data;
 
   g_return_val_if_fail (PINOS_IS_NODE (node), SPA_RESULT_INVALID_ARGUMENTS);
   priv = node->priv;
@@ -1283,15 +1289,12 @@ pinos_node_set_state (PinosNode      *node,
   if (SPA_RESULT_IS_ERROR (res))
     return res;
 
-  data.node = node;
-  data.state = state;
-
   pinos_main_loop_defer (priv->main_loop,
                          node,
                          res,
                          (PinosDeferFunc) on_state_complete,
-                         g_memdup (&data, sizeof (StateData)),
+                         GINT_TO_POINTER (state),
-                         g_free);
+                         NULL);
 
   return res;
 }

spa/include/spa/poll.h

@@ -61,7 +61,7 @@ typedef struct {
   unsigned int n_fds;
 } SpaPollNotifyData;
 
-typedef int (*SpaPollNotify) (SpaPollNotifyData *data);
+typedef SpaResult (*SpaPollNotify) (SpaPollNotifyData *data);
 
 /**
  * SpaPollItem:
@@ -86,6 +86,14 @@ typedef struct {
   void          *user_data;
 } SpaPollItem;
 
+
+typedef SpaResult (*SpaPollInvokeFunc)  (SpaPoll        *poll,
+                                         uint32_t        seq,
+                                         size_t          size,
+                                         void           *data,
+                                         void           *user_data);
+
+
 /**
  * SpaPoll:
  *
@@ -122,11 +130,20 @@ struct _SpaPoll {
 
   SpaResult   (*remove_item)         (SpaPoll         *poll,
                                       SpaPollItem     *item);
+
+
+  SpaResult   (*invoke)              (SpaPoll            *poll,
+                                      SpaPollInvokeFunc   func,
+                                      uint32_t            seq,
+                                      size_t              size,
+                                      void               *data,
+                                      void               *user_data);
 };
 
 #define spa_poll_add_item(n,...)          (n)->add_item((n),__VA_ARGS__)
 #define spa_poll_update_item(n,...)       (n)->update_item((n),__VA_ARGS__)
 #define spa_poll_remove_item(n,...)       (n)->remove_item((n),__VA_ARGS__)
+#define spa_poll_invoke(n,...)            (n)->invoke((n),__VA_ARGS__)
 
 #ifdef __cplusplus
 }  /* extern "C" */

spa/plugins/v4l2/v4l2-source.c

@@ -86,6 +86,7 @@ typedef struct {
 
 typedef struct {
   SpaLog *log;
+  SpaPoll *main_loop;
   SpaPoll *data_loop;
 
   bool export_buf;
@@ -134,6 +135,8 @@ struct _SpaV4l2Source {
   SpaIDMap *map;
   SpaLog *log;
 
+  uint32_t seq;
+
   SpaV4l2SourceProps props[2];
 
   SpaNodeEventCallback event_cb;
@@ -221,6 +224,73 @@ spa_v4l2_source_node_set_props (SpaNode         *node,
   return res;
 }
 
+static SpaResult
+do_command_complete (SpaPoll        *poll,
+                     uint32_t        seq,
+                     size_t          size,
+                     void           *data,
+                     void           *user_data)
+{
+  SpaV4l2Source *this = user_data;
+  SpaNodeEvent event;
+
+  event.type = SPA_NODE_EVENT_TYPE_ASYNC_COMPLETE;
+  event.size = size;
+  event.data = data;
+  this->event_cb (&this->node, &event, this->user_data);
+
+  return SPA_RESULT_OK;
+}
+
+static SpaResult
+do_start (SpaPoll        *poll,
+          uint32_t        seq,
+          size_t          size,
+          void           *data,
+          void           *user_data)
+{
+  SpaV4l2Source *this = user_data;
+  SpaResult res;
+  SpaNodeEventAsyncComplete ac;
+
+  res = spa_v4l2_start (this);
+
+  ac.seq = seq;
+  ac.res = res;
+  spa_poll_invoke (this->state[0].main_loop,
+                   do_command_complete,
+                   seq,
+                   sizeof (ac),
+                   &ac,
+                   this);
+  return res;
+}
+
+static SpaResult
+do_pause (SpaPoll        *poll,
+          uint32_t        seq,
+          size_t          size,
+          void           *data,
+          void           *user_data)
+{
+  SpaV4l2Source *this = user_data;
+  SpaResult res;
+  SpaNodeEventAsyncComplete ac;
+
+  res = spa_v4l2_pause (this);
+
+  ac.seq = seq;
+  ac.res = res;
+  spa_poll_invoke (this->state[0].main_loop,
+                   do_command_complete,
+                   seq,
+                   sizeof (ac),
+                   &ac,
+                   this);
+  return res;
+}
+
+
 static SpaResult
 spa_v4l2_source_node_send_command (SpaNode        *node,
                                    SpaNodeCommand *command)
@@ -246,7 +316,12 @@ spa_v4l2_source_node_send_command (SpaNode        *node,
       if (state->n_buffers == 0)
         return SPA_RESULT_NO_BUFFERS;
 
-      return spa_v4l2_start (this);
+      return spa_poll_invoke (this->state[0].data_loop,
+                              do_start,
+                              ++this->seq,
+                              0,
+                              NULL,
+                              this);
     }
     case SPA_NODE_COMMAND_PAUSE:
     {
@@ -258,7 +333,12 @@ spa_v4l2_source_node_send_command (SpaNode        *node,
       if (state->n_buffers == 0)
         return SPA_RESULT_NO_BUFFERS;
 
-      return spa_v4l2_pause (this);
+      return spa_poll_invoke (this->state[0].data_loop,
+                              do_pause,
+                              ++this->seq,
+                              0,
+                              NULL,
+                              this);
     }
     case SPA_NODE_COMMAND_FLUSH:
     case SPA_NODE_COMMAND_DRAIN:
@@ -841,6 +921,8 @@ v4l2_source_init (const SpaHandleFactory  *factory,
       this->map = support[i].data;
     else if (strcmp (support[i].uri, SPA_LOG_URI) == 0)
       this->log = support[i].data;
+    else if (strcmp (support[i].uri, SPA_POLL__MainLoop) == 0)
+      this->state[0].main_loop = support[i].data;
     else if (strcmp (support[i].uri, SPA_POLL__DataLoop) == 0)
       this->state[0].data_loop = support[i].data;
   }
@@ -848,6 +930,10 @@ v4l2_source_init (const SpaHandleFactory  *factory,
     spa_log_error (this->log, "an id-map is needed");
     return SPA_RESULT_ERROR;
   }
+  if (this->state[0].main_loop == NULL) {
+    spa_log_error (this->log, "a main_loop is needed");
+    return SPA_RESULT_ERROR;
+  }
   if (this->state[0].data_loop == NULL) {
     spa_log_error (this->log, "a data_loop is needed");
     return SPA_RESULT_ERROR;