lib: Flag all flows down as the IRMd exits

On exit of the IRMd all flows will now be flagged as down, so external
applications will not hang anymore. Note: reads keep work from flows
that are down until there are no more remaining packets in the buffer,
but no more packets can be written.

When the RIB is used, the external application may exit a bit later
than the IRMd, so I added a brief sleep before the IRMd tries to
remove the fuse main directory.

Signed-off-by: Dimitri Staessens <dimitri@ouroboros.rocks>
Signed-off-by: Sander Vrijders <sander@ouroboros.rocks>

3 files changed, 26 insertions, 13 deletions

diff --git a/src/lib/shm_rbuff.c b/src/lib/shm_rbuff.c
index 361d5bb0..d491c69a 100644
--- a/src/lib/shm_rbuff.c
+++ b/src/lib/shm_rbuff.c
@@ -69,20 +69,11 @@ struct shm_rbuff {
         int               flow_id;  /* flow_id of the flow           */
 };
 
-void shm_rbuff_close(struct shm_rbuff * rb)
-{
-        assert(rb);
-
-        munmap(rb->shm_base, SHM_RB_FILE_SIZE);
-
-        free(rb);
-}
-
 #define MM_FLAGS (PROT_READ | PROT_WRITE)
 
-struct shm_rbuff * rbuff_create(pid_t pid,
-                                int   flow_id,
-                                int   flags)
+static struct shm_rbuff * rbuff_create(pid_t pid,
+                                       int   flow_id,
+                                       int   flags)
 {
         struct shm_rbuff * rb;
         int                fd;
@@ -130,6 +121,13 @@ struct shm_rbuff * rbuff_create(pid_t pid,
         return NULL;
 }
 
+static void rbuff_destroy(struct shm_rbuff * rb)
+{
+        munmap(rb->shm_base, SHM_RB_FILE_SIZE);
+
+        free(rb);
+}
+
 struct shm_rbuff * shm_rbuff_create(pid_t pid,
                                     int   flow_id)
 {
@@ -202,6 +200,13 @@ struct shm_rbuff * shm_rbuff_open(pid_t pid,
         return rbuff_create(pid, flow_id, O_RDWR);
 }
 
+void shm_rbuff_close(struct shm_rbuff * rb)
+{
+        assert(rb);
+
+        rbuff_destroy(rb);
+}
+
 #if (defined(SHM_RBUFF_LOCKLESS) &&                            \
      (defined(__GNUC__) || defined (__clang__)))
 #include "shm_rbuff_ll.c"
diff --git a/src/lib/shm_rbuff_ll.c b/src/lib/shm_rbuff_ll.c
index eef8a2fb..880d81dc 100644
--- a/src/lib/shm_rbuff_ll.c
+++ b/src/lib/shm_rbuff_ll.c
@@ -31,6 +31,11 @@ void shm_rbuff_destroy(struct shm_rbuff * rb)
 
         sprintf(fn, SHM_RBUFF_PREFIX "%d.%d", rb->pid, rb->flow_id);
 
+        __sync_bool_compare_and_swap(rb->acl, *rb->acl, ACL_FLOWDOWN);
+
+        pthread_cond_broadcast(rb->del);
+        pthread_cond_broadcast(rb->add);
+
         shm_rbuff_close(rb);
 
         shm_unlink(fn);
diff --git a/src/lib/shm_rbuff_pthr.c b/src/lib/shm_rbuff_pthr.c
index 5a3e8c42..e41d31f8 100644
--- a/src/lib/shm_rbuff_pthr.c
+++ b/src/lib/shm_rbuff_pthr.c
@@ -29,7 +29,10 @@ void shm_rbuff_destroy(struct shm_rbuff * rb)
 #ifdef CONFIG_OUROBOROS_DEBUG
         pthread_mutex_lock(rb->lock);
 
-        assert(shm_rbuff_empty(rb));
+        *rb->acl = *rb->acl & ACL_FLOWDOWN;
+
+        pthread_cond_broadcast(rb->del);
+        pthread_cond_broadcast(rb->add);
 
         pthread_mutex_unlock(rb->lock);
 #endif