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