[PATCH] RB: to be safer save the read and write pointers at the top of the blackbox
by Angus Salkeld
Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com>
---
lib/ringbuffer.c | 35 +++++++++++++++++------------------
1 file changed, 17 insertions(+), 18 deletions(-)
diff --git a/lib/ringbuffer.c b/lib/ringbuffer.c
index 9650979..eda9d9b 100644
--- a/lib/ringbuffer.c
+++ b/lib/ringbuffer.c
@@ -657,13 +657,6 @@ qb_rb_write_to_file(struct qb_ringbuffer_s * rb, int32_t fd)
}
written_size += result;
- result = write(fd, rb->shared_data,
- rb->shared_hdr->word_size * sizeof(uint32_t));
- if (result != rb->shared_hdr->word_size * sizeof(uint32_t)) {
- return -errno;
- }
- written_size += result;
-
/*
* store the read & write pointers
*/
@@ -678,6 +671,13 @@ qb_rb_write_to_file(struct qb_ringbuffer_s * rb, int32_t fd)
}
written_size += result;
+ result = write(fd, rb->shared_data,
+ rb->shared_hdr->word_size * sizeof(uint32_t));
+ if (result != rb->shared_hdr->word_size * sizeof(uint32_t)) {
+ return -errno;
+ }
+ written_size += result;
+
qb_util_log(LOG_DEBUG, " writing total of: %zd\n", written_size);
return written_size;
@@ -706,13 +706,22 @@ qb_rb_create_from_file(int32_t fd, uint32_t flags)
}
total_read += n_read;
- n_required = (word_size * sizeof(uint32_t));
+ n_read = read(fd, &write_pt, sizeof(uint32_t));
+ assert(n_read == sizeof(uint32_t));
+ total_read += n_read;
+ n_read = read(fd, &read_pt, sizeof(uint32_t));
+ assert(n_read == sizeof(uint32_t));
+ total_read += n_read;
+
+ n_required = (word_size * sizeof(uint32_t));
rb = qb_rb_open("create_from_file", n_required,
QB_RB_FLAG_CREATE | QB_RB_FLAG_NO_SEMAPHORE, 0);
if (rb == NULL) {
goto cleanup_fail2;
}
+ rb->shared_hdr->read_pt = read_pt;
+ rb->shared_hdr->write_pt = write_pt;
n_read = read(fd, rb->shared_data, n_required);
if (n_read < 0) {
@@ -727,16 +736,6 @@ qb_rb_create_from_file(int32_t fd, uint32_t flags)
goto cleanup_fail;
}
- n_read = read(fd, &write_pt, sizeof(uint32_t));
- assert(n_read == sizeof(uint32_t));
- rb->shared_hdr->write_pt = write_pt;
- total_read += n_read;
-
- n_read = read(fd, &read_pt, sizeof(uint32_t));
- assert(n_read == sizeof(uint32_t));
- rb->shared_hdr->read_pt = read_pt;
- total_read += n_read;
-
qb_util_log(LOG_DEBUG, "read total of: %zd", total_read);
print_header(rb);
--
1.7.10.1
11 years, 11 months
[PATCH] avoid unwarranted use of strncpy: use memcpy instead
by Jim Meyering
* lib/log_format.c (qb_vsnprintf_deserialize): We know that there are
len or more non-NUL bytes in the source string, so it's clearer to
use memcpy.
---
Hi Angus,
No big deal, but this does make the code a little clearer,
and it really is best to avoid strncpy altogether, as it seems
you already know ;-)
lib/log_format.c | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/lib/log_format.c b/lib/log_format.c
index 6a74fbb..ff09b26 100644
--- a/lib/log_format.c
+++ b/lib/log_format.c
@@ -547,7 +547,7 @@ qb_vsnprintf_deserialize(char *string, size_t str_len, const char *buf)
}
/* copy from current to the next % */
len = p - format;
- strncpy(&string[location], format, len);
+ memcpy(&string[location], format, len);
location += len;
format = p;
--
1.7.10.2.565.gbd578b5
11 years, 11 months
[PATCH 1/2] RB: add an option to not use any semaphores
by Angus Salkeld
(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
11 years, 11 months
[PATCH] LOG: tweak the blackbox format string
by Angus Salkeld
Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com>
---
lib/log_blackbox.c | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/lib/log_blackbox.c b/lib/log_blackbox.c
index 09db385..dc08294 100644
--- a/lib/log_blackbox.c
+++ b/lib/log_blackbox.c
@@ -202,7 +202,7 @@ qb_log_blackbox_print_from_file(const char *bb_filename)
message[len] = '\0';
len--;
}
- printf("%s %s():%d %s\n",
+ printf("%s %s(%d): %s\n",
time_buf, function, lineno, message);
}
} while (bytes_read > 0);
--
1.7.10.1
11 years, 11 months
[PATCH] LOG: accept NULL strings into the blackbox
by Angus Salkeld
Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com>
---
examples/simplelog.c | 6 ++++++
lib/log_format.c | 6 +++++-
2 files changed, 11 insertions(+), 1 deletion(-)
diff --git a/examples/simplelog.c b/examples/simplelog.c
index 8fb0941..17e3b61 100644
--- a/examples/simplelog.c
+++ b/examples/simplelog.c
@@ -226,6 +226,12 @@ main(int32_t argc, char *argv[])
qb_log(LOG_DEBUG, "hello");
qb_log(LOG_INFO, "this is an info");
qb_log(LOG_NOTICE, "hello - notice?");
+ {
+ char * str = NULL;
+ qb_log(LOG_ERR,
+ "%s ------%s------------- %u?",
+ NULL, str, 56);
+ }
func_one();
func_two();
diff --git a/lib/log_format.c b/lib/log_format.c
index 0cbb543..6a74fbb 100644
--- a/lib/log_format.c
+++ b/lib/log_format.c
@@ -494,7 +494,11 @@ reprocess:
{
char *arg_string;
arg_string = va_arg(ap, char *);
- p = stpcpy(&serialize[location], arg_string);
+ if (arg_string == NULL) {
+ p = stpcpy(&serialize[location], "(null)");
+ } else {
+ p = stpcpy(&serialize[location], arg_string);
+ }
location += p - &serialize[location] + 1;
break;
}
--
1.7.10.1
11 years, 11 months
[PATCH] LOG: protect close and reload from calling log
by Angus Salkeld
Logging during these operations is temporarly disabled.
Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com>
---
lib/log.c | 4 ++++
1 file changed, 4 insertions(+)
diff --git a/lib/log.c b/lib/log.c
index 4056b1f..97c274b 100644
--- a/lib/log.c
+++ b/lib/log.c
@@ -894,7 +894,9 @@ qb_log_custom_close(int32_t t)
target = qb_log_target_get(t);
if (target->close) {
+ in_logger = QB_TRUE;
target->close(t);
+ in_logger = QB_FALSE;
}
qb_log_target_free(target);
}
@@ -987,7 +989,9 @@ qb_log_ctl(int32_t t, enum qb_log_conf c, int32_t arg)
rc = -EINVAL;
}
if (rc == 0 && need_reload && conf[t].reload) {
+ in_logger = QB_TRUE;
conf[t].reload(t);
+ in_logger = QB_FALSE;
}
return rc;
}
--
1.7.10.1
11 years, 11 months
Re: [libqb] [corosync] Fwd: Re: cpg latency
by Steven Dake
On 05/18/2012 01:42 PM, Steven Dake wrote:
>
> Taking back to list...
>
> odd that more cpg clients reduce latency..
>
> Assuming the instrumentation is correct (everything I read says don't
> trust the TSC), the results are:
>
> rbreader->rbwriter = 10-20 usec
> corosync 1 client without totem = 70-120 usec
> corosync 1 client with totem = 150-160 usec
> corosync 4 client without totem = 40-50 usec
> corosync 4 client with totem = 50-60 usec
>
> This implies corosync adds 40usec of latency on top of the ringbuffer
> implementation.
>
> Running rbwriter 1000 times in a loop with a rbreader running shows the
> following oprofile results:
>
> rb-reader:
>
> 2282834 62.0242 libc-2.14.90.so __memcpy_ssse3_back
> 495055 13.4506 libqb.so.0.13.0 qb_rb_space_used
> 378187 10.2753 libpthread-2.14.90.so sem_timedwait
> 207293 5.6321 libqb.so.0.13.0 qb_rb_chunk_check
> 82163 2.2324 libqb.so.0.13.0 qb_rb_chunk_read
> 78788 2.1407 libqb.so.0.13.0 qb_rb_chunk_reclaim
> 42200 1.1466 libqb.so.0.13.0 my_posix_sem_timedwait
> 35622 0.9678 [vdso] (tgid:20388
> range:0x7fff2e5ff000-0x7fff2e600000) [vdso]
> (tgid:20388 range:0x7fff2e5ff000-0x7fff2e600000)
> 32311 0.8779 librt-2.14.90.so clock_gettime
>
> rb-writer:
> 852329 45.5852 libc-2.14.90.so __memcpy_ssse3_back
> 413116 22.0947 libpthread-2.14.90.so sem_post
> 219898 11.7608 libqb.so.0.13.0 qb_rb_space_free
> 161334 8.6286 libqb.so.0.13.0 qb_rb_chunk_commit
> 113573 6.0742 libqb.so.0.13.0 qb_rb_chunk_alloc
> 69525 3.7184 libqb.so.0.13.0 qb_rb_chunk_write
> 13529 0.7236 lt-rbwriter main
>
> rb-reader spending 4x as much time in memcpy - possible cache alignment
> issue.
>
Angus,
It might make more sense to store the used and free space in the header
to avoid their 11% cpu utilization. The routines are a bit complex and
take alot of clock cycles with the math + comparisons + jumping. If
stored in the header, each process is responsible for maintaining its
values (write responsible for setting used, read responsible for setting
free) which would convert those to a simple addition and subtraction for
each message.
I prototyped this change by changing rbreader not to call
qb_rbspace_used (they are only error checks atm), and throughput dropped...
I tried a different change of adding a printf for each message received
in rbreader, and throughput increased 4x.
The conclusion about these points is sem_wait enters the kernel via
syscall (futex) more often because it is often waiting on new data from
the writer.
In the writer flooding the ring buffer faster then the reader can
process them, the user space semaphore doesn't need to context switch
(sem_wait returns immediately). This results in 550mb/sec for the small
message sizes (vs 100mb/sec when always hitting the context switch).
This would also explain the more clients = less latency pattern seen,
since more clients will cause the server's ring buffers to back up
Hope this helps.
Vladimir,
I am pretty sure there is no way to implement a process shared ring
buffer without a semaphore (or some other dec&test mechanism) that
doesn't result in excessive cpu utilization (a nieve approach would be
for reader to spin on reading). In the case of the other program you
spoke about, they batch up 1000 messages. This allows the reader to
read all 1000 messages without having to context switch into the kernel
at inopportune times. Essentially their ring buffer backs up a slight bit.
As an example, consider the following (in the libqb context):
1) writer writes a message into the ring buffer
2) writer sem_posts
3) writer writes a message into the ring buffer
4) reader sitting in futex (sem_wait) awakens
5) reader reads message
6) reader sem_waits, but because step 3 didn't finish its sem_post, it
enters a kernel wait queue
7) writer sem_posts
8) reader awakens and processes message
In a scenario where there are batched messages, step 6 doesn't result in
a context switch, but instead an immediate processing of the message on
multiprocessors.
It would be interesting to see what the latency is like without the
above example scenario. I think I got this mostly to happen by simply
adding a printf (to slightly delay rbreader) with the attached patch.
Regards
-steve
> Regards
> -steve
>
> -------- Original Message --------
> Subject: Re: [corosync] cpg latency
> Date: Fri, 18 May 2012 12:19:25 +0400
> From: Voznesensky Vladimir <voznesensky(a)rts.ru>
> To: Steven Dake <sdake(a)redhat.com>
>
> We had tested it.
>
> 1 process sending 1 message per second gave:
> 70-120 microseconds latency without totem;
> 150-160 microseconds latency with totem.
>
> 4 processes each sending 1 message per second gave:
> 40-50 microseconds latency without totem;
> 50-60 microseconds latency with totem.
>
> Thank you.
> - VV
>
>
>
> On 18.05.2012 11:03, Steven Dake wrote:
>> I have sOn 05/17/2012 08:19 AM, Voznesensky Vladimir wrote:
>>> I've just tested libqb. For every message I've:
>>> - Instrumented a client with tcs reading before qb_ipcc_send and after
>>> qb_ipcc_recv;
>>> - Instrumented a client with CPU frequency testing.
>>> - Commented out qb_log in s1_msg_process_fn of a server.
>>>
>>> So, it took 0.000140 sec - 0.000156 sec for every message to pass and
>>> return.
>>> As I understood, I'ts very large number.
>>> Compare, for instance, with
>>> http://code.google.com/p/disruptor/
>>>
>>> Thanks.
>>>
>> I have sent a test patch which short-circuits totem for cpg messaging
>> single node.
>>
>> Could you see how the latency looks (or post a sample latency tester).
>>
>> Regards
>> -steve
>>
>
> _______________________________________________
> discuss mailing list
> discuss(a)corosync.org
> http://lists.corosync.org/mailman/listinfo/discuss
11 years, 11 months
[PATCH] Correctly display timestamp in blackbox
by Jan Friesse
On little endian architectures where time_t is 64-bit long integer could
happen, that instead of displaying correct timestamp, random data was
used, because of memcpy.
Solution is to memcpy to real 32-bit variable and then assing this value
to timestamp. This will let compiler decide endianity and store number
correctly.
Signed-off-by: Jan Friesse <jfriesse(a)redhat.com>
---
lib/log_blackbox.c | 4 +++-
1 files changed, 3 insertions(+), 1 deletions(-)
diff --git a/lib/log_blackbox.c b/lib/log_blackbox.c
index e2a1f7e..09db385 100644
--- a/lib/log_blackbox.c
+++ b/lib/log_blackbox.c
@@ -165,6 +165,7 @@ qb_log_blackbox_print_from_file(const char *bb_filename)
uint32_t len;
time_t timestamp;
char message[QB_LOG_MAX_LEN];
+ uint32_t u32;
bytes_read = qb_rb_chunk_read(instance, chunk, 512, 0);
ptr = chunk;
@@ -182,7 +183,8 @@ qb_log_blackbox_print_from_file(const char *bb_filename)
ptr += fn_size;
/* timestamp size & content */
- memcpy(×tamp, ptr, sizeof(uint32_t));
+ memcpy(&u32, ptr, sizeof(uint32_t));
+ timestamp = u32;
ptr += sizeof(time_t);
tm = localtime(×tamp);
if (tm) {
--
1.7.1
11 years, 12 months
ANNOUNCE release of libqb-0.12.0
by Angus Salkeld
I am pleased to release version 0.12.0.
It has bug fixes mainly for socket IPC that Andrew Beekhof
has been reporting (pacemaker will soon be using libqb IPC).
It also has a lot of bsd portabiltiy changes.
I have added support for kqueues and more consistent suppport for
semaphores. I have tested on freebsd9 and it works well there.
I'll am going to get it working better on darwin/mac.
If you are super keen on a particular bsd flavour let me know.
As usual the release bits are here:
https://github.com/asalkeld/libqb/wiki#wiki-Releases
-Angus
11 years, 12 months