From cfea9d5fb2def10978e09c94af00abbbe11867ad Mon Sep 17 00:00:00 2001
From: Kamil Trzcinski <ayufan@ayufan.eu>
Date: Tue, 5 Apr 2022 00:25:58 +0200
Subject: [PATCH] Improve buffer handling list

---
 buffer_list.c  | 18 +++++++++---------
 buffer_list.h  |  1 +
 buffer_queue.c | 13 +++++++++++++
 links.c        | 45 ++++++++++++++++++++++++++++++++++++++-------
 4 files changed, 61 insertions(+), 16 deletions(-)

diff --git a/buffer_list.c b/buffer_list.c
index 7e061e4..7985688 100644
--- a/buffer_list.c
+++ b/buffer_list.c
@@ -171,16 +171,16 @@ int buffer_list_stream(buffer_list_t *buf_list, bool do_on)
     buffer_t *buf = buf_list->bufs[i];
 
     // dequeue buffers (when stream off)
-    if (buf->enqueued) {
-      if (buf->mmap_source) {
-        buf->mmap_source->used = 0;
-        buffer_consumed(buf->mmap_source);
-        buf->mmap_source = NULL;
-      }
+    // if (buf->enqueued) {
+    //   if (buf->mmap_source) {
+    //     buf->mmap_source->used = 0;
+    //     buffer_consumed(buf->mmap_source);
+    //     buf->mmap_source = NULL;
+    //   }
 
-      buf->enqueued = false;
-      buf->mmap_reflinks = 1;
-    }
+    //   buf->enqueued = false;
+    //   buf->mmap_reflinks = 1;
+    // }
 
     // re-enqueue buffers on stream start
     if (buf_list->streaming && buf_list->do_capture && !buf->enqueued && buf->mmap_reflinks == 1) {
diff --git a/buffer_list.h b/buffer_list.h
index 0b2ce97..63a28ef 100644
--- a/buffer_list.h
+++ b/buffer_list.h
@@ -31,5 +31,6 @@ int buffer_list_stream(buffer_list_t *buf_list, bool do_on);
 
 buffer_t *buffer_list_find_slot(buffer_list_t *buf_list);
 buffer_t *buffer_list_dequeue(buffer_list_t *buf_list);
+int buffer_list_count_enqueued(buffer_list_t *buf_list);
 
 int buffer_list_enqueue(buffer_list_t *buf_list, buffer_t *dma_buf);
diff --git a/buffer_queue.c b/buffer_queue.c
index 7efc70a..55bdd0f 100644
--- a/buffer_queue.c
+++ b/buffer_queue.c
@@ -91,6 +91,19 @@ buffer_t *buffer_list_find_slot(buffer_list_t *buf_list)
   return buf;
 }
 
+int buffer_list_count_enqueued(buffer_list_t *buf_list)
+{
+  int n = 0;
+
+  for (int i = 0; i < buf_list->nbufs; i++) {
+    if (buf_list->bufs[i]->enqueued) {
+      n++;
+    }
+  }
+
+  return n;
+}
+
 int buffer_list_enqueue(buffer_list_t *buf_list, buffer_t *dma_buf)
 {
   if (!buf_list->do_mmap && !dma_buf->buf_list->do_mmap) {
diff --git a/links.c b/links.c
index 8e008f5..130563e 100644
--- a/links.c
+++ b/links.c
@@ -22,11 +22,7 @@ int _build_fds(link_t *all_links, struct pollfd *fds, link_t **links, buffer_lis
       continue;
     }
 
-    struct pollfd fd = {link->capture->fd, POLLIN};
-    fds[n] = fd;
-    buf_lists[n] = link->capture->capture_list;
-    links[n] = link;
-    n++;
+    bool can_consume = true;
 
     for (int j = 0; link->outputs[j]; j++) {
       device_t *output = link->outputs[j];
@@ -38,12 +34,33 @@ int _build_fds(link_t *all_links, struct pollfd *fds, link_t **links, buffer_lis
         continue;
       }
 
+      int count_enqueued = buffer_list_count_enqueued(output->output_list);
+
+      if (count_enqueued == output->output_list->nbufs) {
+        E_LOG_DEBUG(link->capture->capture_list, "Cannot consume due to %s using %d of %d",
+          output->output_list->name, count_enqueued, output->output_list->nbufs);
+        can_consume = false;
+      }
+
+      // Can something be dequeued?
+      if (count_enqueued == 0) {
+        continue;
+      }
+
       struct pollfd fd = {output->fd, POLLOUT};
       fds[n] = fd;
       buf_lists[n] = output->output_list;
       links[n] = link;
       n++;
     }
+
+    if (can_consume) {
+      struct pollfd fd = {link->capture->fd, POLLIN};
+      fds[n] = fd;
+      buf_lists[n] = link->capture->capture_list;
+      links[n] = link;
+      n++;
+    }
   }
 
   return n;
@@ -56,15 +73,21 @@ int links_step(link_t *all_links, int timeout)
   buffer_list_t *buf_lists[N_FDS];
   buffer_t *buf;
   int n = _build_fds(all_links, fds, links, buf_lists, N_FDS);
+  int ret = poll(fds, n, timeout);
 
-  if (poll(fds, n, timeout) < 0 && errno != EINTR) {
+  if (ret < 0 && errno != EINTR) {
     return errno;
   }
 
+  printf("links_step n=%d, ret=%d\n", n, ret);
+
   for (int i = 0; i < n; i++) {
     buffer_list_t *buf_list = buf_lists[i];
     link_t *link = links[i];
 
+    E_LOG_DEBUG(buf_list, "pool i=%d revents=%08x streaming=%d enqueued=%d/%d", i, fds[i].revents, buf_list->streaming,
+      buffer_list_count_enqueued(buf_list), buf_list->nbufs);
+
     if (fds[i].revents & POLLIN) {
       E_LOG_DEBUG(buf_list, "POLLIN");
       if (buf = buffer_list_dequeue(buf_list)) {
@@ -83,6 +106,13 @@ int links_step(link_t *all_links, int timeout)
       E_LOG_DEBUG(buf_list, "POLLOUT");
       buffer_list_dequeue(buf_list);
     }
+    if (fds[i].revents & POLLERR) {
+      E_LOG_DEBUG(buf_list, "POLLERR");
+      device_consume_event(buf_list->device);
+    }
+    if (fds[i].revents & ~(POLLIN|POLLOUT|POLLERR)) {
+      E_LOG_DEBUG(buf_list, "POLL%08x", fds[i].revents);
+    }
   }
   return 0;
 }
@@ -181,15 +211,16 @@ int links_loop(link_t *all_links, bool *running)
 {
   *running = true;
 
-  while(*running) {
     if (links_stream(all_links, true) < 0) {
       return -1;
     }
 
+  while(*running) {
     if (links_step(all_links, 1000) < 0) {
       links_stream(all_links, false);
       return -1;
     }
+    //usleep(100*1000);
   }
 
   links_stream(all_links, false);