Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com>
---
include/qb/Makefile.am | 2 +-
include/qb/qbrb.h | 158 +++++++++++++++++++
include/qb/qbutil.h | 51 ++++++-
lib/Makefile.am | 2 +-
lib/ringbuffer.c | 395 ++++++++++++++++++++++++++++++++++++++++++++++++
lib/util.c | 74 +++++++++
libqb.spec.in | 1 +
tests/Makefile.am | 8 +-
tests/check_rb.c | 207 +++++++++++++++++++++++++
9 files changed, 889 insertions(+), 9 deletions(-)
create mode 100644 include/qb/qbrb.h
create mode 100644 lib/ringbuffer.c
create mode 100644 tests/check_rb.c
diff --git a/include/qb/Makefile.am b/include/qb/Makefile.am
index aed2cc0..6bd55e9 100644
--- a/include/qb/Makefile.am
+++ b/include/qb/Makefile.am
@@ -24,5 +24,5 @@ inst_HEADERS = qbtsafe.h qbhdb.h qblist.h qbpoll.h \
qbtimer.h qbplugin.h qbplugin_comp.h \
qbipcc.h qbipcs.h qbwthread.h qbqueue.h \
qblogsys.h qbhash.h qbipc_common.h \
- qbutil.h
+ qbrb.h qbutil.h
diff --git a/include/qb/qbrb.h b/include/qb/qbrb.h
new file mode 100644
index 0000000..c119c6f
--- /dev/null
+++ b/include/qb/qbrb.h
@@ -0,0 +1,158 @@
+/*
+ * Copyright (C) 2010 Red Hat, Inc.
+ *
+ * Author: Angus Salkeld <asalkeld(a)redhat.com>
+ *
+ * This file is part of libqb.
+ *
+ * libqb is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 2.1 of the License, or
+ * (at your option) any later version.
+ *
+ * libqb is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with libqb. If not, see <
http://www.gnu.org/licenses/>.
+ */
+#ifndef QB_RB_DEFINED
+#define QB_RB_DEFINED
+
+#include <sys/types.h>
+#include <stdint.h>
+
+/**
+ * @file qbrb.h
+ * @author Angus Salkeld <asalkeld(a)redhat.com>
+ *
+ * This implements a ring buffer that works in "chunks" not bytes.
+ * So you write/read a complete chunk or not at all.
+ * There are two types of ring buffer normal and overwrite.
+ * Overwrite will reclaim the oldest chunks inorder to make way for new ones,
+ * the normal version will refuse to write a new chunk if the ring buffer
+ * is full.
+ */
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef enum {
+ QB_RB_NORMAL,
+ QB_RB_OVERWRITE,
+} qb_ringbuffer_type;
+
+struct qb_ringbuffer_s;
+typedef struct qb_ringbuffer_s qb_ringbuffer_t;
+
+/**
+ * Create the ring buffer with the given type.
+ *
+ * This creates allocates a ring buffer in shared memory.
+ *
+ * @param size the requested size.
+ * @param type QB_RB_OVERWRITE or QB_RB_NORMAL
+ * @note the actual size will be rounded up to the next page size.
+ * @return a new ring buffer or NULL if there was a problem.
+ */
+qb_ringbuffer_t* qb_rb_create (size_t size, qb_ringbuffer_type type);
+
+/**
+ * Write a chunk to the ring buffer.
+ *
+ * This simply calls qb_rb_chunk_writable_alloc() and then
+ * qb_rb_chunk_writable_commit().
+ *
+ * @param data (in) the data to write
+ * @param len (in) the size of the chunk.
+ * @return the amount of bytes actually buffered (either len or 0).
+ *
+ * @see qb_rb_chunk_writable_alloc()
+ * @see qb_rb_chunk_writable_commit()
+ */
+size_t qb_rb_chunk_write (qb_ringbuffer_t *rb, const void* data, size_t len);
+
+/**
+ * Allocate space for a chunk of the given size.
+ *
+ * If type == QB_RB_OVERWRITE then this will always return non-null
+ * but if it's type is QB_RB_NORMAL then when there is not enough space then
+ * it will return NULL.
+ *
+ * @param len (in) the size to allocate.
+ * @return pointer to chunk to write to, or NULL (if no space).
+ *
+ * @see qb_rb_chunk_writable_alloc()
+ */
+void* qb_rb_chunk_writable_alloc (qb_ringbuffer_t *rb, size_t len);
+
+/**
+ * finalize the chunk.
+ */
+void qb_rb_chunk_writable_commit (qb_ringbuffer_t *rb, size_t len);
+
+/**
+ * Read (without reclaiming) the last chunk.
+ *
+ * This function is a way of accessing the next chunk without a memcpy().
+ * You can read the chunk data in place.
+ *
+ * @note This function will not "pop" the chunk, you will need to call
+ * qb_rb_chunk_reclaim().
+ * @param data_out (out) a pointer to the next chunk to read (not copied).
+ *
+ * @return the size of the chunk (0 if buffer empty).
+ */
+size_t qb_rb_chunk_peek (qb_ringbuffer_t *rb, void **data_out);
+
+/**
+ * Reclaim the oldest chunk.
+ * You will need to call this if using qb_rb_chunk_peek().
+ */
+void qb_rb_chunk_reclaim (qb_ringbuffer_t *rb);
+
+/**
+ * Read the oldest chunk into data_out.
+ *
+ * This is the same as qb_rb_chunk_peek() memcpy() and qb_rb_chunk_reclaim().
+ *
+ * @param data_out (in/out) the chunk will be memcpy'ed into this.
+ * @param len (in) the size of data_out.
+ * @return the size of the chunk.
+ */
+size_t qb_rb_chunk_read (qb_ringbuffer_t *rb, void *data_out, size_t len);
+
+/**
+ * The amount of free space in the ring buffer.
+ *
+ * @note Some of this space will be consumed by the chunk headers.
+ */
+size_t qb_rb_space_free (qb_ringbuffer_t *rb);
+
+/**
+ * The total amount of data in the buffer.
+ *
+ * @note This includes the chunk headers (8 bytes per chunk).
+ */
+size_t qb_rb_space_used (qb_ringbuffer_t *rb);
+
+/**
+ * Write the contents of the Ring Buffer to file.
+ */
+size_t qb_rb_write_to_file (qb_ringbuffer_t *rb, int32_t fd);
+
+/**
+ * Load the saved ring buffer from file into tempory memory.
+ */
+qb_ringbuffer_t *qb_rb_create_from_file (int32_t fd, qb_ringbuffer_type type);
+
+
+#ifdef __cplusplus
+}
+#endif /* __cplusplus */
+
+#endif /* QB_RB_DEFINED */
+
diff --git a/include/qb/qbutil.h b/include/qb/qbutil.h
index 11c7a1f..440ea20 100644
--- a/include/qb/qbutil.h
+++ b/include/qb/qbutil.h
@@ -20,10 +20,22 @@
#ifndef QB_UTIL_H_DEFINED
#define QB_UTIL_H_DEFINED
+/**
+ * @file qbutil.h
+ * @author Angus Salkeld <asalkeld(a)redhat.com>
+ *
+ * These are some convience functions used throughout libqb.
+ */
+
#ifdef __cplusplus
extern "C" {
#endif
+/**
+ * @typedef qb_thread_lock_type_t
+ * QB_THREAD_LOCK_SHORT is a short term lock (spinlock if available on your system)
+ * QB_THREAD_LOCK_LONG is a mutex
+ */
typedef enum {
QB_THREAD_LOCK_SHORT,
QB_THREAD_LOCK_LONG,
@@ -33,7 +45,7 @@ struct qb_thread_lock_s;
typedef struct qb_thread_lock_s qb_thread_lock_t;
/**
- * create a new lock.
+ * Create a new lock of the given type..
* @param type QB_THREAD_LOCK_SHORT == spinlock (where available, else mutex)
* QB_THREAD_LOCK_LONG == mutex
* @return pointer to qb_thread_lock_type_t or NULL on error.
@@ -41,17 +53,17 @@ typedef struct qb_thread_lock_s qb_thread_lock_t;
qb_thread_lock_t* qb_thread_lock_create (qb_thread_lock_type_t type);
/**
- * calls either pthread_mutex_lock() or pthread_spin_lock.
+ * Calls either pthread_mutex_lock() or pthread_spin_lock().
*/
int32_t qb_thread_lock (qb_thread_lock_t* tl);
/**
- * calls either pthread_mutex_trylock() or pthread_spin_trylock.
+ * Calls either pthread_mutex_trylock() or pthread_spin_trylock().
*/
int32_t qb_thread_trylock (qb_thread_lock_t* tl);
/**
- * calls either pthread_mutex_unlock() or pthread_spin_unlock.
+ * Calls either pthread_mutex_unlock() or pthread_spin_unlock.
*/
int32_t qb_thread_unlock (qb_thread_lock_t* tl);
@@ -62,10 +74,39 @@ typedef void (* qb_util_log_fn_t)(const char *file_name,
const char *msg);
/**
- * Use this function output libqb internal log message as you wish.
+ * Use this function to output libqb internal log message as you wish.
*/
void qb_util_set_log_function (qb_util_log_fn_t fn);
+/**
+ * Create a file to be used to back shared memory.
+ *
+ * @param path (out) the final absolute path of the file.
+ * @param file (in) the name of the file to be used.
+ * @param bytes the size to truncate the file to.
+ * @return 0 (success) or -1 (error)
+ */
+int32_t qb_util_circular_mmap_file_create (char *path, const char *file, size_t bytes);
+
+/**
+ * Open an existing file to be used to back shared memory.
+ *
+ * @param path (in) the absolute path to the file.
+ * @param bytes the size to truncate the file to.
+ * @return 0 (success) or -1 (error)
+ */
+int32_t qb_util_circular_mmap_file_open (const char *path, size_t bytes);
+
+/**
+ * Create a shared mamory circular buffer.
+ *
+ * @param fd an open file to use to back the shared memory.
+ * @param buf (out) the pointer to the start of the memory.
+ * @param bytes the size of the shared memory.
+ * @return 0 (success) or -1 (error)
+ */
+int32_t qb_util_circular_mmap (int32_t fd, void **buf, size_t bytes);
+
#ifdef __cplusplus
}
#endif /* __cplusplus */
diff --git a/lib/Makefile.am b/lib/Makefile.am
index 9a44a89..602016f 100644
--- a/lib/Makefile.am
+++ b/lib/Makefile.am
@@ -43,7 +43,7 @@ lib_LTLIBRARIES = libqb.la
libqb_la_CPPFLAGS = -I$(top_builddir)/include -I$(top_srcdir)/include
libqb_la_LDFLAGS = -version-info 0:0:0
libqb_la_SOURCES = util.c tsafe.c hash.c poll.c timer.c wthread.c \
- ipcc.c ipcs.c logsys.c
+ ipcc.c ipcs.c logsys.c ringbuffer.c
pkgconfigdir = $(libdir)/pkgconfig
pkgconfig_DATA = libqb.pc
diff --git a/lib/ringbuffer.c b/lib/ringbuffer.c
new file mode 100644
index 0000000..2d67329
--- /dev/null
+++ b/lib/ringbuffer.c
@@ -0,0 +1,395 @@
+/*
+ * Copyright (C) 2010 Red Hat, Inc.
+ *
+ * Author: Angus Salkeld <asalkeld(a)redhat.com>
+ *
+ * This file is part of libqb.
+ *
+ * libqb is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 2.1 of the License, or
+ * (at your option) any later version.
+ *
+ * libqb is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with libqb. If not, see <
http://www.gnu.org/licenses/>.
+ */
+#include <config.h>
+
+#include <os_base.h>
+#include <sys/mman.h>
+#include <util_int.h>
+#include <qb/qbutil.h>
+#include <qb/qbrb.h>
+
+struct qb_ringbuffer_s {
+ qb_ringbuffer_type type;
+ volatile uint32_t write_pt;
+ volatile uint32_t read_pt;
+ uint32_t size;
+ uint32_t *buf;
+};
+
+#ifdef CRAZY_DEBUG_PRINTFS
+#define DEBUG_PRINTF(args...) \
+do { \
+ printf (##args); \
+} while(0)
+#else
+#define DEBUG_PRINTF(args...)
+#endif /* CRAZY_DEBUG_PRINTFS */
+
+#define ROUNDUP(x, y) ((((x) + ((y) - 1)) / (y)) * (y))
+
+/* the chunk header is two words
+ * 1) the chunk data size
+ * 2) the magic number
+ */
+#define RB_CHUNK_HEADER_WORDS 2
+#define RB_CHUNK_HEADER_SIZE (sizeof(uint32_t) * RB_CHUNK_HEADER_WORDS)
+//#define RB_CHUNK_MAGIC 0xdeadbeef
+#define RB_CHUNK_MAGIC 0xAAAAAAAA
+#define FDHEAD_INDEX (rb->size)
+#define FDTAIL_INDEX (rb->size + 1)
+
+#define idx_step(idx) \
+do { \
+ if (idx > (rb->size - 1)) { \
+ idx = ((idx) % (rb->size)); \
+ } \
+} while (0);
+
+static void check_chunk (qb_ringbuffer_t *rb);
+
+qb_ringbuffer_t* qb_rb_create (size_t size, qb_ringbuffer_type type)
+{
+ struct qb_ringbuffer_s* rb = malloc (sizeof (struct qb_ringbuffer_s));
+ size_t real_size = ROUNDUP(size, sysconf(_SC_PAGESIZE));
+ char path[256];
+ int32_t fd;
+
+ rb->buf = NULL;
+
+ fd = qb_util_circular_mmap_file_create (path, "qb-ringbuffer-XXXXXX",
real_size);
+ if (fd == -1) {
+ qb_util_log (LOG_ERR, "couldn't create file for mmap");
+ return NULL;
+ }
+ if (qb_util_circular_mmap (fd, (void**)&rb->buf, real_size) != 0) {
+ close (fd);
+ unlink (path);
+ return NULL;
+ }
+
+ memset (rb->buf, 0, real_size * 2);
+
+ rb->type = type;
+ /*
+ * rb->size tracks data by ints and not bytes/chars.
+ */
+ rb->size = real_size / sizeof (uint32_t);
+ /*
+ * First record starts at zero
+ * Last record ends at zero
+ */
+ rb->write_pt = 0;
+ rb->read_pt = 0;
+
+ return rb;
+}
+
+size_t qb_rb_space_free (qb_ringbuffer_t *rb)
+{
+ uint32_t write_size;
+ uint32_t read_size;
+ size_t space_free = 0;
+
+ write_size = rb->write_pt;
+ read_size = rb->read_pt;
+
+ if (write_size > read_size) {
+ space_free = (read_size - write_size + rb->size) - 1;
+ }
+ else if (write_size < read_size) {
+ space_free = (read_size - write_size) - 1;
+ }
+ else {
+ space_free = rb->size;
+ }
+ /* word -> bytes */
+ return (space_free * sizeof (uint32_t));
+}
+
+size_t qb_rb_space_used (qb_ringbuffer_t *rb)
+{
+ uint32_t write_size;
+ uint32_t read_size;
+ size_t space_used;
+
+ write_size = rb->write_pt;
+ read_size = rb->read_pt;
+
+ if (write_size > read_size) {
+ space_used = write_size - read_size;
+ } else
+ if (write_size < read_size) {
+ space_used = (write_size - read_size + rb->size) - 1;
+ } else {
+ space_used = 0;
+ }
+ /* word -> bytes */
+ return (space_used * sizeof (uint32_t));
+}
+
+void* qb_rb_chunk_writable_alloc (qb_ringbuffer_t *rb, size_t len)
+{
+ uint32_t idx;
+
+ /*
+ * Reclaim data if we are over writing and we need space
+ */
+ if (rb->type == QB_RB_OVERWRITE) {
+ while (qb_rb_space_free (rb) < (len + RB_CHUNK_HEADER_SIZE)) {
+ qb_rb_chunk_reclaim (rb);
+ }
+ } else {
+ if (qb_rb_space_free (rb) < (len + RB_CHUNK_HEADER_SIZE)) {
+ return NULL;
+ }
+ }
+ idx = rb->write_pt;
+ /*
+ * insert the chunk header
+ */
+ rb->buf[idx++] = 0;
+ idx_step(idx);
+ rb->buf[idx++] = RB_CHUNK_MAGIC;
+ idx_step(idx);
+
+ /*
+ * return a pointer to the begining of the chunk data
+ */
+ return (void*)&rb->buf[idx];
+
+}
+
+void qb_rb_chunk_writable_commit (qb_ringbuffer_t *rb, size_t len)
+{
+ uint32_t idx = rb->write_pt;
+
+ /*
+ * skip over the chunk header
+ */
+ rb->buf[idx++] = len;
+ idx_step(idx);
+ rb->buf[idx++] = RB_CHUNK_MAGIC;
+ idx_step(idx);
+
+ /*
+ * skip over the user's chunk.
+ */
+ idx += (len / sizeof (uint32_t));
+ if ((len % sizeof (uint32_t)) != 0) {
+ idx++;
+ }
+ idx_step (idx);
+
+ /*
+ * commit the write_pt
+ */
+ rb->write_pt = idx;
+ DEBUG_PRINTF("%s: read: %u, write: %u, cs:%u, used:%zu\n", __func__,
+ rb->read_pt, rb->write_pt, len, qb_rb_space_used (rb));
+}
+
+size_t qb_rb_chunk_write (qb_ringbuffer_t *rb, const void* data, size_t len)
+{
+ char *dest = qb_rb_chunk_writable_alloc (rb, len);
+
+ if (dest == NULL) {
+ return 0;
+ }
+
+ /*
+ * copy the data
+ */
+ memcpy (dest, data, len);
+
+ qb_rb_chunk_writable_commit (rb, len);
+
+ return len;
+}
+
+void qb_rb_chunk_reclaim (qb_ringbuffer_t *rb)
+{
+ uint32_t words_needed = 0;
+ uint32_t chunk_size = rb->buf[rb->read_pt];
+ uint32_t idx;
+
+ if (qb_rb_space_used (rb) == 0) {
+ return;
+ }
+ check_chunk (rb);
+
+ words_needed = (chunk_size + RB_CHUNK_HEADER_SIZE) / sizeof (uint32_t);
+ if ((chunk_size % sizeof (uint32_t)) != 0) {
+ words_needed += 1;
+ }
+ /*
+ * clear the header
+ */
+ rb->buf[rb->read_pt] = 0;
+ rb->buf[rb->read_pt + 1] = 0;
+
+ idx = rb->read_pt + words_needed;
+ idx_step(idx);
+ rb->read_pt = idx;
+ DEBUG_PRINTF ("%s : read: %u, write: %u, cs:%u, used:%zu\n", __func__,
+ rb->read_pt, rb->write_pt, chunk_size, qb_rb_space_used (rb));
+}
+
+size_t qb_rb_chunk_peek (qb_ringbuffer_t *rb, void **data_out)
+{
+ uint32_t chunk_size = rb->buf[rb->read_pt];
+ uint32_t chunk_magic = rb->buf[(rb->read_pt + 1) % rb->size];
+
+ *data_out = &rb->buf[rb->read_pt + RB_CHUNK_HEADER_WORDS];
+
+ if (chunk_magic != RB_CHUNK_MAGIC) {
+ return 0;
+ } else {
+ return chunk_size;
+ }
+}
+
+size_t qb_rb_chunk_read (qb_ringbuffer_t *rb, void *data_out, size_t len)
+{
+ uint32_t chunk_size = rb->buf[rb->read_pt];
+
+ if (qb_rb_space_used (rb) == 0) {
+ return 0;
+ }
+
+ check_chunk (rb);
+
+ if (len < chunk_size) {
+ return 0;
+ }
+
+ memcpy (data_out, &rb->buf[rb->read_pt + RB_CHUNK_HEADER_WORDS], chunk_size);
+
+ qb_rb_chunk_reclaim (rb);
+
+ return chunk_size;
+}
+
+static void print_header (qb_ringbuffer_t *rb)
+{
+ printf ("Ringbuffer: \n");
+ if (rb->type == QB_RB_OVERWRITE) {
+ printf (" ->OVERWRITE\n");
+ } else {
+ printf (" ->NORMAL\n");
+ }
+ printf (" ->write_pt [%d]\n", rb->write_pt);
+ printf (" ->read_pt [%d]\n", rb->read_pt);
+ printf (" ->size [%d words]\n", rb->size);
+
+ printf (" =>free [%zu bytes]\n", qb_rb_space_free (rb));
+ printf (" =>used [%zu bytes]\n", qb_rb_space_used (rb));
+}
+
+static void check_chunk (qb_ringbuffer_t *rb)
+{
+ uint32_t chunk_size = rb->buf[rb->read_pt];
+ uint32_t chunk_magic = rb->buf[(rb->read_pt + 1) % rb->size];
+
+ if (chunk_magic != RB_CHUNK_MAGIC) {
+ printf("size: %x\n", chunk_size);
+ printf("magic: %x\n", chunk_magic);
+ print_header (rb);
+ assert(chunk_magic == RB_CHUNK_MAGIC);
+ }
+}
+
+size_t qb_rb_write_to_file (qb_ringbuffer_t *rb, int32_t fd)
+{
+ ssize_t result = 0;
+ ssize_t written_size = 0;
+
+ print_header (rb);
+
+ result = write (fd, &rb->size, sizeof (uint32_t));
+ if ((result < 0) || (result != sizeof (uint32_t))) {
+ return -1;
+ }
+ written_size += result;
+
+ result = write (fd, rb->buf, rb->size * sizeof (uint32_t));
+ if ((result < 0) || (result != rb->size * sizeof (uint32_t))) {
+ return -1;
+ }
+ written_size += result;
+
+ /*
+ * store the read & write pointers
+ */
+ result += write (fd, (void*)&rb->write_pt, sizeof (uint32_t));
+ if ((result < 0) || (result != sizeof (uint32_t))) {
+ return -1;
+ }
+ written_size += result;
+ result += write (fd, (void*)&rb->read_pt, sizeof (uint32_t));
+ if ((result < 0) || (result != sizeof (uint32_t))) {
+ return -1;
+ }
+ written_size += result;
+
+ return written_size;
+}
+
+qb_ringbuffer_t *qb_rb_create_from_file (int32_t fd, qb_ringbuffer_type type)
+{
+ ssize_t n_read;
+ size_t n_required;
+ qb_ringbuffer_t *rb = malloc (sizeof (qb_ringbuffer_t));
+
+ rb->type = type;
+
+ n_required = sizeof (uint32_t);
+ n_read = read (fd, &rb->size, n_required);
+ if (n_read != n_required) {
+ fprintf (stderr, "Unable to read fdata header\n");
+ return NULL;
+ }
+
+ n_required = ((rb->size + 2) * sizeof (uint32_t));
+
+ if ((rb->buf = malloc (n_required)) == NULL) {
+ fprintf (stderr, "exhausted virtual memory\n");
+ return NULL;
+ }
+ n_read = read (fd, rb->buf, n_required);
+ if (n_read < 0) {
+ fprintf (stderr, "reading file failed: %s\n",
+ strerror (errno));
+ return NULL;
+ }
+
+ if (n_read != n_required) {
+ printf ("Warning: read %lu bytes, but expected %lu\n",
+ (unsigned long) n_read, (unsigned long) n_required);
+ }
+
+ rb->write_pt = rb->buf[FDHEAD_INDEX];
+ rb->read_pt = rb->buf[FDTAIL_INDEX];
+
+ print_header (rb);
+
+ return rb;
+}
+
diff --git a/lib/util.c b/lib/util.c
index f9acd5b..dc2213e 100644
--- a/lib/util.c
+++ b/lib/util.c
@@ -21,6 +21,7 @@
#include <config.h>
#include "os_base.h"
+#include <syslog.h>
#include "util_int.h"
#include <stdarg.h>
#include <sys/shm.h>
@@ -139,4 +140,77 @@ void qb_util_set_log_function (qb_util_log_fn_t fn)
real_log_fn = fn;
}
+/*
+ * ---------------------------------------------------
+ * shared memory functions.
+ */
+int32_t qb_util_circular_mmap_file_create (char *path, const char *file, size_t bytes)
+{
+ int32_t fd;
+
+ sprintf (path, "/dev/shm/%s", file);
+
+ fd = mkstemp (path);
+ if (fd == -1) {
+ sprintf (path, LOCALSTATEDIR "/run/%s", file);
+ fd = mkstemp (path);
+ if (fd == -1) {
+ return (-1);
+ }
+ }
+
+ ftruncate (fd, bytes);
+ return fd;
+}
+
+int32_t qb_util_circular_mmap_file_open (const char *path, size_t bytes)
+{
+ int32_t fd;
+
+ fd = open (path, O_RDWR, 0600);
+ if (fd == -1) {
+ return (-1);
+ }
+
+ ftruncate (fd, bytes);
+ return fd;
+}
+
+int32_t qb_util_circular_mmap (int32_t fd, void **buf, size_t bytes)
+{
+ void *addr_orig;
+ void *addr;
+ int32_t res;
+
+ addr_orig = mmap (NULL, bytes << 1, PROT_NONE,
+ MAP_ANONYMOUS | MAP_PRIVATE, -1, 0);
+
+ if (addr_orig == MAP_FAILED) {
+ return (-1);
+ }
+
+ addr = mmap (addr_orig, bytes, PROT_READ | PROT_WRITE,
+ MAP_FIXED | MAP_SHARED, fd, 0);
+
+ if (addr != addr_orig) {
+ return (-1);
+ }
+#ifdef QB_BSD
+ madvise(addr_orig, bytes, MADV_NOSYNC);
+#endif
+
+ addr = mmap (((char *)addr_orig) + bytes,
+ bytes, PROT_READ | PROT_WRITE,
+ MAP_FIXED | MAP_SHARED, fd, 0);
+#ifdef QB_BSD
+ madvise(((char *)addr_orig) + bytes, bytes, MADV_NOSYNC);
+#endif
+
+ res = close (fd);
+ if (res) {
+ return (-1);
+ }
+ *buf = addr_orig;
+ return (0);
+}
diff --git a/libqb.spec.in b/libqb.spec.in
index d8f1441..06ba9cd 100644
--- a/libqb.spec.in
+++ b/libqb.spec.in
@@ -81,6 +81,7 @@ The Quarterback APIs.
%{_includedir}/qb/qbwthread.h
%{_includedir}/qb/qblogsys.h
%{_includedir}/qb/qbhash.h
+%{_includedir}/qb/qbrb.h
%{_includedir}/qb/qbutil.h
%{_mandir}/man3/qb*3*
diff --git a/tests/Makefile.am b/tests/Makefile.am
index 579a9da..47bc4a4 100644
--- a/tests/Makefile.am
+++ b/tests/Makefile.am
@@ -45,12 +45,16 @@ bms_LDADD = -lrt $(top_builddir)/lib/libqb.la
if HAVE_CHECK
-TESTS = check_hash
+TESTS = check_rb check_hash
#check_plugin
-check_PROGRAMS = check_hash
+check_PROGRAMS = check_rb check_hash
#check_plugin
+check_rb_SOURCES = check_rb.c $(top_builddir)/include/qb/qbrb.h
+check_rb_CFLAGS = @CHECK_CFLAGS@ -I$(top_srcdir)/include
+check_rb_LDADD = $(top_builddir)/lib/libqb.la -lrt @CHECK_LIBS@
+
check_hash_SOURCES = check_hash.c $(top_builddir)/include/qb/qbhash.h
check_hash_CFLAGS = @CHECK_CFLAGS@ -I$(top_srcdir)/include
check_hash_LDADD = $(top_builddir)/lib/libqb.la -lrt @CHECK_LIBS@
diff --git a/tests/check_rb.c b/tests/check_rb.c
new file mode 100644
index 0000000..f548c26
--- /dev/null
+++ b/tests/check_rb.c
@@ -0,0 +1,207 @@
+/*
+ * Copyright (c) 2010 Red Hat, Inc.
+ *
+ * All rights reserved.
+ *
+ * Author: Angus Salkeld <asalkeld(a)redhat.com>
+ *
+ * This file is part of libqb.
+ *
+ * libqb is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 2.1 of the License, or
+ * (at your option) any later version.
+ *
+ * libqb is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with libqb. If not, see <
http://www.gnu.org/licenses/>.
+ */
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <check.h>
+#include <qb/qbrb.h>
+#include <qb/qbipc_common.h>
+
+START_TEST (test_ring_buffer1)
+{
+ char my_buf[512];
+ qb_ipc_request_header_t *hdr;
+ char *str;
+ qb_ringbuffer_t * rb;
+ int i;
+ int b;
+ size_t actual;
+ size_t avail;
+
+ rb = qb_rb_create (200, QB_RB_NORMAL);
+
+ for (b = 0; b < 3; b++) {
+ hdr = (qb_ipc_request_header_t*)my_buf;
+ str = my_buf + sizeof (qb_ipc_request_header_t);
+
+ for (i = 0; i < 900; i++) {
+ hdr->id = __LINE__ + i;
+ hdr->size = sprintf (str, "ID: %d (%s + i(%d)) -- %s-%s!",
+ hdr->id, "actually the line number", i,
+ __func__, __FILE__) + 1;
+ hdr->size += sizeof (qb_ipc_request_header_t);
+ avail = qb_rb_space_free (rb);
+ actual = qb_rb_chunk_write (rb, hdr, hdr->size);
+ if (avail < (hdr->size + (2*sizeof(uint32_t)))) {
+ ck_assert_int_eq (actual, 0);
+ } else {
+ ck_assert_int_eq (actual, hdr->size);
+ }
+ }
+
+ memset (my_buf, 0, sizeof(my_buf));
+
+ hdr = (qb_ipc_request_header_t*)my_buf;
+ str = my_buf + sizeof (qb_ipc_request_header_t);
+
+ for (i = 0; i < 15; i++) {
+ actual = qb_rb_chunk_read (rb, hdr, 512);
+ if (actual == 0) {
+ break;
+ }
+ str[actual - sizeof (qb_ipc_request_header_t)] = '\0';
+
+ ck_assert_int_eq (actual, hdr->size);
+ }
+ }
+}
+END_TEST
+
+/*
+ * nice size (int64)
+ */
+START_TEST (test_ring_buffer2)
+{
+ qb_ringbuffer_t *t;
+ int32_t i;
+ int64_t v = 7891034;
+ int64_t *new_data;
+ ssize_t l;
+
+ t = qb_rb_create (10, QB_RB_OVERWRITE);
+ for (i = 0; i < 9000; i++) {
+ l = qb_rb_chunk_write (t, &v, sizeof (v));
+ ck_assert_int_eq (l, sizeof (v));
+ }
+ for (i = 0; i < 2000; i++) {
+ l = qb_rb_chunk_peek (t, (void**)&new_data);
+ if (l == 0) {
+ /* no more to read */
+ break;
+ }
+ ck_assert_int_eq (l, sizeof (v));
+ fail_unless (v == *new_data);
+ qb_rb_chunk_reclaim (t);
+ }
+}
+END_TEST
+
+/*
+ * odd size (10)
+ */
+START_TEST (test_ring_buffer3)
+{
+ qb_ringbuffer_t *t;
+ int32_t i;
+ char v[] = "1234567891";
+ char out[32];
+ ssize_t l;
+ size_t len = strlen (v) + 1;
+
+ t = qb_rb_create (10, QB_RB_OVERWRITE);
+ for (i = 0; i < 9000; i++) {
+ l = qb_rb_chunk_write (t, v, len);
+ ck_assert_int_eq (l, len);
+ }
+ for (i = 0; i < 2000; i++) {
+ l = qb_rb_chunk_read (t, (void*)out, 32);
+ if (l == 0) {
+ /* no more to read */
+ break;
+ }
+ ck_assert_int_eq (l, len);
+ ck_assert_str_eq (v, out);
+ }
+}
+END_TEST
+
+START_TEST (test_ring_buffer4)
+{
+ qb_ringbuffer_t *t;
+ char data[] = "1234567891";
+ int i;
+ char *new_data;
+ ssize_t l;
+
+ t = qb_rb_create (10, QB_RB_OVERWRITE);
+ for (i = 0; i < 2000; i++) {
+ l = qb_rb_chunk_write (t, data, strlen (data));
+ if (l == 0) {
+ break;
+ }
+
+ fprintf (stderr,
+ "[%d] Bytes written: %zu, free %zu\n",
+ i, l, qb_rb_space_free (t));
+
+ if (i == 0) {
+ data[0]='b';
+ }
+ }
+ for (i = 0; i < 2000; i++) {
+ l = qb_rb_chunk_peek (t, (void**)&new_data);
+ fprintf (stderr, "Get [%d] %zd: %s\n", i, l, new_data);
+
+ if (l == 0) {
+ fprintf (stderr, "EE\n");
+ exit (1);
+ }
+ if (l != strlen("1234567891")) {
+ fprintf (stderr, "EE %zd\n", l);
+ exit (1);
+ for (i = 0; i < l; i++) {
+ fprintf (stderr, "%x ", (int)new_data[i]);
+ }
+ fprintf(stderr, "\n");
+ }
+ qb_rb_chunk_reclaim (t);
+ }
+}
+END_TEST
+
+static Suite *rb_suite (void)
+{
+ TCase *tc_load;
+ Suite *s = suite_create ("ringbuffer");
+
+ tc_load = tcase_create ("test01");
+ tcase_add_test (tc_load, test_ring_buffer1);
+ tcase_add_test (tc_load, test_ring_buffer2);
+ tcase_add_test (tc_load, test_ring_buffer3);
+ suite_add_tcase (s, tc_load);
+
+ return s;
+}
+
+int main (void)
+{
+ int number_failed;
+
+ Suite *s = rb_suite ();
+ SRunner *sr = srunner_create (s);
+ srunner_run_all (sr, CK_NORMAL);
+ number_failed = srunner_ntests_failed (sr);
+ srunner_free (sr);
+ return (number_failed == 0) ? EXIT_SUCCESS : EXIT_FAILURE;
+}
+
--
1.6.6.1