Page MenuHomeClusterLabs Projects
Files F3152015

No OneTemporary
Actions

View Options

diff --git a/lib/ipc_shm.c b/lib/ipc_shm.c
index 3171827..f15e326 100644
--- a/lib/ipc_shm.c
+++ b/lib/ipc_shm.c
@@ -1,311 +1,306 @@
/*
* Copyright (C) 2010 Red Hat, Inc.
*
* Author: Angus Salkeld <asalkeld@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 "os_base.h"
#include "ipc_int.h"
#include "util_int.h"
#include <qb/qbdefs.h>
#include <qb/qbloop.h>
#include <qb/qbrb.h>
/*
* utility functions
* --------------------------------------------------------
*/
/*
* client functions
* --------------------------------------------------------
*/
static void qb_ipcc_shm_disconnect(struct qb_ipcc_connection *c)
{
qb_rb_close(c->request.u.shm.rb);
qb_rb_close(c->response.u.shm.rb);
qb_rb_close(c->event.u.shm.rb);
}
static ssize_t qb_ipc_shm_send(struct qb_ipc_one_way *one_way,
const void *msg_ptr, size_t msg_len)
{
return qb_rb_chunk_write(one_way->u.shm.rb, msg_ptr, msg_len);
}
static ssize_t qb_ipc_shm_sendv(struct qb_ipc_one_way *one_way,
const struct iovec* iov,
size_t iov_len)
{
char *dest;
int32_t res = 0;
int32_t total_size = 0;
int32_t i;
char *pt = NULL;
if (one_way->u.shm.rb == NULL) {
return -ENOTCONN;
}
for (i = 0; i < iov_len; i++) {
total_size += iov[i].iov_len;
}
dest = qb_rb_chunk_alloc(one_way->u.shm.rb, total_size);
if (dest == NULL) {
return -errno;
}
pt = dest;
for (i = 0; i < iov_len; i++) {
memcpy(pt, iov[i].iov_base, iov[i].iov_len);
pt += iov[i].iov_len;
}
res = qb_rb_chunk_commit(one_way->u.shm.rb, total_size);
if (res < 0) {
return res;
}
return total_size;
}
static ssize_t qb_ipc_shm_recv(struct qb_ipc_one_way *one_way,
void *msg_ptr,
size_t msg_len,
int32_t ms_timeout)
{
if (one_way->u.shm.rb == NULL) {
return -ENOTCONN;
}
return qb_rb_chunk_read(one_way->u.shm.rb,
(void *)msg_ptr,
msg_len,
ms_timeout);
}
static ssize_t qb_ipc_shm_peek(struct qb_ipc_one_way *one_way, void **data_out, int32_t ms_timeout)
{
if (one_way->u.shm.rb == NULL) {
return -ENOTCONN;
}
return qb_rb_chunk_peek(one_way->u.shm.rb,
data_out,
ms_timeout);
}
static void qb_ipc_shm_reclaim(struct qb_ipc_one_way *one_way)
{
if (one_way->u.shm.rb != NULL) {
qb_rb_chunk_reclaim(one_way->u.shm.rb);
}
}
static void qb_ipc_shm_fc_set(struct qb_ipc_one_way *one_way,
int32_t fc_enable)
{
int32_t *fc;
fc = qb_rb_shared_user_data_get(one_way->u.shm.rb);
*fc = fc_enable;
}
static int32_t qb_ipc_shm_fc_get(struct qb_ipc_one_way *one_way)
{
int32_t *fc;
int32_t rc = qb_rb_refcount_get(one_way->u.shm.rb);
if (rc != 2) {
return -ENOTCONN;
}
fc = qb_rb_shared_user_data_get(one_way->u.shm.rb);
return *fc;
}
static ssize_t qb_ipc_shm_q_len_get(struct qb_ipc_one_way *one_way)
{
if (one_way->u.shm.rb == NULL) {
return -ENOTCONN;
}
return qb_rb_chunks_used(one_way->u.shm.rb);
}
int32_t qb_ipcc_shm_connect(struct qb_ipcc_connection *c,
struct qb_ipc_connection_response *response)
{
int32_t res = 0;
c->funcs.send = qb_ipc_shm_send;
c->funcs.sendv = qb_ipc_shm_sendv;
c->funcs.recv = qb_ipc_shm_recv;
c->funcs.fc_get = qb_ipc_shm_fc_get;
c->funcs.disconnect = qb_ipcc_shm_disconnect;
c->needs_sock_for_poll = QB_TRUE;
if (strlen(c->name) > (NAME_MAX - 20)) {
errno = EINVAL;
return -errno;
}
c->request.u.shm.rb = qb_rb_open(response->request, c->request.max_msg_size,
QB_RB_FLAG_SHARED_PROCESS,
sizeof(int32_t));
if (c->request.u.shm.rb == NULL) {
res = -errno;
goto return_error;
}
c->response.u.shm.rb = qb_rb_open(response->response,
c->response.max_msg_size,
QB_RB_FLAG_SHARED_PROCESS, 0);
if (c->response.u.shm.rb == NULL) {
res = -errno;
perror("qb_rb_open:RESPONSE");
goto cleanup_request;
}
c->event.u.shm.rb = qb_rb_open(response->event,
c->response.max_msg_size,
QB_RB_FLAG_SHARED_PROCESS, 0);
if (c->event.u.shm.rb == NULL) {
res = -errno;
perror("qb_rb_open:EVENT");
goto cleanup_request_response;
}
return 0;
cleanup_request_response:
qb_rb_close(c->response.u.shm.rb);
cleanup_request:
qb_rb_close(c->request.u.shm.rb);
return_error:
errno = -res;
qb_util_perror(LOG_ERR, "connection failed");
return res;
}
/*
* service functions
* --------------------------------------------------------
*/
static void qb_ipcs_shm_disconnect(struct qb_ipcs_connection *c)
{
struct qb_ipc_response_header msg;
- int32_t peer_alive = QB_TRUE;
-
- if (c->setup.u.us.sock == -1) {
- peer_alive = QB_FALSE;
- }
msg.id = QB_IPC_MSG_DISCONNECT;
msg.size = sizeof(msg);
msg.error = 0;
if (c->response.u.shm.rb) {
qb_rb_close(c->response.u.shm.rb);
c->response.u.shm.rb = NULL;
}
if (c->event.u.shm.rb) {
qb_rb_close(c->event.u.shm.rb);
c->event.u.shm.rb = NULL;
}
if (c->request.u.shm.rb) {
qb_rb_close(c->request.u.shm.rb);
c->request.u.shm.rb = NULL;
}
}
static int32_t qb_ipcs_shm_connect(struct qb_ipcs_service *s,
struct qb_ipcs_connection *c,
struct qb_ipc_connection_response *r)
{
int32_t res;
qb_util_log(LOG_DEBUG, "connecting to client [%d]", c->pid);
snprintf(r->request, NAME_MAX, "qb-%s-request-%d-%d", s->name, c->pid, c->setup.u.us.sock);
snprintf(r->response, NAME_MAX, "qb-%s-response-%d-%d", s->name, c->pid, c->setup.u.us.sock);
snprintf(r->event, NAME_MAX, "qb-%s-event-%d-%d", s->name, c->pid, c->setup.u.us.sock);
c->request.u.shm.rb = qb_rb_open(r->request,
c->request.max_msg_size,
QB_RB_FLAG_CREATE |
QB_RB_FLAG_SHARED_PROCESS,
sizeof(int32_t));
if (c->request.u.shm.rb == NULL) {
res = -errno;
perror("qb_rb_open:REQUEST");
goto cleanup;
}
res = qb_rb_chown(c->request.u.shm.rb, c->euid, c->egid);
c->response.u.shm.rb = qb_rb_open(r->response,
c->response.max_msg_size,
QB_RB_FLAG_CREATE |
QB_RB_FLAG_SHARED_PROCESS, 0);
if (c->response.u.shm.rb == NULL) {
res = -errno;
perror("qb_rb_open:RESPONSE");
goto cleanup_request;
}
res = qb_rb_chown(c->response.u.shm.rb, c->euid, c->egid);
c->event.u.shm.rb = qb_rb_open(r->event,
c->event.max_msg_size,
QB_RB_FLAG_CREATE |
QB_RB_FLAG_SHARED_PROCESS, 0);
if (c->event.u.shm.rb == NULL) {
res = -errno;
perror("qb_rb_open:EVENT");
goto cleanup_request_response;
}
res = qb_rb_chown(c->event.u.shm.rb, c->euid, c->egid);
r->hdr.error = 0;
return 0;
cleanup_request_response:
qb_rb_close(c->request.u.shm.rb);
cleanup_request:
qb_rb_close(c->response.u.shm.rb);
cleanup:
r->hdr.error = res;
errno = -res;
qb_util_perror(LOG_ERR, "shm connection FAILED");
return res;
}
void qb_ipcs_shm_init(struct qb_ipcs_service *s)
{
s->funcs.connect = qb_ipcs_shm_connect;
s->funcs.disconnect = qb_ipcs_shm_disconnect;
s->funcs.recv = qb_ipc_shm_recv;
s->funcs.peek = qb_ipc_shm_peek;
s->funcs.reclaim = qb_ipc_shm_reclaim;
s->funcs.send = qb_ipc_shm_send;
s->funcs.sendv = qb_ipc_shm_sendv;
s->funcs.fc_set = qb_ipc_shm_fc_set;
s->funcs.q_len_get = qb_ipc_shm_q_len_get;
s->needs_sock_for_poll = QB_TRUE;
}
diff --git a/lib/log_blackbox.c b/lib/log_blackbox.c
index 2e50afe..7e39a40 100644
--- a/lib/log_blackbox.c
+++ b/lib/log_blackbox.c
@@ -1,185 +1,183 @@
/*
* Copyright (C) 2011 Red Hat, Inc.
*
* All rights reserved.
*
* Author: Angus Salkeld <asalkeld@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 "os_base.h"
#include <qb/qbrb.h>
#include "log_int.h"
static void _blackbox_reload(struct qb_log_target *t)
{
if (t->instance == NULL) {
return;
}
qb_rb_close(t->instance);
t->instance = qb_rb_open(t->name, t->size,
QB_RB_FLAG_CREATE | QB_RB_FLAG_OVERWRITE, 0);
}
/* <u32> file lineno
* <u32> function name length
* <string> function name
* <u32> buffer lenght
* <string> buffer
*/
static void _blackbox_logger(struct qb_log_target *t,
struct qb_log_callsite *cs,
time_t timestamp, const char *buffer)
{
size_t size = sizeof(uint32_t);
size_t fn_size;
size_t buf_size;
char *chunk;
if (t->instance == NULL) {
return;
}
fn_size = strlen(cs->function) + 1;
buf_size = strlen(buffer) + 1;
size += 2 * sizeof(uint32_t) + fn_size + buf_size + sizeof(time_t);
chunk = qb_rb_chunk_alloc(t->instance, size);
/* line number */
memcpy(chunk, &cs->lineno, sizeof(uint32_t));
chunk += sizeof(uint32_t);
/* function name */
memcpy(chunk, &fn_size, sizeof(uint32_t));
chunk += sizeof(uint32_t);
memcpy(chunk, cs->function, fn_size);
chunk += fn_size;
/* timestamp */
memcpy(chunk, &timestamp, sizeof(time_t));
chunk += sizeof(time_t);
/* log message */
memcpy(chunk, &buf_size, sizeof(uint32_t));
chunk += sizeof(uint32_t);
memcpy(chunk, buffer, buf_size);
(void)qb_rb_chunk_commit(t->instance, size);
}
static void _blackbox_close(struct qb_log_target *t)
{
if (t->instance) {
qb_rb_close(t->instance);
t->instance = NULL;
}
}
int32_t qb_log_blackbox_open(struct qb_log_target *t)
{
if (t->size < 1024) {
return -EINVAL;
}
t->instance = qb_rb_open(t->name, t->size,
QB_RB_FLAG_CREATE | QB_RB_FLAG_OVERWRITE, 0);
if (t->instance == NULL) {
return -errno;
}
t->logger = _blackbox_logger;
t->reload = _blackbox_reload;
t->close = _blackbox_close;
return 0;
}
ssize_t qb_log_blackbox_write_to_file(const char *filename)
{
ssize_t written_size = 0;
struct qb_log_target *t;
int fd = open(filename, O_CREAT | O_RDWR, 0700);
if (fd < 0) {
return -errno;
}
t = qb_log_target_get(QB_LOG_BLACKBOX);
if (t->instance) {
written_size = qb_rb_write_to_file(t->instance, fd);
} else {
written_size = -ENOENT;
}
close(fd);
return written_size;
}
void qb_log_blackbox_print_from_file(const char *bb_filename)
{
qb_ringbuffer_t *instance;
ssize_t bytes_read;
char chunk[512];
int fd;
char time_buf[64];
fd = open(bb_filename, O_CREAT | O_RDWR, 0700);
if (fd < 0) {
qb_perror(LOG_ERR, "qb_log_blackbox_print_from_file");
return;
}
instance = qb_rb_create_from_file(fd, 0);
close(fd);
if (instance == NULL) {
return;
}
do {
char *ptr;
uint32_t *lineno;
uint32_t *fn_size;
char *function;
time_t *timestamp;
- uint32_t *log_size;
- char *logmsg;
+ /*uint32_t *log_size;*/
bytes_read = qb_rb_chunk_read(instance, chunk, 512, 0);
ptr = chunk;
if (bytes_read > 0) {
/* lineno */
lineno = (uint32_t *) ptr;
ptr += sizeof(uint32_t);
/* function size & name */
fn_size = (uint32_t *) ptr;
ptr += sizeof(uint32_t);
function = ptr;
ptr += *fn_size;
/* timestamp size & content */
timestamp = (time_t *) ptr;
ptr += sizeof(time_t);
(void)strftime(time_buf, sizeof(time_buf), "%b %d %T",
localtime(timestamp));
/* message size & content */
- log_size = (uint32_t *) ptr;
+ /* log_size = (uint32_t *) ptr; */
ptr += sizeof(uint32_t);
- logmsg = ptr;
printf("%s %s():%d %s\n", time_buf, function, *lineno,
- logmsg);
+ ptr);
}
} while (bytes_read > 0);
}
diff --git a/tests/bmcpt.c b/tests/bmcpt.c
index b7b1369..6b4924b 100644
--- a/tests/bmcpt.c
+++ b/tests/bmcpt.c
@@ -1,200 +1,190 @@
/*
* Copyright (c) 2009 Red Hat, Inc.
*
* All rights reserved.
*
* Author: Steven Dake (sdake@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 <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <errno.h>
#include <assert.h>
#include <stdio.h>
#include <stdint.h>
#include <string.h>
#include <sys/time.h>
#include <time.h>
#include <signal.h>
#include <qb/qbdefs.h>
#include <qb/qbipcc.h>
#define ITERATIONS 10000000
#define THREADS 4
struct bm_ctx {
qb_ipcc_connection_t *conn;
struct timeval tv1;
struct timeval tv2;
struct timeval tv_elapsed;
float mbs;
int32_t multi;
uint32_t counter;
};
#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 bm_start(struct bm_ctx *ctx)
{
gettimeofday(&ctx->tv1, NULL);
}
static void bm_finish(struct bm_ctx *ctx, const char *operation, int32_t size)
{
- float ops_per_sec;
- float mbs_per_sec;
-
gettimeofday(&ctx->tv2, NULL);
timersub(&ctx->tv2, &ctx->tv1, &ctx->tv_elapsed);
- ops_per_sec =
- ((float)ctx->counter) / (((float)ctx->tv_elapsed.tv_sec) +
- (((float)ctx->tv_elapsed.tv_usec) /
- 1000000.0));
-
- mbs_per_sec =
+ ctx->mbs =
((((float)ctx->counter) * size) /
(((float)ctx->tv_elapsed.tv_sec) +
(((float)ctx->tv_elapsed.tv_usec) / 1000000.0))) / (1024.0 *
1024.0);
-
- ctx->mbs = ops_per_sec;
}
static void bmc_connect(struct bm_ctx *ctx)
{
ctx->conn = qb_ipcc_connect("bm1", QB_MAX(1000 * (100 + THREADS),
1024*1024));
if (ctx->conn == NULL) {
perror("qb_ipcc_connect");
exit(-1);
}
}
static void bmc_disconnect(struct bm_ctx *ctx)
{
qb_ipcc_disconnect(ctx->conn);
}
static char buffer[1024 * 1024];
static int32_t bmc_send_nozc(struct bm_ctx *ctx, uint32_t size)
{
struct qb_ipc_request_header *req_header = (struct qb_ipc_request_header *)buffer;
struct qb_ipc_response_header res_header;
int32_t res;
req_header->id = QB_IPC_MSG_USER_START + 3;
req_header->size = sizeof(struct qb_ipc_request_header) + size;
repeat_send:
res = qb_ipcc_send(ctx->conn, req_header, req_header->size);
if (res < 0) {
if (res == -EAGAIN) {
goto repeat_send;
} else if (res == -EINVAL || res == -EINTR) {
perror("qb_ipcc_send");
return -1;
} else {
errno = -res;
perror("qb_ipcc_send");
goto repeat_send;
}
}
res = qb_ipcc_recv(ctx->conn, &res_header,
sizeof(struct qb_ipc_response_header), -1);
if (res == -EINTR) {
return -1;
}
if (res < 0) {
perror("qb_ipcc_recv");
}
assert(res == sizeof(struct qb_ipc_response_header));
assert(res_header.id == 13);
assert(res_header.size == sizeof(struct qb_ipc_response_header));
return 0;
}
uint32_t alarm_notice = 0;
static void sigalrm_handler(int32_t num)
{
alarm_notice = 1;
}
static void *benchmark(void *ctx)
{
struct bm_ctx *bm_ctx = (struct bm_ctx *)ctx;
int32_t res;
bmc_connect(bm_ctx);
bm_start(bm_ctx);
for (;;) {
bm_ctx->counter++;
res = bmc_send_nozc(bm_ctx, 1000 * bm_ctx->multi);
if (alarm_notice || res == -1) {
bm_finish(bm_ctx, "send_nozc", 1000 * bm_ctx->multi);
bmc_disconnect(bm_ctx);
return (NULL);
}
}
}
int32_t main(void)
{
struct bm_ctx bm_ctx[THREADS];
pthread_t threads[THREADS];
pthread_attr_t thread_attr[THREADS];
int32_t i, j;
float total_mbs;
void *retval;
signal(SIGALRM, sigalrm_handler);
for (j = 0; j < 500; j++) {
alarm_notice = 0;
alarm(3);
for (i = 0; i < THREADS; i++) {
bm_ctx[i].multi = j + 100;
bm_ctx[i].counter = 0;
pthread_attr_init(&thread_attr[i]);
pthread_attr_setdetachstate(&thread_attr[i],
PTHREAD_CREATE_JOINABLE);
pthread_create(&threads[i], &thread_attr[i], benchmark,
&bm_ctx[i]);
}
for (i = 0; i < THREADS; i++) {
pthread_join(threads[i], &retval);
}
total_mbs = 0;
for (i = 0; i < THREADS; i++) {
total_mbs = total_mbs + bm_ctx[i].mbs;
}
printf("%d ", 1000 * bm_ctx[0].multi);
printf("%9.3f\n", total_mbs);
}
return EXIT_SUCCESS;
}

File Metadata

Mime Type
text/x-diff
Expires
Mon, Feb 24, 2:41 PM (8 h, 17 m ago)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
1464269
Default Alt Text
(17 KB)

Event Timeline