Add ring buffer edge IPC.

This patch adds a ring buffer IPC to implement very fast data
passing between nodes within the same address space (same process).
The ring buffer can be used effectively to pass a primative type
value (e.g., int or pointer) or struct between producer and consumer.

There are restrictions:
1) Producer and consumer must be within the same process.
2) The same amount of data must be produced and consumed each
   node invocation. That is, a producer cannot write 4 bytes
   one invocation and then 8 bytes the next.

TODO: Support ring buffer IPC on shared memory (allow producer
   and consumer to be in different processes).

5 files changed, 694 insertions, 28 deletions

diff --git a/Makefile b/Makefile
index 571eedb..8bf0591 100644
--- a/Makefile
+++ b/Makefile
@@ -23,7 +23,7 @@ LIBPGM = .
 # compiler flags
 flags-std      = -std=gnu++11
 flags-optim    = -O2 -march=native
-flags-debug    = -Wall -Werror -Wno-unused-function
+flags-debug    = -Wall -Werror -Wno-unused-function -Wno-sign-compare
 flags-api      = -D_XOPEN_SOURCE=600 -D_GNU_SOURCE -pthread
 
 # Comment out 'flags-litmus' to disable Litmus support
@@ -81,7 +81,7 @@ AR  := ${CROSS_COMPILE}${AR}
 # Targets
 
 all     = lib ${tools}
-tools   = cvtest basictest datapassingtest sockstreamtest pingpong depthtest pgmrt
+tools   = cvtest ringtest basictest datapassingtest sockstreamtest pingpong depthtest pgmrt
 
 .PHONY: all lib clean dump-config TAGS tags cscope help
 
@@ -130,6 +130,9 @@ vpath %.cpp tools
 obj-cvtest = cvtest.o
 lib-cvtest = -lpthread -lm -lrt -lboost_graph -lboost_filesystem -lboost_system ${liblitmus-flags}
 
+obj-ringtest = ringtest.o
+lib-ringtest = -lpthread -lm -lrt -lboost_graph -lboost_filesystem -lboost_system ${liblitmus-flags}
+
 obj-basictest = basictest.o
 lib-basictest = -lpthread -lm -lrt -lboost_graph -lboost_filesystem -lboost_system ${liblitmus-flags}
 
@@ -146,7 +149,7 @@ obj-pingpong = pingpong.o
 lib-pingpong = -lpthread -lm -lrt -lboost_graph -lboost_system -lboost_thread ${liblitmus-flags}
 
 obj-depthtest = depthtest.o
-lib-depthtest = -lm -lrt -lboost_graph -lboost_filesystem -lboost_system ${liblitmus-flags}
+lib-depthtest = -lpthread -lm -lrt -lboost_graph -lboost_filesystem -lboost_system ${liblitmus-flags}
 
 # ##############################################################################
 # Build everything that depends on liblitmus.
diff --git a/include/pgm.h b/include/pgm.h
index efd85f8..d570faf 100644
--- a/include/pgm.h
+++ b/include/pgm.h
@@ -24,7 +24,7 @@
 #define __PGM_EDGE_CV          0x00000001
 #define __PGM_EDGE_FIFO        0x00000002
 #define __PGM_EDGE_MQ          0x00000004
-#define __PGM_EDGE_CV_BUF      0x00000008 /* reserved */
+#define __PGM_EDGE_RING        0x00000008
 #define __PGM_EDGE_SOCK_STREAM 0x00000010
 
 typedef enum
@@ -34,6 +34,9 @@ typedef enum
 
         /* condition variable-based IPC */
         pgm_cv_edge             = (__PGM_EDGE_CV   | __PGM_SIGNALED),
+        /* Very low overhead IPC for passing POD data between nodes.
+           Currently limited to nodes within the same process. */
+        pgm_ring_edge   = (__PGM_EDGE_RING | __PGM_SIGNALED | __PGM_DATA_PASSING),
         /* named FIFO IPC */
         pgm_fast_fifo_edge      = (__PGM_EDGE_FIFO | __PGM_SIGNALED | __PGM_DATA_PASSING),
         /* POSIX message queue IPC */
@@ -90,6 +93,13 @@ typedef struct pgm_edge_attr
         /* edge-type specific params */
         union
         {
+                struct /* Ring buffer params */
+                {
+                        /* nr_produce/nr_consume interpreted as size of members */
+
+                        /* Number of elements in ring buffer */
+                        size_t nmemb;
+                };
                 struct /* POSIX message queue params */
                 {
                         /* Values > 10 may require root privilege. */
diff --git a/include/ring.h b/include/ring.h
new file mode 100644
index 0000000..2800bb2
--- /dev/null
+++ b/include/ring.h
@@ -0,0 +1,321 @@
+// Copyright (c) 2014, Glenn Elliott
+// All rights reserved.
+
+#pragma once
+
+#include <stdlib.h>
+#include <string.h>
+
+#ifndef __cplusplus
+
+#define nleading_unset_bits(x) \
+__builtin_choose_expr((sizeof(x) == 8), \
+        __builtin_clzl(x), \
+        __builtin_clz(x))
+
+#else
+
+/*
+  g++ doesn't have __builtin_choose_expr().
+  Switch between types using meta programming...
+*/
+namespace __ring
+{
+        template<bool c, typename T, typename F>
+        struct conditional { typedef T type; };
+        template<typename T, typename F>
+        struct conditional<false, T, F> { typedef F type; };
+
+        struct clz32
+        {
+                template <class T>
+                static inline int clz(T x)
+                {
+                        return __builtin_clz(x);
+                }
+        };
+
+        struct clz64
+        {
+                template <class T>
+                static inline int clz(T x)
+                {
+                        return __builtin_clzl(x);
+                }
+        };
+}
+#define nleading_unset_bits(x) \
+        __ring::conditional<(sizeof(x) == 8), __ring::clz64, __ring::clz32>::type::clz(x)
+
+#endif
+
+typedef enum
+{
+        SLOT_FREE = 0,  /* must be zero */
+        SLOT_READY
+} slot_state_t;
+
+struct ring
+{
+        size_t nmemb;
+        size_t memb_sz;
+        volatile size_t nfree;
+
+        size_t widx;
+        size_t ridx;
+
+        char* slots;
+        char* buf;
+
+        char user_managed_buffers;
+};
+
+/*
+   Initialize a ring buffer struct.
+     [in] r: Pointer to ring buffer instance.
+     [in] min_count: Minimum number of elements in ring buffer.
+              (value is rounded UP to nearest power of two)
+     [in] size: size of ring buffer element.
+
+   Return: 0 on success. -1 on error.
+ */
+static int init_ring(struct ring* r, size_t min_count, size_t size)
+{
+        size_t count;
+
+        if (!r || min_count == 0 || size == 0)
+                return -1;
+
+        /* round min_count up to the nearest power of two */
+        count = (min_count > 2) ?
+                (~((~(size_t)0)>>1)) >> (nleading_unset_bits(min_count - 1) - 1) :
+                min_count;
+
+        /* overflow! too big! */
+        if (count == 0)
+                return -1;
+
+        r->nmemb = count;
+        r->memb_sz = size;
+        r->nfree = count;
+        r->widx = 0;
+        r->ridx = 0;
+        /* calloc() initializes slots to SLOT_FREE */
+        r->slots = (char*)calloc(count, sizeof(char));
+        r->buf = (char*)malloc(count*size);
+
+        r->user_managed_buffers = 0;
+
+        return 0;
+}
+
+/*
+  Varient to allow user to specify their own (possibly static) buffers.
+  Assumptions:
+    1) slot_buf and data_buf have been allocated and are of sufficient size.
+    2) 'count' is a power of two.
+*/
+static void __init_ring(struct ring* r, size_t count, size_t size,
+        char* slot_buf, void* data_buf)
+{
+        r->nmemb = count;
+        r->memb_sz = size;
+        r->nfree = count;
+        r->widx = 0;
+        r->ridx = 0;
+        r->slots = slot_buf;
+        r->buf = (char*)data_buf;
+
+        /* init slots to SLOT_FREE */
+        memset(r->slots, 0, count);
+
+        r->user_managed_buffers = 1;
+}
+
+/*
+   Free ring buffer resources.
+ */
+static void free_ring(struct ring* r)
+{
+        if (!r)
+                return;
+        if (r->user_managed_buffers)
+                return;
+
+        free(r->slots);
+        free(r->buf);
+}
+
+static inline int is_ring_empty(struct ring* r)
+{
+        return (r->nfree == r->nmemb);
+}
+
+static inline int is_ring_full(struct ring* r)
+{
+        return (r->nfree == 0);
+}
+
+#if 0
+/* multi-writer varients for adding elements to ring buf */
+static inline void* __begin_write_ring(struct ring* r)
+{
+        ssize_t nfree;
+        size_t idx;
+        void* dst;
+
+        if(*((volatile ssize_t*)(&r->nfree)) <= 0)
+                return NULL;
+
+        nfree = __sync_fetch_and_sub(&r->nfree, 1);
+        if (nfree <= 0)
+        {
+                __sync_fetch_and_add(&r->nfree, 1);
+                return NULL;
+        }
+
+        /* roll over idx */
+        idx = __sync_fetch_and_add(&r->widx, 1) % r->nmemb;
+        dst = r->buf + idx*r->memb_sz;
+
+        return (void*)dst;
+}
+
+static inline void __end_ring_write(struct ring* r, void* addr)
+{
+        size_t idx = ((char*)addr - r->buf) / r->memb_sz;
+        r->slots[idx] = SLOT_READY;
+        __sync_synchronize();
+}
+#endif
+
+
+/* single-writer varients */
+
+static inline void* __begin_write_ring(struct ring* r)
+{
+        size_t idx;
+        void* dst;
+
+        if(r->nfree <= 0)
+                return NULL;
+
+        /* roll over idx */
+        idx = r->widx++ % r->nmemb;
+        dst = r->buf + idx*r->memb_sz;
+
+        return (void*)dst;
+}
+
+static inline void __end_ring_write(struct ring* r, void* addr)
+{
+        size_t idx = ((char*)addr - r->buf) / r->memb_sz;
+        r->slots[idx] = SLOT_READY;
+        __sync_fetch_and_sub(&r->nfree, 1); /* memory barrier */
+}
+
+
+/* single-reader varients */
+
+static inline void* __begin_read_ring(struct ring* r)
+{
+        size_t idx;
+
+        if (r->nfree == r->nmemb)
+                return NULL;
+
+        idx = r->ridx % r->nmemb;
+        if (*(volatile char*)&(r->slots[idx]) == SLOT_READY)
+                return r->buf + idx * r->memb_sz;
+        else
+                return NULL;
+}
+
+static inline void __end_read_ring(struct ring* r, void* addr)
+{
+        size_t idx = ((char*)addr - r->buf) / r->memb_sz;
+        r->slots[idx] = SLOT_FREE;
+        r->ridx++;
+        __sync_fetch_and_add(&r->nfree, 1); /* memory barrier */
+}
+
+
+#ifdef NDEBUG
+#define check_size(r, ptr) \
+        do { assert(sizeof(*ptr) == r->memb_sz); } while(0)
+#define check_size_vec(r, sz) \
+        do { assert(sz == r->memb_sz); } while(0)
+#else
+#define check_size(r, size)
+#define check_size_vec(r, sz)
+#endif
+
+/*
+   The following macros are used for enqueuing and dequeuing data from
+   the provided ring buffer
+
+   BEWARE!!!! Reads and writes are done by simple assignment (by the '='
+   operator) for the sake of performance (vs using memcpy). Types are
+   deduced from the second parameter to each macro. You must be consistent
+   with your types. For example, mixing int32_t with int64_t will result
+   in corrupted ring buffers.
+
+   ----> Be very careful when passing literals!!! <----
+
+   TODO: Make C++-template versions for safer typing.
+*/
+
+/*
+   Macro for enqueuing an element to the ring buffer.
+   [in] r: Pointer to struct ring
+   [in] src: Value to write (not pointer to value!)
+ */
+#define write_ring(r, src) \
+do{ \
+        struct ring* __r = (r); \
+        typeof((src))* __dst; \
+        check_size(__r, __dst); \
+        do { __dst = (typeof(__dst)) __begin_write_ring(__r); } \
+                while (__dst == NULL); \
+        *__dst = (src); \
+        __end_ring_write(__r, __dst); \
+}while(0)
+
+/*
+   Macro for dequeuing an element from the ring buffer.
+   [in] r: Pointer to struct ring
+   [in/out] dst_ptr: Pointer to where read element is to be written.
+ */
+#define read_ring(r, dst_ptr) \
+do{ \
+        struct ring* __r = (r); \
+        typeof((dst_ptr)) __src; \
+        check_size(__r, __src); \
+        do { __src = (typeof(__src)) __begin_read_ring(__r); } \
+                while (__src == NULL); \
+        *(dst_ptr) = *__src; \
+        __end_read_ring(__r, __src); \
+}while(0)
+
+
+/* Write/Read routines for plain vector/array types. These use memcpy. */
+
+#define write_vec_ring(r, src_vec, sz) \
+do{ \
+        struct ring* __r = (r); \
+        void* __dst; \
+        check_size_vec(__r, sz); \
+        do { __dst = __begin_write_ring(__r); } while (__dst == NULL); \
+        memcpy(__dst, src_vec, sz); \
+        __end_ring_write(__r, __dst); \
+}while(0)
+
+#define read_vec_ring(r, dst_vec, sz) \
+do{ \
+        struct ring* __r = (r); \
+        void* __src; \
+        check_size_vec(__r, sz); \
+        do { __src = __begin_read_ring(__r); } while (__src == NULL); \
+        memcpy(dst_vec, __src, sz); \
+        __end_read_ring(__r, __src); \
+}while(0)
diff --git a/src/pgm.cpp b/src/pgm.cpp
index c841fbc..8057c36 100644
--- a/src/pgm.cpp
+++ b/src/pgm.cpp
@@ -37,6 +37,8 @@ typedef ticketlock_t    pgm_lock_t;
 typedef cv_t            pgm_cv_t;
 #endif
 
+#include "ring.h"
+
 using namespace std;
 using namespace boost;
 using namespace boost::interprocess;
@@ -141,21 +143,34 @@ struct pgm_edge
 
         // the remaining fields are used by data-passing edges
 
-        // fd_out and fd_in may be the same
-        // if different ends of FIFOs are
-        // opened by different processes.
-        int fd_out;
-        int fd_in;
+        union
+        {
+                // fields for standard data-passing IPCs
+                struct
+                {
+                        // fd_out and fd_in may be the same
+                        // if different ends of FIFOs are
+                        // opened by different processes.
+                        int fd_out;
+                        int fd_in;
+
+                        // counter for determining location of the message
+                        // header contained within received data.
+                        size_t next_tag;
+                };
+                // fields for ring buffer IPC
+                struct
+                {
+                        struct ring ringbuf;
+                        volatile pgm_command_t ring_cmd;
+                };
+        };
 
         // buffer for sending data
         struct pgm_memory_hdr* buf_out;
 
         // buffer for receiving data
         struct pgm_memory_hdr* buf_in;
-
-        // counter for determining location of the message
-        // header contained within received data.
-        size_t next_tag;
 };
 
 static inline bool is_signal_driven(const struct pgm_edge_attr* attr)
@@ -267,6 +282,126 @@ static ssize_t dummy_edge_write(struct pgm_edge* e, const void* buf, size_t nbyt
         return 0;
 }
 
+/************* RING IPC ROUTINES *****************/
+static int ring_init(pgm_graph* g,
+                                pgm_node* producer, pgm_node* consumer,
+                                pgm_edge* edge)
+{
+        int ret = -1;
+
+        /* nr_produce/nr_consume interpreted as bytes passed between
+           producer and consumer, per invocation. Impl. requires that
+           nr_produce == nr_consume. */
+        if (edge->attr.nr_produce != edge->attr.nr_consume)
+                goto out;
+
+        ret = init_ring(&edge->ringbuf, edge->attr.nmemb, edge->attr.nr_produce);
+
+out:
+        return ret;
+}
+
+static int ring_open_consumer(pgm_graph* g,
+                                pgm_node* producer, pgm_node* consumer,
+                                pgm_edge* edge)
+{
+        edge->buf_in = __pgm_malloc_edge_buf(g, edge, false);
+        return 0;
+}
+
+static int ring_open_producer(pgm_graph* g,
+                                pgm_node* producer, pgm_node* consumer,
+                                pgm_edge* edge)
+{
+        edge->buf_out = __pgm_malloc_edge_buf(g, edge, true);
+        return 0;
+}
+
+static int ring_close_consumer(pgm_edge* edge)
+{
+        free(edge->buf_in);
+        edge->buf_in = 0;
+        return 0;
+}
+
+static int ring_close_producer(pgm_edge* edge)
+{
+        free(edge->buf_out);
+        edge->buf_out = 0;
+        return 0;
+}
+
+static int ring_destroy(pgm_graph* g,
+                                pgm_node* producer, pgm_node* consumer,
+                                pgm_edge* edge)
+{
+        free_ring(&edge->ringbuf);
+        return 0;
+}
+
+static ssize_t ring_read(struct pgm_edge* e, void* buf, size_t nbytes)
+{
+        assert(nbytes == e->attr.nr_produce);
+        assert(nbytes == e->attr.nr_consume);
+
+        switch(nbytes)
+        {
+        case 8:
+                read_ring(&e->ringbuf, (uint64_t*)buf);
+                break;
+        case 4:
+                read_ring(&e->ringbuf, (uint32_t*)buf);
+                break;
+        case 2:
+                read_ring(&e->ringbuf, (uint16_t*)buf);
+                break;
+        case 1:
+                read_ring(&e->ringbuf, (uint8_t*)buf);
+                break;
+        default:
+                read_vec_ring(&e->ringbuf, buf, nbytes);
+        }
+        return nbytes; /* assume always successful */
+}
+
+static ssize_t ring_write(struct pgm_edge* e, const void* buf, size_t nbytes)
+{
+        assert(nbytes == e->attr.nr_produce);
+        assert(nbytes == e->attr.nr_consume);
+
+        switch(nbytes)
+        {
+        case 8:
+                write_ring(&e->ringbuf, *(uint64_t*)buf);
+                break;
+        case 4:
+                write_ring(&e->ringbuf, *(uint32_t*)buf);
+                break;
+        case 2:
+                write_ring(&e->ringbuf, *(uint16_t*)buf);
+                break;
+        case 1:
+                write_ring(&e->ringbuf, *(uint8_t*)buf);
+                break;
+        default:
+                write_vec_ring(&e->ringbuf, buf, nbytes);
+        }
+        return nbytes; /* assume always successful */
+}
+
+static const struct pgm_edge_ops pgm_ring_edge_ops =
+{
+        .init = ring_init,
+        .open_consumer = ring_open_consumer,
+        .open_producer = ring_open_producer,
+        .close_consumer = ring_close_consumer,
+        .close_producer = ring_close_producer,
+        .destroy = ring_destroy,
+        .read = ring_read,
+        .write = ring_write,
+};
+
+
 /************* FIFO IPC ROUTINES *****************/
 static std::string fifo_name(pgm_graph* g,
                                 pgm_node* producer, pgm_node* consumer,
@@ -858,7 +993,7 @@ static inline int is_producer_buf(void* userptr)
         if(!ptr)
                 return 0;
         if(ptr->assigned_edge.graph == BAD_EDGE.graph &&
-       ptr->assigned_edge.edge == BAD_EDGE.edge)
+           ptr->assigned_edge.edge == BAD_EDGE.edge)
                 return 0;
         return (ptr->producer_flag == 1);
 }
@@ -869,7 +1004,7 @@ static inline int is_consumer_buf(void* userptr)
         if(!ptr)
                 return 0;
         if(ptr->assigned_edge.graph == BAD_EDGE.graph &&
-       ptr->assigned_edge.edge == BAD_EDGE.edge)
+           ptr->assigned_edge.edge == BAD_EDGE.edge)
                 return 0;
         return (ptr->producer_flag == 0);
 }
@@ -880,7 +1015,7 @@ static inline int is_buf_assigned(void* userptr, edge_t* e = NULL)
         if(!ptr)
                 return 0;
         if(ptr->assigned_edge.graph == BAD_EDGE.graph &&
-       ptr->assigned_edge.edge == BAD_EDGE.edge)
+           ptr->assigned_edge.edge == BAD_EDGE.edge)
                 return 0;
         if(e)
                 *e = ptr->assigned_edge;
@@ -1081,7 +1216,7 @@ void* __pgm_swap_edge_buf(edge_t edge, void* new_uptr, bool swap_producer)
                 goto out;
         }
         if(hdr->assigned_edge.graph != BAD_EDGE.graph &&
-       hdr->assigned_edge.edge != BAD_EDGE.edge)
+           hdr->assigned_edge.edge != BAD_EDGE.edge)
         {
                 E("%p is already in use by an edge.\n", new_uptr);
                 goto out;
@@ -1093,7 +1228,7 @@ void* __pgm_swap_edge_buf(edge_t edge, void* new_uptr, bool swap_producer)
         if(hdr->usersize != old_hdr->usersize)
         {
                 E("Buffer %p is the wrong size. (is %lu, expected %lu)\n",
-          new_uptr, hdr->usersize, old_hdr->usersize);
+                  new_uptr, hdr->usersize, old_hdr->usersize);
                 goto out;
         }
 
@@ -1151,7 +1286,7 @@ int pgm_swap_edge_bufs(void* a, void* b)
         if(hdra->usersize != hdrb->usersize)
         {
                 E("Buffers are not the same size: %p:%lu vs %p:%lu\n",
-          a, hdra->usersize, b, hdrb->usersize);
+                  a, hdra->usersize, b, hdrb->usersize);
                 goto out;
         }
 
@@ -1731,6 +1866,19 @@ int pgm_init_edge(edge_t* edge,
                 E("Produce amnt. must equal consume amnt. for POSIX msg queues.\n");
                 goto out;
         }
+        if(attr->type & __PGM_EDGE_RING)
+        {
+                if(attr->nr_produce != attr->nr_consume)
+                {
+                        E("Produce amnt. must equal consume amnt. for ring buffers.\n");
+                        goto out;
+                }
+                if(attr->nmemb == 0)
+                {
+                        E("nmemb cannot be zero.\n");
+                        goto out;
+                }
+        }
 
         if(attr->nr_threshold < attr->nr_consume)
                 goto out;
@@ -1784,14 +1932,18 @@ int pgm_init_edge(edge_t* edge,
         e->consumer = consumer.node;
         e->attr = *attr;
 
-        if(attr->type & __PGM_EDGE_FIFO)
+        if     (attr->type & __PGM_EDGE_CV)
+                e->ops = &pgm_cv_edge_ops;
+        else if(attr->type & __PGM_EDGE_FIFO)
                 e->ops = &pgm_fifo_edge_ops;
         else if(attr->type & __PGM_EDGE_MQ)
                 e->ops = &pgm_mq_edge_ops;
+        else if(attr->type & __PGM_EDGE_RING)
+                e->ops = &pgm_ring_edge_ops;
         else if(attr->type & __PGM_EDGE_SOCK_STREAM)
                 e->ops = &pgm_sock_stream_edge_ops;
         else
-                e->ops = &pgm_cv_edge_ops;
+                goto out_unlock;
 
         ret = e->ops->init(g, np, nc, e);
 
@@ -2738,7 +2890,7 @@ static eWaitStatus pgm_wait_for_tokens(struct pgm_graph* g, struct pgm_node* n)
         pgm_lock(&n->lock, flags);
         do
         {
-            // recheck the condition
+                // recheck the condition
                 nr_ready = pgm_nr_ready_edges(g, n);
                 if(nr_ready == n->nr_in_signaled)
                         break;
@@ -2760,9 +2912,9 @@ static void pgm_consume_tokens(struct pgm_graph* g, struct pgm_node* n)
 {
         for(int i = 0; i < n->nr_in; ++i)
         {
-            struct pgm_edge* e = &g->edges[n->in[i]];
+                struct pgm_edge* e = &g->edges[n->in[i]];
                 if(is_signal_driven(e))
-                    __sync_fetch_and_sub(&e->nr_pending, e->attr.nr_consume);
+                        __sync_fetch_and_sub(&e->nr_pending, e->attr.nr_consume);
         }
 }
 
@@ -2790,7 +2942,7 @@ typedef uint32_t pgm_fd_mask_t;
 
 static const unsigned char PGM_NORMAL = 0x01;
 
-static int pgm_send_data(struct pgm_edge* e, pgm_command_t tag = PGM_NORMAL)
+static int pgm_send_std_data(struct pgm_edge* e, pgm_command_t tag)
 {
         // only the tag is sent if this is a terminate message
 
@@ -2834,6 +2986,23 @@ static int pgm_send_data(struct pgm_edge* e, pgm_command_t tag = PGM_NORMAL)
         return ret;
 }
 
+static int pgm_send_ring_data(struct pgm_edge* e, pgm_command_t tag)
+{
+        if(!(tag & PGM_TERMINATE))
+                e->ops->write(e, pgm_get_user_ptr(e->buf_out), e->attr.nr_produce);
+        else
+                e->ring_cmd = tag;
+        return 0;
+}
+
+static int pgm_send_data(struct pgm_edge* e, pgm_command_t tag = PGM_NORMAL)
+{
+        if(!(e->attr.type & __PGM_EDGE_RING))
+                return pgm_send_std_data(e, tag);
+        else
+                return pgm_send_ring_data(e, tag);
+}
+
 static eWaitStatus pgm_wait_for_data(pgm_fd_mask_t* to_wait,
                                 struct pgm_graph* g, struct pgm_node* n)
 {
@@ -2880,7 +3049,7 @@ static eWaitStatus pgm_wait_for_data(pgm_fd_mask_t* to_wait,
                                 ++scanned;
                         }
                 }
-            ++num_looped;
+                ++num_looped;
         }
 
         return wait_status;
@@ -2946,6 +3115,16 @@ wait_for_data: // jump here if we would block on read
                 if(!is_data_passing(e))
                         continue;
 
+                if(e->attr.type & __PGM_EDGE_RING)
+                {
+                        /* short-cut for the simple ring buffer IPC */
+                        if(!((e->ring_cmd & PGM_TERMINATE) && is_ring_empty(&e->ringbuf)))
+                                e->ops->read(e, dest_ptrs[i], e->attr.nr_consume);
+                        else
+                                n->nr_terminate_msgs++;
+                        continue;
+                }
+
 read_more: // jump to here if we need to read more bytes into our buffer
 
                 remaining = e->attr.nr_consume - (dest_ptrs[i] - (char*)pgm_get_user_ptr(e->buf_in));
@@ -3094,14 +3273,14 @@ int pgm_wait(node_t node)
         }
 
         if(token_status == WaitExhaustedAndTerminate &&
-                        data_status == WaitExhaustedAndTerminate)
+           data_status == WaitExhaustedAndTerminate)
         {
                 ret = PGM_TERMINATE;
         }
         else
         {
                 ret = (token_status == WaitTimeout || token_status == WaitError ||
-                           data_status  == WaitTimeout || data_status  == WaitError) ?
+                       data_status  == WaitTimeout || data_status  == WaitError) ?
                                 -1 : 0;
         }
 
diff --git a/tools/ringtest.cpp b/tools/ringtest.cpp
new file mode 100644
index 0000000..d6850ed
--- /dev/null
+++ b/tools/ringtest.cpp
@@ -0,0 +1,153 @@
+// Copyright (c) 2014, Glenn Elliott
+// All rights reserved.
+
+/* A program for testing the basic ring-based edge. */
+
+#include <iostream>
+#include <unistd.h>
+#include <errno.h>
+#include <pthread.h>
+#include <string.h>
+#include <stdlib.h>
+
+#include "pgm.h"
+
+int errors = 0;
+pthread_barrier_t init_barrier;
+
+__thread char __errstr[80] = {0};
+
+#define CheckError(e) \
+do { int __ret = (e); \
+if(__ret < 0) { \
+        errors++; \
+        char* errstr = strerror_r(errno, __errstr, sizeof(errstr)); \
+        fprintf(stderr, "%lu: Error %d (%s (%d)) @ %s:%s:%d\n",  \
+                pthread_self(), __ret, errstr, errno, __FILE__, __FUNCTION__, __LINE__); \
+}}while(0)
+
+int TOTAL_ITERATIONS = 10000000;
+
+void* thread(void* _graph_t)
+{
+        char tabbuf[] = "\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t";
+        int iterations = 0;
+        int ret = 0;
+
+        const graph_t& graph = *((graph_t*)_graph_t);
+        node_t node;
+
+        CheckError(pgm_claim_any_node(graph, &node));
+        tabbuf[node.node] = '\0';
+
+        int out_degree = pgm_get_degree_out(node);
+        int in_degree = pgm_get_degree_in(node);
+        edge_t* out_edges = (edge_t*)calloc(out_degree, sizeof(edge_t));
+        edge_t* in_edges = (edge_t*)calloc(in_degree, sizeof(edge_t));
+
+        bool is_src = (in_degree == 0);
+        uint32_t* buf;
+        if(is_src)
+                buf = (uint32_t*)pgm_get_edge_buf_p(out_edges[0]);
+        else
+                buf = (uint32_t*)pgm_get_edge_buf_c(in_edges[0]);
+
+        uint64_t sum = 0;
+
+        pthread_barrier_wait(&init_barrier);
+
+        if(!errors)
+        {
+                do {
+                        if(!is_src)
+                        {
+                                ret = pgm_wait(node);
+                        }
+                        else
+                        {
+                                if(iterations > TOTAL_ITERATIONS)
+                                        ret = PGM_TERMINATE;
+                        }
+
+                        if(ret != PGM_TERMINATE)
+                        {
+                                CheckError(ret);
+
+                                if(is_src)
+                                {
+//                                      usleep(500*1000);
+//                                      fprintf(stdout, "%s%d fires\n", tabbuf, node.node);
+                                        *buf = (uint32_t)iterations;
+                                        sum += iterations;
+                                }
+                                else
+                                {
+//                                      fprintf(stdout, "%s%d fires. read:%d\n", tabbuf, node.node, *buf);
+                                        sum += *buf;
+
+                                        // slow down the consumer a little bit to induce backlog in token buffer
+                                        if(rand()%5 == 0)
+                                                sched_yield();
+                                }
+
+                                CheckError(pgm_complete(node));
+                                iterations++;
+                        }
+                        else
+                        {
+                                fprintf(stdout, "%s%d terminates: sum: %lu\n", tabbuf, node.node, sum);
+
+                                if(is_src)
+                                        CheckError(pgm_terminate(node));
+                        }
+
+                } while(ret != PGM_TERMINATE);
+        }
+
+        pthread_barrier_wait(&init_barrier);
+
+        CheckError(pgm_release_node(node));
+
+        free(out_edges);
+        free(in_edges);
+
+        pthread_exit(0);
+}
+
+int main(void)
+{
+        graph_t g;
+        node_t  n0, n1;
+        edge_t  e0_1;
+
+        pthread_t t0, t1;
+
+        edge_attr_t ring_attr;
+        memset(&ring_attr, 0, sizeof(ring_attr));
+        ring_attr.type = pgm_ring_edge;
+        ring_attr.nr_produce = sizeof(uint32_t);
+        ring_attr.nr_consume = sizeof(uint32_t);
+        ring_attr.nr_threshold = sizeof(uint32_t);
+        ring_attr.nmemb = 32;
+
+        CheckError(pgm_init("/tmp/graphs", 1));
+        CheckError(pgm_init_graph(&g, "demo"));
+
+        CheckError(pgm_init_node(&n0, g, "n0"));
+        CheckError(pgm_init_node(&n1, g, "n1"));
+
+        CheckError(pgm_init_edge(&e0_1, n0, n1, "e0_1", &ring_attr));
+
+        pthread_barrier_init(&init_barrier, 0, 1);
+        pthread_create(&t0, 0, thread, &g);
+        pthread_create(&t1, 0, thread, &g);
+
+        pthread_join(t0, 0);
+        pthread_join(t1, 0);
+
+        CheckError(pgm_destroy_graph(g));
+
+        CheckError(pgm_destroy());
+
+        return 0;
+}