TEsting ring buffer for live streaming

meson.build

@@ -27,7 +27,7 @@ uca_headers = include_directories('@0@/@1@/uca'.format(get_option('prefix'), 'in
 plugindir = uca_dep.get_pkgconfig_variable('plugindir')
 
 phantom = library('ucaphantom',
-    sources: ['uca-phantom-communicate.c', 'uca-phantom-camera.c'],
+    sources: ['ringbuf.c', 'uca-phantom-communicate.c', 'uca-phantom-camera.c'],
     dependencies: deps,
     include_directories: headers,
     install: true,

ringbuf.c

@@ -0,0 +1,215 @@
+/*
+ * ringbuf.c - C ring buffer (FIFO) implementation.
+ *
+ * Written in 2011 by Drew Hess <dhess-src@bothan.net>.
+ *
+ * To the extent possible under law, the author(s) have dedicated all
+ * copyright and related and neighboring rights to this software to
+ * the public domain worldwide. This software is distributed without
+ * any warranty.
+ *
+ * You should have received a copy of the CC0 Public Domain Dedication
+ * along with this software. If not, see
+ * <http://creativecommons.org/publicdomain/zero/1.0/>.
+ */
+
+#include "ringbuf.h"
+
+struct ringbuf_t {
+    gpointer buf, head, tail;
+    gsize element_size, total_size;
+    GMutex mutex;
+};
+
+ringbuf_t ringbuf_new (gsize size, guint capacity) {
+    ringbuf_t rb = g_new0(struct ringbuf_t, 1);
+    if (rb) {
+        /* One byte is used for detecting the full condition. */
+        rb->element_size = size;
+        rb->total_size = (capacity + 1) * size;
+        rb->buf = g_malloc(rb->total_size);
+        if (rb->buf)
+            ringbuf_reset(rb);
+        else {
+            g_free(rb);
+            return 0;
+        }
+        g_mutex_init(&rb->mutex);
+    }
+    return rb;
+}
+
+gsize ringbuf_buffer_size (const struct ringbuf_t *rb) {
+    return rb->total_size;
+}
+
+void ringbuf_reset (ringbuf_t rb) {
+    gpointer *current_buf = NULL;
+    g_mutex_lock(&rb->mutex);
+    current_buf = rb->buf;
+    rb->head = current_buf;
+    rb->tail = current_buf;
+    g_mutex_unlock(&rb->mutex);
+}
+
+void ringbuf_free (ringbuf_t *rb) {
+    g_assert(rb && *rb);
+    g_mutex_clear(&(*rb)->mutex);
+    g_free((*rb)->buf);
+    (*rb)->buf = NULL;
+    g_free(*rb);
+    *rb = NULL;
+}
+
+gsize ringbuf_capacity (const struct ringbuf_t *rb) {
+    return ringbuf_buffer_size(rb) - 1;
+}
+
+/*
+ * Return a pointer to one-past-the-end of the ring buffer's
+ * contiguous buffer. You shouldn't normally need to use this function
+ * unless you're writing a new ringbuf_* function.
+ */
+static gconstpointer ringbuf_end (const struct ringbuf_t *rb) {
+    return (guint8 *)(rb->buf) + ringbuf_buffer_size(rb);
+}
+
+gsize ringbuf_bytes_free (struct ringbuf_t *rb) {
+    gsize retval = 0;
+    const guint8 *tail = ringbuf_tail(rb);
+    const guint8 *head = ringbuf_head(rb);
+
+    if (head >= tail)
+        retval = ringbuf_capacity(rb) - (gsize)(head - tail);
+    else
+        retval =  (gsize)(tail - head) - 1;
+
+    return retval;
+}
+
+gsize ringbuf_bytes_used (struct ringbuf_t *rb) {
+    return ringbuf_capacity(rb) - ringbuf_bytes_free(rb);
+}
+
+gboolean ringbuf_is_full (struct ringbuf_t *rb) {
+    return (ringbuf_bytes_free(rb) == 0);
+}
+
+gboolean ringbuf_is_empty (struct ringbuf_t *rb) {
+    return (ringbuf_bytes_free (rb) == ringbuf_capacity (rb));
+}
+
+gconstpointer ringbuf_tail (struct ringbuf_t *rb) {
+    gpointer retval = NULL;
+    g_mutex_lock(&rb->mutex);
+    retval = rb->tail;
+    g_mutex_unlock(&rb->mutex);
+    return retval;
+}
+
+gconstpointer ringbuf_head (struct ringbuf_t *rb) {
+    gpointer retval = NULL;
+    g_mutex_lock(&rb->mutex);
+    retval = rb->head;
+    g_mutex_unlock(&rb->mutex);
+    return retval;
+}
+
+/*
+ * Given a ring buffer rb and a pointer to a location within its
+ * contiguous buffer, return the a pointer to the next logical
+ * location in the ring buffer.
+ */
+static gpointer ringbuf_nextp (ringbuf_t rb, gconstpointer p) {
+    /*
+     * The assert guarantees the expression (++p - rb->buf) is
+     * non-negative; therefore, the modulus operation is safe and
+     * portable.
+     */
+    guint8 *p_uint8 = (gpointer)p;
+    guint8 *buf_uint8 = rb->buf;
+    p_uint8 += rb->element_size;
+
+    g_return_val_if_fail ((p_uint8 >= buf_uint8) && (p_uint8 < (guint8 *)ringbuf_end(rb)), NULL);
+
+    return buf_uint8 + ((p_uint8 - buf_uint8) % ringbuf_buffer_size(rb));
+}
+
+gpointer ringbuf_memcpy_into(ringbuf_t dst, gconstpointer src, gsize count) {
+    const guint8 *u8src = src;
+    const guint8 *bufend = ringbuf_end(dst);
+    guint8 *dsthead = ringbuf_head(dst);
+
+    gboolean overflow = count > ringbuf_bytes_free(dst);
+    gsize nread = 0, n, diff = 0;
+
+    while (nread != count) {
+        /* don't copy beyond the end of the buffer */
+        if (bufend <= dsthead)
+            break;
+
+        diff = bufend - dsthead;
+        n = MIN(diff, count - nread);
+        memcpy(dsthead, u8src + nread, n);
+        dsthead += n;
+        nread += n;
+
+        /* wrap? */
+        if (dsthead == bufend)
+            dsthead = dst->buf;
+    }
+
+    g_mutex_lock(&dst->mutex);
+    dst->head = dsthead;
+    g_mutex_unlock(&dst->mutex);
+
+    if (overflow) {
+        /* mark the ring buffer as full */
+        g_mutex_lock(&dst->mutex);
+        dst->tail = ringbuf_nextp(dst, dsthead);
+        g_mutex_unlock(&dst->mutex);
+
+        if (!ringbuf_is_full(dst)) {
+            return NULL;
+        }
+    }
+
+    return dsthead;
+}
+
+gpointer ringbuf_memcpy_from (gpointer dst, ringbuf_t src, gsize count) {
+    gsize bytes_used = ringbuf_bytes_used(src), n = 0, diff = 0;
+    if (count > bytes_used) {
+        return NULL;
+    }
+
+    guint8 *u8dst = dst;
+    guint8 *bufend = ringbuf_end(src);
+    guint8 *tail = ringbuf_tail(src);
+
+    gsize nwritten = 0;
+    while (nwritten != count) {
+        if (bufend <= tail)
+            break;
+
+        diff = bufend - tail;
+        n = MIN(diff, count - nwritten);
+        memcpy(u8dst + nwritten, tail, n);
+        tail += n;
+        nwritten += n;
+
+        /* wrap ? */
+        if (tail == bufend)
+            tail = src->buf;
+    }
+
+    g_mutex_lock(&src->mutex);
+    src->tail = tail;
+    g_mutex_unlock(&src->mutex);
+
+    if (count + ringbuf_bytes_used(src) != bytes_used) {
+        return NULL;
+    }
+
+    return tail;
+}

ringbuf.h

@@ -0,0 +1,148 @@
+#ifndef INCLUDED_RINGBUF_H
+#define INCLUDED_RINGBUF_H
+
+/*
+ * ringbuf.h - C ring buffer (FIFO) interface.
+ *
+ * Written in 2011 by Drew Hess <dhess-src@bothan.net>.
+ *
+ * To the extent possible under law, the author(s) have dedicated all
+ * copyright and related and neighboring rights to this software to
+ * the public domain worldwide. This software is distributed without
+ * any warranty.
+ *
+ * You should have received a copy of the CC0 Public Domain Dedication
+ * along with this software. If not, see
+ * <http://creativecommons.org/publicdomain/zero/1.0/>.
+ */
+
+/*
+ * A byte-addressable ring buffer FIFO implementation.
+ *
+ * The ring buffer's head pointer points to the starting location
+ * where data should be written when copying data *into* the buffer
+ * (e.g., with ringbuf_read). The ring buffer's tail pointer points to
+ * the starting location where data should be read when copying data
+ * *from* the buffer (e.g., with ringbuf_write).
+ */
+
+#include <glib.h>
+
+typedef struct ringbuf_t *ringbuf_t;
+
+/*
+ * Create a new ring buffer with the given capacity (usable
+ * bytes). Note that the actual internal buffer size may be one or
+ * more bytes larger than the usable capacity, for bookkeeping.
+ *
+ * Returns the new ring buffer object, or 0 if there's not enough
+ * memory to fulfill the request for the given capacity.
+ */
+ringbuf_t ringbuf_new (gsize size, guint capacity);
+
+/*
+ * The size of the internal buffer, in bytes. One or more bytes may be
+ * unusable in order to distinguish the "buffer full" state from the
+ * "buffer empty" state.
+ *
+ * For the usable capacity of the ring buffer, use the
+ * ringbuf_capacity function.
+ */
+gsize ringbuf_buffer_size(const struct ringbuf_t *rb);
+
+/*
+ * Deallocate a ring buffer, and, as a side effect, set the pointer to
+ * 0.
+ */
+void ringbuf_free(ringbuf_t *rb);
+
+/*
+ * Reset a ring buffer to its initial state (empty).
+ */
+void ringbuf_reset(ringbuf_t rb);
+
+/*
+ * The usable capacity of the ring buffer, in bytes. Note that this
+ * value may be less than the ring buffer's internal buffer size, as
+ * returned by ringbuf_buffer_size.
+ */
+gsize ringbuf_capacity(const struct ringbuf_t *rb);
+
+/*
+ * The number of free/available bytes in the ring buffer. This value
+ * is never larger than the ring buffer's usable capacity.
+ */
+gsize ringbuf_bytes_free(struct ringbuf_t *rb);
+
+/*
+ * The number of bytes currently being used in the ring buffer. This
+ * value is never larger than the ring buffer's usable capacity.
+ */
+gsize ringbuf_bytes_used(struct ringbuf_t *rb);
+
+int ringbuf_is_full(struct ringbuf_t *rb);
+
+int ringbuf_is_empty(struct ringbuf_t *rb);
+
+/*
+ * access to the head and tail pointers of the ring buffer.
+ */
+gconstpointer ringbuf_tail(struct ringbuf_t *rb);
+
+gconstpointer ringbuf_head(struct ringbuf_t *rb);
+
+/*
+ * Copy n bytes from a contiguous memory area src into the ring buffer
+ * dst. Returns the ring buffer's new head pointer.
+ *
+ * It is possible to copy more data from src than is available in the
+ * buffer; i.e., it's possible to overflow the ring buffer using this
+ * function. When an overflow occurs, the state of the ring buffer is
+ * guaranteed to be consistent, including the head and tail pointers;
+ * old data will simply be overwritten in FIFO fashion, as
+ * needed. However, note that, if calling the function results in an
+ * overflow, the value of the ring buffer's tail pointer may be
+ * different than it was before the function was called.
+ */
+gpointer ringbuf_memcpy_into(ringbuf_t dst, gconstpointer src, gsize count);
+
+/*
+ * Copy n bytes from the ring buffer src, starting from its tail
+ * pointer, into a contiguous memory area dst. Returns the value of
+ * src's tail pointer after the copy is finished.
+ *
+ * Note that this copy is destructive with respect to the ring buffer:
+ * the n bytes copied from the ring buffer are no longer available in
+ * the ring buffer after the copy is complete, and the ring buffer
+ * will have n more free bytes than it did before the function was
+ * called.
+ *
+ * This function will *not* allow the ring buffer to underflow. If
+ * count is greater than the number of bytes used in the ring buffer,
+ * no bytes are copied, and the function will return 0.
+ */
+gpointer ringbuf_memcpy_from(gpointer dst, ringbuf_t src, gsize count);
+
+/*
+ * This convenience function calls write(2) on the file descriptor fd,
+ * using the ring buffer rb as the source buffer for writing (starting
+ * at the ring buffer's tail pointer), and returns the value returned
+ * by write(2). It will only call write(2) once, and may return a
+ * short count.
+ *
+ * Note that this copy is destructive with respect to the ring buffer:
+ * any bytes written from the ring buffer to the file descriptor are
+ * no longer available in the ring buffer after the copy is complete,
+ * and the ring buffer will have N more free bytes than it did before
+ * the function was called, where N is the value returned by the
+ * function (unless N is < 0, in which case an error occurred and no
+ * bytes were written).
+ *
+ * This function will *not* allow the ring buffer to underflow. If
+ * count is greater than the number of bytes used in the ring buffer,
+ * no bytes are written to the file descriptor, and the function will
+ * return 0.
+ */
+gssize ringbuf_write(int fd, ringbuf_t rb, gsize count);
+
+#endif /* INCLUDED_RINGBUF_H */

test/meson.build

@@ -12,7 +12,7 @@
 
 test_network = executable(
     'test_unit_Variables', 
-    sources: ['test_unit_variables.c', '../uca-phantom-communicate.c'],
+    sources: ['test_unit_variables.c', '../uca-phantom-communicate.c', '../ringbuf.c'],
     dependencies: deps,
     # link_with: phantom,
     c_args: c_args,