irmd: Remove shm_sanitizer thread

This removes the sanitizer thread in the IRMd to avoid the IRMd eating
the CPU when the buffer is full. The processes will clean the head PDU
if there is a broken lock in the rdrbuff. Chances for a lingering tail
PDU are slim.

Signed-off-by: Dimitri Staessens <dimitri.staessens@ugent.be>
Signed-off-by: Sander Vrijders <sander.vrijders@ugent.be>

1 files changed, 40 insertions, 90 deletions

diff --git a/src/lib/shm_rdrbuff.c b/src/lib/shm_rdrbuff.c
index 1dfcb2ca..5ae2085d 100644
--- a/src/lib/shm_rdrbuff.c
+++ b/src/lib/shm_rdrbuff.c
@@ -46,29 +46,17 @@
                        + sizeof(pthread_mutex_t) + 2 * sizeof(pthread_cond_t)  \
                        + sizeof(pid_t))
 
-#ifndef SHM_RDRB_MULTI_BLOCK
-#define WAIT_BLOCKS 1
-#else
-#define WAIT_BLOCKS ((SHM_BUFFER_SIZE) >> 4)
-#if WAIT_BLOCKS == 0
-#undef WAIT_BLOCKS
-#define WAIT_BLOCKS 1
-#endif
-#endif
-
 #define get_head_ptr(rdrb)                                                     \
-        ((struct shm_du_buff *) (rdrb->shm_base + (*rdrb->head                 \
-                                                   * SHM_RDRB_BLOCK_SIZE)))
+        idx_to_du_buff_ptr(rdrb, *rdrb->head)
 
 #define get_tail_ptr(rdrb)                                                     \
-        ((struct shm_du_buff *) (rdrb->shm_base + (*rdrb->tail                 \
-                                                   * SHM_RDRB_BLOCK_SIZE)))
+        idx_to_du_buff_ptr(rdrb, *rdrb->tail)
 
 #define idx_to_du_buff_ptr(rdrb, idx)                                          \
         ((struct shm_du_buff *) (rdrb->shm_base + idx * SHM_RDRB_BLOCK_SIZE))
 
 #define shm_rdrb_used(rdrb)                                                    \
-        ((*rdrb->head + (SHM_BUFFER_SIZE) - *rdrb->tail)                       \
+        (((*rdrb->head + (SHM_BUFFER_SIZE) - *rdrb->tail) + 1)                 \
          & ((SHM_BUFFER_SIZE) - 1))
 
 #define shm_rdrb_free(rdrb, i)                                                 \
@@ -118,6 +106,13 @@ static void garbage_collect(struct shm_rdrbuff * rdrb)
         pthread_cond_broadcast(rdrb->healthy);
 }
 
+static void sanitize(struct shm_rdrbuff * rdrb)
+{
+        get_head_ptr(rdrb)->flags = SDB_NULL;
+        garbage_collect(rdrb);
+        pthread_mutex_consistent(rdrb->lock);
+}
+
 static char * rdrb_filename(void)
 {
         char * str;
@@ -282,51 +277,6 @@ struct shm_rdrbuff * shm_rdrbuff_open()
         return rdrb_create(O_RDWR);
 }
 
-int shm_rdrbuff_wait_full(struct shm_rdrbuff * rdrb,
-                          struct timespec *    timeo)
-{
-        struct timespec abstime;
-
-        if (timeo != NULL) {
-                clock_gettime(PTHREAD_COND_CLOCK, &abstime);
-                ts_add(&abstime, timeo, &abstime);
-        }
-
-#ifndef HAVE_ROBUST_MUTEX
-        pthread_mutex_lock(rdrb->lock);
-#else
-        if (pthread_mutex_lock(rdrb->lock) == EOWNERDEAD)
-                pthread_mutex_consistent(rdrb->lock);
-#endif
-
-        while (shm_rdrb_free(rdrb, WAIT_BLOCKS)) {
-#ifndef HAVE_ROBUST_MUTEX
-                if (pthread_cond_timedwait(rdrb->full,
-                                           rdrb->lock,
-                                           &abstime) == ETIMEDOUT) {
-                        pthread_mutex_unlock(rdrb->lock);
-                        return -ETIMEDOUT;
-                }
-#else
-                int ret = pthread_cond_timedwait(rdrb->full,
-                                                 rdrb->lock,
-                                                 &abstime);
-                if (ret == EOWNERDEAD)
-                        pthread_mutex_consistent(rdrb->lock);
-                if (ret == ETIMEDOUT) {
-                        pthread_mutex_unlock(rdrb->lock);
-                        return -ETIMEDOUT;
-                }
-#endif
-        }
-
-        garbage_collect(rdrb);
-
-        pthread_mutex_unlock(rdrb->lock);
-
-        return 0;
-}
-
 void shm_rdrbuff_purge(void)
 {
         char * shm_rdrb_fn;
@@ -358,19 +308,19 @@ ssize_t shm_rdrbuff_write(struct shm_rdrbuff * rdrb,
 #ifndef SHM_RDRB_MULTI_BLOCK
         if (sz > SHM_RDRB_BLOCK_SIZE)
                 return -EMSGSIZE;
+#else
+        while (sz > 0) {
+                sz -= SHM_RDRB_BLOCK_SIZE;
+                ++blocks;
+        }
 #endif
 #ifndef HAVE_ROBUST_MUTEX
         pthread_mutex_lock(rdrb->lock);
 #else
         if (pthread_mutex_lock(rdrb->lock) == EOWNERDEAD)
-                pthread_mutex_consistent(rdrb->lock);
+                sanitize(rdrb);
 #endif
 #ifdef SHM_RDRB_MULTI_BLOCK
-        while (sz > 0) {
-                sz -= SHM_RDRB_BLOCK_SIZE;
-                ++blocks;
-        }
-
         if (blocks + *rdrb->head > (SHM_BUFFER_SIZE))
                 padblocks = (SHM_BUFFER_SIZE) - *rdrb->head;
 
@@ -426,39 +376,41 @@ ssize_t shm_rdrbuff_write_b(struct shm_rdrbuff *    rdrb,
                             const struct timespec * abstime)
 {
         struct shm_du_buff * sdb;
-        size_t               size = headspace + len + tailspace;
+        size_t               size      = headspace + len + tailspace;
 #ifdef SHM_RDRB_MULTI_BLOCK
-        size_t               blocks = 0;
+        size_t               blocks    = 0;
         size_t               padblocks = 0;
+        size_t               rblocks;
 #endif
-        ssize_t              sz = size + sizeof(*sdb);
-        int                  ret = 0;
+        ssize_t              sz        = size + sizeof(*sdb);
+        int                  ret       = 0;
 
         assert(rdrb);
 
 #ifndef SHM_RDRB_MULTI_BLOCK
         if (sz > SHM_RDRB_BLOCK_SIZE)
                 return -EMSGSIZE;
+#else
+        while (sz > 0) {
+                sz -= SHM_RDRB_BLOCK_SIZE;
+                ++blocks;
+        }
+        rblocks = blocks;
 #endif
 #ifndef HAVE_ROBUST_MUTEX
         pthread_mutex_lock(rdrb->lock);
 #else
         if (pthread_mutex_lock(rdrb->lock) == EOWNERDEAD)
-                pthread_mutex_consistent(rdrb->lock);
+                sanitize(rdrb);
 #endif
         pthread_cleanup_push((void (*) (void *)) pthread_mutex_unlock,
                              (void *) rdrb->lock);
 
 #ifdef SHM_RDRB_MULTI_BLOCK
-        while (sz > 0) {
-                sz -= SHM_RDRB_BLOCK_SIZE;
-                ++blocks;
-        }
+        if (blocks > 1)
+                rblocks = blocks << 1;
 
-        if (blocks + *rdrb->head > (SHM_BUFFER_SIZE))
-                padblocks = (SHM_BUFFER_SIZE) - *rdrb->head;
-
-        while (!shm_rdrb_free(rdrb, (blocks + padblocks)) && ret != ETIMEDOUT) {
+        while (!shm_rdrb_free(rdrb, rblocks) && ret != ETIMEDOUT) {
 #else
         while (!shm_rdrb_free(rdrb, 1) && ret != ETIMEDOUT) {
 #endif
@@ -469,6 +421,11 @@ ssize_t shm_rdrbuff_write_b(struct shm_rdrbuff *    rdrb,
                                                      abstime);
                 else
                         ret = pthread_cond_wait(rdrb->healthy, rdrb->lock);
+
+#ifdef SHM_RDRB_MULTI_BLOCK
+                if (blocks + *rdrb->head > (SHM_BUFFER_SIZE))
+                        padblocks = (SHM_BUFFER_SIZE) - *rdrb->head;
+#endif
         }
 
         if (ret != ETIMEDOUT) {
@@ -547,21 +504,14 @@ int shm_rdrbuff_remove(struct shm_rdrbuff * rdrb,
         pthread_mutex_lock(rdrb->lock);
 #else
         if (pthread_mutex_lock(rdrb->lock) == EOWNERDEAD)
-                pthread_mutex_consistent(rdrb->lock);
+                sanitize(rdrb);
 #endif
-        if (shm_rdrb_empty(rdrb)) {
-                pthread_mutex_unlock(rdrb->lock);
-                return -1;
-        }
+        assert(!shm_rdrb_empty(rdrb));
 
         idx_to_du_buff_ptr(rdrb, idx)->flags = SDB_NULL;
 
-        if (idx != *rdrb->tail) {
-                pthread_mutex_unlock(rdrb->lock);
-                return 0;
-        }
-
-        garbage_collect(rdrb);
+        if (idx == *rdrb->tail)
+                garbage_collect(rdrb);
 
         pthread_mutex_unlock(rdrb->lock);