(brought over from the "speed" branch)
Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com>
---
include/qb/qbrb.h | 7 ++
lib/ringbuffer.c | 189 +++++++++++++++++++++++++++--------------------
lib/ringbuffer_helper.c | 8 +-
lib/ringbuffer_int.h | 2 +-
4 files changed, 124 insertions(+), 82 deletions(-)
diff --git a/include/qb/qbrb.h b/include/qb/qbrb.h
index c3a99ae..8a012bd 100644
--- a/include/qb/qbrb.h
+++ b/include/qb/qbrb.h
@@ -99,6 +99,13 @@ extern "C" {
*/
#define QB_RB_FLAG_SHARED_PROCESS 0x08
+/**
+ * Don't use semaphores, only atomic ops.
+ * This mean that the timeout passed into qb_rb_chunk_read()
+ * will be ignored.
+ */
+#define QB_RB_FLAG_NO_SEMAPHORE 0x10
+
struct qb_ringbuffer_s;
typedef struct qb_ringbuffer_s qb_ringbuffer_t;
diff --git a/lib/ringbuffer.c b/lib/ringbuffer.c
index 67b2fb4..b37d9cb 100644
--- a/lib/ringbuffer.c
+++ b/lib/ringbuffer.c
@@ -34,6 +34,35 @@ do { \
#define DEBUG_PRINTF(format, args...)
#endif /* CRAZY_DEBUG_PRINTFS */
+/*
+ * move the write pointer to the next 128 byte boundary
+ * write_pt goes in 4 bytes (sizeof(uint32_t))
+ * #define USE_CACHE_LINE_ALIGNMENT 1
+ */
+#ifdef USE_CACHE_LINE_ALIGNMENT
+#define QB_CACHE_LINE_SIZE 128
+#define QB_CACHE_LINE_WORDS (QB_CACHE_LINE_SIZE/sizeof(uint32_t))
+#define idx_cache_line_step(idx) \
+do { \
+ if (idx % QB_CACHE_LINE_WORDS) { \
+ idx += (QB_CACHE_LINE_WORDS - (idx % QB_CACHE_LINE_WORDS)); \
+ } \
+ if (idx > (rb->shared_hdr->word_size - 1)) { \
+ idx = ((idx) % (rb->shared_hdr->word_size)); \
+ } \
+} while (0)
+#else
+#define QB_CACHE_LINE_SIZE 0
+#define QB_CACHE_LINE_WORDS 0
+#define idx_cache_line_step(idx) \
+do { \
+ if (idx > (rb->shared_hdr->word_size - 1)) { \
+ idx = ((idx) % (rb->shared_hdr->word_size)); \
+ } \
+} while (0)
+#endif
+
+
/* the chunk header is two words
* 1) the chunk data size
* 2) the magic number
@@ -46,51 +75,42 @@ do { \
* So:
* qb_rb_space_free() >= QB_RB_CHUNK_MARGIN + new data chunk
* The extra word size is to allow for non word sized data chunks.
+ * QB_CACHE_LINE_WORDS is to make sure we have space to align the
+ * chunk.
*/
-#define QB_RB_CHUNK_MARGIN (sizeof(uint32_t) * (QB_RB_CHUNK_HEADER_WORDS + 1))
-#define QB_RB_CHUNK_MAGIC 0xAAAAAAAA
+#define QB_RB_CHUNK_MARGIN (sizeof(uint32_t) * (QB_RB_CHUNK_HEADER_WORDS +\
+ 1 + QB_CACHE_LINE_WORDS))
+#define QB_RB_CHUNK_MAGIC 0xA1A1A1A1
+#define QB_RB_CHUNK_MAGIC_DEAD 0xD0D0D0D0
+#define QB_RB_CHUNK_MAGIC_ALLOC 0xA110CED0
#define QB_RB_CHUNK_SIZE_GET(rb, pointer) \
rb->shared_data[pointer]
#define QB_RB_CHUNK_MAGIC_GET(rb, pointer) \
rb->shared_data[(pointer + 1) % rb->shared_hdr->word_size]
-
-#define QB_RB_WRITE_PT_INDEX (rb->shared_hdr->word_size)
-#define QB_RB_READ_PT_INDEX (rb->shared_hdr->word_size + 1)
-
-#define idx_step(idx) \
+#define QB_RB_CHUNK_MAGIC_SET(rb, pointer, new_val) \
do { \
- if (idx > (rb->shared_hdr->word_size - 1)) { \
- idx = ((idx) % (rb->shared_hdr->word_size)); \
- } \
+ if ((pointer + 1) > (rb->shared_hdr->word_size - 1)) { \
+ rb->shared_data[(pointer + 1) % rb->shared_hdr->word_size] = new_val; \
+ } else { \
+ rb->shared_data[(pointer + 1)] = new_val; \
+ } \
} while (0)
-/*
- * move the write pointer to the next 32 byte boundary
- * 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->word_size - 1)) { \
- idx = ((idx) % (rb->shared_hdr->word_size)); \
- } \
+#define QB_MAGIC_ASSERT(_ptr_) \
+do { \
+ uint32_t chunk_magic = QB_RB_CHUNK_MAGIC_GET(rb, _ptr_); \
+ if (chunk_magic != QB_RB_CHUNK_MAGIC) print_header(rb); \
+ assert(chunk_magic == QB_RB_CHUNK_MAGIC); \
} while (0)
-#else
-#define idx_cache_line_step(idx) \
+
+#define idx_step(idx) \
do { \
if (idx > (rb->shared_hdr->word_size - 1)) { \
idx = ((idx) % (rb->shared_hdr->word_size)); \
} \
} while (0)
-#endif
-static void qb_rb_chunk_check(struct qb_ringbuffer_s * rb, uint32_t pointer);
+static void print_header(struct qb_ringbuffer_s * rb);
qb_ringbuffer_t *
qb_rb_open(const char *name, size_t size, uint32_t flags,
@@ -221,7 +241,9 @@ cleanup_hdr:
}
if (rb && (flags & QB_RB_FLAG_CREATE)) {
unlink(rb->shared_hdr->hdr_path);
- (void)rb->sem_destroy_fn(rb);
+ if (rb->sem_destroy_fn) {
+ (void)rb->sem_destroy_fn(rb);
+ }
}
if (rb && (rb->shared_hdr != MAP_FAILED && rb->shared_hdr !=
NULL)) {
munmap(rb->shared_hdr, sizeof(struct qb_ringbuffer_shared_s));
@@ -240,7 +262,9 @@ qb_rb_close(struct qb_ringbuffer_s * rb)
(void)qb_atomic_int_dec_and_test(&rb->shared_hdr->ref_count);
if (rb->flags & QB_RB_FLAG_CREATE) {
- (void)rb->sem_destroy_fn(rb);
+ if (rb->sem_destroy_fn) {
+ (void)rb->sem_destroy_fn(rb);
+ }
unlink(rb->shared_hdr->data_path);
unlink(rb->shared_hdr->hdr_path);
qb_util_log(LOG_DEBUG,
@@ -262,7 +286,9 @@ qb_rb_force_close(struct qb_ringbuffer_s * rb)
return;
}
- (void)rb->sem_destroy_fn(rb);
+ if (rb->sem_destroy_fn) {
+ (void)rb->sem_destroy_fn(rb);
+ }
unlink(rb->shared_hdr->data_path);
unlink(rb->shared_hdr->hdr_path);
qb_util_log(LOG_DEBUG,
@@ -311,9 +337,6 @@ qb_rb_space_free(struct qb_ringbuffer_s * rb)
return -EINVAL;
}
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) {
@@ -360,7 +383,11 @@ qb_rb_chunks_used(struct qb_ringbuffer_s *rb)
if (rb == NULL) {
return -EINVAL;
}
- return rb->sem_getvalue_fn(rb);
+ if (rb->sem_getvalue_fn) {
+ return rb->sem_getvalue_fn(rb);
+ } else {
+ return -ENOTSUP;
+ }
}
void *
@@ -369,6 +396,7 @@ qb_rb_chunk_alloc(struct qb_ringbuffer_s * rb, size_t len)
uint32_t write_pt;
if (rb == NULL) {
+ errno = EINVAL;
return NULL;
}
/*
@@ -389,9 +417,9 @@ qb_rb_chunk_alloc(struct qb_ringbuffer_s * rb, size_t len)
/*
* insert the chunk header
*/
- rb->shared_data[write_pt++] = 0;
- idx_step(write_pt);
- rb->shared_data[write_pt++] = QB_RB_CHUNK_MAGIC;
+ rb->shared_data[write_pt] = 0;
+ QB_RB_CHUNK_MAGIC_SET(rb, write_pt, QB_RB_CHUNK_MAGIC_ALLOC);
+ write_pt += QB_RB_CHUNK_HEADER_WORDS;
idx_step(write_pt);
/*
@@ -409,7 +437,6 @@ qb_rb_chunk_step(struct qb_ringbuffer_s * rb, uint32_t pointer)
* skip over the chunk header
*/
pointer += QB_RB_CHUNK_HEADER_WORDS;
- idx_step(pointer);
/*
* skip over the user's data.
@@ -437,12 +464,12 @@ qb_rb_chunk_commit(struct qb_ringbuffer_s * rb, size_t len)
*/
old_write_pt = rb->shared_hdr->write_pt;
rb->shared_data[old_write_pt] = len;
- rb->shared_data[old_write_pt + 1] = QB_RB_CHUNK_MAGIC;
/*
* commit the new write pointer
*/
rb->shared_hdr->write_pt = qb_rb_chunk_step(rb, old_write_pt);
+ QB_RB_CHUNK_MAGIC_SET(rb, old_write_pt, QB_RB_CHUNK_MAGIC);
DEBUG_PRINTF("%s: read: %u, write: %u (was:%u)\n", __func__,
rb->shared_hdr->read_pt, rb->shared_hdr->write_pt,
@@ -451,7 +478,11 @@ qb_rb_chunk_commit(struct qb_ringbuffer_s * rb, size_t len)
/*
* post the notification to the reader
*/
- return rb->sem_post_fn(rb);
+ if (rb->sem_post_fn) {
+ return rb->sem_post_fn(rb);
+ } else {
+ return 0;
+ }
}
ssize_t
@@ -478,16 +509,13 @@ qb_rb_chunk_write(struct qb_ringbuffer_s * rb, const void *data,
size_t len)
return len;
}
-void
-qb_rb_chunk_reclaim(struct qb_ringbuffer_s * rb)
+static void
+_rb_chunk_reclaim(struct qb_ringbuffer_s * rb)
{
uint32_t old_read_pt;
- if (rb == NULL || qb_rb_space_used(rb) == 0) {
- return;
- }
old_read_pt = rb->shared_hdr->read_pt;
- qb_rb_chunk_check(rb, old_read_pt);
+ QB_MAGIC_ASSERT(old_read_pt);
rb->shared_hdr->read_pt = qb_rb_chunk_step(rb, old_read_pt);
@@ -495,42 +523,53 @@ qb_rb_chunk_reclaim(struct qb_ringbuffer_s * rb)
* clear the header
*/
rb->shared_data[old_read_pt] = 0;
- rb->shared_data[old_read_pt + 1] = 0;
+ QB_RB_CHUNK_MAGIC_SET(rb, old_read_pt, QB_RB_CHUNK_MAGIC_DEAD);
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(struct qb_ringbuffer_s * rb)
+{
+ if (rb == NULL || qb_rb_space_used(rb) == 0) {
+ return;
+ }
+ _rb_chunk_reclaim(rb);
+}
+
ssize_t
qb_rb_chunk_peek(struct qb_ringbuffer_s * rb, void **data_out, int32_t timeout)
{
uint32_t read_pt;
uint32_t chunk_size;
uint32_t chunk_magic;
- int32_t res;
+ int32_t res = 0;
if (rb == NULL) {
return -EINVAL;
}
- res = rb->sem_timedwait_fn(rb, timeout);
+ if (rb->sem_timedwait_fn) {
+ res = rb->sem_timedwait_fn(rb, timeout);
+ }
if (res < 0 && res != -EIDRM) {
- if (res != -ETIMEDOUT) {
+ if (res == -ETIMEDOUT) {
+ return 0;
+ } else {
+ errno = -res;
qb_util_perror(LOG_ERR, "sem_timedwait");
}
return res;
}
read_pt = rb->shared_hdr->read_pt;
- chunk_size = QB_RB_CHUNK_SIZE_GET(rb, read_pt);
chunk_magic = QB_RB_CHUNK_MAGIC_GET(rb, read_pt);
- *data_out = &rb->shared_data[read_pt + QB_RB_CHUNK_HEADER_WORDS];
-
if (chunk_magic != QB_RB_CHUNK_MAGIC) {
- errno = ENOMSG;
return 0;
- } else {
- return chunk_size;
}
+ chunk_size = QB_RB_CHUNK_SIZE_GET(rb, read_pt);
+ *data_out = &rb->shared_data[read_pt + QB_RB_CHUNK_HEADER_WORDS];
+ return chunk_size;
}
ssize_t
@@ -539,6 +578,7 @@ qb_rb_chunk_read(struct qb_ringbuffer_s * rb, void *data_out, size_t
len,
{
uint32_t read_pt;
uint32_t chunk_size;
+ uint32_t chunk_magic;
int32_t res = 0;
if (rb == NULL) {
@@ -554,14 +594,18 @@ qb_rb_chunk_read(struct qb_ringbuffer_s * rb, void *data_out, size_t
len,
return res;
}
- if (qb_rb_space_used(rb) == 0) {
- return -ENOMSG;
+ read_pt = rb->shared_hdr->read_pt;
+ chunk_magic = QB_RB_CHUNK_MAGIC_GET(rb, read_pt);
+
+ if (chunk_magic != QB_RB_CHUNK_MAGIC) {
+ if (rb->sem_timedwait_fn == NULL) {
+ return -ETIMEDOUT;
+ } else {
+ return -EBADMSG;
+ }
}
- read_pt = rb->shared_hdr->read_pt;
- qb_rb_chunk_check(rb, read_pt);
chunk_size = QB_RB_CHUNK_SIZE_GET(rb, read_pt);
-
if (len < chunk_size) {
qb_util_log(LOG_ERR,
"trying to recv chunk of size %d but %d available",
@@ -573,7 +617,7 @@ qb_rb_chunk_read(struct qb_ringbuffer_s * rb, void *data_out, size_t
len,
&rb->shared_data[read_pt + QB_RB_CHUNK_HEADER_WORDS],
chunk_size);
- qb_rb_chunk_reclaim(rb);
+ _rb_chunk_reclaim(rb);
return chunk_size;
}
@@ -596,21 +640,6 @@ print_header(struct qb_ringbuffer_s * rb)
#endif /* S_SPLINT_S */
}
-static void
-qb_rb_chunk_check(struct qb_ringbuffer_s * rb, uint32_t pointer)
-{
- uint32_t chunk_size;
- uint32_t chunk_magic = QB_RB_CHUNK_MAGIC_GET(rb, pointer);
-
- if (chunk_magic != QB_RB_CHUNK_MAGIC) {
- chunk_size = QB_RB_CHUNK_SIZE_GET(rb, pointer);
- printf("size: %x\n", chunk_size);
- printf("magic: %x\n", chunk_magic);
- print_header(rb);
- assert(0);
- }
-}
-
ssize_t
qb_rb_write_to_file(struct qb_ringbuffer_s * rb, int32_t fd)
{
diff --git a/lib/ringbuffer_helper.c b/lib/ringbuffer_helper.c
index c3c3428..ec84764 100644
--- a/lib/ringbuffer_helper.c
+++ b/lib/ringbuffer_helper.c
@@ -281,7 +281,13 @@ qb_rb_sem_create(struct qb_ringbuffer_s * rb, uint32_t flags)
#endif /* HAVE_SEMTIMEDOP */
#endif /* HAVE_SEM_TIMEDWAIT */
}
- if (use_posix) {
+ if (flags & QB_RB_FLAG_NO_SEMAPHORE) {
+ rc = 0;
+ rb->sem_timedwait_fn = NULL;
+ rb->sem_post_fn = NULL;
+ rb->sem_getvalue_fn = NULL;
+ rb->sem_destroy_fn = NULL;
+ } else if (use_posix) {
rc = my_posix_sem_create(rb, flags);
rb->sem_timedwait_fn = my_posix_sem_timedwait;
rb->sem_post_fn = my_posix_sem_post;
diff --git a/lib/ringbuffer_int.h b/lib/ringbuffer_int.h
index 35283c7..762dc44 100644
--- a/lib/ringbuffer_int.h
+++ b/lib/ringbuffer_int.h
@@ -55,7 +55,7 @@ struct qb_ringbuffer_shared_s {
int32_t ref_count;
rpl_sem_t posix_sem;
char user_data[1];
-};
+} __attribute__ ((aligned(8)));
struct qb_ringbuffer_s {
uint32_t flags;
--
1.7.10.1