optimize data transport

Remove queue and ringbuffer between nodes. transfer the buffer id
directly between the io areas when possible.
Let only pinos send push or pull requests for now.
Allow polling multiple fds, like how alsa wants it
Remove port_id from events.

pinos/server/client-node.c

@@ -171,13 +171,12 @@ send_need_input (SpaProxy *this)
 {
   PinosNode *pnode = this->pnode;
   SpaNodeEventNeedInput ni;
-  uint8_t cmd = 0;
+  uint64_t cmd = 1;
 
   ni.event.type = SPA_NODE_EVENT_TYPE_NEED_INPUT;
   ni.event.size = sizeof (ni);
-  ni.port_id = 0;
   pinos_transport_add_event (pnode->transport, &ni.event);
-  write (this->data_source.fd, &cmd, 1);
+  write (this->data_source.fd, &cmd, 8);
 }
 
 static void
@@ -185,13 +184,12 @@ send_have_output (SpaProxy *this)
 {
   PinosNode *pnode = this->pnode;
   SpaNodeEventHaveOutput ho;
-  uint8_t cmd = 0;
+  uint64_t cmd = 1;
 
   ho.event.type = SPA_NODE_EVENT_TYPE_HAVE_OUTPUT;
   ho.event.size = sizeof (ho);
-  ho.port_id = 0;
   pinos_transport_add_event (pnode->transport, &ho.event);
-  write (this->data_source.fd, &cmd, 1);
+  write (this->data_source.fd, &cmd, 8);
 }
 
 static SpaResult
@@ -857,7 +855,7 @@ spa_proxy_node_port_reuse_buffer (SpaNode         *node,
   SpaProxy *this;
   SpaNodeEventReuseBuffer rb;
   PinosNode *pnode;
-  uint8_t cmd = 0;
+  //uint64_t cmd = 1;
 
   if (node == NULL)
     return SPA_RESULT_INVALID_ARGUMENTS;
@@ -873,7 +871,7 @@ spa_proxy_node_port_reuse_buffer (SpaNode         *node,
   rb.port_id = port_id;
   rb.buffer_id = buffer_id;
   pinos_transport_add_event (pnode->transport, &rb.event);
-  write (this->data_source.fd, &cmd, 1);
+  //write (this->data_source.fd, &cmd, 8);
 
   return SPA_RESULT_OK;
 }
@@ -1060,9 +1058,9 @@ proxy_on_data_fd_events (SpaSource *source)
 
   if (source->rmask & SPA_IO_IN) {
     SpaNodeEvent event;
-    uint8_t cmd;
+    uint64_t cmd;
 
-    read (this->data_source.fd, &cmd, 1);
+    read (this->data_source.fd, &cmd, 8);
 
     while (pinos_transport_next_event (pnode->transport, &event) == SPA_RESULT_OK) {
       SpaNodeEvent *ev = alloca (event.size);
@@ -1338,15 +1336,22 @@ pinos_client_node_get_data_socket (PinosClientNode  *this,
   PinosClientNodeImpl *impl = SPA_CONTAINER_OF (this, PinosClientNodeImpl, this);
 
   if (impl->data_fd == -1) {
+#if 1
     int fd[2];
 
-    if (socketpair (AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0, fd) != 0)
+    if (socketpair (AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0, fd) != 0)
       return SPA_RESULT_ERRNO;
 
     impl->proxy.data_source.fd = fd[0];
-    spa_loop_add_source (impl->proxy.data_loop, &impl->proxy.data_source);
-    pinos_log_debug ("client-node %p: add data fd %d", this, fd[0]);
     impl->data_fd = fd[1];
+#else
+
+    impl->proxy.data_source.fd = eventfd (0, EFD_CLOEXEC | EFD_NONBLOCK);
+    impl->data_fd = eventfd (0, EFD_CLOEXEC | EFD_NONBLOCK);
+#endif
+
+    spa_loop_add_source (impl->proxy.data_loop, &impl->proxy.data_source);
+    pinos_log_debug ("client-node %p: add data fd %d", this, impl->proxy.data_source.fd);
   }
   *fd = impl->data_fd;
   return SPA_RESULT_OK;

pinos/server/data-loop.c

@@ -42,6 +42,7 @@ typedef struct
   pthread_t thread;
 } PinosDataLoopImpl;
 
+
 static void
 make_realtime (PinosDataLoop *this)
 {
@@ -57,7 +58,7 @@ make_realtime (PinosDataLoop *this)
   spa_zero (sp);
   sp.sched_priority = rtprio;
 
-  if (pthread_setschedparam (pthread_self(), SCHED_RR|SCHED_RESET_ON_FORK, &sp) == 0) {
+  if (pthread_setschedparam (pthread_self(), SCHED_OTHER|SCHED_RESET_ON_FORK, &sp) == 0) {
     pinos_log_debug ("SCHED_OTHER|SCHED_RESET_ON_FORK worked.");
     return;
   }

pinos/server/link.c

@@ -716,8 +716,6 @@ pinos_link_activate (PinosLink *this)
 {
   PinosLinkImpl *impl = SPA_CONTAINER_OF (this, PinosLinkImpl, this);
 
-  spa_ringbuffer_init (&this->ringbuffer, SPA_N_ELEMENTS (this->queue));
-
   pinos_work_queue_add (impl->work,
                         this,
                         SPA_RESULT_WAIT_SYNC,
@@ -729,7 +727,6 @@ pinos_link_activate (PinosLink *this)
 bool
 pinos_pinos_link_deactivate (PinosLink *this)
 {
-  spa_ringbuffer_clear (&this->ringbuffer);
   return true;
 }
 

pinos/server/link.h

@@ -68,9 +68,6 @@ struct _PinosLink {
                                 PinosLink     *link,
                                 PinosPort     *port));
 
-  uint32_t      queue[64];
-  SpaRingbuffer ringbuffer;
-
   struct {
     unsigned int   in_ready;
     PinosPort     *input;

pinos/server/node.c

@@ -246,37 +246,6 @@ send_clock_update (PinosNode *this)
     pinos_log_debug ("got error %d", res);
 }
 
-static SpaResult
-do_read_link (SpaLoop        *loop,
-              bool            async,
-              uint32_t        seq,
-              size_t          size,
-              void           *data,
-              void           *user_data)
-{
-  PinosNode *this = user_data;
-  PinosLink *link = ((PinosLink**)data)[0];
-  size_t offset;
-  SpaResult res;
-
-  if (link->rt.input == NULL)
-    return SPA_RESULT_OK;
-
-  while (link->rt.in_ready > 0 && spa_ringbuffer_get_read_offset (&link->ringbuffer, &offset) > 0) {
-    SpaPortInput *input = &this->transport->inputs[link->rt.input->port_id];
-
-    input->buffer_id = link->queue[offset];
-
-    if ((res = spa_node_process_input (link->rt.input->node->node)) < 0)
-      pinos_log_warn ("node %p: error pushing buffer: %d, %d", this, res, input->status);
-
-    spa_ringbuffer_read_advance (&link->ringbuffer, 1);
-    link->rt.in_ready--;
-  }
-  return SPA_RESULT_OK;
-}
-
-
 static void
 on_node_event (SpaNode *node, SpaNodeEvent *event, void *user_data)
 {
@@ -303,59 +272,84 @@ on_node_event (SpaNode *node, SpaNodeEvent *event, void *user_data)
 
     case SPA_NODE_EVENT_TYPE_NEED_INPUT:
     {
-      SpaNodeEventNeedInput *ni = (SpaNodeEventNeedInput *) event;
-      PinosPort *port = this->input_port_map[ni->port_id];
-      PinosLink *link;
+      SpaResult res;
+      int i;
+      bool processed = false;
 
-      spa_list_for_each (link, &port->rt.links, rt.input_link) {
-        if (link->rt.input == NULL || link->rt.output == NULL)
+//      pinos_log_debug ("node %p: need input", this);
+
+      for (i = 0; i < this->transport->area->n_inputs; i++) {
+        PinosLink *link;
+        PinosPort *inport, *outport;
+        SpaPortInput *pi;
+        SpaPortOutput *po;
+
+        pi = &this->transport->inputs[i];
+        if (pi->buffer_id != SPA_ID_INVALID)
           continue;
 
-        link->rt.in_ready++;
-        pinos_loop_invoke (link->rt.input->node->data_loop->loop,
-                           do_read_link,
-                           SPA_ID_INVALID,
-                           sizeof (PinosLink *),
-                           &link,
-                           link->rt.input->node);
+        inport = this->input_port_map[i];
+        spa_list_for_each (link, &inport->rt.links, rt.input_link) {
+          if (link->rt.input == NULL || link->rt.output == NULL)
+            continue;
+
+          outport = link->rt.output;
+          po = &outport->node->transport->outputs[outport->port_id];
+
+          if (po->buffer_id != SPA_ID_INVALID) {
+            processed = true;
+
+            pi->buffer_id = po->buffer_id;
+            po->buffer_id = SPA_ID_INVALID;
+          }
+          if ((res = spa_node_process_output (outport->node->node)) < 0)
+            pinos_log_warn ("node %p: got process output %d", outport->node, res);
+        }
+      }
+      if (processed) {
+        if ((res = spa_node_process_input (this->node)) < 0)
+          pinos_log_warn ("node %p: got process input %d", this, res);
       }
       break;
     }
     case SPA_NODE_EVENT_TYPE_HAVE_OUTPUT:
     {
-      SpaNodeEventHaveOutput *ho = (SpaNodeEventHaveOutput *) event;
       SpaResult res;
-      bool pushed = false;
-      SpaPortOutput *po = &this->transport->outputs[ho->port_id];
-      PinosPort *port = this->output_port_map[ho->port_id];
-      PinosLink *link;
+      int i;
+      bool processed = false;
 
-      spa_list_for_each (link, &port->rt.links, rt.output_link) {
-        size_t offset;
+//      pinos_log_debug ("node %p: have output", this);
 
-        if (link->rt.input == NULL || link->rt.output == NULL)
+      for (i = 0; i < this->transport->area->n_outputs; i++) {
+        PinosLink *link;
+        PinosPort *inport, *outport;
+        SpaPortInput *pi;
+        SpaPortOutput *po;
+
+        po = &this->transport->outputs[i];
+        if (po->buffer_id == SPA_ID_INVALID)
           continue;
 
-        if (spa_ringbuffer_get_write_offset (&link->ringbuffer, &offset) > 0) {
-          link->queue[offset] = po->buffer_id;
-          spa_ringbuffer_write_advance (&link->ringbuffer, 1);
+        outport = this->output_port_map[i];
+        spa_list_for_each (link, &outport->rt.links, rt.output_link) {
+          if (link->rt.input == NULL || link->rt.output == NULL)
+            continue;
 
-          pinos_loop_invoke (link->rt.input->node->data_loop->loop,
-                             do_read_link,
-                             SPA_ID_INVALID,
-                             sizeof (PinosLink *),
-                             &link,
-                             link->rt.input->node);
-          pushed = true;
+          inport = link->rt.input;
+          pi = &inport->node->transport->inputs[inport->port_id];
+
+          processed = true;
+
+          pi->buffer_id = po->buffer_id;
+
+          if ((res = spa_node_process_input (inport->node->node)) < 0)
+            pinos_log_warn ("node %p: got process input %d", inport->node, res);
         }
+        po->buffer_id = SPA_ID_INVALID;
       }
-      if (!pushed) {
-        if ((res = spa_node_port_reuse_buffer (node, ho->port_id, po->buffer_id)) < 0)
-          pinos_log_warn ("node %p: error reuse buffer: %d", node, res);
-      }
-      if ((res = spa_node_process_output (node)) < 0) {
-        pinos_log_warn ("node %p: got pull error %d, %d", this, res, po->status);
-        break;
+      if (processed) {
+        if ((res = spa_node_process_output (this->node)) < 0)
+          pinos_log_warn ("node %p: got process output %d", this, res);
       }
       break;
     }
@@ -366,6 +360,8 @@ on_node_event (SpaNode *node, SpaNodeEvent *event, void *user_data)
       PinosPort *port = this->input_port_map[rb->port_id];
       PinosLink *link;
 
+//      pinos_log_debug ("node %p: reuse buffer %u", this, rb->buffer_id);
+
       spa_list_for_each (link, &port->rt.links, rt.input_link) {
         if (link->rt.input == NULL || link->rt.output == NULL)
           continue;