[PATCH] Add a ringbuffer based off the one in logsys.
by Angus Salkeld
This ringbuffer is usable across processes.
the point is to use this for IPC to provide async
connections from client to server, but with inherient
flow control.
This still needs a bit of clean up, but committing now
for feedback and as it is quite functional.
Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com>
---
.gitignore | 3 +-
docs/html.dox.in | 8 +-
include/qb/Makefile.am | 2 +-
include/qb/qbhdb.h | 114 +++------
include/qb/qbrb.h | 211 ++++++++++++++++
include/qb/qbutil.h | 49 ++++-
lib/Makefile.am | 7 +-
lib/ringbuffer.c | 626 +++++++++++++++++++++++++++++++++++++++++++++++
lib/ringbuffer.h | 88 +++++++
lib/ringbuffer_helper.c | 378 ++++++++++++++++++++++++++++
lib/util.c | 119 ++++++++--
libqb.spec.in | 1 +
tests/.gitignore | 3 +-
tests/Makefile.am | 30 ++-
tests/check_rb.c | 227 +++++++++++++++++
tests/rbreader.c | 54 ++++
tests/rbwriter.c | 168 +++++++++++++
17 files changed, 1969 insertions(+), 119 deletions(-)
create mode 100644 include/qb/qbrb.h
create mode 100644 lib/ringbuffer.c
create mode 100644 lib/ringbuffer.h
create mode 100644 lib/ringbuffer_helper.c
create mode 100644 tests/check_rb.c
create mode 100644 tests/rbreader.c
create mode 100644 tests/rbwriter.c
diff --git a/.gitignore b/.gitignore
index 3179bbf..6a6517e 100644
--- a/.gitignore
+++ b/.gitignore
@@ -25,7 +25,6 @@ missing
.libs
m4
libtool
+compile
ltmain.sh
libqb-*
-mbsec
-opsec
diff --git a/docs/html.dox.in b/docs/html.dox.in
index 82ba027..7755e6c 100644
--- a/docs/html.dox.in
+++ b/docs/html.dox.in
@@ -131,8 +131,8 @@ QHP_SECT_FILTER_ATTRS =
QHG_LOCATION =
DISABLE_INDEX = NO
ENUM_VALUES_PER_LINE = 4
-GENERATE_TREEVIEW = YES
-USE_INLINE_TREES = YES
+GENERATE_TREEVIEW = NO
+USE_INLINE_TREES = NO
TREEVIEW_WIDTH = 250
FORMULA_FONTSIZE = 10
SEARCHENGINE = YES
@@ -199,8 +199,8 @@ UML_LOOK = YES
TEMPLATE_RELATIONS = YES
INCLUDE_GRAPH = YES
INCLUDED_BY_GRAPH = YES
-CALL_GRAPH = YES
-CALLER_GRAPH = YES
+CALL_GRAPH = NO
+CALLER_GRAPH = NO
GRAPHICAL_HIERARCHY = YES
DIRECTORY_GRAPH = YES
DOT_IMAGE_FORMAT = png
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/qbhdb.h b/include/qb/qbhdb.h
index bcd20fa..87fbbd2 100644
--- a/include/qb/qbhdb.h
+++ b/include/qb/qbhdb.h
@@ -33,6 +33,7 @@
#include <pthread.h>
#include <stdint.h>
#include <inttypes.h>
+#include <qb/qbutil.h>
typedef uint64_t qb_hdb_handle_t;
@@ -60,55 +61,10 @@ struct qb_hdb_handle_database {
struct qb_hdb_handle *handles;
unsigned int iterator;
void (*destructor) (void *);
-#if defined(HAVE_PTHREAD_SPIN_LOCK)
- pthread_spinlock_t lock;
-#else
- pthread_mutex_t lock;
-#endif
+ qb_thread_lock_t *lock;
unsigned int first_run;
};
-#if defined(HAVE_PTHREAD_SPIN_LOCK)
-static inline void qb_hdb_database_lock (pthread_spinlock_t *spinlock)
-{
- pthread_spin_lock (spinlock);
-}
-
-static inline void qb_hdb_database_unlock (pthread_spinlock_t *spinlock)
-{
- pthread_spin_unlock (spinlock);
-}
-static inline void qb_hdb_database_lock_init (pthread_spinlock_t *spinlock)
-{
- pthread_spin_init (spinlock, 0);
-}
-
-static inline void qb_hdb_database_lock_destroy (pthread_spinlock_t *spinlock)
-{
- pthread_spin_destroy (spinlock);
-}
-
-#else
-static inline void qb_hdb_database_lock (pthread_mutex_t *mutex)
-{
- pthread_mutex_lock (mutex);
-}
-
-static inline void qb_hdb_database_unlock (pthread_mutex_t *mutex)
-{
- pthread_mutex_unlock (mutex);
-}
-static inline void qb_hdb_database_lock_init (pthread_mutex_t *mutex)
-{
- pthread_mutex_init (mutex, NULL);
-}
-
-static inline void qb_hdb_database_lock_destroy (pthread_mutex_t *mutex)
-{
- pthread_mutex_destroy (mutex);
-}
-#endif
-
#define DECLARE_HDB_DATABASE(database_name,destructor_function) \
static struct qb_hdb_handle_database (database_name) = { \
.handle_count = 0, \
@@ -122,14 +78,14 @@ static inline void qb_hdb_create (
struct qb_hdb_handle_database *handle_database)
{
memset (handle_database, 0, sizeof (struct qb_hdb_handle_database));
- qb_hdb_database_lock_init (&handle_database->lock);
+ handle_database->lock = qb_thread_lock_create (QB_THREAD_LOCK_SHORT);
}
static inline void qb_hdb_destroy (
struct qb_hdb_handle_database *handle_database)
{
free (handle_database->handles);
- qb_hdb_database_lock_destroy (&handle_database->lock);
+ qb_thread_lock_destroy (handle_database->lock);
memset (handle_database, 0, sizeof (struct qb_hdb_handle_database));
}
@@ -148,9 +104,9 @@ static inline int qb_hdb_handle_create (
if (handle_database->first_run == 1) {
handle_database->first_run = 0;
- qb_hdb_database_lock_init (&handle_database->lock);
+ handle_database->lock = qb_thread_lock_create (QB_THREAD_LOCK_SHORT);
}
- qb_hdb_database_lock (&handle_database->lock);
+ qb_thread_lock (handle_database->lock);
for (handle = 0; handle < handle_database->handle_count; handle++) {
if (handle_database->handles[handle].state == QB_HDB_HANDLE_STATE_EMPTY) {
@@ -164,7 +120,7 @@ static inline int qb_hdb_handle_create (
new_handles = (struct qb_hdb_handle *)realloc (handle_database->handles,
sizeof (struct qb_hdb_handle) * handle_database->handle_count);
if (new_handles == NULL) {
- qb_hdb_database_unlock (&handle_database->lock);
+ qb_thread_unlock (handle_database->lock);
errno = ENOMEM;
return (-1);
}
@@ -202,7 +158,7 @@ static inline int qb_hdb_handle_create (
*handle_id_out = (((unsigned long long)(check)) << 32) | handle;
- qb_hdb_database_unlock (&handle_database->lock);
+ qb_thread_unlock (handle_database->lock);
return (0);
}
@@ -217,19 +173,19 @@ static inline int qb_hdb_handle_get (
if (handle_database->first_run == 1) {
handle_database->first_run = 0;
- qb_hdb_database_lock_init (&handle_database->lock);
+ handle_database->lock = qb_thread_lock_create (QB_THREAD_LOCK_SHORT);
}
- qb_hdb_database_lock (&handle_database->lock);
+ qb_thread_lock (handle_database->lock);
*instance = NULL;
if (handle >= handle_database->handle_count) {
- qb_hdb_database_unlock (&handle_database->lock);
+ qb_thread_unlock (handle_database->lock);
errno = EBADF;
return (-1);
}
if (handle_database->handles[handle].state != QB_HDB_HANDLE_STATE_ACTIVE) {
- qb_hdb_database_unlock (&handle_database->lock);
+ qb_thread_unlock (handle_database->lock);
errno = EBADF;
return (-1);
}
@@ -237,7 +193,7 @@ static inline int qb_hdb_handle_get (
if (check != 0xffffffff &&
check != handle_database->handles[handle].check) {
- qb_hdb_database_unlock (&handle_database->lock);
+ qb_thread_unlock (handle_database->lock);
errno = EBADF;
return (-1);
}
@@ -246,7 +202,7 @@ static inline int qb_hdb_handle_get (
handle_database->handles[handle].ref_count += 1;
- qb_hdb_database_unlock (&handle_database->lock);
+ qb_thread_unlock (handle_database->lock);
return (0);
}
@@ -260,19 +216,19 @@ static inline int qb_hdb_handle_get_always (
if (handle_database->first_run == 1) {
handle_database->first_run = 0;
- qb_hdb_database_lock_init (&handle_database->lock);
+ handle_database->lock = qb_thread_lock_create (QB_THREAD_LOCK_SHORT);
}
- qb_hdb_database_lock (&handle_database->lock);
+ qb_thread_lock (handle_database->lock);
*instance = NULL;
if (handle >= handle_database->handle_count) {
- qb_hdb_database_unlock (&handle_database->lock);
+ qb_thread_unlock (handle_database->lock);
errno = EBADF;
return (-1);
}
if (handle_database->handles[handle].state == QB_HDB_HANDLE_STATE_EMPTY) {
- qb_hdb_database_unlock (&handle_database->lock);
+ qb_thread_unlock (handle_database->lock);
errno = EBADF;
return (-1);
}
@@ -280,7 +236,7 @@ static inline int qb_hdb_handle_get_always (
if (check != 0xffffffff &&
check != handle_database->handles[handle].check) {
- qb_hdb_database_unlock (&handle_database->lock);
+ qb_thread_unlock (handle_database->lock);
errno = EBADF;
return (-1);
}
@@ -289,7 +245,7 @@ static inline int qb_hdb_handle_get_always (
handle_database->handles[handle].ref_count += 1;
- qb_hdb_database_unlock (&handle_database->lock);
+ qb_thread_unlock (handle_database->lock);
return (0);
}
@@ -302,12 +258,12 @@ static inline int qb_hdb_handle_put (
if (handle_database->first_run == 1) {
handle_database->first_run = 0;
- qb_hdb_database_lock_init (&handle_database->lock);
+ handle_database->lock = qb_thread_lock_create (QB_THREAD_LOCK_SHORT);
}
- qb_hdb_database_lock (&handle_database->lock);
+ qb_thread_lock (handle_database->lock);
if (handle >= handle_database->handle_count) {
- qb_hdb_database_unlock (&handle_database->lock);
+ qb_thread_unlock (handle_database->lock);
errno = EBADF;
return (-1);
@@ -316,7 +272,7 @@ static inline int qb_hdb_handle_put (
if (check != 0xffffffff &&
check != handle_database->handles[handle].check) {
- qb_hdb_database_unlock (&handle_database->lock);
+ qb_thread_unlock (handle_database->lock);
errno = EBADF;
return (-1);
}
@@ -331,7 +287,7 @@ static inline int qb_hdb_handle_put (
free (handle_database->handles[handle].instance);
memset (&handle_database->handles[handle], 0, sizeof (struct qb_hdb_handle));
}
- qb_hdb_database_unlock (&handle_database->lock);
+ qb_thread_unlock (handle_database->lock);
return (0);
}
@@ -345,12 +301,12 @@ static inline int qb_hdb_handle_destroy (
if (handle_database->first_run == 1) {
handle_database->first_run = 0;
- qb_hdb_database_lock_init (&handle_database->lock);
+ handle_database->lock = qb_thread_lock_create (QB_THREAD_LOCK_SHORT);
}
- qb_hdb_database_lock (&handle_database->lock);
+ qb_thread_lock (handle_database->lock);
if (handle >= handle_database->handle_count) {
- qb_hdb_database_unlock (&handle_database->lock);
+ qb_thread_unlock (handle_database->lock);
errno = EBADF;
return (-1);
@@ -358,13 +314,13 @@ static inline int qb_hdb_handle_destroy (
if (check != 0xffffffff &&
check != handle_database->handles[handle].check) {
- qb_hdb_database_unlock (&handle_database->lock);
+ qb_thread_unlock (handle_database->lock);
errno = EBADF;
return (-1);
}
handle_database->handles[handle].state = QB_HDB_HANDLE_STATE_PENDINGREMOVAL;
- qb_hdb_database_unlock (&handle_database->lock);
+ qb_thread_unlock (handle_database->lock);
res = qb_hdb_handle_put (handle_database, handle_in);
return (res);
}
@@ -380,26 +336,26 @@ static inline int qb_hdb_handle_refcount_get (
if (handle_database->first_run == 1) {
handle_database->first_run = 0;
- qb_hdb_database_lock_init (&handle_database->lock);
+ handle_database->lock = qb_thread_lock_create (QB_THREAD_LOCK_SHORT);
}
- qb_hdb_database_lock (&handle_database->lock);
+ qb_thread_lock (handle_database->lock);
if (handle >= handle_database->handle_count) {
- qb_hdb_database_unlock (&handle_database->lock);
+ qb_thread_unlock (handle_database->lock);
errno = EBADF;
return (-1);
}
if (check != 0xffffffff &&
check != handle_database->handles[handle].check) {
- qb_hdb_database_unlock (&handle_database->lock);
+ qb_thread_unlock (handle_database->lock);
errno = EBADF;
return (-1);
}
refcount = handle_database->handles[handle].ref_count;
- qb_hdb_database_unlock (&handle_database->lock);
+ qb_thread_unlock (handle_database->lock);
return (refcount);
}
diff --git a/include/qb/qbrb.h b/include/qb/qbrb.h
new file mode 100644
index 0000000..c5fcf80
--- /dev/null
+++ b/include/qb/qbrb.h
@@ -0,0 +1,211 @@
+/*
+ * 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
+
+/**
+ * create a ring buffer (rather than open and existing one)
+ * @see qb_rb_open()
+ */
+#define QB_RB_FLAG_CREATE 0x01
+/**
+ * New calls to qb_rb_chunk_write() will call qb_rb_chunk_reclaim()
+ * if there is not enough space.
+ * If this is not set then new writes will be refused.
+ * @see qb_rb_open()
+ */
+#define QB_RB_FLAG_OVERWRITE 0x02
+/**
+ * The ringbuffer will be shared between pthreads not processes.
+ * This effects the type of locks/semaphores that are used.
+ * @see qb_rb_open()
+ */
+#define QB_RB_FLAG_SHARED_THREAD 0x03
+/**
+ * The ringbuffer will be shared between processes.
+ * This effects the type of locks/semaphores that are used.
+ * @see qb_rb_open()
+ */
+#define QB_RB_FLAG_SHARED_PROCESS 0x04
+/*
+PTHREAD_LOCK_SHARED
+PROCESS_LOCK_SHARED
+SEMAPHORE_SIGNAL
+FD_SIGNAL
+*/
+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 name the unique name of this ringbuffer.
+ * @param size the requested size.
+ * @param flags or'ed flags
+ * @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.
+ * @see QB_RB_FLAG_CREATE, QB_RB_FLAG_OVERWRITE, QB_RB_FLAG_SHARED_THREAD, QB_RB_FLAG_SHARED_PROCESS
+ */
+qb_ringbuffer_t* qb_rb_open (const char *name, size_t size, uint32_t flags);
+
+/**
+ * Dereference the ringbuffer and if we are the last user destroy it.
+ *
+ * All files, mmaped memory, semaphores and locks will be destroyed.
+ *
+ * @param rb ringbuffer instance
+ */
+void qb_rb_close (qb_ringbuffer_t *rb);
+
+/**
+ * Write a chunk to the ring buffer.
+ *
+ * This simply calls qb_rb_chunk_writable_alloc() and then
+ * qb_rb_chunk_writable_commit().
+ *
+ * @param rb ringbuffer instance
+ * @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 -1).
+ *
+ * @see qb_rb_chunk_writable_alloc()
+ * @see qb_rb_chunk_writable_commit()
+ */
+ssize_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 rb ringbuffer instance
+ * @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.
+ * @param rb ringbuffer instance
+ * @param len (in) the size of the chunk.
+ */
+int32_t 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 rb ringbuffer instance
+ * @param data_out (out) a pointer to the next chunk to read (not copied).
+ *
+ * @return the size of the chunk (0 if buffer empty).
+ */
+ssize_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().
+ * @param rb ringbuffer instance
+ */
+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 rb ringbuffer instance
+ * @param data_out (in/out) the chunk will be memcpy'ed into this.
+ * @param len (in) the size of data_out.
+ * @param ms_timeout the amount od time to wait for new data.
+ * @return the size of the chunk, or error.
+ */
+ssize_t qb_rb_chunk_read (qb_ringbuffer_t *rb, void *data_out, size_t len, int32_t ms_timeout);
+
+/**
+ * The amount of free space in the ring buffer.
+ *
+ * @note Some of this space will be consumed by the chunk headers.
+ * @param rb ringbuffer instance
+ */
+ssize_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).
+ * @param rb ringbuffer instance
+ */
+ssize_t qb_rb_space_used (qb_ringbuffer_t *rb);
+
+/**
+ * Write the contents of the Ring Buffer to file.
+ * @param fd open file to write the ringbuffer data to.
+ * @param rb ringbuffer instance
+ * @see qb_rb_create_from_file()
+ */
+ssize_t qb_rb_write_to_file (qb_ringbuffer_t *rb, int32_t fd);
+
+/**
+ * Load the saved ring buffer from file into tempory memory.
+ * @param fd file with saved ringbuffer data.
+ * @param flags same flags as passed into qb_rb_open()
+ * @return new ringbuffer instance
+ * @see qb_rb_write_to_file()
+ */
+qb_ringbuffer_t *qb_rb_create_from_file (int32_t fd, uint32_t flags);
+
+
+#ifdef __cplusplus
+}
+#endif /* __cplusplus */
+
+#endif /* QB_RB_DEFINED */
+
diff --git a/include/qb/qbutil.h b/include/qb/qbutil.h
index 11c7a1f..d61f650 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,20 +53,25 @@ 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);
+/**
+ * Calls either pthread_mutex_destro() or pthread_spin_destroy().
+ */
+int32_t qb_thread_lock_destroy (qb_thread_lock_t* tl);
+
typedef void (* qb_util_log_fn_t)(const char *file_name,
int32_t file_line,
@@ -62,10 +79,32 @@ 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.
+ * @param file_flags same as passed into open()
+ * @return 0 (success) or -1 (error)
+ */
+int32_t qb_util_mmap_file_open (char *path, const char *file, size_t bytes, uint32_t file_flags);
+
+
+/**
+ * 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..f1e1d03 100644
--- a/lib/Makefile.am
+++ b/lib/Makefile.am
@@ -21,11 +21,12 @@
MAINTAINERCLEANFILES = Makefile.in
-noinst_HEADERS = ipc_int.h util_int.h
+noinst_HEADERS = ipc_int.h util_int.h ringbuffer.h
lib_LIBRARIES = libqbplugin.a
-libqbplugin_a_SOURCES = plugin_loader.c
+libqbplugin_a_SOURCES = plugin_loader.c $(top_builddir)/include/qb/qbrb.h $(top_builddir)/include/qb/qbutil.h
+libqbplugin_a_CPPFLAGS = -I$(top_builddir)/include -I$(top_srcdir)/include
#
# Here are a set of rules to help you update your library version information:
@@ -43,7 +44,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 ringbuffer_helper.c
pkgconfigdir = $(libdir)/pkgconfig
pkgconfig_DATA = libqb.pc
diff --git a/lib/ringbuffer.c b/lib/ringbuffer.c
new file mode 100644
index 0000000..c031f33
--- /dev/null
+++ b/lib/ringbuffer.c
@@ -0,0 +1,626 @@
+/*
+ * 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 "ringbuffer.h"
+
+//#define CRAZY_DEBUG_PRINTFS 1
+#ifdef CRAZY_DEBUG_PRINTFS
+#define DEBUG_PRINTF(format, args...) \
+do { \
+ printf(format, ##args); \
+} while(0)
+#else
+#define DEBUG_PRINTF(format, 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 0xAAAAAAAA
+#define CHUNK_SIZE_GET(rb, pointer) rb->shared_data[pointer]
+#define CHUNK_MAGIC_GET(rb, pointer) rb->shared_data[(pointer + 1) % rb->shared_hdr->size]
+
+#define FDHEAD_INDEX (rb->shared_hdr->size)
+#define FDTAIL_INDEX (rb->shared_hdr->size + 1)
+
+#define idx_step(idx) \
+do { \
+ if (idx > (rb->shared_hdr->size - 1)) { \
+ idx = ((idx) % (rb->shared_hdr->size)); \
+ } \
+} while (0)
+
+/*
+ * move the write pointer to the next 32 byte boundry
+ * write_pt goes in 4 bytes (sizeof(uint32_t))
+ */
+//#define USE_CACHE_LINE_ALIGNMENT 1
+#ifdef USE_CACHE_LINE_ALIGNMENT
+#define idx_cache_line_step(idx) \
+do { \
+ if (idx % 8) { \
+ idx += (8 - (idx % 8)); \
+ } \
+ if (idx > (rb->shared_hdr->size - 1)) { \
+ idx = ((idx) % (rb->shared_hdr->size)); \
+ } \
+} while (0)
+#else
+#define idx_cache_line_step(idx) \
+do { \
+ if (idx > (rb->shared_hdr->size - 1)) { \
+ idx = ((idx) % (rb->shared_hdr->size)); \
+ } \
+} while (0)
+#endif
+
+static void _check_chunk_locked_ (qb_ringbuffer_t * rb, uint32_t pointer);
+static size_t _qb_rb_space_used_locked_ (qb_ringbuffer_t * rb);
+static void _qb_rb_chunk_reclaim_locked_ (qb_ringbuffer_t * rb);
+
+qb_ringbuffer_t *qb_rb_open (const char *name, size_t size, uint32_t flags)
+{
+ struct qb_ringbuffer_s *rb = malloc (sizeof (struct qb_ringbuffer_s));
+ size_t real_size = ROUNDUP (size, sysconf (_SC_PAGESIZE));
+ char path[PATH_MAX];
+ int32_t fd_hdr;
+ int32_t fd_data;
+ uint32_t file_flags = O_RDWR;
+
+ if (flags & QB_RB_FLAG_CREATE) {
+ file_flags |= O_CREAT | O_TRUNC;
+ }
+ /*
+ * Create a shared_hdr memory segment for the header.
+ */
+ fd_hdr = qb_util_mmap_file_open (path, name,
+ sizeof (struct qb_ringbuffer_shared_s),
+ file_flags);
+ if (fd_hdr == -1) {
+ qb_util_log (LOG_ERR, "couldn't create file for mmap");
+ return NULL;
+ }
+
+ rb->shared_hdr = mmap (0,
+ sizeof (struct qb_ringbuffer_shared_s),
+ PROT_READ | PROT_WRITE, MAP_SHARED, fd_hdr, 0);
+
+ if (rb->shared_hdr == MAP_FAILED) {
+ qb_util_log (LOG_ERR, "couldn't create mmap for header");
+ goto cleanup_hdr;
+ }
+
+ rb->flags = flags;
+
+ /*
+ * create the semaphore
+ */
+ if (flags & QB_RB_FLAG_CREATE) {
+ rb->shared_data = NULL;
+ /* rb->shared_hdr->size tracks data by ints and not bytes/chars. */
+ rb->shared_hdr->size = real_size / sizeof (uint32_t);
+ rb->shared_hdr->write_pt = 0;
+ rb->shared_hdr->read_pt = 0;
+ rb->shared_hdr->count = 0;
+ strncpy (rb->shared_hdr->hdr_path, path, PATH_MAX);
+ }
+ if (my_lock_it_create (rb, flags) == -1) {
+ qb_util_log (LOG_ERR, "couldn't get a shared lock %s",
+ strerror (errno));
+ goto cleanup_hdr;
+ }
+
+ if (my_sem_create (rb, flags) == -1) {
+ qb_util_log (LOG_ERR, "couldn't get a semaphore %s",
+ strerror (errno));
+ goto cleanup_hdr;
+ }
+
+ /* Create the shared_data memory segment for the actual ringbuffer.
+ * They have to be seperate.
+ */
+ if (flags & QB_RB_FLAG_CREATE) {
+ fd_data = qb_util_mmap_file_open (path,
+ "qb-ringbuffer-XXXXXX",
+ real_size, file_flags);
+ strncpy (rb->shared_hdr->data_path, path, PATH_MAX);
+ } else {
+ fd_data = qb_util_mmap_file_open (path,
+ rb->shared_hdr->data_path,
+ real_size, file_flags);
+ }
+ if (fd_data == -1) {
+ qb_util_log (LOG_ERR, "couldn't create file for mmap");
+ goto cleanup_hdr;
+ }
+
+ qb_util_log (LOG_INFO,
+ "shm \n size:%zd\n real_size:%zd\n rb->size:%d\n", size,
+ real_size, rb->shared_hdr->size);
+
+ if (qb_util_circular_mmap (fd_data,
+ (void **)&rb->shared_data, real_size) != 0) {
+ goto cleanup_data;
+ }
+
+ if (flags & QB_RB_FLAG_CREATE) {
+ memset (rb->shared_data, 0, real_size);
+ rb->shared_data[rb->shared_hdr->size] = 5;
+ rb->shared_hdr->ref_count = 1;
+ } else {
+ my_lock_it (rb);
+ rb->shared_hdr->ref_count++;
+ my_unlock_it (rb);
+ }
+
+ return rb;
+
+cleanup_data:
+ close (fd_data);
+ if (flags & QB_RB_FLAG_CREATE) {
+ unlink (rb->shared_hdr->data_path);
+ }
+
+cleanup_hdr:
+ close (fd_hdr);
+ if (flags & QB_RB_FLAG_CREATE) {
+ unlink (rb->shared_hdr->hdr_path);
+ my_lock_it_destroy (rb);
+ my_sem_destroy (rb);
+ }
+ if (rb && (rb->shared_hdr != MAP_FAILED && rb->shared_hdr != NULL)) {
+ munmap (rb->shared_hdr, sizeof (struct qb_ringbuffer_shared_s));
+ }
+ free (rb);
+ return NULL;
+}
+
+void qb_rb_close (qb_ringbuffer_t * rb)
+{
+ int32_t destroy_it = 0;
+
+ my_lock_it (rb);
+ rb->shared_hdr->ref_count--;
+ qb_util_log (LOG_INFO, "ref_count:%d", rb->shared_hdr->ref_count);
+ if (rb->shared_hdr->ref_count == 0) {
+ destroy_it = 1;
+ }
+ my_unlock_it (rb);
+
+ if (destroy_it) {
+ qb_util_log (LOG_INFO, "Destroying ringbuffer");
+ my_lock_it_destroy (rb);
+ my_sem_destroy (rb);
+
+ unlink (rb->shared_hdr->data_path);
+ unlink (rb->shared_hdr->hdr_path);
+ }
+
+ munmap (rb->shared_data, rb->shared_hdr->size);
+ munmap (rb->shared_hdr, sizeof (struct qb_ringbuffer_shared_s));
+ free (rb);
+}
+
+static size_t _qb_rb_space_free_locked_ (qb_ringbuffer_t * rb)
+{
+ uint32_t write_size;
+ uint32_t read_size;
+ size_t space_free = 0;
+
+ write_size = rb->shared_hdr->write_pt;
+ // TODO idx_cache_line_step (write_size);
+ read_size = rb->shared_hdr->read_pt;
+
+ if (write_size > read_size) {
+ space_free =
+ (read_size - write_size + rb->shared_hdr->size) - 1;
+ } else if (write_size < read_size) {
+ space_free = (read_size - write_size) - 1;
+ } else {
+ space_free = rb->shared_hdr->size;
+ }
+
+ /* word -> bytes */
+ return (space_free * sizeof (uint32_t));
+}
+
+ssize_t qb_rb_space_free (qb_ringbuffer_t * rb)
+{
+ size_t space_free;
+
+ if (my_lock_it (rb) == -1) {
+ return -1;
+ }
+ space_free = _qb_rb_space_free_locked_ (rb);
+ if (my_unlock_it (rb) == -1) {
+ /* aarg stuck locked! */
+ qb_util_log (LOG_ERR, "failed to unlock ringbuffer lock %s",
+ strerror (errno));
+ return -1;
+ }
+ return space_free;
+}
+
+/*
+ * TODO write a function that returns the number of chunks
+ * in the rb.
+ *
+ */
+
+static size_t _qb_rb_space_used_locked_ (qb_ringbuffer_t * rb)
+{
+ uint32_t write_size;
+ uint32_t read_size;
+ size_t space_used;
+
+ write_size = rb->shared_hdr->write_pt;
+ read_size = rb->shared_hdr->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->shared_hdr->size) - 1;
+ } else {
+ space_used = 0;
+ }
+ /* word -> bytes */
+ return (space_used * sizeof (uint32_t));
+}
+
+ssize_t qb_rb_space_used (qb_ringbuffer_t * rb)
+{
+ size_t used;
+
+ if (my_lock_it (rb) == -1) {
+ return -1;
+ }
+ used = _qb_rb_space_used_locked_ (rb);
+ if (my_unlock_it (rb) == -1) {
+ /* aarg stuck locked! */
+ qb_util_log (LOG_ERR, "failed to unlock ringbuffer lock %s",
+ strerror (errno));
+ return -1;
+ }
+ return used;
+}
+
+void *qb_rb_chunk_writable_alloc (qb_ringbuffer_t * rb, size_t len)
+{
+ uint32_t idx;
+
+ my_lock_it (rb);
+ /*
+ * Reclaim data if we are over writing and we need space
+ */
+ if (rb->flags & QB_RB_FLAG_OVERWRITE) {
+ while (_qb_rb_space_free_locked_ (rb) <
+ (len + RB_CHUNK_HEADER_SIZE + 4)) {
+ _qb_rb_chunk_reclaim_locked_ (rb);
+ }
+ } else {
+ if (_qb_rb_space_free_locked_ (rb) <
+ (len + RB_CHUNK_HEADER_SIZE + 4)) {
+ my_unlock_it (rb);
+ return NULL;
+ }
+ }
+
+ idx = rb->shared_hdr->write_pt;
+ /*
+ * insert the chunk header
+ */
+ rb->shared_data[idx++] = 0;
+ idx_step (idx);
+ rb->shared_data[idx++] = RB_CHUNK_MAGIC;
+ idx_step (idx);
+
+ /*
+ * return a pointer to the begining of the chunk data
+ */
+ return (void *)&rb->shared_data[idx];
+
+}
+
+static uint32_t
+_qb_rb_chunk_step_to_next_locked_ (qb_ringbuffer_t * rb, uint32_t pointer)
+{
+ uint32_t chunk_size = CHUNK_SIZE_GET (rb, pointer);
+ /*
+ * skip over the chunk header
+ */
+ pointer += RB_CHUNK_HEADER_WORDS;
+ idx_step (pointer);
+
+ /*
+ * skip over the user's data.
+ */
+ pointer += (chunk_size / sizeof (uint32_t));
+ /* make allowance for non-word sizes */
+ if ((chunk_size % sizeof (uint32_t)) != 0) {
+ pointer++;
+ }
+
+ idx_cache_line_step (pointer);
+ return pointer;
+}
+
+int32_t qb_rb_chunk_writable_commit (qb_ringbuffer_t * rb, size_t len)
+{
+ uint32_t old_write_pt = rb->shared_hdr->write_pt;
+ /*
+ * commit the magic & chunk_size
+ */
+ rb->shared_data[old_write_pt] = len;
+ rb->shared_data[old_write_pt + 1] = RB_CHUNK_MAGIC;
+ rb->shared_hdr->count++;
+
+ /*
+ * commit the new write pointer
+ */
+ rb->shared_hdr->write_pt =
+ _qb_rb_chunk_step_to_next_locked_ (rb, rb->shared_hdr->write_pt);
+
+ DEBUG_PRINTF ("%s: read: %u, write: %u (was:%u)\n", __func__,
+ rb->shared_hdr->read_pt, rb->shared_hdr->write_pt,
+ old_write_pt);
+
+ if (my_unlock_it (rb) != 0) {
+ qb_util_log (LOG_ERR, "failed to unlock ringbuffer lock %s",
+ strerror (errno));
+ return -1;
+ }
+
+ /*
+ * post the notification to the reader
+ */
+ return my_sem_post (rb);
+}
+
+ssize_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 -1;
+ }
+
+ /*
+ * copy the data
+ */
+ memcpy (dest, data, len);
+
+ qb_rb_chunk_writable_commit (rb, len);
+
+ return len;
+}
+
+static void _qb_rb_chunk_reclaim_locked_ (qb_ringbuffer_t * rb)
+{
+ uint32_t old_read_pt = rb->shared_hdr->read_pt;
+
+ if (_qb_rb_space_used_locked_ (rb) == 0) {
+ return;
+ }
+
+ _check_chunk_locked_ (rb, rb->shared_hdr->read_pt);
+
+ rb->shared_hdr->read_pt =
+ _qb_rb_chunk_step_to_next_locked_ (rb, rb->shared_hdr->read_pt);
+ rb->shared_hdr->count--;
+
+ /*
+ * clear the header
+ */
+ rb->shared_data[old_read_pt] = 0;
+ rb->shared_data[old_read_pt + 1] = 0;
+
+ DEBUG_PRINTF ("%s: read: %u (was:%u), write: %u\n", __func__,
+ rb->shared_hdr->read_pt, old_read_pt,
+ rb->shared_hdr->write_pt);
+}
+
+void qb_rb_chunk_reclaim (qb_ringbuffer_t * rb)
+{
+ my_lock_it (rb);
+ _qb_rb_chunk_reclaim_locked_ (rb);
+ my_unlock_it (rb);
+}
+
+ssize_t qb_rb_chunk_peek (qb_ringbuffer_t * rb, void **data_out)
+{
+ uint32_t chunk_size = CHUNK_SIZE_GET (rb, rb->shared_hdr->read_pt);
+ uint32_t chunk_magic = CHUNK_MAGIC_GET (rb, rb->shared_hdr->read_pt);
+
+ *data_out =
+ &rb->shared_data[rb->shared_hdr->read_pt + RB_CHUNK_HEADER_WORDS];
+
+ if (chunk_magic != RB_CHUNK_MAGIC) {
+ errno = ENODATA;
+ return -1;
+ } else {
+ return chunk_size;
+ }
+}
+
+ssize_t
+qb_rb_chunk_read (qb_ringbuffer_t * rb, void *data_out, size_t len,
+ int32_t timeout)
+{
+ uint32_t chunk_size;
+ int32_t res;
+
+ res = my_sem_timedwait (rb, timeout);
+ if (res == -1 && errno != EIDRM) {
+ printf ("%s:%d %s\n", __func__, __LINE__, strerror (errno));
+ return -1;
+ }
+
+ if (my_lock_it (rb) == -1) {
+ res = errno;
+ qb_util_log (LOG_ERR, "could not lock ringbuffer %s",
+ strerror (errno));
+ errno = res;
+ return -1;
+ }
+ if (_qb_rb_space_used_locked_ (rb) == 0) {
+ printf ("%s:%d count:%d\n", __func__, __LINE__,
+ rb->shared_hdr->count);
+ my_unlock_it (rb);
+ errno = ENODATA;
+ return -1;
+ }
+
+ _check_chunk_locked_ (rb, rb->shared_hdr->read_pt);
+ chunk_size = CHUNK_SIZE_GET (rb, rb->shared_hdr->read_pt);
+
+ if (len < chunk_size) {
+ printf ("%s:%d\n", __func__, __LINE__);
+ my_unlock_it (rb);
+ return 0;
+ }
+
+ memcpy (data_out,
+ &rb->shared_data[rb->shared_hdr->read_pt +
+ RB_CHUNK_HEADER_WORDS], chunk_size);
+
+ _qb_rb_chunk_reclaim_locked_ (rb);
+ my_unlock_it (rb);
+
+ return chunk_size;
+}
+
+static void print_header (qb_ringbuffer_t * rb)
+{
+ printf ("Ringbuffer: \n");
+ if (rb->flags & QB_RB_FLAG_OVERWRITE) {
+ printf (" ->OVERWRITE\n");
+ } else {
+ printf (" ->NORMAL\n");
+ }
+ printf (" ->write_pt [%d]\n", rb->shared_hdr->write_pt);
+ printf (" ->read_pt [%d]\n", rb->shared_hdr->read_pt);
+ printf (" ->size [%d words]\n", rb->shared_hdr->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_locked_ (qb_ringbuffer_t * rb, uint32_t pointer)
+{
+ uint32_t chunk_size = CHUNK_SIZE_GET (rb, pointer);
+ uint32_t chunk_magic = CHUNK_MAGIC_GET (rb, pointer);
+
+ if (chunk_magic != RB_CHUNK_MAGIC) {
+ printf ("size: %x\n", chunk_size);
+ printf ("magic: %x\n", chunk_magic);
+ print_header (rb);
+ my_unlock_it (rb);
+ assert (0);
+ }
+}
+
+ssize_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->shared_hdr->size, sizeof (uint32_t));
+ if ((result < 0) || (result != sizeof (uint32_t))) {
+ return -1;
+ }
+ written_size += result;
+
+ result = write (fd, rb->shared_data,
+ rb->shared_hdr->size * sizeof (uint32_t));
+ if ((result < 0)
+ || (result != rb->shared_hdr->size * sizeof (uint32_t))) {
+ return -1;
+ }
+ written_size += result;
+
+ /*
+ * store the read & write pointers
+ */
+ result +=
+ write (fd, (void *)&rb->shared_hdr->write_pt, sizeof (uint32_t));
+ if ((result < 0) || (result != sizeof (uint32_t))) {
+ return -1;
+ }
+ written_size += result;
+ result +=
+ write (fd, (void *)&rb->shared_hdr->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, uint32_t flags)
+{
+ ssize_t n_read;
+ size_t n_required;
+ qb_ringbuffer_t *rb = malloc (sizeof (qb_ringbuffer_t));
+ rb->shared_hdr = malloc (sizeof (struct qb_ringbuffer_shared_s));
+
+ rb->flags = flags;
+
+ n_required = sizeof (uint32_t);
+ n_read = read (fd, &rb->shared_hdr->size, n_required);
+ if (n_read != n_required) {
+ fprintf (stderr, "Unable to read fdata header\n");
+ return NULL;
+ }
+
+ n_required = ((rb->shared_hdr->size + 2) * sizeof (uint32_t));
+
+ if ((rb->shared_data = malloc (n_required)) == NULL) {
+ fprintf (stderr, "exhausted virtual memory\n");
+ return NULL;
+ }
+ n_read = read (fd, rb->shared_data, 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->shared_hdr->write_pt = rb->shared_data[FDHEAD_INDEX];
+ rb->shared_hdr->read_pt = rb->shared_data[FDTAIL_INDEX];
+
+ print_header (rb);
+
+ return rb;
+}
diff --git a/lib/ringbuffer.h b/lib/ringbuffer.h
new file mode 100644
index 0000000..88f38dc
--- /dev/null
+++ b/lib/ringbuffer.h
@@ -0,0 +1,88 @@
+/*
+ * 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 _RINGBUFFER_H_
+#define _RINGBUFFER_H_
+
+#include <config.h>
+
+#ifndef _GNU_SOURCE
+#define _GNU_SOURCE 1
+#endif
+
+#include "os_base.h"
+
+#include <limits.h>
+#include <sys/mman.h>
+#include <sys/sem.h>
+#include <sys/ipc.h>
+#include <pthread.h>
+#include <semaphore.h>
+
+#include "util_int.h"
+#include <qb/qbutil.h>
+#include <qb/qbrb.h>
+
+struct qb_ringbuffer_shared_s {
+ volatile uint32_t write_pt;
+ volatile uint32_t read_pt;
+ uint32_t size;
+ uint32_t count;
+ char hdr_path[PATH_MAX];
+ char data_path[PATH_MAX];
+ int32_t ref_count;
+#if _POSIX_THREAD_PROCESS_SHARED > 0
+ sem_t posix_sem;
+ pthread_spinlock_t spinlock;
+#endif
+};
+
+struct qb_ringbuffer_s {
+ uint32_t flags;
+#if _POSIX_THREAD_PROCESS_SHARED < 1
+ int32_t lock_id;
+ int32_t sem_id;
+#endif
+ struct qb_ringbuffer_shared_s *shared_hdr;
+ uint32_t *shared_data;
+};
+
+#if defined(_SEM_SEMUN_UNDEFINED)
+union semun {
+ int val;
+ struct semid_ds *buf;
+ unsigned short int *array;
+ struct seminfo *__buf;
+};
+#endif /* _SEM_SEMUN_UNDEFINED */
+
+#define RB_NS_IN_MSEC 1000000ULL
+
+int32_t my_lock_it_create (qb_ringbuffer_t * rb, uint32_t flags);
+int32_t my_lock_it (qb_ringbuffer_t * rb);
+int32_t my_unlock_it (qb_ringbuffer_t * rb);
+int32_t my_lock_it_destroy (qb_ringbuffer_t * rb);
+
+int32_t my_sem_create (qb_ringbuffer_t * rb, uint32_t flags);
+int32_t my_sem_post (qb_ringbuffer_t * rb);
+int32_t my_sem_timedwait (qb_ringbuffer_t * rb, int32_t ms_timeout);
+int32_t my_sem_destroy (qb_ringbuffer_t * rb);
+
+#endif /* _RINGBUFFER_H_ */
diff --git a/lib/ringbuffer_helper.c b/lib/ringbuffer_helper.c
new file mode 100644
index 0000000..71d0640
--- /dev/null
+++ b/lib/ringbuffer_helper.c
@@ -0,0 +1,378 @@
+/*
+ * 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 "ringbuffer.h"
+
+#if _POSIX_THREAD_PROCESS_SHARED > 0
+static int32_t my_posix_sem_create (qb_ringbuffer_t * rb, uint32_t flags)
+{
+ int32_t pshared = 0;
+ if (flags & QB_RB_FLAG_SHARED_PROCESS) {
+ if ((flags & QB_RB_FLAG_CREATE) == 0) {
+ return 0;
+ }
+ pshared = 1;
+ }
+ return sem_init (&rb->shared_hdr->posix_sem, pshared, 0);
+}
+#else
+static int32_t my_sysv_sem_create (qb_ringbuffer_t * rb, uint32_t flags)
+{
+ union semun options;
+ int32_t res;
+ key_t sem_key;
+
+ sem_key = ftok (rb->shared_hdr->hdr_path, (rb->shared_hdr->size + 1));
+
+ if (sem_key == -1) {
+ qb_util_log (LOG_ERR, "couldn't get a sem id %s",
+ strerror (errno));
+ return -1;
+ }
+
+ if (flags & QB_RB_FLAG_CREATE) {
+ rb->sem_id = semget (sem_key, 1, IPC_CREAT | IPC_EXCL | 0600);
+ if (rb->sem_id == -1) {
+ qb_util_log (LOG_ERR, "couldn't create a semaphore %s",
+ strerror (errno));
+ return -1;
+ }
+ options.val = 0;
+ res = semctl (rb->sem_id, 0, SETVAL, options);
+ } else {
+ rb->sem_id = semget (sem_key, 0, 0600);
+ if (rb->sem_id == -1) {
+ qb_util_log (LOG_ERR, "couldn't get a sem id %s",
+ strerror (errno));
+ return -1;
+ }
+ res = 0;
+ }
+ qb_util_log (LOG_INFO, "sem key:%d, id:%d, value:%d",
+ sem_key, rb->sem_id, semctl (rb->sem_id, 0, GETVAL, 0));
+
+ return res;
+}
+#endif
+
+int32_t my_sem_create (qb_ringbuffer_t * rb, uint32_t flags)
+{
+ if ((rb->flags & QB_RB_FLAG_SHARED_PROCESS) == 0) {
+ return 0;
+ }
+#if _POSIX_THREAD_PROCESS_SHARED > 0
+ return my_posix_sem_create (rb, flags);
+#else
+ return my_sysv_sem_create (rb, flags);
+#endif
+}
+
+#if _POSIX_THREAD_PROCESS_SHARED > 0
+static int32_t my_posix_sem_post (qb_ringbuffer_t * rb)
+{
+ return sem_post (&rb->shared_hdr->posix_sem);
+}
+#else
+static int32_t my_sysv_sem_post (qb_ringbuffer_t * rb)
+{
+ struct sembuf sops[1];
+
+ if ((rb->flags & QB_RB_FLAG_SHARED_PROCESS) == 0) {
+ return 0;
+ }
+
+ sops[0].sem_num = 0;
+ sops[0].sem_op = 1;
+ sops[0].sem_flg = 0;
+
+semop_again:
+ if (semop (rb->sem_id, sops, 1) == -1) {
+ if (errno == EINTR) {
+ goto semop_again;
+ } else {
+ qb_util_log (LOG_ERR,
+ "could not increment semaphore : %s",
+ strerror (errno));
+ }
+
+ return -1;
+ }
+ return 0;
+}
+#endif
+
+int32_t my_sem_post (qb_ringbuffer_t * rb)
+{
+ if ((rb->flags & QB_RB_FLAG_SHARED_PROCESS) == 0) {
+ return 0;
+ }
+#if _POSIX_THREAD_PROCESS_SHARED > 0
+ return my_posix_sem_post (rb);
+#else
+ return my_sysv_sem_post (rb);
+#endif
+}
+
+#if _POSIX_THREAD_PROCESS_SHARED > 0
+static int32_t my_posix_sem_timedwait (qb_ringbuffer_t * rb, int32_t ms_timeout)
+{
+ struct timespec ts_timeout;
+
+ if (ms_timeout >= 0) {
+ ts_timeout.tv_sec = time (NULL);
+ ts_timeout.tv_sec += (ms_timeout / 1000);
+ ts_timeout.tv_nsec = (ms_timeout % 1000) * RB_NS_IN_MSEC;
+ return sem_timedwait (&rb->shared_hdr->posix_sem, &ts_timeout);
+ } else {
+ return sem_wait (&rb->shared_hdr->posix_sem);
+ }
+}
+
+#else
+
+static int32_t my_sysv_sem_timedwait (qb_ringbuffer_t * rb, int32_t ms_timeout)
+{
+ struct sembuf sops[1];
+ struct timespec ts_timeout;
+ struct timespec *ts_pt;
+
+ if (ms_timeout >= 0) {
+ /*
+ * Note: sem_timedwait takes an absolute time where as semtimedop
+ * takes a relative time.
+ */
+ ts_timeout.tv_sec = 0;
+ ts_timeout.tv_sec += (ms_timeout / 1000);
+ ts_timeout.tv_nsec = (ms_timeout % 1000) * RB_NS_IN_MSEC;
+ ts_pt = &ts_timeout;
+ } else {
+ ts_pt = NULL;
+ }
+
+ /*
+ * wait for sem post.
+ */
+ sops[0].sem_num = 0;
+ sops[0].sem_op = -1;
+ sops[0].sem_flg = 0;
+
+semop_again:
+ if (semtimedop (rb->sem_id, sops, 1, ts_pt) == -1) {
+ if (errno == EINTR) {
+ goto semop_again;
+ } else if (errno == EAGAIN) {
+ /* make consistent with sem_timedwait */
+ errno = ETIMEDOUT;
+ } else {
+ qb_util_log (LOG_ERR,
+ "error waiting for semaphore : %s",
+ strerror (errno));
+ }
+
+ return -1;
+ }
+
+ return 0;
+}
+#endif
+
+int32_t my_sem_timedwait (qb_ringbuffer_t * rb, int32_t ms_timeout)
+{
+ if ((rb->flags & QB_RB_FLAG_SHARED_PROCESS) == 0) {
+ return 0;
+ }
+#if _POSIX_THREAD_PROCESS_SHARED > 0
+ return my_posix_sem_timedwait (rb, ms_timeout);
+#else
+ return my_sysv_sem_timedwait (rb, ms_timeout);
+#endif
+}
+
+int32_t my_sem_destroy (qb_ringbuffer_t * rb)
+{
+#if _POSIX_THREAD_PROCESS_SHARED > 0
+ return sem_destroy (&rb->shared_hdr->posix_sem);
+#else
+ return semctl (rb->sem_id, 0, IPC_RMID, 0);
+#endif
+}
+
+#if _POSIX_THREAD_PROCESS_SHARED > 0
+static int32_t my_posix_lock_it_create (qb_ringbuffer_t * rb, uint32_t flags)
+{
+ if (flags & QB_RB_FLAG_CREATE) {
+ return pthread_spin_init (&rb->shared_hdr->spinlock,
+ PTHREAD_PROCESS_SHARED);
+ } else {
+ return 0;
+ }
+}
+#else
+static int32_t my_sysv_lock_it_create (qb_ringbuffer_t * rb, uint32_t flags)
+{
+ union semun options;
+ int32_t res;
+ key_t sem_key;
+
+ sem_key = ftok (rb->shared_hdr->hdr_path, rb->shared_hdr->size);
+
+ if (sem_key == -1) {
+ qb_util_log (LOG_ERR, "couldn't get a sem id %s",
+ strerror (errno));
+ return -1;
+ }
+
+ if (flags & QB_RB_FLAG_CREATE) {
+ rb->lock_id = semget (sem_key, 1, IPC_CREAT | IPC_EXCL | 0600);
+ if (rb->lock_id == -1) {
+ qb_util_log (LOG_ERR, "couldn't create a semaphore %s",
+ strerror (errno));
+ return -1;
+ }
+ options.val = 0;
+ res = semctl (rb->lock_id, 0, SETVAL, options);
+ } else {
+ rb->lock_id = semget (sem_key, 0, 0600);
+ if (rb->lock_id == -1) {
+ qb_util_log (LOG_ERR, "couldn't get a sem id %s",
+ strerror (errno));
+ return -1;
+ }
+ res = 0;
+ }
+ qb_util_log (LOG_INFO, "sem key:%d, id:%d, value:%d",
+ sem_key, rb->lock_id, semctl (rb->lock_id, 0, GETVAL, 0));
+
+ return res;
+}
+#endif
+
+int32_t my_lock_it_create (qb_ringbuffer_t * rb, uint32_t flags)
+{
+ if ((rb->flags & QB_RB_FLAG_SHARED_PROCESS) == 0) {
+ return 0;
+ }
+#if _POSIX_THREAD_PROCESS_SHARED > 0
+ return my_posix_lock_it_create (rb, flags);
+#else
+ return my_sysv_lock_it_create (rb, flags);
+#endif
+}
+
+#if _POSIX_THREAD_PROCESS_SHARED > 0
+static int32_t my_posix_lock_it (qb_ringbuffer_t * rb)
+{
+ return pthread_spin_lock (&rb->shared_hdr->spinlock);
+}
+#else
+static int32_t my_sysv_lock_it (qb_ringbuffer_t * rb)
+{
+ struct sembuf sops[2];
+
+ /*
+ * atomically wait for sem to get to 0 and then incr.
+ */
+ sops[0].sem_num = 0;
+ sops[0].sem_op = 0;
+ sops[0].sem_flg = 0;
+
+ sops[1].sem_num = 0;
+ sops[1].sem_op = 1;
+ sops[1].sem_flg = 0;
+
+semop_again:
+ if (semop (rb->lock_id, sops, 2) == -1) {
+ if (errno == EINTR) {
+ goto semop_again;
+ } else {
+ qb_util_log (LOG_ERR, "could not lock it : %s",
+ strerror (errno));
+ }
+ return -1;
+ }
+ return 0;
+}
+#endif
+
+int32_t my_lock_it (qb_ringbuffer_t * rb)
+{
+ if ((rb->flags & QB_RB_FLAG_SHARED_PROCESS) == 0) {
+ return 0;
+ }
+#if _POSIX_THREAD_PROCESS_SHARED > 0
+ return my_posix_lock_it (rb);
+#else
+ return my_sysv_lock_it (rb);
+#endif
+}
+
+#if _POSIX_THREAD_PROCESS_SHARED > 0
+static int32_t my_posix_unlock_it (qb_ringbuffer_t * rb)
+{
+ return pthread_spin_unlock (&rb->shared_hdr->spinlock);
+}
+#else
+static int32_t my_sysv_unlock_it (qb_ringbuffer_t * rb)
+{
+ struct sembuf lock_it;
+
+ lock_it.sem_num = 0;
+ lock_it.sem_op = -1;
+ lock_it.sem_flg = IPC_NOWAIT;
+
+semop_again:
+ if (semop (rb->lock_id, &lock_it, 1) == -1) {
+ if (errno == EINTR) {
+ goto semop_again;
+ } else {
+ qb_util_log (LOG_ERR, "could not unlock it : %s",
+ strerror (errno));
+ }
+ return -1;
+ }
+ return 0;
+}
+#endif
+
+int32_t my_unlock_it (qb_ringbuffer_t * rb)
+{
+ if ((rb->flags & QB_RB_FLAG_SHARED_PROCESS) == 0) {
+ return 0;
+ }
+#if _POSIX_THREAD_PROCESS_SHARED > 0
+ return my_posix_unlock_it (rb);
+#else
+ return my_sysv_unlock_it (rb);
+#endif
+}
+
+int32_t my_lock_it_destroy (qb_ringbuffer_t * rb)
+{
+ if ((rb->flags & QB_RB_FLAG_SHARED_PROCESS) == 0) {
+ return 0;
+ }
+#if _POSIX_THREAD_PROCESS_SHARED > 0
+ return pthread_spin_destroy (&rb->shared_hdr->spinlock);
+#else
+ return semctl (rb->lock_id, 0, IPC_RMID, 0);
+#endif
+}
diff --git a/lib/util.c b/lib/util.c
index f9acd5b..5bb7f40 100644
--- a/lib/util.c
+++ b/lib/util.c
@@ -22,6 +22,7 @@
#include "os_base.h"
#include "util_int.h"
+#include <syslog.h>
#include <stdarg.h>
#include <sys/shm.h>
#include <sys/mman.h>
@@ -31,6 +32,9 @@
#else
#include <sys/sem.h>
#endif
+#include <sys/stat.h>
+#include <fcntl.h>
+#include <limits.h>
#include <qb/qbutil.h>
struct qb_thread_lock_s {
@@ -39,13 +43,11 @@ struct qb_thread_lock_s {
pthread_spinlock_t spinlock;
#endif
pthread_mutex_t mutex;
-
};
-
-qb_thread_lock_t* qb_thread_lock_create (qb_thread_lock_type_t type)
+qb_thread_lock_t *qb_thread_lock_create (qb_thread_lock_type_t type)
{
- struct qb_thread_lock_s* tl = malloc (sizeof (struct qb_thread_lock_s));
+ struct qb_thread_lock_s *tl = malloc (sizeof (struct qb_thread_lock_s));
int res;
#if defined(HAVE_PTHREAD_SPIN_LOCK)
@@ -66,7 +68,7 @@ qb_thread_lock_t* qb_thread_lock_create (qb_thread_lock_type_t type)
}
}
-int32_t qb_thread_lock (qb_thread_lock_t* tl)
+int32_t qb_thread_lock (qb_thread_lock_t * tl)
{
int res;
#if defined(HAVE_PTHREAD_SPIN_LOCK)
@@ -77,10 +79,10 @@ int32_t qb_thread_lock (qb_thread_lock_t* tl)
{
res = pthread_mutex_lock (&tl->mutex);
}
- return (int32_t)res;
+ return res;
}
-int32_t qb_thread_unlock (qb_thread_lock_t* tl)
+int32_t qb_thread_unlock (qb_thread_lock_t * tl)
{
int res;
#if defined(HAVE_PTHREAD_SPIN_LOCK)
@@ -91,10 +93,10 @@ int32_t qb_thread_unlock (qb_thread_lock_t* tl)
{
res = pthread_mutex_unlock (&tl->mutex);
}
- return (int32_t)res;
+ return res;
}
-int32_t qb_thread_trylock (qb_thread_lock_t* tl)
+int32_t qb_thread_trylock (qb_thread_lock_t * tl)
{
int res;
#if defined(HAVE_PTHREAD_SPIN_LOCK)
@@ -105,9 +107,23 @@ int32_t qb_thread_trylock (qb_thread_lock_t* tl)
{
res = pthread_mutex_trylock (&tl->mutex);
}
- return (int32_t)res;
+ return res;
}
+int32_t qb_thread_lock_destroy (qb_thread_lock_t * tl)
+{
+ int32_t res;
+#if defined(HAVE_PTHREAD_SPIN_LOCK)
+ if (tl->type == QB_THREAD_LOCK_SHORT) {
+ res = pthread_spin_destroy (&tl->spinlock);
+ } else
+#endif
+ {
+ res = pthread_mutex_destroy (&tl->mutex);
+ }
+ free (tl);
+ return res;
+}
/*
* ---------------------------------------------------
@@ -116,10 +132,7 @@ int32_t qb_thread_trylock (qb_thread_lock_t* tl)
static qb_util_log_fn_t real_log_fn = NULL;
void _qb_util_log (const char *file_name,
- int32_t file_line,
- int32_t severity,
- const char *format,
- ...)
+ int32_t file_line, int32_t severity, const char *format, ...)
{
if (real_log_fn) {
va_list ap;
@@ -133,10 +146,86 @@ void _qb_util_log (const char *file_name,
}
}
-
void qb_util_set_log_function (qb_util_log_fn_t fn)
{
real_log_fn = fn;
}
+static int32_t open_mmap_file (char *path, uint32_t file_flags)
+{
+ if (strstr (path, "XXXXXX") != NULL) {
+ return mkstemp (path);
+ }
+
+ return open (path, file_flags, 0600);
+}
+
+/*
+ * ---------------------------------------------------
+ * shared memory functions.
+ */
+int32_t qb_util_mmap_file_open (char *path, const char *file, size_t bytes,
+ uint32_t file_flags)
+{
+ int32_t fd;
+ char *is_absolute = strchr (file, '/');;
+
+ if (is_absolute) {
+ strcpy (path, file);
+ } else {
+ snprintf (path, PATH_MAX, "/dev/shm/%s", file);
+ }
+ fd = open_mmap_file (path, file_flags);
+ if (fd == -1 && !is_absolute) {
+ qb_util_log (LOG_ERR, "couldn't open file %s error: %s",
+ path, strerror (errno));
+ snprintf (path, PATH_MAX, LOCALSTATEDIR "/run/%s", file);
+ fd = open_mmap_file (path, file_flags);
+ if (fd == -1) {
+ return -1;
+ }
+ }
+
+ if (fd != -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/.gitignore b/tests/.gitignore
index 16f6463..7cfb5d3 100644
--- a/tests/.gitignore
+++ b/tests/.gitignore
@@ -3,4 +3,5 @@ check_hash
bmc
bmcpt
bms
-
+rbreader
+rbwriter
diff --git a/tests/Makefile.am b/tests/Makefile.am
index 579a9da..bcb05aa 100644
--- a/tests/Makefile.am
+++ b/tests/Makefile.am
@@ -18,20 +18,28 @@
# along with libqb. If not, see <http://www.gnu.org/licenses/>.
#
-noinst_PROGRAMS = bmc bmcpt bms
+noinst_PROGRAMS = bmc bmcpt bms rbwriter rbreader
-bmc_SOURCES = bmc.c
-bmc_CFLAGS = -I$(top_srcdir)/include
+bmc_SOURCES = bmc.c $(top_builddir)/include/qb/qbipcc.h
+bmc_CPPFLAGS = -I$(top_builddir)/include -I$(top_srcdir)/include
bmc_LDADD = -lrt $(top_builddir)/lib/libqb.la
-bmcpt_SOURCES = bmcpt.c
-bmcpt_CFLAGS = -I$(top_srcdir)/include
+bmcpt_SOURCES = bmcpt.c $(top_builddir)/include/qb/qbipcc.h
+bmcpt_CPPFLAGS = -I$(top_builddir)/include -I$(top_srcdir)/include
bmcpt_LDADD = -lrt $(top_builddir)/lib/libqb.la
-bms_SOURCES = bms.c
-bms_CFLAGS = -I$(top_srcdir)/include
+bms_SOURCES = bms.c $(top_builddir)/include/qb/qbipcs.h
+bms_CPPFLAGS = -I$(top_builddir)/include -I$(top_srcdir)/include
bms_LDADD = -lrt $(top_builddir)/lib/libqb.la
+rbwriter_SOURCES = rbwriter.c $(top_builddir)/include/qb/qbrb.h
+rbwriter_CPPFLAGS = -I$(top_builddir)/include -I$(top_srcdir)/include
+rbwriter_LDADD = -lrt $(top_builddir)/lib/libqb.la
+
+rbreader_SOURCES = rbreader.c $(top_builddir)/include/qb/qbrb.h
+rbreader_CPPFLAGS = -I$(top_builddir)/include -I$(top_srcdir)/include
+rbreader_LDADD = -lrt $(top_builddir)/lib/libqb.la
+
#noinst_libdir = $(abs_top_builddir)/tests
#noinst_lib_LTLIBRARIES = libtest_a.la libtest_b.la
#
@@ -45,12 +53,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..1bc4d11
--- /dev/null
+++ b/tests/check_rb.c
@@ -0,0 +1,227 @@
+/*
+ * 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 <errno.h>
+#include <check.h>
+#include <qb/qbrb.h>
+#include <qb/qbipc_common.h>
+#include <qb/qbutil.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;
+ ssize_t actual;
+ ssize_t avail;
+
+ rb = qb_rb_open ("test1", 200, QB_RB_FLAG_CREATE);
+ fail_if (rb == NULL);
+
+ 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, -1);
+ } 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, 0);
+ if (actual == -1) {
+ break;
+ }
+ str[actual - sizeof (qb_ipc_request_header_t)] = '\0';
+
+ ck_assert_int_eq (actual, hdr->size);
+ }
+ }
+ qb_rb_close (rb);
+}
+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_open ("test2", 200 * sizeof(int64_t), QB_RB_FLAG_CREATE);
+ fail_if (t == NULL);
+ for (i = 0; i < 200; i++) {
+ l = qb_rb_chunk_write (t, &v, sizeof (v));
+ ck_assert_int_eq (l, sizeof (v));
+ }
+ for (i = 0; i < 100; i++) {
+ l = qb_rb_chunk_peek (t, (void**)&new_data);
+ if (l == -1) {
+ /* no more to read */
+ break;
+ }
+ ck_assert_int_eq (l, sizeof (v));
+ fail_unless (v == *new_data);
+ qb_rb_chunk_reclaim (t);
+ }
+ for (i = 0; i < 100; i++) {
+ l = qb_rb_chunk_write (t, &v, sizeof (v));
+ ck_assert_int_eq (l, sizeof (v));
+ }
+ for (i = 0; i < 100; i++) {
+ l = qb_rb_chunk_peek (t, (void**)&new_data);
+ if (l == -1) {
+ /* no more to read */
+ break;
+ }
+ ck_assert_int_eq (l, sizeof (v));
+ fail_unless (v == *new_data);
+ qb_rb_chunk_reclaim (t);
+ }
+ qb_rb_close (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_open ("test3", 10, QB_RB_FLAG_CREATE | QB_RB_FLAG_OVERWRITE);
+ fail_if (t == NULL);
+ 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, 0);
+ if (l == -1) {
+ /* no more to read */
+ break;
+ }
+ ck_assert_int_eq (l, len);
+ ck_assert_str_eq (v, out);
+ }
+ qb_rb_close (t);
+}
+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_open ("test4", 10, QB_RB_FLAG_CREATE | QB_RB_FLAG_OVERWRITE);
+ fail_if (t == NULL);
+ for (i = 0; i < 2000; i++) {
+ l = qb_rb_chunk_write (t, data, strlen (data));
+ ck_assert_int_eq (l, strlen (data));
+ if (i == 0) {
+ data[0]='b';
+ }
+ }
+ for (i = 0; i < 2000; i++) {
+ l = qb_rb_chunk_peek (t, (void**)&new_data);
+ if (l == -1 && errno == ENODATA) {
+ break;
+ }
+ ck_assert_int_eq (l, strlen (data));
+ qb_rb_chunk_reclaim (t);
+ }
+ qb_rb_close (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);
+ tcase_add_test (tc_load, test_ring_buffer4);
+ suite_add_tcase (s, tc_load);
+
+ return s;
+}
+
+static void libqb_log_fn(const char *file_name,
+ int32_t file_line,
+ int32_t severity,
+ const char *msg)
+{
+ printf ("libqb: %s:%d %s\n", file_name, file_line, msg);
+}
+
+
+int main (void)
+{
+ int number_failed;
+
+ Suite *s = rb_suite ();
+ SRunner *sr = srunner_create (s);
+
+ qb_util_set_log_function (libqb_log_fn);
+
+ srunner_run_all (sr, CK_NORMAL);
+ number_failed = srunner_ntests_failed (sr);
+ srunner_free (sr);
+ return (number_failed == 0) ? EXIT_SUCCESS : EXIT_FAILURE;
+}
+
diff --git a/tests/rbreader.c b/tests/rbreader.c
new file mode 100644
index 0000000..84f0756
--- /dev/null
+++ b/tests/rbreader.c
@@ -0,0 +1,54 @@
+
+#include <stdlib.h>
+#include <stdio.h>
+#include <string.h>
+#include <unistd.h>
+#include <signal.h>
+#include <qb/qbrb.h>
+#include <qb/qbutil.h>
+
+#define BUFFER_CHUNK_SIZE (50*50*10)
+static qb_ringbuffer_t *rb = NULL;
+
+static void sigterm_handler (int num)
+{
+ printf ("reader: %s(%d)\n", __func__, num);
+ qb_rb_close (rb);
+ exit (0);
+}
+
+static void libqb_log_writer (const char *file_name,
+ int32_t file_line,
+ int32_t severity, const char *msg)
+{
+ printf ("libqb:reader: %s:%d %s\n", file_name, file_line, msg);
+}
+
+int main (int argc, char *argv[])
+{
+ ssize_t num_read;
+ int8_t buffer[BUFFER_CHUNK_SIZE];
+ int32_t keep_reading = 1;
+
+ signal (SIGINT, sigterm_handler);
+
+ qb_util_set_log_function (libqb_log_writer);
+
+ rb = qb_rb_open ("tester", BUFFER_CHUNK_SIZE * 3,
+ QB_RB_FLAG_SHARED_PROCESS | QB_RB_FLAG_CREATE);
+
+ if (rb == NULL) {
+ printf ("reader: failed to create ringbuffer\n");
+ return -1;
+ }
+ while (keep_reading) {
+ num_read =
+ qb_rb_chunk_read (rb, buffer, BUFFER_CHUNK_SIZE, 5500);
+ if (num_read == -1) {
+ printf ("reader: nothing to read\n");
+ //usleep(1);
+ }
+ }
+ qb_rb_close (rb);
+ return 0;
+}
diff --git a/tests/rbwriter.c b/tests/rbwriter.c
new file mode 100644
index 0000000..a32a36b
--- /dev/null
+++ b/tests/rbwriter.c
@@ -0,0 +1,168 @@
+/*
+ * Copyright (c) 2009 Red Hat, Inc.
+ *
+ * All rights reserved.
+ *
+ * Author: Steven Dake (sdake(a)redhat.com)
+ *
+ * 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 <sys/types.h>
+#include <sys/socket.h>
+#include <qb/qbipcc.h>
+#include <stdio.h>
+#include <stdint.h>
+#include <string.h>
+#include <sys/time.h>
+#include <time.h>
+#include <unistd.h>
+#include <signal.h>
+#include <qb/qbrb.h>
+#include <qb/qbutil.h>
+
+static qb_ringbuffer_t *rb = NULL;
+#define ITERATIONS 100000
+#define BUFFER_CHUNK_SIZE (50*50*10)
+
+static struct timeval tv1, tv2, tv_elapsed;
+
+#define timersub(a, b, result) \
+do { \
+ (result)->tv_sec = (a)->tv_sec - (b)->tv_sec; \
+ (result)->tv_usec = (a)->tv_usec - (b)->tv_usec; \
+ if ((result)->tv_usec < 0) { \
+ --(result)->tv_sec; \
+ (result)->tv_usec += 1000000; \
+ } \
+} while (0)
+
+static void sigterm_handler (int num)
+{
+ printf ("writer: %s(%d)\n", __func__, num);
+ qb_rb_close (rb);
+ exit (0);
+}
+
+static void libqb_log_writer (const char *file_name,
+ int32_t file_line,
+ int32_t severity, const char *msg)
+{
+ printf ("libqb:writer: %s:%d %s\n", file_name, file_line, msg);
+}
+
+static void bm_start (void)
+{
+ gettimeofday (&tv1, NULL);
+}
+
+static void bm_finish (const char *operation, int size)
+{
+ float ops_per_sec;
+ float mbs_per_sec;
+
+ gettimeofday (&tv2, NULL);
+ timersub (&tv2, &tv1, &tv_elapsed);
+
+ ops_per_sec =
+ ((float)ITERATIONS) / (((float)tv_elapsed.tv_sec) +
+ (((float)tv_elapsed.tv_usec) / 1000000.0));
+
+ mbs_per_sec =
+ ((((float)ITERATIONS) * size) /
+ (((float)tv_elapsed.tv_sec) +
+ (((float)tv_elapsed.tv_usec) / 1000000.0))) / (1024.0 * 1024.0);
+
+ printf ("write size %d OPs/sec %9.3f ", size, ops_per_sec);
+ printf ("MB/sec %9.3f\n", mbs_per_sec);
+}
+
+static void bmc_connect (void)
+{
+ rb = qb_rb_open ("tester", BUFFER_CHUNK_SIZE * 3,
+ QB_RB_FLAG_SHARED_PROCESS);
+
+ if (rb == NULL) {
+ printf ("writer: failed to create ringbuffer\n");
+ exit (1);
+ }
+
+}
+
+static char buffer[1024 * 1024];
+static void bmc_send_nozc (unsigned int size)
+{
+ ssize_t res = 0;
+
+repeat_send:
+ res = qb_rb_chunk_write (rb, buffer, size);
+ if (res < size) {
+ //usleep(1);
+ goto repeat_send;
+ }
+}
+
+static void show_usage (const char *name)
+{
+ printf ("usage: \n");
+ printf ("%s <options>\n", name);
+ printf ("\n");
+ printf (" options:\n");
+ printf ("\n");
+ printf (" -n non-blocking ipc (default blocking)\n");
+ printf (" -v verbose\n");
+ printf (" -h show this help text\n");
+ printf ("\n");
+}
+
+int main (int argc, char *argv[])
+{
+ const char *options = "vh";
+ int opt;
+ int i, j;
+ int size;
+
+ while ((opt = getopt (argc, argv, options)) != -1) {
+ switch (opt) {
+ case 'v':
+ qb_util_set_log_function (libqb_log_writer);
+ break;
+ case 'h':
+ default:
+ show_usage (argv[0]);
+ exit (0);
+ break;
+ }
+ }
+
+ signal (SIGINT, sigterm_handler);
+
+ bmc_connect ();
+
+ for (j = 1; j < 49; j++) {
+ bm_start ();
+ size = 7 * (j + 1) * j;
+
+ if (size > BUFFER_CHUNK_SIZE) {
+ size = BUFFER_CHUNK_SIZE;
+ }
+
+ for (i = 0; i < ITERATIONS; i++) {
+ bmc_send_nozc (size);
+ }
+ bm_finish ("ringbuffer", size);
+ }
+ qb_rb_close (rb);
+ return EXIT_SUCCESS;
+}
--
1.6.6.1
13 years, 10 months
[PATCH] Add a ringbuffer based off the one in logsys.
by Angus Salkeld
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
13 years, 11 months