lib: updated shm_du_map to be a hybrid ring buffer

Elements must be created/destroyed in order, but IPCPs can access PCI
while the PDU is in the ring buffer.
Test updated, no more waits are needed.

3 files changed, 87 insertions, 171 deletions

diff --git a/src/lib/shm_du_map.c b/src/lib/shm_du_map.c
index 052b8868..630cfa42 100644
--- a/src/lib/shm_du_map.c
+++ b/src/lib/shm_du_map.c
@@ -34,23 +34,28 @@
 
 
 #define SHM_DU_BLOCK_DATA_SIZE (SHM_DU_BUFF_BLOCK_SIZE -                       \
-                                sizeof (struct shm_block))
+                                sizeof(struct shm_block))
 #define SHM_BLOCKS_IN_MAP (SHM_DU_MAP_SIZE / SHM_DU_BUFF_BLOCK_SIZE)
 #define SHM_BLOCKS_SIZE (SHM_DU_BUFF_BLOCK_SIZE * SHM_BLOCKS_IN_MAP)
 #define SHM_BUFFS_SIZE (SHM_BLOCKS_IN_MAP * sizeof (struct shm_du_buff))
-#define SHM_MGT_MAP_SIZE (SHM_BLOCKS_IN_MAP * sizeof (uint8_t))
 #define SHM_FILE_SIZE (SHM_BLOCKS_IN_MAP * (SHM_DU_BUFF_BLOCK_SIZE             \
-                                            + sizeof (struct shm_du_buff)      \
-                                            + sizeof (uint8_t))                \
+                                            + sizeof(struct shm_du_buff)       \
+                                            + sizeof(uint8_t))                 \
+                       + 2 * sizeof (size_t)                                   \
                        + sizeof(pthread_mutex_t))
 
-#define idx_to_block_ptr(dum, i) ((struct shm_block *)                        \
+#define idx_to_block_ptr(dum, i) ((struct shm_block *)                         \
                                   (dum->shm_base + i * SHM_DU_BUFF_BLOCK_SIZE))
 #define idx_to_du_buff_ptr(dum, i) (dum->ptr_du_buff + i)
 #define du_buff_ptr_to_idx(dum, sdb) ((sdb - dum->ptr_du_buff) / sizeof *sdb)
 #define block_ptr_to_idx(dum, sdb) (((uint8_t *)sdb - dum->shm_base)           \
                                     / SHM_DU_BUFF_BLOCK_SIZE)
 
+#define shm_map_used(dum) ((*(dum->ptr_head) + SHM_BLOCKS_IN_MAP -             \
+                            *(dum->ptr_tail)) % SHM_BLOCKS_IN_MAP)
+
+#define shm_map_free(dum, i)(shm_map_used(dum) + i < SHM_BLOCKS_IN_MAP)
+
 struct shm_block {
         size_t size;
         long   next;
@@ -64,9 +69,10 @@ struct shm_du_buff {
 };
 
 struct shm_du_map {
-        uint8_t            * shm_base;    /* start of buffer blocks */
+        uint8_t            * shm_base;    /* start of blocks */
         struct shm_du_buff * ptr_du_buff; /* start of du_buff structs */
-        uint8_t            * ptr_map;     /* start of management map */
+        size_t             * ptr_head;    /* start of ringbuffer head */
+        size_t             * ptr_tail;    /* start of ringbuffer tail */
         pthread_mutex_t    * shm_mutex;   /* lock all free space in shm */
         int                  fd;
 };
@@ -76,9 +82,7 @@ struct shm_du_map * shm_du_map_create()
         struct shm_du_map * dum;
         int                 shm_fd;
         uint8_t           * shm_base;
-        uint8_t           * ptr;
         pthread_mutexattr_t attr;
-        int                 i;
 
         dum = malloc(sizeof *dum);
         if (dum == NULL) {
@@ -129,17 +133,19 @@ struct shm_du_map * shm_du_map_create()
         dum->shm_base = shm_base;
         dum->ptr_du_buff = (struct shm_du_buff *)
                 ((uint8_t *) dum->shm_base + SHM_BLOCKS_SIZE);
-        dum->ptr_map = (uint8_t *) dum->ptr_du_buff + SHM_BUFFS_SIZE;
+        dum->ptr_head = (size_t *)
+                ((uint8_t *) dum->ptr_du_buff + SHM_BUFFS_SIZE);
+        dum->ptr_tail = (size_t *)
+                ((uint8_t *) dum->ptr_head + sizeof(size_t));
         dum->shm_mutex = (pthread_mutex_t *)
-                ((uint8_t *) dum->ptr_map + SHM_MGT_MAP_SIZE);
+                ((uint8_t *) dum->ptr_tail + sizeof(size_t));
 
         pthread_mutexattr_init(&attr);
         pthread_mutexattr_setpshared(&attr, PTHREAD_PROCESS_SHARED);
         pthread_mutex_init(dum->shm_mutex, &attr);
 
-        ptr = dum->ptr_map;
-        for (i = 0; i < SHM_BLOCKS_IN_MAP; ++i)
-                *(ptr++) = 0;
+        *dum->ptr_head = 0;
+        *dum->ptr_tail = 0;
 
         dum->fd = shm_fd;
 
@@ -185,9 +191,12 @@ struct shm_du_map * shm_du_map_open()
         dum->shm_base = shm_base;
         dum->ptr_du_buff = (struct shm_du_buff *)
                 ((uint8_t *) dum->shm_base + SHM_BLOCKS_SIZE);
-        dum->ptr_map = (uint8_t *) dum->ptr_du_buff + SHM_BUFFS_SIZE;
+        dum->ptr_head = (size_t *)
+                ((uint8_t *) dum->ptr_du_buff + SHM_BUFFS_SIZE);
+        dum->ptr_tail = (size_t *)
+                ((uint8_t *) dum->ptr_head + sizeof(size_t));
         dum->shm_mutex = (pthread_mutex_t *)
-                ((uint8_t *) dum->ptr_map + SHM_MGT_MAP_SIZE);
+                ((uint8_t *) dum->ptr_tail + sizeof(size_t));
 
         return dum;
 }
@@ -208,53 +217,6 @@ void shm_du_map_close(struct shm_du_map * dum)
         free(dum);
 }
 
-static struct shm_du_buff * alloc_du_buff(struct shm_du_map * dum)
-{
-        uint8_t * ptr = dum->ptr_map;
-
-        pthread_mutex_lock(dum->shm_mutex);
-
-        while (*ptr == 1 && ptr < (uint8_t *) dum->shm_mutex)
-                ++ptr;
-
-        if (ptr < (uint8_t *) dum->shm_mutex) {
-                *ptr = 1;
-        } else {
-                pthread_mutex_unlock(dum->shm_mutex);
-                return NULL;
-        }
-
-        pthread_mutex_unlock(dum->shm_mutex);
-
-        return dum->ptr_du_buff + (ptr - dum->ptr_map);
-}
-
-static struct shm_block * alloc_block(struct shm_du_map * dum)
-{
-
-        pthread_mutex_lock(dum->shm_mutex);
-
-        uint8_t * ptr = dum->ptr_map;
-
-        while (*ptr == 1 && ptr < (uint8_t *) dum->shm_mutex)
-                ++ptr;
-
-        if (ptr < (uint8_t *) dum->shm_mutex) {
-                *ptr = 1;
-        }
-        else {
-                pthread_mutex_unlock(dum->shm_mutex);
-                return NULL;
-        }
-
-        pthread_mutex_unlock(dum->shm_mutex);
-
-
-        return (struct shm_block *)
-                (dum->shm_base + ((ptr - dum->ptr_map)
-                                  * SHM_DU_BUFF_BLOCK_SIZE));
-}
-
 struct shm_du_buff * shm_create_du_buff(struct shm_du_map * dum,
                                         size_t              size,
                                         size_t              headspace,
@@ -262,11 +224,11 @@ struct shm_du_buff * shm_create_du_buff(struct shm_du_map * dum,
                                         size_t              len)
 {
         struct shm_du_buff * sdb;
-        long                 index;
         long                 prev_index = -1;
         size_t               remaining = size;
         size_t               ts = size - (headspace + len);
         uint8_t            * read_pos = data;
+        size_t               blocks;
 
         if (dum == NULL || data == NULL) {
                 LOG_DBGF("Bogus input, bugging out.");
@@ -288,51 +250,42 @@ struct shm_du_buff * shm_create_du_buff(struct shm_du_map * dum,
                 return NULL;
         }
 
-        sdb = alloc_du_buff(dum);
-        if (sdb == NULL) {
+        pthread_mutex_lock(dum->shm_mutex);
+
+        blocks = size / SHM_DU_BLOCK_DATA_SIZE;
+        if (size % SHM_DU_BLOCK_DATA_SIZE > 0)
+                ++blocks;
+
+        if (!shm_map_free(dum, blocks)) {
+                pthread_mutex_unlock(dum->shm_mutex);
                 LOG_DBGF("Allocation failed, Out of Memory.");
                 return NULL;
         }
 
-        index = du_buff_ptr_to_idx(dum, sdb);
+        sdb = dum->ptr_du_buff + *dum->ptr_head;
+
         sdb->size = size;
         sdb->du_head = headspace;
         sdb->du_tail = sdb->du_head + len;
 
         while (remaining > 0) {
                 struct shm_block * shm_buf;
-                unsigned long       bytes_to_copy = len;
-                uint8_t           * write_pos;
+                long               bytes_to_copy = len;
+                uint8_t          * write_pos;
 
-                if (prev_index == -1)
-                        shm_buf = idx_to_block_ptr(dum, index);
-                else
-                        shm_buf = alloc_block(dum);
-
-                if (shm_buf == NULL) {
-                        shm_release_du_buff(dum, sdb);
-                        return NULL;
-                }
+                shm_buf = idx_to_block_ptr(dum, *(dum->ptr_head));
 
                 write_pos = (uint8_t *) shm_buf + sizeof *shm_buf;
 
-                index = block_ptr_to_idx(dum, shm_buf);
+                shm_buf->size = remaining < SHM_DU_BLOCK_DATA_SIZE ?
+                        remaining : SHM_DU_BLOCK_DATA_SIZE;
 
-                if (size > SHM_DU_BLOCK_DATA_SIZE
-                    && remaining - ts <=  SHM_DU_BLOCK_DATA_SIZE
-                    && remaining != ts) {
+                bytes_to_copy = shm_buf->size;
+
+                if (remaining <= SHM_DU_BLOCK_DATA_SIZE)
+                        bytes_to_copy -= ts;
+                else if (remaining - ts <= SHM_DU_BLOCK_DATA_SIZE)
                         shm_buf->size = remaining - ts;
-                        bytes_to_copy = shm_buf->size;
-                } else if (size > SHM_DU_BLOCK_DATA_SIZE && remaining == ts) {
-                        shm_buf->size = ts;
-                        bytes_to_copy = 0;
-                } else {
-                        shm_buf->size = remaining < SHM_DU_BLOCK_DATA_SIZE ?
-                                remaining : SHM_DU_BLOCK_DATA_SIZE;
-                        bytes_to_copy =shm_buf->size;
-                        if (remaining == shm_buf->size)
-                                bytes_to_copy -= ts;
-                }
 
                 remaining -= shm_buf->size;
 
@@ -340,16 +293,17 @@ struct shm_du_buff * shm_create_du_buff(struct shm_du_map * dum,
 #ifdef CONFIG_OUROBOROS_DEBUG
                         memset(write_pos, 0, sdb->du_head);
 #endif
-
                         write_pos += sdb->du_head;
                         bytes_to_copy -= sdb->du_head;
                 }
 
                 if (prev_index != -1)
-                        idx_to_block_ptr(dum, prev_index)->next = index;
-
-                memcpy(write_pos, read_pos, bytes_to_copy);
+                        idx_to_block_ptr(dum, prev_index)->next =
+                                *(dum->ptr_head);
 
+                if (len > 0) {
+                        memcpy(write_pos, read_pos, bytes_to_copy);
+                }
                 read_pos += bytes_to_copy;
 #ifdef CONFIG_OUROBOROS_DEBUG
                 if (remaining == 0) {
@@ -360,52 +314,38 @@ struct shm_du_buff * shm_create_du_buff(struct shm_du_map * dum,
                 shm_buf->next = -1;
                 shm_buf->prev = prev_index;
 
-                prev_index = index;
+                prev_index = *dum->ptr_head;
+
+                *(dum->ptr_head) = (*dum->ptr_head + 1) % SHM_BLOCKS_IN_MAP;
         }
 
+        pthread_mutex_unlock(dum->shm_mutex);
+
         return sdb;
 }
 
-void shm_release_du_buff(struct shm_du_map * dum, struct shm_du_buff * sdb)
+int shm_release_du_buff(struct shm_du_map * dum)
 {
-        long index = 0;
-
-        if (sdb == NULL) {
-                LOG_DBGF("Bogus input, bugging out.");
-                return;
-        }
-
-        if (sdb < dum->ptr_du_buff || sdb > dum->ptr_du_buff + SHM_BUFFS_SIZE) {
-                LOG_DBGF("Refused to free sdb outside of region.");
-                return;
-        }
-
-        if ((sdb - dum->ptr_du_buff) % sizeof *sdb != 0) {
-                LOG_DBGF("Refused to free sdb at incorrect offset.");
-                return;
-        }
-
-        index = du_buff_ptr_to_idx(dum, sdb);
+        int released = 0;
 
         pthread_mutex_lock(dum->shm_mutex);
 
-        if (dum->ptr_map[index] == 0) {
-                LOG_DBGF("Attempt to free unused sdb. Nothing to do.");
+        if (*dum->ptr_head == *dum->ptr_tail) {
+                LOG_DBGF("Attempt to free empty ringbuffer. Nothing to do.");
                 pthread_mutex_unlock(dum->shm_mutex);
-                return;
+                return -1;
         }
 
-        /* release all blocks in the structure */
-        dum->ptr_map[index] = 0;
-
-        index = idx_to_block_ptr(dum, index)->next;
-        while (index != -1) {
-                dum->ptr_map[index] = 0;
-                index = idx_to_block_ptr(dum, index)->next;
+        while (idx_to_block_ptr(dum, *dum->ptr_tail)->next != -1) {
+                *(dum->ptr_tail) = (*dum->ptr_tail + 1) % SHM_BLOCKS_IN_MAP;
+                released++;
         }
 
+        *(dum->ptr_tail) = (*dum->ptr_tail + 1) % SHM_BLOCKS_IN_MAP;
+
         pthread_mutex_unlock(dum->shm_mutex);
 
+        return 0;
 }
 
 uint8_t * shm_du_buff_head_alloc(struct shm_du_map * dum,
diff --git a/src/lib/tests/shm_du_map_test_du_buff_create.c b/src/lib/tests/shm_du_map_test_du_buff_create.c
index 9f66b20c..6787a1bc 100644
--- a/src/lib/tests/shm_du_map_test_du_buff_create.c
+++ b/src/lib/tests/shm_du_map_test_du_buff_create.c
@@ -54,7 +54,9 @@ int shm_du_map_test_du_buff_create(int argc, char ** argv)
         j_inc = MAX(1, SHM_DU_BLOCK_DATA_SIZE / 8);
         k_inc = MAX(1, SHM_DU_BLOCK_DATA_SIZE / 16);
 
-        for (i = SHM_DU_BUFF_BLOCK_SIZE / 4; i <= TEST_BUFF_SIZE; i += i_inc) {
+        for (i = SHM_DU_BLOCK_DATA_SIZE / 4;
+             i <= TEST_BUFF_SIZE;
+             i += i_inc) {
                 for (j = 0; j < i; j += j_inc) {
                         for (k = 0; k < i - j; k += k_inc) {
                                 if (k > SHM_DU_BLOCK_DATA_SIZE)
@@ -73,7 +75,7 @@ int shm_du_map_test_du_buff_create(int argc, char ** argv)
                                         shm_du_map_close(dum);
                                         return -1;
                                 }
-                                shm_release_du_buff(dum, dub);
+                                shm_release_du_buff(dum);
                         }
                 }
         }
diff --git a/src/lib/tests/shm_du_map_test_prod_cons.c b/src/lib/tests/shm_du_map_test_prod_cons.c
index 945104c1..ff9dfb0a 100644
--- a/src/lib/tests/shm_du_map_test_prod_cons.c
+++ b/src/lib/tests/shm_du_map_test_prod_cons.c
@@ -27,12 +27,12 @@
 #include <pthread.h>
 #include "shm_du_map.c"
 
-#define TEST_BUFF_SIZE (3 * SHM_DU_BLOCK_DATA_SIZE)
+#define TEST_BUFF_SIZE (SHM_DU_BLOCK_DATA_SIZE)
 
 #define MAX(a,b) (a > b ? a : b)
 #define MIN(a,b) (a < b ? a : b)
 
-int * sync;
+int sync;
 
 void * produce()
 {
@@ -46,31 +46,26 @@ void * produce()
         struct timespec     starttime;
         struct timespec     stoptime;
         double              elapsed;
-        long                overruns = 0;
 
         dum = shm_du_map_open();
         if (dum == NULL)
                 return (void *)-1;
 
         srand(time(NULL));
-        clock_gettime(CLOCK_MONOTONIC, &starttime);
 
         test_values = malloc (sizeof *test_values * TEST_BUFF_SIZE);
         for (i = 0; i < TEST_BUFF_SIZE; i++)
                 test_values[i] = 170;
 
-        for (i = 0; i < 4 * SHM_BLOCKS_IN_MAP; i++) {
+        clock_gettime(CLOCK_MONOTONIC, &starttime);
+        for (i = 0; i < SHM_BLOCKS_IN_MAP; i++) {
                 struct shm_du_buff * sdb;
                 size_t               len;
-                struct timespec      ts;
-
-                test_buf_size = rand() % (TEST_BUFF_SIZE - 512) + 512;
 
-                headspace     = MAX(4, rand() % 64);
-                tailspace     = MAX(1, rand() % 24);
+                test_buf_size = TEST_BUFF_SIZE;
 
-                ts.tv_sec     = 0;
-                ts.tv_nsec    = rand() % 90000;
+                headspace     = 32;
+                tailspace     = 8;
 
                 len = test_buf_size - (headspace + tailspace);
 
@@ -81,23 +76,12 @@ void * produce()
                                          len);
 
                 if (sdb != NULL) {
-                        sync[i] = du_buff_ptr_to_idx(dum, sdb);
                         bytes_written += len;
                 }
                 else {
-                        i--;
-                        ++overruns;
-                        ts.tv_nsec = 10000;
-                        nanosleep(&ts, NULL);
-                }
-                nanosleep(&ts, NULL);
-
-                if (overruns > 100) {
-                        LOG_INFO("Bugging out due to overruns.");
-                        sync[i+1] = -2;
+                        sync = -2;
                         break;
                 }
-
         }
 
         clock_gettime(CLOCK_MONOTONIC, &stoptime);
@@ -110,6 +94,9 @@ void * produce()
 
         free(test_values);
         shm_du_map_close(dum);
+
+        sync = -1;
+
         return 0;
 }
 
@@ -117,24 +104,19 @@ void * consume()
 {
         struct shm_du_map * dum;
 
-        long                i;
-
         struct timespec     ts;
 
         ts.tv_sec = 0;
-        ts.tv_nsec = 5000;
+        ts.tv_nsec = 1000;
 
         dum = shm_du_map_open();
 
         if (dum == NULL)
                 pthread_exit((void *) -1);
 
-        for (i = 0; i < 4 * SHM_BLOCKS_IN_MAP; i++) {
-                while (sync[i] == -1)
-                        nanosleep(&ts, NULL); /* wait for the producer */
-                if (sync[i] == -2)
-                        break;
-                shm_release_du_buff(dum, idx_to_du_buff_ptr(dum, sync[i]));
+        while (!sync) {
+                while (!shm_release_du_buff(dum));
+                nanosleep(&ts, NULL);
         }
 
         shm_du_map_close(dum);
@@ -148,11 +130,8 @@ int shm_du_map_test_prod_cons(int argc, char ** argv)
 
         int res1;
 
-        int i;
-
         pthread_t producer;
         pthread_t consumer;
-
         shm_unlink(SHM_DU_MAP_FILENAME);
 
         dum = shm_du_map_create();
@@ -160,10 +139,7 @@ int shm_du_map_test_prod_cons(int argc, char ** argv)
         if (dum == NULL)
                 return -1;
 
-        sync = malloc(sizeof *sync * 4 * SHM_BLOCKS_IN_MAP);
-
-        for (i = 0; i < 4 * SHM_BLOCKS_IN_MAP; i++)
-                sync[i] = -1;
+        sync = 0;
 
         res1 = (int) pthread_create(&producer, NULL, produce, NULL);
         pthread_create(&consumer, NULL, consume, NULL);
@@ -171,8 +147,6 @@ int shm_du_map_test_prod_cons(int argc, char ** argv)
         pthread_join(producer, NULL);
         pthread_join(consumer, NULL);
 
-        free(sync);
-
         shm_du_map_close(dum);
 
         return res1;