Merged in dstaesse/ouroboros/be-rdrbuff (pull request #303)

lib: Remove dst_api field from rdrbuff blocks

5 files changed, 138 insertions, 192 deletions

diff --git a/include/ouroboros/shm_rdrbuff.h b/include/ouroboros/shm_rdrbuff.h
index b16e2530..c4e1e7b0 100644
--- a/include/ouroboros/shm_rdrbuff.h
+++ b/include/ouroboros/shm_rdrbuff.h
@@ -43,18 +43,16 @@ void                 shm_rdrbuff_close(struct shm_rdrbuff * rdrb);
 
 void                 shm_rdrbuff_destroy(struct shm_rdrbuff * rdrb);
 
-void *               shm_rdrbuff_sanitize(void * o);
+void                 shm_rdrbuff_wait_full(struct shm_rdrbuff * rdrb);
 
 /* returns the index of the buffer in the DU map */
 ssize_t              shm_rdrbuff_write(struct shm_rdrbuff * rdrb,
-                                       pid_t                dst_api,
                                        size_t               headspace,
                                        size_t               tailspace,
                                        uint8_t *            data,
                                        size_t               data_len);
 
 ssize_t              shm_rdrbuff_write_b(struct shm_rdrbuff * rdrb,
-                                         pid_t                dst_api,
                                          size_t               headspace,
                                          size_t               tailspace,
                                          uint8_t *            data,
diff --git a/src/irmd/main.c b/src/irmd/main.c
index 055e9d8f..13bfa052 100644
--- a/src/irmd/main.c
+++ b/src/irmd/main.c
@@ -55,6 +55,7 @@
 #include <sys/wait.h>
 
 #define IRMD_CLEANUP_TIMER ((IRMD_FLOW_TIMEOUT / 20) * MILLION) /* ns */
+#define SHM_SAN_HOLDOFF 1000 /* ms */
 
 struct ipcp_entry {
         struct list_head next;
@@ -1612,6 +1613,46 @@ void irmd_sig_handler(int sig, siginfo_t * info, void * c)
         }
 }
 
+void * shm_sanitize(void * o)
+{
+        struct list_head * p = NULL;
+        struct timespec ts = {SHM_SAN_HOLDOFF / 1000,
+                              (SHM_SAN_HOLDOFF % 1000) * MILLION};
+        ssize_t idx;
+
+        (void) o;
+
+        while (true) {
+                shm_rdrbuff_wait_full(irmd->rdrb);
+
+                pthread_rwlock_rdlock(&irmd->state_lock);
+                pthread_rwlock_wrlock(&irmd->flows_lock);
+
+                list_for_each(p, &irmd->irm_flows) {
+                        struct irm_flow * f =
+                                list_entry(p, struct irm_flow, next);
+                        if (kill(f->n_api, 0) < 0) {
+                                while ((idx = shm_rbuff_read(f->n_rb)) >= 0)
+                                        shm_rdrbuff_remove(irmd->rdrb, idx);
+                                continue;
+                        }
+
+                        if (kill(f->n_1_api, 0) < 0) {
+                                while ((idx = shm_rbuff_read(f->n_1_rb)) >= 0)
+                                        shm_rdrbuff_remove(irmd->rdrb, idx);
+                                continue;
+                        }
+                }
+
+                pthread_rwlock_unlock(&irmd->flows_lock);
+                pthread_rwlock_unlock(&irmd->state_lock);
+
+                nanosleep(&ts, NULL);
+        }
+
+        return (void *) 0;
+}
+
 void * irm_sanitize(void * o)
 {
         struct timespec now;
@@ -2153,7 +2194,7 @@ int main(int argc, char ** argv)
 
         pthread_create(&irmd->irm_sanitize, NULL, irm_sanitize, NULL);
         pthread_create(&irmd->shm_sanitize, NULL,
-                       shm_rdrbuff_sanitize, irmd->rdrb);
+                       shm_sanitize, irmd->rdrb);
 
         /* wait for (all of them) to return */
         for (t = 0; t < IRMD_THREADPOOL_SIZE; ++t)
diff --git a/src/lib/dev.c b/src/lib/dev.c
index fc8739a2..1c0d73a1 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -536,7 +536,7 @@ int flow_alloc(char * dst_name, char * src_ae_name, struct qos_spec * qos)
                 return -1;
         }
 
-        ai.flows[fd].rx_rb   = shm_rbuff_open(ai.api, recv_msg->port_id);
+        ai.flows[fd].rx_rb = shm_rbuff_open(ai.api, recv_msg->port_id);
         if (ai.flows[fd].rx_rb == NULL) {
                 reset_flow(fd);
                 bmp_release(ai.fds, fd);
@@ -746,7 +746,6 @@ ssize_t flow_write(int fd, void * buf, size_t count)
 
         if (ai.flows[fd].oflags & FLOW_O_NONBLOCK) {
                 idx = shm_rdrbuff_write(ai.rdrb,
-                                       ai.flows[fd].api,
                                        DU_BUFF_HEADSPACE,
                                        DU_BUFF_TAILSPACE,
                                        buf,
@@ -766,7 +765,6 @@ ssize_t flow_write(int fd, void * buf, size_t count)
         } else { /* blocking */
                 struct shm_rdrbuff * rdrb = ai.rdrb;
                 struct shm_rbuff * tx_rb = ai.flows[fd].tx_rb;
-                pid_t api  = ai.flows[fd].api;
 
                 pthread_rwlock_unlock(&ai.flows_lock);
                 pthread_rwlock_unlock(&ai.data_lock);
@@ -774,7 +772,6 @@ ssize_t flow_write(int fd, void * buf, size_t count)
                 assert(tx_rb);
 
                 idx = shm_rdrbuff_write_b(rdrb,
-                                          api,
                                           DU_BUFF_HEADSPACE,
                                           DU_BUFF_TAILSPACE,
                                           buf,
diff --git a/src/lib/shm_rbuff.c b/src/lib/shm_rbuff.c
index 301669e7..c0901ab1 100644
--- a/src/lib/shm_rbuff.c
+++ b/src/lib/shm_rbuff.c
@@ -43,12 +43,12 @@
 #include <stdbool.h>
 
 #define FN_MAX_CHARS 255
-#define RB_CLOSED -1
 #define RB_OPEN 0
+#define RB_CLOSED 1
 
 #define SHM_RBUFF_FILE_SIZE ((SHM_BUFFER_SIZE) * sizeof(ssize_t)          \
-                             + 2 * sizeof(size_t) + sizeof(int8_t)      \
-                             + sizeof(pthread_mutex_t)                  \
+                             + 3 * sizeof(size_t)                         \
+                             + sizeof(pthread_mutex_t)                    \
                              + 2 * sizeof (pthread_cond_t))
 
 #define shm_rbuff_used(rb) ((*rb->head + (SHM_BUFFER_SIZE) - *rb->tail)   \
@@ -62,7 +62,7 @@ struct shm_rbuff {
         ssize_t *         shm_base; /* start of entry                */
         size_t *          head;     /* start of ringbuffer head      */
         size_t *          tail;     /* start of ringbuffer tail      */
-        int8_t *          acl;      /* access control                */
+        size_t *          acl;      /* access control                */
         pthread_mutex_t * lock;     /* lock all free space in shm    */
         pthread_cond_t *  add;      /* SDU arrived                   */
         pthread_cond_t *  del;      /* SDU removed                   */
@@ -126,7 +126,7 @@ struct shm_rbuff * shm_rbuff_create(pid_t api, int port_id)
         rb->shm_base = shm_base;
         rb->head     = (size_t *) (rb->shm_base + (SHM_BUFFER_SIZE));
         rb->tail     = rb->head + 1;
-        rb->acl      = (int8_t *) (rb->tail + 1);
+        rb->acl      = rb->tail + 1;
         rb->lock     = (pthread_mutex_t *) (rb->acl + 1);
         rb->add      = (pthread_cond_t *) (rb->lock + 1);
         rb->del      = rb->add + 1;
@@ -153,9 +153,6 @@ struct shm_rbuff * shm_rbuff_create(pid_t api, int port_id)
         rb->api = api;
         rb->port_id = port_id;
 
-        if (munmap(rb->shm_base, SHM_RBUFF_FILE_SIZE) == -1)
-                LOG_DBG("Couldn't unmap shared memory.");
-
         return rb;
 }
 
@@ -202,7 +199,7 @@ struct shm_rbuff * shm_rbuff_open(pid_t api, int port_id)
         rb->shm_base = shm_base;
         rb->head     = (size_t *) (rb->shm_base + (SHM_BUFFER_SIZE));
         rb->tail     = rb->head + 1;
-        rb->acl      = (int8_t *) (rb->tail + 1);
+        rb->acl      = rb->tail + 1;
         rb->lock     = (pthread_mutex_t *) (rb->acl + 1);
         rb->add      = (pthread_cond_t *) (rb->lock + 1);
         rb->del      = rb->add + 1;
@@ -225,13 +222,16 @@ void shm_rbuff_close(struct shm_rbuff * rb)
 
 void shm_rbuff_destroy(struct shm_rbuff * rb)
 {
-        char fn[25];
+        char fn[FN_MAX_CHARS];
 
         if (rb == NULL)
                 return;
 
         sprintf(fn, SHM_RBUFF_PREFIX "%d.%d", rb->api, rb->port_id);
 
+        if (munmap(rb->shm_base, SHM_RBUFF_FILE_SIZE) == -1)
+                LOG_DBG("Couldn't unmap shared memory.");
+
         if (shm_unlink(fn) == -1)
                 LOG_DBG("Failed to unlink shm %s.", fn);
 
@@ -251,7 +251,7 @@ int shm_rbuff_write(struct shm_rbuff * rb, size_t idx)
                 pthread_mutex_consistent(rb->lock);
         }
 #endif
-        if (*rb->acl) {
+        if (*rb->acl == RB_CLOSED) {
                 pthread_mutex_unlock(rb->lock);
                 return -ENOTALLOC;
         }
diff --git a/src/lib/shm_rdrbuff.c b/src/lib/shm_rdrbuff.c
index dc1feb10..a8245447 100644
--- a/src/lib/shm_rdrbuff.c
+++ b/src/lib/shm_rdrbuff.c
@@ -42,16 +42,26 @@
 #include <ouroboros/logs.h>
 
 #define SHM_BLOCKS_SIZE ((SHM_BUFFER_SIZE) * SHM_RDRB_BLOCK_SIZE)
-#define SHM_FILE_SIZE (SHM_BLOCKS_SIZE + 3 * sizeof(size_t)                    \
+#define SHM_FILE_SIZE (SHM_BLOCKS_SIZE + 2 * sizeof(size_t)                    \
                        + 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->ptr_head             \
+        ((struct shm_du_buff *) (rdrb->shm_base + (*rdrb->head                 \
                                                    * SHM_RDRB_BLOCK_SIZE)))
 
 #define get_tail_ptr(rdrb)                                                     \
-        ((struct shm_du_buff *) (rdrb->shm_base + (*rdrb->ptr_tail             \
+        ((struct shm_du_buff *) (rdrb->shm_base + (*rdrb->tail                 \
                                                    * SHM_RDRB_BLOCK_SIZE)))
 
 #define idx_to_du_buff_ptr(rdrb, idx)                                          \
@@ -61,13 +71,19 @@
         (((uint8_t *)sdb - rdrb->shm_base) / SHM_RDRB_BLOCK_SIZE)
 
 #define shm_rdrb_used(rdrb)                                                    \
-        ((*rdrb->ptr_head + (SHM_BUFFER_SIZE) - *rdrb->ptr_tail)               \
+        ((*rdrb->head + (SHM_BUFFER_SIZE) - *rdrb->tail)                       \
          & ((SHM_BUFFER_SIZE) - 1))
+
 #define shm_rdrb_free(rdrb, i)                                                 \
         (shm_rdrb_used(rdrb) + i < (SHM_BUFFER_SIZE))
 
 #define shm_rdrb_empty(rdrb)                                                   \
-        (*rdrb->ptr_tail == *rdrb->ptr_head)
+        (*rdrb->tail == *rdrb->head)
+
+enum shm_du_buff_flags {
+        SDB_VALID = 0,
+        SDB_NULL
+};
 
 struct shm_du_buff {
         size_t size;
@@ -76,20 +92,18 @@ struct shm_du_buff {
 #endif
         size_t du_head;
         size_t du_tail;
-        pid_t  dst_api;
+        size_t flags;
         size_t idx;
 };
 
 struct shm_rdrbuff {
-        uint8_t *         shm_base;    /* start of blocks */
-        size_t *          ptr_head;    /* start of ringbuffer head */
-        size_t *          ptr_tail;    /* start of ringbuffer tail */
-        pthread_mutex_t * lock;        /* lock all free space in shm */
-        size_t *          choked;      /* stale sdu detection */
-        pthread_cond_t *  healthy;     /* du map is healthy */
-        pthread_cond_t *  full;        /* run sanitizer when buffer full */
-        pid_t *           api;         /* api of the irmd owner */
-        enum qos_cube     qos;         /* qos id which this buffer serves */
+        uint8_t *         shm_base; /* start of blocks */
+        size_t *          head;     /* start of ringbuffer head */
+        size_t *          tail;     /* start of ringbuffer tail */
+        pthread_mutex_t * lock;     /* lock all free space in shm */
+        pthread_cond_t *  full;     /* flag when full */
+        pthread_cond_t *  healthy;  /* flag when SDU is read */
+        pid_t *           api;      /* api of the irmd owner */
 };
 
 static void garbage_collect(struct shm_rdrbuff * rdrb)
@@ -97,61 +111,31 @@ static void garbage_collect(struct shm_rdrbuff * rdrb)
 #ifdef SHM_RDRB_MULTI_BLOCK
         struct shm_du_buff * sdb;
         while (!shm_rdrb_empty(rdrb) &&
-               (sdb = get_tail_ptr(rdrb))->dst_api == -1)
-                *rdrb->ptr_tail = (*rdrb->ptr_tail + sdb->blocks)
+               (sdb = get_tail_ptr(rdrb))->flags == SDB_NULL)
+                *rdrb->tail = (*rdrb->tail + sdb->blocks)
                         & ((SHM_BUFFER_SIZE) - 1);
 #else
-        while (!shm_rdrb_empty(rdrb) && get_tail_ptr(rdrb)->dst_api == -1)
-                *rdrb->ptr_tail =
-                        (*rdrb->ptr_tail + 1) & ((SHM_BUFFER_SIZE) - 1);
-
+        while (!shm_rdrb_empty(rdrb) && get_tail_ptr(rdrb)->flags == SDB_NULL)
+                *rdrb->tail = (*rdrb->tail + 1) & ((SHM_BUFFER_SIZE) - 1);
 #endif
+        pthread_cond_broadcast(rdrb->healthy);
 }
 
-static void clean_sdus(struct shm_rdrbuff * rdrb, pid_t api)
-{
-        size_t idx = *rdrb->ptr_tail;
-        struct shm_du_buff * buf;
-
-        while (idx != *rdrb->ptr_head) {
-                buf = idx_to_du_buff_ptr(rdrb, idx);
-                if (buf->dst_api == api)
-                        buf->dst_api = -1;
-#ifdef SHM_RDRB_MULTI_BLOCK
-                idx = (idx + buf->blocks) & ((SHM_BUFFER_SIZE) - 1);
-#else
-                idx = (idx + 1) & ((SHM_BUFFER_SIZE) - 1);
-#endif
-        }
-
-        garbage_collect(rdrb);
-
-        *rdrb->choked = 0;
-}
-
-static char * rdrb_filename(enum qos_cube qos)
+static char * rdrb_filename(void)
 {
-        size_t chars = 0;
         char * str;
-        int qm = QOS_MAX;
 
-        do {
-                qm /= 10;
-                ++chars;
-        } while (qm > 0);
-
-        str = malloc(strlen(SHM_RDRB_PREFIX) + chars + 1);
+        str = malloc(strlen(SHM_RDRB_PREFIX) + 1);
         if (str == NULL) {
                 LOG_ERR("Failed to create shm_rdrbuff: Out of Memory.");
                 return NULL;
         }
 
-        sprintf(str, "%s%d", SHM_RDRB_PREFIX, (int) qos);
+        sprintf(str, "%s", SHM_RDRB_PREFIX);
 
         return str;
 }
 
-/* FIXME: create a ringbuffer for each qos cube in the system */
 struct shm_rdrbuff * shm_rdrbuff_create()
 {
         struct shm_rdrbuff * rdrb;
@@ -160,8 +144,7 @@ struct shm_rdrbuff * shm_rdrbuff_create()
         uint8_t *            shm_base;
         pthread_mutexattr_t  mattr;
         pthread_condattr_t   cattr;
-        enum qos_cube        qos = QOS_CUBE_BE;
-        char *               shm_rdrb_fn = rdrb_filename(qos);
+        char *               shm_rdrb_fn = rdrb_filename();
         if (shm_rdrb_fn == NULL) {
                 LOG_ERR("Could not create rdrbuff. Out of Memory");
                 return NULL;
@@ -212,14 +195,12 @@ struct shm_rdrbuff * shm_rdrbuff_create()
         }
 
         rdrb->shm_base = shm_base;
-        rdrb->ptr_head = (size_t *)
-                ((uint8_t *) rdrb->shm_base + SHM_BLOCKS_SIZE);
-        rdrb->ptr_tail = rdrb->ptr_head + 1;
-        rdrb->lock = (pthread_mutex_t *) (rdrb->ptr_tail + 1);
-        rdrb->choked = (size_t *) (rdrb->lock + 1);
-        rdrb->healthy = (pthread_cond_t *) (rdrb->choked + 1);
-        rdrb->full = rdrb->healthy + 1;
-        rdrb->api = (pid_t *) (rdrb->full + 1);
+        rdrb->head = (size_t *) ((uint8_t *) rdrb->shm_base + SHM_BLOCKS_SIZE);
+        rdrb->tail = rdrb->head + 1;
+        rdrb->lock = (pthread_mutex_t *) (rdrb->tail + 1);
+        rdrb->full = (pthread_cond_t *) (rdrb->lock + 1);
+        rdrb->healthy = rdrb->full + 1;
+        rdrb->api = (pid_t *) (rdrb->healthy + 1);
 
         pthread_mutexattr_init(&mattr);
         pthread_mutexattr_setpshared(&mattr, PTHREAD_PROCESS_SHARED);
@@ -236,29 +217,22 @@ struct shm_rdrbuff * shm_rdrbuff_create()
         pthread_cond_init(rdrb->full, &cattr);
         pthread_cond_init(rdrb->healthy, &cattr);
 
-        *rdrb->ptr_head = 0;
-        *rdrb->ptr_tail = 0;
-
-        *rdrb->choked = 0;
+        *rdrb->head = 0;
+        *rdrb->tail = 0;
 
         *rdrb->api = getpid();
 
-        rdrb->qos = qos;
-
         free(shm_rdrb_fn);
 
         return rdrb;
 }
 
-/* FIXME: open a ringbuffer for each qos cube in the system */
 struct shm_rdrbuff * shm_rdrbuff_open()
 {
         struct shm_rdrbuff * rdrb;
         int                  shm_fd;
         uint8_t *            shm_base;
-
-        enum qos_cube        qos = QOS_CUBE_BE;
-        char *               shm_rdrb_fn = rdrb_filename(qos);
+        char *               shm_rdrb_fn = rdrb_filename();
         if (shm_rdrb_fn == NULL) {
                 LOG_ERR("Could not create rdrbuff. Out of Memory");
                 return NULL;
@@ -297,32 +271,20 @@ struct shm_rdrbuff * shm_rdrbuff_open()
         }
 
         rdrb->shm_base = shm_base;
-        rdrb->ptr_head = (size_t *)
-                ((uint8_t *) rdrb->shm_base + SHM_BLOCKS_SIZE);
-        rdrb->ptr_tail = rdrb->ptr_head + 1;
-        rdrb->lock = (pthread_mutex_t *) (rdrb->ptr_tail + 1);
-        rdrb->choked = (size_t *) (rdrb->lock + 1);
-        rdrb->healthy = (pthread_cond_t *) (rdrb->choked + 1);
-        rdrb->full = rdrb->healthy + 1;
-        rdrb->api = (pid_t *) (rdrb->full + 1);
-
-        rdrb->qos = qos;
+        rdrb->head = (size_t *) ((uint8_t *) rdrb->shm_base + SHM_BLOCKS_SIZE);
+        rdrb->tail = rdrb->head + 1;
+        rdrb->lock = (pthread_mutex_t *) (rdrb->tail + 1);
+        rdrb->full = (pthread_cond_t *) (rdrb->lock + 1);
+        rdrb->healthy = rdrb->full + 1;
+        rdrb->api = (pid_t *) (rdrb->healthy + 1);
 
         free(shm_rdrb_fn);
 
         return rdrb;
 }
 
-void * shm_rdrbuff_sanitize(void * o)
+void shm_rdrbuff_wait_full(struct shm_rdrbuff * rdrb)
 {
-        struct shm_rdrbuff * rdrb = (struct shm_rdrbuff *) o;
-        struct timespec intv
-                = {SHM_DU_TIMEOUT_MICROS / MILLION,
-                   (SHM_DU_TIMEOUT_MICROS % MILLION) * 1000};
-
-        pid_t   api;
-
-        assert(o);
 
 #ifdef __APPLE__
         pthread_mutex_lock(rdrb->lock);
@@ -332,14 +294,10 @@ void * shm_rdrbuff_sanitize(void * o)
                 pthread_mutex_consistent(rdrb->lock);
         }
 #endif
-
         pthread_cleanup_push((void (*)(void *)) pthread_mutex_unlock,
                              (void *) rdrb->lock);
 
-        while (true) {
-                int ret = 0;
-                struct timespec now;
-                struct timespec dl;
+        while (shm_rdrb_free(rdrb, WAIT_BLOCKS)) {
 #ifdef __APPLE__
                 pthread_cond_wait(rdrb->full, rdrb->lock);
 #else
@@ -348,49 +306,11 @@ void * shm_rdrbuff_sanitize(void * o)
                         pthread_mutex_consistent(rdrb->lock);
                 }
 #endif
-                *rdrb->choked = 1;
-
-                garbage_collect(rdrb);
-
-                if (shm_rdrb_empty(rdrb)) {
-                        pthread_cond_broadcast(rdrb->healthy);
-                        continue;
-                }
-
-                api = get_tail_ptr(rdrb)->dst_api;
-
-                if (kill(api, 0)) {
-                        LOG_DBGF("Dead process %d left stale sdu.", api);
-                        clean_sdus(rdrb, api);
-                        pthread_cond_broadcast(rdrb->healthy);
-                        continue;
-                }
-
-                clock_gettime(CLOCK_REALTIME, &now);
-                ts_add(&now, &intv, &dl);
-                while (*rdrb->choked) {
-                        ret = pthread_cond_timedwait(rdrb->healthy,
-                                                     rdrb->lock,
-                                                     &dl);
-                        if (!ret)
-                                continue;
-#ifndef __APPLE__
-                        if (ret == EOWNERDEAD) {
-                                LOG_WARN("Recovering dead mutex.");
-                                pthread_mutex_consistent(rdrb->lock);
-                        }
-#endif
-                        if (ret == ETIMEDOUT) {
-                                LOG_DBGF("SDU timed out (dst: %d).", api);
-                                clean_sdus(rdrb, api);
-                        }
-                }
-                pthread_cond_broadcast(rdrb->healthy);
         }
 
-        pthread_cleanup_pop(true);
+        garbage_collect(rdrb);
 
-        return (void *) 0;
+        pthread_cleanup_pop(true);
 }
 
 void shm_rdrbuff_close(struct shm_rdrbuff * rdrb)
@@ -417,7 +337,7 @@ void shm_rdrbuff_destroy(struct shm_rdrbuff * rdrb)
         if (munmap(rdrb->shm_base, SHM_FILE_SIZE) == -1)
                 LOG_DBG("Couldn't unmap shared memory.");
 
-        shm_rdrb_fn = rdrb_filename(rdrb->qos);
+        shm_rdrb_fn = rdrb_filename();
         if (shm_rdrb_fn == NULL) {
                 LOG_ERR("Could not create rdrbuff. Out of Memory");
                 return;
@@ -431,7 +351,6 @@ void shm_rdrbuff_destroy(struct shm_rdrbuff * rdrb)
 }
 
 ssize_t shm_rdrbuff_write(struct shm_rdrbuff * rdrb,
-                          pid_t                dst_api,
                           size_t               headspace,
                           size_t               tailspace,
                           uint8_t *            data,
@@ -444,7 +363,6 @@ ssize_t shm_rdrbuff_write(struct shm_rdrbuff * rdrb,
         size_t               padblocks = 0;
 #endif
         ssize_t              sz = size + sizeof(*sdb);
-        uint8_t *            write_pos;
 
         assert(rdrb);
         assert(data);
@@ -469,14 +387,15 @@ ssize_t shm_rdrbuff_write(struct shm_rdrbuff * rdrb,
                 ++blocks;
         }
 
-        if (blocks + *rdrb->ptr_head > (SHM_BUFFER_SIZE))
-                padblocks = (SHM_BUFFER_SIZE) - *rdrb->ptr_head;
+        if (blocks + *rdrb->head > (SHM_BUFFER_SIZE))
+                padblocks = (SHM_BUFFER_SIZE) - *rdrb->head;
 
         if (!shm_rdrb_free(rdrb, blocks + padblocks)) {
 #else
         if (!shm_rdrb_free(rdrb, 1)) {
 #endif
-                pthread_cond_signal(rdrb->full);
+                LOG_DBG("buffer full, idx = %ld.", *rdrb->tail);
+                pthread_cond_broadcast(rdrb->full);
                 pthread_mutex_unlock(rdrb->lock);
                 return -1;
         }
@@ -486,31 +405,29 @@ ssize_t shm_rdrbuff_write(struct shm_rdrbuff * rdrb,
                 sdb = get_head_ptr(rdrb);
                 sdb->size    = 0;
                 sdb->blocks  = padblocks;
-                sdb->dst_api = -1;
+                sdb->flags   = SDB_NULL;
                 sdb->du_head = 0;
                 sdb->du_tail = 0;
-                sdb->idx     = *rdrb->ptr_head;
+                sdb->idx     = *rdrb->head;
 
-                *rdrb->ptr_head = 0;
+                *rdrb->head = 0;
         }
 #endif
         sdb          = get_head_ptr(rdrb);
         sdb->size    = size;
-        sdb->dst_api = dst_api;
+        sdb->flags   = SDB_VALID;
         sdb->du_head = headspace;
         sdb->du_tail = sdb->du_head + len;
 #ifdef  SHM_RDRB_MULTI_BLOCK
         sdb->blocks  = blocks;
 #endif
-        write_pos = ((uint8_t *) (sdb + 1)) + headspace;
-
-        memcpy(write_pos, data, len);
+        memcpy(((uint8_t *) (sdb + 1)) + headspace, data, len);
 
-        sdb->idx = *rdrb->ptr_head;
+        sdb->idx = *rdrb->head;
 #ifdef SHM_RDRB_MULTI_BLOCK
-        *rdrb->ptr_head = (*rdrb->ptr_head + blocks) & ((SHM_BUFFER_SIZE) - 1);
+        *rdrb->head = (*rdrb->head + blocks) & ((SHM_BUFFER_SIZE) - 1);
 #else
-        *rdrb->ptr_head = (*rdrb->ptr_head + 1) & ((SHM_BUFFER_SIZE) - 1);
+        *rdrb->head = (*rdrb->head + 1) & ((SHM_BUFFER_SIZE) - 1);
 #endif
         pthread_mutex_unlock(rdrb->lock);
 
@@ -518,7 +435,6 @@ ssize_t shm_rdrbuff_write(struct shm_rdrbuff * rdrb,
 }
 
 ssize_t shm_rdrbuff_write_b(struct shm_rdrbuff * rdrb,
-                           pid_t                 dst_api,
                            size_t                headspace,
                            size_t                tailspace,
                            uint8_t *             data,
@@ -531,7 +447,6 @@ ssize_t shm_rdrbuff_write_b(struct shm_rdrbuff * rdrb,
         size_t               padblocks = 0;
 #endif
         ssize_t              sz = size + sizeof(*sdb);
-        uint8_t *            write_pos;
 
         assert(rdrb);
         assert(data);
@@ -559,14 +474,14 @@ ssize_t shm_rdrbuff_write_b(struct shm_rdrbuff * rdrb,
                 ++blocks;
         }
 
-        if (blocks + *rdrb->ptr_head > (SHM_BUFFER_SIZE))
-                padblocks = (SHM_BUFFER_SIZE) - *rdrb->ptr_head;
+        if (blocks + *rdrb->head > (SHM_BUFFER_SIZE))
+                padblocks = (SHM_BUFFER_SIZE) - *rdrb->head;
 
         while (!shm_rdrb_free(rdrb, (blocks + padblocks))) {
 #else
         while (!shm_rdrb_free(rdrb, 1)) {
 #endif
-                pthread_cond_signal(rdrb->full);
+                pthread_cond_broadcast(rdrb->full);
                 pthread_cond_wait(rdrb->healthy, rdrb->lock);
         }
 
@@ -575,31 +490,29 @@ ssize_t shm_rdrbuff_write_b(struct shm_rdrbuff * rdrb,
                 sdb = get_head_ptr(rdrb);
                 sdb->size    = 0;
                 sdb->blocks  = padblocks;
-                sdb->dst_api = -1;
+                sdb->flags   = SDB_NULL;
                 sdb->du_head = 0;
                 sdb->du_tail = 0;
-                sdb->idx     = *rdrb->ptr_head;
+                sdb->idx     = *rdrb->head;
 
-                *rdrb->ptr_head = 0;
+                *rdrb->head = 0;
         }
 #endif
         sdb          = get_head_ptr(rdrb);
         sdb->size    = size;
-        sdb->dst_api = dst_api;
+        sdb->flags   = SDB_VALID;
         sdb->du_head = headspace;
         sdb->du_tail = sdb->du_head + len;
 #ifdef  SHM_RDRB_MULTI_BLOCK
         sdb->blocks  = blocks;
 #endif
-        write_pos = ((uint8_t *) (sdb + 1)) + headspace;
-
-        memcpy(write_pos, data, len);
+        memcpy(((uint8_t *) (sdb + 1)) + headspace, data, len);
 
-        sdb->idx = *rdrb->ptr_head;
+        sdb->idx = *rdrb->head;
 #ifdef SHM_RDRB_MULTI_BLOCK
-        *rdrb->ptr_head = (*rdrb->ptr_head + blocks) & ((SHM_BUFFER_SIZE) - 1);
+        *rdrb->head = (*rdrb->head + blocks) & ((SHM_BUFFER_SIZE) - 1);
 #else
-        *rdrb->ptr_head = (*rdrb->ptr_head + 1) & ((SHM_BUFFER_SIZE) - 1);
+        *rdrb->head = (*rdrb->head + 1) & ((SHM_BUFFER_SIZE) - 1);
 #endif
         pthread_cleanup_pop(true);
 
@@ -684,18 +597,15 @@ int shm_rdrbuff_remove(struct shm_rdrbuff * rdrb, size_t idx)
                 return -1;
         }
 
-        idx_to_du_buff_ptr(rdrb, idx)->dst_api = -1;
+        idx_to_du_buff_ptr(rdrb, idx)->flags = SDB_NULL;
 
-        if (idx != *rdrb->ptr_tail) {
+        if (idx != *rdrb->tail) {
                 pthread_mutex_unlock(rdrb->lock);
                 return 0;
         }
 
         garbage_collect(rdrb);
 
-        *rdrb->choked = 0;
-
-        pthread_cond_broadcast(rdrb->healthy);
         pthread_mutex_unlock(rdrb->lock);
 
         return 0;