[PATCH 1/9] POLL: check for stop_requested before poll() as well.

[Angus Salkeld]

Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com&gt;
---
 lib/poll.c |    3 +++
 1 files changed, 3 insertions(+), 0 deletions(-)

diff --git a/lib/poll.c b/lib/poll.c
index 36e061a..58dd237 100644
--- a/lib/poll.c
+++ b/lib/poll.c
@@ -441,6 +441,9 @@ int32_t qb_poll_run(qb_handle_t handle)
 		}
 
 retry_poll:
+		if (poll_instance->stop_requested) {
+			return (0);
+		}
 		res = poll(poll_instance->ufds,
 			   poll_instance->poll_entry_count,
 			   expire_timeout_msec);
-- 
1.7.2.3

[PATCH 2/9] IPC: un-const the data pointer in recv()

[Angus Salkeld]

Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com&gt;
---
 include/qb/qbipcc.h |    2 +-
 lib/ipc_int.h       |    2 +-
 lib/ipc_posix_mq.c  |    2 +-
 lib/ipc_shm.c       |    2 +-
 lib/ipc_sysv_mq.c   |    2 +-
 lib/ipcc.c          |    2 +-
 6 files changed, 6 insertions(+), 6 deletions(-)

diff --git a/include/qb/qbipcc.h b/include/qb/qbipcc.h
index b02c8c6..0b92194 100644
--- a/include/qb/qbipcc.h
+++ b/include/qb/qbipcc.h
@@ -41,7 +41,7 @@ qb_ipcc_connect(const char *name);
 
 int32_t qb_ipcc_send(qb_ipcc_connection_t* c, const void *msg_ptr,
                      size_t msg_len);
-ssize_t qb_ipcc_recv(qb_ipcc_connection_t* c, const void *msg_ptr,
+ssize_t qb_ipcc_recv(qb_ipcc_connection_t* c, void *msg_ptr,
                      size_t msg_len);
 
 void qb_ipcc_disconnect(qb_ipcc_connection_t* c);
diff --git a/lib/ipc_int.h b/lib/ipc_int.h
index cd95723..c1e50ca 100644
--- a/lib/ipc_int.h
+++ b/lib/ipc_int.h
@@ -55,7 +55,7 @@ struct qb_ipcc_connection;
 struct qb_ipcc_funcs {
 	int32_t (*send)(struct qb_ipcc_connection* c, const void *msg_ptr,
 		size_t msg_len);
-	ssize_t (*recv)(struct qb_ipcc_connection* c, const void *msg_ptr,
+	ssize_t (*recv)(struct qb_ipcc_connection* c, void *msg_ptr,
 		size_t msg_len);
 	void (*disconnect)(struct qb_ipcc_connection* c);
 };
diff --git a/lib/ipc_posix_mq.c b/lib/ipc_posix_mq.c
index 57f7350..293f8c4 100644
--- a/lib/ipc_posix_mq.c
+++ b/lib/ipc_posix_mq.c
@@ -140,7 +140,7 @@ static int32_t qb_ipcc_pmq_send(struct qb_ipcc_connection *c,
 }
 
 static ssize_t qb_ipcc_pmq_recv(struct qb_ipcc_connection *c,
-				const void *msg_ptr, size_t msg_len)
+				void *msg_ptr, size_t msg_len)
 {
 	uint32_t msg_prio;
 	ssize_t res = mq_receive(c->u.pmq.response.q, (char *)msg_ptr, c->max_msg_size,
diff --git a/lib/ipc_shm.c b/lib/ipc_shm.c
index b5ccd21..9747b58 100644
--- a/lib/ipc_shm.c
+++ b/lib/ipc_shm.c
@@ -50,7 +50,7 @@ static int32_t qb_ipcc_shm_send(struct qb_ipcc_connection *c,
 }
 
 static ssize_t qb_ipcc_shm_recv(struct qb_ipcc_connection *c,
-				const void *msg_ptr, size_t msg_len)
+				void *msg_ptr, size_t msg_len)
 {
 	ssize_t res =
 	    qb_rb_chunk_read(c->u.shm.response.rb, (void *)msg_ptr, msg_len, 0);
diff --git a/lib/ipc_sysv_mq.c b/lib/ipc_sysv_mq.c
index 7bdb94d..51fb4cd 100644
--- a/lib/ipc_sysv_mq.c
+++ b/lib/ipc_sysv_mq.c
@@ -106,7 +106,7 @@ static int32_t qb_ipcc_smq_send(struct qb_ipcc_connection *c,
 }
 
 static ssize_t qb_ipcc_smq_recv(struct qb_ipcc_connection *c,
-				const void *msg_ptr, size_t msg_len)
+				void *msg_ptr, size_t msg_len)
 {
 	ssize_t res;
 
diff --git a/lib/ipcc.c b/lib/ipcc.c
index fab9d49..7cc8b10 100644
--- a/lib/ipcc.c
+++ b/lib/ipcc.c
@@ -119,7 +119,7 @@ int32_t qb_ipcc_send(struct qb_ipcc_connection * c, const void
*msg_ptr,
 	return res;
 }
 
-ssize_t qb_ipcc_recv(struct qb_ipcc_connection * c, const void *msg_ptr,
+ssize_t qb_ipcc_recv(struct qb_ipcc_connection * c, void *msg_ptr,
 		     size_t msg_len)
 {
 	return c->funcs.recv(c, msg_ptr, msg_len);
-- 
1.7.2.3

[PATCH 3/9] IPC: rename qb_ipcs_connection_pt -> qb_ipcs_connection_handle_t

[Angus Salkeld]

Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com&gt;
---
 include/qb/qbipcs.h |   12 ++++++------
 lib/ipc_int.h       |    2 +-
 lib/ipcs.c          |    4 ++--
 tests/bms.c         |    8 ++++----
 tests/check_ipc.c   |    2 +-
 5 files changed, 14 insertions(+), 14 deletions(-)

diff --git a/include/qb/qbipcs.h b/include/qb/qbipcs.h
index 5e5a7b5..c5d791c 100644
--- a/include/qb/qbipcs.h
+++ b/include/qb/qbipcs.h
@@ -33,7 +33,7 @@ extern "C" {
 #endif
 /* *INDENT-ON* */
 
-typedef qb_handle_t qb_ipcs_connection_pt;
+typedef qb_handle_t qb_ipcs_connection_handle_t;
 typedef qb_handle_t qb_ipcs_service_pt;
 
 typedef int32_t (*qb_ipcs_dispatch_fn_t) (qb_ipcs_service_pt s, int32_t fd, int32_t
revents,
@@ -49,10 +49,10 @@ struct qb_ipcs_poll_handlers {
 	qb_ipcs_dispatch_rm_fn dispatch_rm;
 };
 
-typedef int32_t (*qb_ipcs_connection_authenticate_fn) (qb_ipcs_connection_pt c, uid_t
uid, gid_t gid);
-typedef void (*qb_ipcs_connection_created_fn) (qb_ipcs_connection_pt c);
-typedef void (*qb_ipcs_connection_destroyed_fn) (qb_ipcs_connection_pt c);
-typedef void (*qb_ipcs_msg_process_fn) (qb_ipcs_connection_pt c,
+typedef int32_t (*qb_ipcs_connection_authenticate_fn) (qb_ipcs_connection_handle_t c,
uid_t uid, gid_t gid);
+typedef void (*qb_ipcs_connection_created_fn) (qb_ipcs_connection_handle_t c);
+typedef void (*qb_ipcs_connection_destroyed_fn) (qb_ipcs_connection_handle_t c);
+typedef void (*qb_ipcs_msg_process_fn) (qb_ipcs_connection_handle_t c,
 		void *data, size_t size);
 
 struct qb_ipcs_service_handlers {
@@ -76,7 +76,7 @@ int32_t qb_ipcs_run(qb_ipcs_service_pt s, qb_handle_t poll);
 
 void qb_ipcs_destroy(qb_ipcs_service_pt s);
 
-ssize_t qb_ipcs_response_send(qb_ipcs_connection_pt c, void *data, size_t size);
+ssize_t qb_ipcs_response_send(qb_ipcs_connection_handle_t c, void *data, size_t size);
 
 /* *INDENT-OFF* */
 #ifdef __cplusplus
diff --git a/lib/ipc_int.h b/lib/ipc_int.h
index c1e50ca..7e30631 100644
--- a/lib/ipc_int.h
+++ b/lib/ipc_int.h
@@ -158,7 +158,7 @@ struct qb_ipcs_service {
 };
 
 struct qb_ipcs_connection {
-	qb_ipcs_connection_pt handle;
+	qb_ipcs_connection_handle_t handle;
 	pid_t pid;
 	uid_t euid;
 	gid_t egid;
diff --git a/lib/ipcs.c b/lib/ipcs.c
index a7db325..81bf60a 100644
--- a/lib/ipcs.c
+++ b/lib/ipcs.c
@@ -148,7 +148,7 @@ static void qb_ipcs_destroy_internal(void *data)
 	free(s->receive_buf);
 }
 
-ssize_t qb_ipcs_response_send(qb_ipcs_connection_pt c, void *data, size_t size)
+ssize_t qb_ipcs_response_send(qb_ipcs_connection_handle_t c, void *data, size_t size)
 {
 	ssize_t res;
 	struct qb_ipcs_connection *con;
@@ -165,7 +165,7 @@ ssize_t qb_ipcs_response_send(qb_ipcs_connection_pt c, void *data,
size_t size)
 
 struct qb_ipcs_connection *qb_ipcs_connection_alloc(struct qb_ipcs_service *s)
 {
-	qb_ipcs_connection_pt h;
+	qb_ipcs_connection_handle_t h;
 	struct qb_ipcs_connection *c;
 
 	qb_hdb_handle_create(&qb_ipc_connections,
diff --git a/tests/bms.c b/tests/bms.c
index 2ccc84c..b95868e 100644
--- a/tests/bms.c
+++ b/tests/bms.c
@@ -52,7 +52,7 @@ int32_t verbose = 0;
 static qb_handle_t bms_poll_handle;
 static qb_ipcs_service_pt s1;
 
-static int32_t s1_connection_authenticate_fn(qb_ipcs_connection_pt conn, uid_t uid, gid_t
gid)
+static int32_t s1_connection_authenticate_fn(qb_ipcs_connection_handle_t conn, uid_t uid,
gid_t gid)
 {
 	if (uid == 0 && gid == 0) {
 		if (verbose) {
@@ -66,20 +66,20 @@ static int32_t s1_connection_authenticate_fn(qb_ipcs_connection_pt
conn, uid_t u
 }
 
 
-static void s1_connection_created_fn(qb_ipcs_connection_pt conn)
+static void s1_connection_created_fn(qb_ipcs_connection_handle_t conn)
 {
 	if (verbose) {
 		printf("%s:%d %s\n", __FILE__, __LINE__, __func__);
 	}
 }
-static void s1_connection_destroyed_fn(qb_ipcs_connection_pt conn)
+static void s1_connection_destroyed_fn(qb_ipcs_connection_handle_t conn)
 {
 	if (verbose) {
 		printf("%s:%d %s\n", __FILE__, __LINE__, __func__);
 	}
 }
 
-static void s1_msg_process_fn(qb_ipcs_connection_pt conn,
+static void s1_msg_process_fn(qb_ipcs_connection_handle_t conn,
 		void *data, size_t size)
 {
 	struct qb_ipc_request_header *req_pt = (struct qb_ipc_request_header *)data;
diff --git a/tests/check_ipc.c b/tests/check_ipc.c
index 2965104..51fcd12 100644
--- a/tests/check_ipc.c
+++ b/tests/check_ipc.c
@@ -67,7 +67,7 @@ static void sigterm_handler(int32_t num)
 	exit(0);
 }
 
-static void s1_msg_process_fn(qb_ipcs_connection_pt c,
+static void s1_msg_process_fn(qb_ipcs_connection_handle_t c,
 		void *data, size_t size)
 {
 	//struct qb_ipc_request_header *req_pt = (struct qb_ipc_request_header *)data;
-- 
1.7.2.3

[PATCH 4/9] IPC: rename dispatch queues to event queues.

[Angus Salkeld]

Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com&gt;
---
 lib/ipc_int.h      |   15 ++++++++-------
 lib/ipc_posix_mq.c |   34 +++++++++++++++++-----------------
 lib/ipc_shm.c      |   40 ++++++++++++++++++++--------------------
 lib/ipc_sysv_mq.c  |   31 +++++++++++++++----------------
 4 files changed, 60 insertions(+), 60 deletions(-)

diff --git a/lib/ipc_int.h b/lib/ipc_int.h
index 7e30631..a2efbdf 100644
--- a/lib/ipc_int.h
+++ b/lib/ipc_int.h
@@ -78,19 +78,19 @@ struct qb_ipcc_shm_one_way {
 struct qb_ipcc_pmq_connection {
 	struct qb_ipcc_pmq_one_way request;
 	struct qb_ipcc_pmq_one_way response;
-	struct qb_ipcc_pmq_one_way dispatch;
+	struct qb_ipcc_pmq_one_way event;
 };
 
 struct qb_ipcc_smq_connection {
 	struct qb_ipcc_smq_one_way request;
 	struct qb_ipcc_smq_one_way response;
-	struct qb_ipcc_smq_one_way dispatch;
+	struct qb_ipcc_smq_one_way event;
 };
 
 struct qb_ipcc_shm_connection {
 	struct qb_ipcc_shm_one_way request;
 	struct qb_ipcc_shm_one_way response;
-	struct qb_ipcc_shm_one_way dispatch;
+	struct qb_ipcc_shm_one_way event;
 };
 
 struct qb_ipcc_connection {
@@ -179,6 +179,7 @@ int32_t qb_ipcs_shm_create(struct qb_ipcs_service *s);
 
 int32_t qb_ipcs_us_publish(struct qb_ipcs_service *s);
 int32_t qb_ipcs_us_withdraw(struct qb_ipcs_service *s);
+
 int32_t qb_ipcs_dispatch_connection_request(qb_handle_t hdb_handle_t,
 	int32_t fd, int32_t revents, void *data);
 int32_t qb_ipcs_dispatch_service_request(qb_handle_t hdb_handle_t,
@@ -187,6 +188,7 @@ struct qb_ipcs_connection* qb_ipcs_connection_alloc(struct
qb_ipcs_service *s);
 
 int32_t qb_ipcs_process_request(struct qb_ipcs_service *s,
 	struct qb_ipc_request_header *hdr);
+
 void qb_ipcs_disconnect(struct qb_ipcs_connection *c);
 
 struct mar_req_initial_setup {
@@ -200,27 +202,26 @@ struct mar_res_initial_setup {
 	uint32_t max_msg_size __attribute__ ((aligned(8)));
 } __attribute__ ((aligned(8)));
 
-
 struct mar_req_shm_setup {
         struct qb_ipc_request_header hdr __attribute__ ((aligned(8)));
 	uint32_t pid __attribute__ ((aligned(8)));
         char request[PATH_MAX] __attribute__ ((aligned(8)));
         char response[PATH_MAX] __attribute__ ((aligned(8)));
-        char dispatch[PATH_MAX] __attribute__ ((aligned(8)));
+        char event[PATH_MAX] __attribute__ ((aligned(8)));
 } __attribute__ ((aligned(8)));
 
 struct mar_req_pmq_setup {
         struct qb_ipc_request_header hdr __attribute__ ((aligned(8)));
 	uint32_t pid __attribute__ ((aligned(8)));
         char response_mq[NAME_MAX] __attribute__ ((aligned(8)));
-        char dispatch_mq[NAME_MAX] __attribute__ ((aligned(8)));
+        char event_mq[NAME_MAX] __attribute__ ((aligned(8)));
 } __attribute__ ((aligned(8)));
 
 struct mar_req_smq_setup {
         struct qb_ipc_request_header hdr __attribute__ ((aligned(8)));
 	uint32_t pid __attribute__ ((aligned(8)));
         int32_t response_key __attribute__ ((aligned(8)));
-        int32_t dispatch_key __attribute__ ((aligned(8)));
+        int32_t event_key __attribute__ ((aligned(8)));
 } __attribute__ ((aligned(8)));
 
 struct mar_res_setup {
diff --git a/lib/ipc_posix_mq.c b/lib/ipc_posix_mq.c
index 293f8c4..ff8c0ed 100644
--- a/lib/ipc_posix_mq.c
+++ b/lib/ipc_posix_mq.c
@@ -26,7 +26,7 @@
 #include "util_int.h"
 #include <qb/qbpoll.h>
 
-static ssize_t qb_ipcs_pmq_dispatch_send(struct qb_ipcs_connection *c,
+static ssize_t qb_ipcs_pmq_event_send(struct qb_ipcs_connection *c,
 					 void *data, size_t size);
 
 /*
@@ -165,8 +165,8 @@ static void qb_ipcc_pmq_disconnect(struct qb_ipcc_connection *c)
 	hdr.size = sizeof(hdr);
 	mq_send(c->u.pmq.request.q, (const char *)&hdr, hdr.size, 30);
 
-	mq_close(c->u.pmq.dispatch.q);
-	mq_unlink(c->u.pmq.dispatch.name);
+	mq_close(c->u.pmq.event.q);
+	mq_unlink(c->u.pmq.event.name);
 
 	mq_close(c->u.pmq.response.q);
 	mq_unlink(c->u.pmq.response.name);
@@ -187,7 +187,7 @@ static int32_t _ipcc_pmq_connect_to_service_(struct qb_ipcc_connection
*c)
 	start.pid = getpid();
 	start.hdr.size = sizeof(struct mar_req_pmq_setup);
 	strcpy(start.response_mq, c->u.pmq.response.name);
-	strcpy(start.dispatch_mq, c->u.pmq.dispatch.name);
+	strcpy(start.event_mq, c->u.pmq.event.name);
 
 	res =
 	    mq_send(c->u.pmq.request.q, (const char *)&start, start.hdr.size,
@@ -272,18 +272,18 @@ int32_t qb_ipcc_pmq_connect(struct qb_ipcc_connection * c)
 	}
 
 	res =
-	    snprintf(c->u.pmq.dispatch.name, NAME_MAX, "/%s-dispatch-%d",
+	    snprintf(c->u.pmq.event.name, NAME_MAX, "/%s-event-%d",
 		     c->name, getpid());
 
 	posix_mq_increase_limits(c->max_msg_size, 10);
-	c->u.pmq.dispatch.q = posix_mq_create(c->u.pmq.dispatch.name,
+	c->u.pmq.event.q = posix_mq_create(c->u.pmq.event.name,
 					      c->max_msg_size,
 					      O_RDONLY | O_CREAT | O_EXCL |
 					      O_NONBLOCK);
 
-	if (c->u.pmq.dispatch.q == (mqd_t)-1) {
+	if (c->u.pmq.event.q == (mqd_t)-1) {
 		res = -errno;
-		perror("mq_open:DISPATCH");
+		perror("mq_open:event");
 		goto cleanup_request_response;
 	}
 
@@ -292,8 +292,8 @@ int32_t qb_ipcc_pmq_connect(struct qb_ipcc_connection * c)
 		return 0;
 	}
 
-	mq_close(c->u.pmq.dispatch.q);
-	mq_unlink(c->u.pmq.dispatch.name);
+	mq_close(c->u.pmq.event.q);
+	mq_unlink(c->u.pmq.event.name);
 
 cleanup_request_response:
 	mq_close(c->u.pmq.response.q);
@@ -319,7 +319,7 @@ static void qb_ipcs_pmq_disconnect(struct qb_ipcs_connection *c)
 	msg.size = sizeof(msg);
 	msg.error = 0;
 
-	qb_ipcs_pmq_dispatch_send(c, &msg, msg.size);
+	qb_ipcs_pmq_event_send(c, &msg, msg.size);
 }
 
 static void qb_ipcs_pmq_destroy(struct qb_ipcs_service *s)
@@ -373,14 +373,14 @@ static int32_t qb_ipcs_pmq_connect(struct qb_ipcs_service *s,
 		return res;
 	}
 
-	/* setup the dispatch message queue
+	/* setup the event message queue
 	 */
 	posix_mq_increase_limits(c->service->max_msg_size, 10);
-	strcpy(c->u.pmq.dispatch.name, init->dispatch_mq);
-	c->u.pmq.dispatch.q = mq_open(c->u.pmq.dispatch.name,
+	strcpy(c->u.pmq.event.name, init->event_mq);
+	c->u.pmq.event.q = mq_open(c->u.pmq.event.name,
 				      O_WRONLY | O_NONBLOCK);
 
-	if (c->u.pmq.dispatch.q == (mqd_t)-1) {
+	if (c->u.pmq.event.q == (mqd_t)-1) {
 		res = -errno;
 		goto cleanup_response;
 	}
@@ -439,10 +439,10 @@ static ssize_t qb_ipcs_pmq_response_send(struct qb_ipcs_connection
*c,
 	return size;
 }
 
-static ssize_t qb_ipcs_pmq_dispatch_send(struct qb_ipcs_connection *c,
+static ssize_t qb_ipcs_pmq_event_send(struct qb_ipcs_connection *c,
 					 void *data, size_t size)
 {
-	if (mq_send(c->u.pmq.dispatch.q, (const char *)data, size, 1) == -1) {
+	if (mq_send(c->u.pmq.event.q, (const char *)data, size, 1) == -1) {
 		return -errno;
 	}
 	return size;
diff --git a/lib/ipc_shm.c b/lib/ipc_shm.c
index 9747b58..bdf9149 100644
--- a/lib/ipc_shm.c
+++ b/lib/ipc_shm.c
@@ -25,7 +25,7 @@
 #include <qb/qbpoll.h>
 #include <qb/qbrb.h>
 
-static ssize_t qb_ipcs_shm_dispatch_send(struct qb_ipcs_connection *c,
+static ssize_t qb_ipcs_shm_event_send(struct qb_ipcs_connection *c,
 					 void *data, size_t size);
 
 /*
@@ -40,7 +40,7 @@ static void qb_ipcc_shm_disconnect(struct qb_ipcc_connection *c)
 {
 	qb_rb_close(c->u.shm.request.rb);
 	qb_rb_close(c->u.shm.response.rb);
-	qb_rb_close(c->u.shm.dispatch.rb);
+	qb_rb_close(c->u.shm.event.rb);
 }
 
 static int32_t qb_ipcc_shm_send(struct qb_ipcc_connection *c,
@@ -72,7 +72,7 @@ static int32_t _ipcc_shm_connect_to_service_(struct qb_ipcc_connection
*c)
 	start.pid = getpid();
 	start.hdr.size = sizeof(struct mar_req_shm_setup);
 	strcpy(start.response, qb_rb_name_get(c->u.shm.response.rb));
-	strcpy(start.dispatch, qb_rb_name_get(c->u.shm.dispatch.rb));
+	strcpy(start.event, qb_rb_name_get(c->u.shm.event.rb));
 
 	c->needs_sock_for_poll = QB_TRUE;
 
@@ -96,7 +96,7 @@ static int32_t _ipcc_shm_connect_to_service_(struct qb_ipcc_connection
*c)
 		perror("_ipcc_shm_connect_to_service_:qb_rb_chunk_read");
 		goto cleanup;
 	}
-	qb_util_log(LOG_DEBUG, "received response from server %zd\n", size);
+	qb_util_log(LOG_DEBUG, "received response from server size:%zd\n", size);
 	msg_res = (struct mar_res_setup *)c->receive_buf;
 	res = msg_res->hdr.error;
 	if (res == 0) {
@@ -145,17 +145,17 @@ int32_t qb_ipcc_shm_connect(struct qb_ipcc_connection * c)
 	}
 
 	res =
-	    snprintf(c->u.shm.dispatch.name, NAME_MAX, "%s-dispatch-%d",
+	    snprintf(c->u.shm.event.name, NAME_MAX, "%s-event-%d",
 		     c->name, getpid());
 
-	c->u.shm.dispatch.rb = qb_rb_open(c->u.shm.dispatch.name,
+	c->u.shm.event.rb = qb_rb_open(c->u.shm.event.name,
 					  c->max_msg_size,
 					  QB_RB_FLAG_CREATE |
 					  QB_RB_FLAG_SHARED_PROCESS);
 
-	if (c->u.shm.dispatch.rb == NULL) {
+	if (c->u.shm.event.rb == NULL) {
 		res = -errno;
-		perror("qb_rb_open:DISPATCH");
+		perror("qb_rb_open:EVENT");
 		goto cleanup_request_response;
 	}
 
@@ -166,7 +166,7 @@ int32_t qb_ipcc_shm_connect(struct qb_ipcc_connection * c)
 
 	qb_util_log(LOG_DEBUG, "connection failed %d\n", res);
 
-	qb_rb_close(c->u.shm.dispatch.rb);
+	qb_rb_close(c->u.shm.event.rb);
 
 cleanup_request_response:
 	qb_rb_close(c->u.shm.response.rb);
@@ -191,11 +191,11 @@ static void qb_ipcs_shm_disconnect(struct qb_ipcs_connection *c)
 	msg.error = 0;
 
 	if (c->u.shm.response.rb) {
-		qb_ipcs_shm_dispatch_send(c, &msg, msg.size);
+		qb_ipcs_shm_event_send(c, &msg, msg.size);
 		qb_rb_close(c->u.shm.response.rb);
 	}
-	if (c->u.shm.dispatch.rb) {
-		qb_rb_close(c->u.shm.dispatch.rb);
+	if (c->u.shm.event.rb) {
+		qb_rb_close(c->u.shm.event.rb);
 	}
 }
 
@@ -247,17 +247,17 @@ static int32_t qb_ipcs_shm_connect(struct qb_ipcs_service *s,
 		return res;
 	}
 
-	/* setup the dispatch message queue
+	/* setup the event message queue
 	 */
-	strcpy(c->u.shm.dispatch.name, init->dispatch);
-	qb_util_log(LOG_DEBUG, "%s:%s", __func__, c->u.shm.dispatch.name);
-	c->u.shm.dispatch.rb = qb_rb_open(c->u.shm.dispatch.name,
+	strcpy(c->u.shm.event.name, init->event);
+	qb_util_log(LOG_DEBUG, "%s:%s", __func__, c->u.shm.event.name);
+	c->u.shm.event.rb = qb_rb_open(c->u.shm.event.name,
 					  s->max_msg_size,
 					  QB_RB_FLAG_SHARED_PROCESS);
 
-	if (c->u.shm.dispatch.rb == NULL) {
+	if (c->u.shm.event.rb == NULL) {
 		res = -errno;
-		perror("mq_open:DISPATCH");
+		perror("mq_open:EVENT");
 		goto cleanup_response;
 	}
 
@@ -305,10 +305,10 @@ static ssize_t qb_ipcs_shm_response_send(struct qb_ipcs_connection
*c,
 				 size);
 }
 
-static ssize_t qb_ipcs_shm_dispatch_send(struct qb_ipcs_connection *c,
+static ssize_t qb_ipcs_shm_event_send(struct qb_ipcs_connection *c,
 					 void *data, size_t size)
 {
-	return qb_rb_chunk_write(c->u.shm.dispatch.rb, (const char *)data,
+	return qb_rb_chunk_write(c->u.shm.event.rb, (const char *)data,
 				 size);
 }
 
diff --git a/lib/ipc_sysv_mq.c b/lib/ipc_sysv_mq.c
index 51fb4cd..fb93be9 100644
--- a/lib/ipc_sysv_mq.c
+++ b/lib/ipc_sysv_mq.c
@@ -30,7 +30,7 @@
 #define MSGMAX  8192
 #endif
 
-static ssize_t qb_ipcs_smq_dispatch_send(struct qb_ipcs_connection *c,
+static ssize_t qb_ipcs_smq_event_send(struct qb_ipcs_connection *c,
 					 void *data, size_t size);
 
 /*
@@ -137,7 +137,7 @@ static void qb_ipcc_smq_disconnect(struct qb_ipcc_connection *c)
 	hdr.size = sizeof(hdr);
 	msgsnd(c->u.smq.request.q, (const char *)&hdr, hdr.size, 0);
 
-	msgctl(c->u.smq.dispatch.q, IPC_RMID, NULL);
+	msgctl(c->u.smq.event.q, IPC_RMID, NULL);
 	msgctl(c->u.smq.response.q, IPC_RMID, NULL);
 }
 
@@ -154,7 +154,7 @@ static int32_t _smq_connect_to_service_(struct qb_ipcc_connection *c)
 	start.pid = getpid();
 	start.hdr.size = sizeof(struct mar_req_smq_setup);
 	start.response_key = c->u.smq.response.key;
-	start.dispatch_key = c->u.smq.dispatch.key;
+	start.event_key = c->u.smq.event.key;
 
 	if (c->needs_sock_for_poll) {
 		qb_ipc_us_send(c->sock, &start, 1);
@@ -236,11 +236,11 @@ int32_t qb_ipcc_smq_connect(struct qb_ipcc_connection * c)
 		goto cleanup_request;
 	}
 
-	/* Create the dispatch message queue.
+	/* Create the event message queue.
 	 */
-	res = sysv_mq_unnamed_create(&c->u.smq.dispatch);
+	res = sysv_mq_unnamed_create(&c->u.smq.event);
 	if (res < 0) {
-		perror("msgget:DISPATCH");
+		perror("msgget:event");
 		goto cleanup_request_response;
 	}
 
@@ -249,13 +249,12 @@ int32_t qb_ipcc_smq_connect(struct qb_ipcc_connection * c)
 		return 0;
 	}
 
-	msgctl(c->u.smq.dispatch.q, IPC_RMID, NULL);
+	msgctl(c->u.smq.event.q, IPC_RMID, NULL);
 
 cleanup_request_response:
 	msgctl(c->u.smq.response.q, IPC_RMID, NULL);
 
 cleanup_request:
-	free(c);
 
 	return res;
 }
@@ -276,7 +275,7 @@ static void qb_ipcs_smq_disconnect(struct qb_ipcs_connection *c)
 	msg.size = sizeof(msg);
 	msg.error = 0;
 
-	qb_ipcs_smq_dispatch_send(c, &msg, msg.size);
+	qb_ipcs_smq_event_send(c, &msg, msg.size);
 }
 
 static void qb_ipcs_smq_destroy(struct qb_ipcs_service *s)
@@ -327,13 +326,13 @@ static int32_t qb_ipcs_smq_connect(struct qb_ipcs_service *s,
 		goto cleanup;
 	}
 
-	/* setup the dispatch message queue
+	/* setup the event message queue
 	 */
-	c->u.smq.dispatch.key = init->dispatch_key;
-	c->u.smq.dispatch.q = msgget(c->u.smq.dispatch.key, IPC_NOWAIT);
-	if (c->u.smq.dispatch.q == -1) {
+	c->u.smq.event.key = init->event_key;
+	c->u.smq.event.q = msgget(c->u.smq.event.key, IPC_NOWAIT);
+	if (c->u.smq.event.q == -1) {
 		res = -errno;
-		perror("msgget:DISPATCH");
+		perror("msgget:event");
 		goto cleanup_response;
 	}
 
@@ -400,10 +399,10 @@ static ssize_t qb_ipcs_smq_response_send(struct qb_ipcs_connection
*c,
 	return size;
 }
 
-static ssize_t qb_ipcs_smq_dispatch_send(struct qb_ipcs_connection *c,
+static ssize_t qb_ipcs_smq_event_send(struct qb_ipcs_connection *c,
 					 void *data, size_t size)
 {
-	ssize_t res = msgsnd(c->u.smq.dispatch.q, (const char *)data, size, 0);
+	ssize_t res = msgsnd(c->u.smq.event.q, (const char *)data, size, 0);
 	if (res == -1) {
 		return -errno;
 	}
-- 
1.7.2.3

[PATCH 5/9] IPC: add events back (were dispatch messages in corosync)

[Angus Salkeld]

Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com&gt;
---
 include/qb/qbipcc.h |    8 +++-
 include/qb/qbipcs.h |    1 +
 lib/ipc_int.h       |    5 ++-
 lib/ipc_shm.c       |   19 +++++++++
 lib/ipcc.c          |   25 ++++++++++++
 lib/ipcs.c          |   20 ++++++++++
 tests/check_ipc.c   |  106 +++++++++++++++++++++++++++++++++++++++++++++-----
 7 files changed, 171 insertions(+), 13 deletions(-)

diff --git a/include/qb/qbipcc.h b/include/qb/qbipcc.h
index 0b92194..2a1eae1 100644
--- a/include/qb/qbipcc.h
+++ b/include/qb/qbipcc.h
@@ -39,12 +39,18 @@ typedef struct qb_ipcc_connection qb_ipcc_connection_t;
 qb_ipcc_connection_t*
 qb_ipcc_connect(const char *name);
 
+void qb_ipcc_disconnect(qb_ipcc_connection_t* c);
+
 int32_t qb_ipcc_send(qb_ipcc_connection_t* c, const void *msg_ptr,
                      size_t msg_len);
 ssize_t qb_ipcc_recv(qb_ipcc_connection_t* c, void *msg_ptr,
                      size_t msg_len);
 
-void qb_ipcc_disconnect(qb_ipcc_connection_t* c);
+int32_t qb_ipcc_fd_get(qb_ipcc_connection_t* c, int32_t * fd);
+
+int32_t qb_ipcc_event_recv(qb_ipcc_connection_t* c, void **data_out, int32_t timeout);
+
+void qb_ipcc_event_release(qb_ipcc_connection_t* c);
 
 
 /* *INDENT-OFF* */
diff --git a/include/qb/qbipcs.h b/include/qb/qbipcs.h
index c5d791c..996c914 100644
--- a/include/qb/qbipcs.h
+++ b/include/qb/qbipcs.h
@@ -77,6 +77,7 @@ int32_t qb_ipcs_run(qb_ipcs_service_pt s, qb_handle_t poll);
 void qb_ipcs_destroy(qb_ipcs_service_pt s);
 
 ssize_t qb_ipcs_response_send(qb_ipcs_connection_handle_t c, void *data, size_t size);
+ssize_t qb_ipcs_event_send(qb_ipcs_connection_handle_t c, void *data, size_t size);
 
 /* *INDENT-OFF* */
 #ifdef __cplusplus
diff --git a/lib/ipc_int.h b/lib/ipc_int.h
index a2efbdf..b43f7a5 100644
--- a/lib/ipc_int.h
+++ b/lib/ipc_int.h
@@ -49,7 +49,6 @@
 #include <semaphore.h>
 #endif
 
-
 struct qb_ipcc_connection;
 
 struct qb_ipcc_funcs {
@@ -57,6 +56,9 @@ struct qb_ipcc_funcs {
 		size_t msg_len);
 	ssize_t (*recv)(struct qb_ipcc_connection* c, void *msg_ptr,
 		size_t msg_len);
+	ssize_t (*event_recv)(struct qb_ipcc_connection* c, void **data_out,
+		int32_t timeout);
+	void (*event_release)(struct qb_ipcc_connection* c);
 	void (*disconnect)(struct qb_ipcc_connection* c);
 };
 
@@ -133,6 +135,7 @@ struct qb_ipcs_funcs {
 	void (*disconnect)(struct qb_ipcs_connection *c);
 	ssize_t (*request_recv)(struct qb_ipcs_service *s, void *buf, size_t buf_size);
 	ssize_t (*response_send)(struct qb_ipcs_connection *c, void *data, size_t size);
+	ssize_t (*event_send)(struct qb_ipcs_connection *c, void *data, size_t size);
 };
 
 struct qb_ipcs_service {
diff --git a/lib/ipc_shm.c b/lib/ipc_shm.c
index bdf9149..b9002ef 100644
--- a/lib/ipc_shm.c
+++ b/lib/ipc_shm.c
@@ -60,6 +60,22 @@ static ssize_t qb_ipcc_shm_recv(struct qb_ipcc_connection *c,
 	return res;
 }
 
+static ssize_t qb_ipcc_shm_event_recv(struct qb_ipcc_connection *c,
+	void **data_out, int32_t timeout)
+{
+	ssize_t res =
+	    qb_rb_chunk_peek(c->u.shm.event.rb, data_out, timeout);
+	if (res == -ETIMEDOUT) {
+		return -EAGAIN;
+	}
+	return res;
+}
+
+static void qb_ipcc_shm_event_release(struct qb_ipcc_connection* c)
+{
+	qb_rb_chunk_reclaim(c->u.shm.event.rb);
+}
+
 static int32_t _ipcc_shm_connect_to_service_(struct qb_ipcc_connection *c)
 {
 	int32_t res;
@@ -113,6 +129,8 @@ int32_t qb_ipcc_shm_connect(struct qb_ipcc_connection * c)
 
 	c->funcs.send = qb_ipcc_shm_send;
 	c->funcs.recv = qb_ipcc_shm_recv;
+	c->funcs.event_recv = qb_ipcc_shm_event_recv;
+	c->funcs.event_release = qb_ipcc_shm_event_release;
 	c->funcs.disconnect = qb_ipcc_shm_disconnect;
 
 	if (strlen(c->name) > (NAME_MAX - 20)) {
@@ -321,6 +339,7 @@ int32_t qb_ipcs_shm_create(struct qb_ipcs_service *s)
 	s->funcs.response_send = qb_ipcs_shm_response_send;
 	s->funcs.connect = qb_ipcs_shm_connect;
 	s->funcs.disconnect = qb_ipcs_shm_disconnect;
+	s->funcs.event_send = qb_ipcs_shm_event_send;
 	s->needs_sock_for_poll = QB_TRUE;
 
 	s->u.rb = qb_rb_open(s->name, s->max_msg_size,
diff --git a/lib/ipcc.c b/lib/ipcc.c
index 7cc8b10..dd01ecb 100644
--- a/lib/ipcc.c
+++ b/lib/ipcc.c
@@ -125,6 +125,31 @@ ssize_t qb_ipcc_recv(struct qb_ipcc_connection * c, void *msg_ptr,
 	return c->funcs.recv(c, msg_ptr, msg_len);
 }
 
+int32_t qb_ipcc_fd_get(struct qb_ipcc_connection* c, int32_t * fd)
+{
+	if (c->needs_sock_for_poll) {
+		*fd = c->sock;
+	} else {
+		*fd = 0; /*TODO??*/
+	}
+	return 0;
+}
+
+int32_t qb_ipcc_event_recv(struct qb_ipcc_connection* c, void **data_out, int32_t
timeout)
+{
+	char one_byte = 1;
+
+	if (c->needs_sock_for_poll) {
+		qb_ipc_us_recv(c->sock, &one_byte, 1);
+	}
+	return c->funcs.event_recv(c, data_out, timeout);
+}
+
+void qb_ipcc_event_release(struct qb_ipcc_connection* c)
+{
+	c->funcs.event_release(c);
+}
+
 void qb_ipcc_disconnect(struct qb_ipcc_connection *c)
 {
 	qb_util_log(LOG_DEBUG, "%s()", __func__);
diff --git a/lib/ipcs.c b/lib/ipcs.c
index 81bf60a..239e0b8 100644
--- a/lib/ipcs.c
+++ b/lib/ipcs.c
@@ -163,6 +163,26 @@ ssize_t qb_ipcs_response_send(qb_ipcs_connection_handle_t c, void
*data, size_t
 	return res;
 }
 
+ssize_t qb_ipcs_event_send(qb_ipcs_connection_handle_t c, void *data, size_t size)
+{
+	ssize_t res;
+	struct qb_ipcs_connection *con;
+
+	res = qb_hdb_handle_get(&qb_ipc_connections, c, (void **)&con);
+	if (res < 0) {
+		return res;
+	}
+	res = con->service->funcs.event_send(con, data, size);
+
+	if (con->service->needs_sock_for_poll) {
+		qb_ipc_us_send(con->sock, data, 1);
+	}
+
+	qb_hdb_handle_put(&qb_ipc_connections, c);
+
+	return res;
+}
+
 struct qb_ipcs_connection *qb_ipcs_connection_alloc(struct qb_ipcs_service *s)
 {
 	qb_ipcs_connection_handle_t h;
diff --git a/tests/check_ipc.c b/tests/check_ipc.c
index 51fcd12..8dbc85d 100644
--- a/tests/check_ipc.c
+++ b/tests/check_ipc.c
@@ -38,6 +38,13 @@
 #define MAX_MSG_SIZE (8192*16)
 static qb_ipcc_connection_t *conn;
 static enum qb_ipc_type ipc_type;
+static qb_handle_t poll_handle;
+
+#define IPC_MSG_REQ_TX_RX	(QB_IPC_MSG_USER_START + 3)
+#define IPC_MSG_RES_TX_RX	(QB_IPC_MSG_USER_START + 13)
+#define IPC_MSG_REQ_DISPATCH	(QB_IPC_MSG_USER_START + 4)
+#define IPC_MSG_RES_DISPATCH	(QB_IPC_MSG_USER_START + 14)
+
 
 /* Test Cases
  *
@@ -70,17 +77,28 @@ static void sigterm_handler(int32_t num)
 static void s1_msg_process_fn(qb_ipcs_connection_handle_t c,
 		void *data, size_t size)
 {
-	//struct qb_ipc_request_header *req_pt = (struct qb_ipc_request_header *)data;
+	struct qb_ipc_request_header *req_pt = (struct qb_ipc_request_header *)data;
 	struct qb_ipc_response_header response;
 	ssize_t res;
 
-	response.size = sizeof(struct qb_ipc_response_header);
-	response.id = 13;
-	response.error = 0;
-	res = qb_ipcs_response_send(c, &response,
-			sizeof(response));
-	if (res < 0) {
-		perror("qb_ipcs_response_send");
+	if (req_pt->id == IPC_MSG_REQ_TX_RX) {
+		response.size = sizeof(struct qb_ipc_response_header);
+		response.id = IPC_MSG_RES_TX_RX;
+		response.error = 0;
+		res = qb_ipcs_response_send(c, &response,
+				sizeof(response));
+		if (res < 0) {
+			perror("qb_ipcs_response_send");
+		}
+	} else if (req_pt->id == IPC_MSG_REQ_DISPATCH) {
+		response.size = sizeof(struct qb_ipc_response_header);
+		response.id = IPC_MSG_RES_DISPATCH;
+		response.error = 0;
+		res = qb_ipcs_event_send(c, &response,
+				sizeof(response));
+		if (res < 0) {
+			perror("qb_ipcs_dispatch_send");
+		}
 	}
 }
 
@@ -147,7 +165,7 @@ static int32_t bmc_send_nozc(uint32_t size)
 	struct qb_ipc_response_header res_header;
 	int32_t res;
 
-	req_header->id = QB_IPC_MSG_USER_START + 3;
+	req_header->id = IPC_MSG_REQ_TX_RX;
 	req_header->size = sizeof(struct qb_ipc_request_header) + size;
 
 repeat_send:
@@ -176,12 +194,11 @@ repeat_send:
 		return -1;
 	}
 	ck_assert_int_eq(res, sizeof(struct qb_ipc_response_header));
-	ck_assert_int_eq(res_header.id, 13);
+	ck_assert_int_eq(res_header.id, IPC_MSG_RES_TX_RX);
 	ck_assert_int_eq(res_header.size, sizeof(struct qb_ipc_response_header));
 	return 0;
 }
 
-
 static void test_ipc_txrx(void)
 {
 	int32_t j;
@@ -191,10 +208,13 @@ static void test_ipc_txrx(void)
 
 	pid = run_function_in_new_process(run_ipc_server);
 	fail_if(pid == -1);
+	sleep(1);
 
 	do {
 		conn = qb_ipcc_connect(IPC_NAME);
 		if (conn == NULL) {
+			j = waitpid(pid, NULL, WNOHANG);
+			ck_assert_int_eq(j, 0);
 			sleep(1);
 			c++;
 		}
@@ -234,6 +254,64 @@ START_TEST(test_ipc_txrx_smq)
 }
 END_TEST
 
+static void test_ipc_dispatch(void)
+{
+	int32_t res;
+	int32_t j;
+	int32_t c = 0;
+	pid_t pid;
+	struct qb_ipc_request_header *req_header = (struct qb_ipc_request_header *)buffer;
+	struct qb_ipc_response_header *res_header;
+
+	pid = run_function_in_new_process(run_ipc_server);
+	fail_if(pid == -1);
+	sleep(1);
+
+	do {
+		conn = qb_ipcc_connect(IPC_NAME);
+		if (conn == NULL) {
+			j = waitpid(pid, NULL, WNOHANG);
+			ck_assert_int_eq(j, 0);
+			sleep(1);
+			c++;
+		}
+	} while (conn == NULL && c < 5);
+	fail_if(conn == NULL);
+
+	req_header->id = IPC_MSG_REQ_DISPATCH;
+	req_header->size = sizeof(struct qb_ipc_request_header);
+
+repeat_send:
+	res = qb_ipcc_send(conn, req_header, req_header->size);
+	if (res < 0) {
+		if (res == -EAGAIN || res == -ENOMEM) {
+			goto repeat_send;
+		} else if (res == -EINVAL || res == -EINTR) {
+			perror("qb_ipcc_send");
+			return;
+		} else {
+			errno = -res;
+			perror("qb_ipcc_send");
+			goto repeat_send;
+		}
+	}
+
+	res = qb_ipcc_event_recv(conn, &res_header, 0);
+	ck_assert_int_eq(res, sizeof(struct qb_ipc_response_header));
+	ck_assert_int_eq(res_header->id, IPC_MSG_RES_DISPATCH);
+	qb_ipcc_event_release(conn);
+
+	qb_ipcc_disconnect(conn);
+	stop_process(pid);
+}
+
+START_TEST(test_ipc_disp_shm)
+{
+	ipc_type = QB_IPC_SHM;
+	test_ipc_dispatch();
+}
+END_TEST
+
 static Suite *ipc_suite(void)
 {
 	TCase *tc;
@@ -241,6 +319,7 @@ static Suite *ipc_suite(void)
 
 	tc = tcase_create("ipc_txrx_shm");
 	tcase_add_test(tc, test_ipc_txrx_shm);
+	tcase_set_timeout(tc, 6);
 	suite_add_tcase(s, tc);
 
 	tc = tcase_create("ipc_txrx_posix_mq");
@@ -252,6 +331,11 @@ static Suite *ipc_suite(void)
 //	tcase_add_test(tc, test_ipc_txrx_smq);
 //	suite_add_tcase(s, tc);
 
+	tc = tcase_create("ipc_dispatch_shm");
+	tcase_add_test(tc, test_ipc_disp_shm);
+	tcase_set_timeout(tc, 16);
+	suite_add_tcase(s, tc);
+
 	return s;
 }
 
-- 
1.7.2.3

[PATCH 6/9] IPC: pack sysv messages into multiple queue items

[Angus Salkeld]

Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com&gt;
---
 lib/ipc_sysv_mq.c |  175 ++++++++++++++++++++++++++++++++++-------------------
 1 files changed, 113 insertions(+), 62 deletions(-)

diff --git a/lib/ipc_sysv_mq.c b/lib/ipc_sysv_mq.c
index fb93be9..2eb8189 100644
--- a/lib/ipc_sysv_mq.c
+++ b/lib/ipc_sysv_mq.c
@@ -29,6 +29,12 @@
 #ifndef MSGMAX
 #define MSGMAX  8192
 #endif
+#define MY_DATA_SIZE 8000
+struct my_msgbuf {
+	int32_t id __attribute__ ((aligned(8)));
+	char data[MY_DATA_SIZE] __attribute__ ((aligned(8)));
+} __attribute__ ((aligned(8)));
+
 
 static ssize_t qb_ipcs_smq_event_send(struct qb_ipcs_connection *c,
 					 void *data, size_t size);
@@ -57,13 +63,13 @@ static int32_t sysv_mq_create(struct qb_ipcs_service *s)
 		return res;
 	}
 
-	info.msg_qbytes = 10 * MSGMAX;
+	info.msg_qbytes = 10 * s->max_msg_size;
 	res = msgctl(s->u.smq.q, IPC_SET, &info);
 	if (res != 0) {
 		res = -errno;
 		perror("msgctl");
-		qb_util_log(LOG_ERR, "error changing msg_qbytes to %d",
-			    10 * MSGMAX);
+		qb_util_log(LOG_ERR, "error changing msg_qbytes to %zu",
+			    10 * s->max_msg_size);
 	}
 	return res;
 }
@@ -90,37 +96,105 @@ static key_t sysv_key_from_name(const char *name)
 	return ftok(key_location, 0);
 }
 
-/*
- * client functions
- * --------------------------------------------------------
- */
+#define PACK_MESSAGES 1
 
-static int32_t qb_ipcc_smq_send(struct qb_ipcc_connection *c,
-				const void *msg_ptr, size_t msg_len)
+static int32_t sysv_send(mqd_t q,
+			const void *msg_ptr, size_t msg_len)
 {
-	int32_t res = msgsnd(c->u.smq.request.q, msg_ptr, msg_len, 0);
+	int32_t res;
+	int32_t sent = 0;
+#ifdef PACK_MESSAGES
+	char *progress = (char*)msg_ptr;
+	struct my_msgbuf buf;
+	size_t to_send_now; /* to send in this message */
+	size_t to_send_next; /* to send in next message */
+
+	do {
+		to_send_now = QB_MIN(msg_len - sent, MY_DATA_SIZE);
+		to_send_next = msg_len - (sent + to_send_now);
+		/* receiver used the ID to check to see if there
+		 * is more to recieve for this message.
+		 */
+		buf.id = to_send_next + 1;
+		memcpy(buf.data, progress, to_send_now);
+		res = msgsnd(q, &buf, to_send_now, IPC_NOWAIT);
+		if (res == 0) {
+			sent += to_send_now;
+			progress += to_send_now;
+		} else {
+			goto return_status;
+		}
+
+	} while (sent < msg_len);
+
+ return_status:
+#else
+	res = msgsnd(q, msg_ptr, msg_len, IPC_NOWAIT);
+	sent = msg_len;
+#endif
 	if (res == -1) {
 		return -errno;
 	}
-	return msg_len;
+	return sent;
 }
 
-static ssize_t qb_ipcc_smq_recv(struct qb_ipcc_connection *c,
-				void *msg_ptr, size_t msg_len)
+static ssize_t sysv_recv(mqd_t q, void *msg_ptr, size_t msg_len)
 {
 	ssize_t res;
-
-	res = msgrcv(c->u.smq.response.q, (char *)msg_ptr,
-		     c->max_msg_size, 0, IPC_NOWAIT);
+	ssize_t received = 0;
+#ifdef PACK_MESSAGES
+	char *progress = (char*)msg_ptr;
+	struct my_msgbuf buf;
+
+	do {
+ try_again:
+		res = msgrcv(q, &buf,
+			MY_DATA_SIZE, 0, IPC_NOWAIT);
+
+		if (res == -1 && errno == ENOMSG) {
+			goto try_again;
+		}
+		//printf("res:%zd, ID:%d\n", res, buf.id);
+		if (res == -1) {
+			goto return_status;
+		}
+		memcpy(progress, buf.data, res);
+		received += res;
+		progress += res;
+	} while (buf.id > 1);
+ return_status:
+#else
+	res = msgrcv(q, msg_ptr, msg_len,
+		0, IPC_NOWAIT);
+	received = res;
+#endif
 	if (res == -1 && errno == ENOMSG) {
 		/* just to be consistent with other IPC types.
 		 */
 		return -EAGAIN;
 	}
 	if (res == -1) {
-	       return -errno;
+		perror(__func__);
+		return -errno;
 	}
-	return res;
+	return received;
+}
+
+
+/*
+ * client functions
+ * --------------------------------------------------------
+ */
+static int32_t qb_ipcc_smq_send(struct qb_ipcc_connection *c,
+				const void *msg_ptr, size_t msg_len)
+{
+	return sysv_send(c->u.smq.request.q, msg_ptr, msg_len);
+}
+
+static ssize_t qb_ipcc_smq_recv(struct qb_ipcc_connection *c,
+				void *msg_ptr, size_t msg_len)
+{
+	return sysv_recv(c->u.smq.response.q, msg_ptr, msg_len);
 }
 
 static void qb_ipcc_smq_disconnect(struct qb_ipcc_connection *c)
@@ -135,7 +209,7 @@ static void qb_ipcc_smq_disconnect(struct qb_ipcc_connection *c)
 	hdr.id = QB_IPC_MSG_DISCONNECT;
 	hdr.session_id = c->session_id;
 	hdr.size = sizeof(hdr);
-	msgsnd(c->u.smq.request.q, (const char *)&hdr, hdr.size, 0);
+	sysv_send(c->u.smq.request.q, (const char *)&hdr, hdr.size);
 
 	msgctl(c->u.smq.event.q, IPC_RMID, NULL);
 	msgctl(c->u.smq.response.q, IPC_RMID, NULL);
@@ -159,29 +233,27 @@ static int32_t _smq_connect_to_service_(struct qb_ipcc_connection
*c)
 	if (c->needs_sock_for_poll) {
 		qb_ipc_us_send(c->sock, &start, 1);
 	}
-	res = msgsnd(c->u.smq.request.q, (const char *)&start,
-		     start.hdr.size, 0);
+	res = sysv_send(c->u.smq.request.q, (const char *)&start,
+		     start.hdr.size);
 
 	if (res == -1) {
-		res = errno;
-		perror("msgsnd");
+		perror("sysv_send");
 		return res;
 	}
 	qb_util_log(LOG_DEBUG, "sent request to server %d\n", res);
 
 	waited = 0;
 mq_recv_again:
-	size = msgrcv(c->u.smq.response.q, c->receive_buf,
-		      c->max_msg_size, 0, IPC_NOWAIT);
+	size = sysv_recv(c->u.smq.response.q, c->receive_buf,
+		      c->max_msg_size);
 
-	if (size == -1 && (errno == EAGAIN || errno == ENOMSG) && waited <
10) {
+	if ((size == -EAGAIN || size == -ENOMSG) && waited < 10) {
 		usleep(100000);
 		waited++;
 		goto mq_recv_again;
 	}
-	if (size == -1) {
-		res = -errno;
-		perror("msgrcv");
+	if (size < 0) {
+		perror("sysv_recv");
 		goto cleanup;
 	}
 	qb_util_log(LOG_DEBUG, "received response from server %zd\n", size);
@@ -207,8 +279,7 @@ int32_t qb_ipcc_smq_connect(struct qb_ipcc_connection * c)
 	c->needs_sock_for_poll = QB_TRUE;
 
 	if (strlen(c->name) > (NAME_MAX - 20)) {
-		free(c);
-		return EINVAL;
+		return -EINVAL;
 	}
 
 	/* Connect to the service's request message queue.
@@ -217,14 +288,12 @@ int32_t qb_ipcc_smq_connect(struct qb_ipcc_connection * c)
 	if (c->u.smq.request.key == -1) {
 		res = -errno;
 		perror("ftok:REQUEST");
-		free(c);
 		return res;
 	}
 	c->u.smq.request.q = msgget(c->u.smq.request.key, IPC_NOWAIT);
 	if (c->u.smq.request.q == -1) {
 		res = -errno;
 		perror("msgget:REQUEST");
-		free(c);
 		return res;
 	}
 
@@ -341,11 +410,10 @@ static int32_t qb_ipcs_smq_connect(struct qb_ipcs_service *s,
 	accept_msg.hdr.error = 0;
 	accept_msg.max_msg_size = s->max_msg_size;
 
-	res = msgsnd(c->u.smq.response.q, (const char *)&accept_msg,
-		     sizeof(struct mar_res_setup), 0);
-	if (res == -1) {
-		res = -errno;
-		perror("msgsnd:RESPONSE");
+	res = sysv_send(c->u.smq.response.q, (const char *)&accept_msg,
+		     sizeof(struct mar_res_setup));
+	if (res < 0) {
+		perror("sysv_send:RESPONSE");
 		goto cleanup_response;
 	}
 
@@ -353,11 +421,10 @@ static int32_t qb_ipcs_smq_connect(struct qb_ipcs_service *s,
 
 cleanup_response:
 	accept_msg.hdr.error = res;
-	msgsnd(c->u.smq.response.q, (const char *)&accept_msg,
-	       sizeof(struct mar_res_setup), 0);
+	sysv_send(c->u.smq.response.q, (const char *)&accept_msg,
+		     sizeof(struct mar_res_setup));
 
 cleanup:
-	free(c);
 	return res;
 }
 
@@ -372,41 +439,25 @@ static int32_t qb_ipcs_smq_is_msg_ready(struct qb_ipcs_service *s)
 		perror("is_msg_ready");
 	}
 
-	return -1;
+	return -errno;
 }
 #endif
 static ssize_t qb_ipcs_smq_request_recv(struct qb_ipcs_service *s, void *buf,
 					size_t buf_size)
 {
-	ssize_t res = msgrcv(s->u.q, buf, buf_size, 0, 0);
-	//qb_util_log(LOG_INFO, "%s() %d", __func__, res);
-	if (res == -1 && errno == ENOMSG) {
-		return 0;
-	}
-	if (res == -1) {
-		return -errno;
-	}
-	return res;
+	return sysv_recv(s->u.q, buf, buf_size);
 }
 
 static ssize_t qb_ipcs_smq_response_send(struct qb_ipcs_connection *c,
 					 void *data, size_t size)
 {
-	ssize_t res = msgsnd(c->u.smq.response.q, (const char *)data, size, 0);
-	if (res == -1) {
-		return -errno;
-	}
-	return size;
+	return sysv_send(c->u.smq.response.q, (const char *)data, size);
 }
 
 static ssize_t qb_ipcs_smq_event_send(struct qb_ipcs_connection *c,
 					 void *data, size_t size)
 {
-	ssize_t res = msgsnd(c->u.smq.event.q, (const char *)data, size, 0);
-	if (res == -1) {
-		return -errno;
-	}
-	return size;
+	return sysv_send(c->u.smq.event.q, (const char *)data, size);
 }
 
 int32_t qb_ipcs_smq_create(struct qb_ipcs_service * s)
@@ -426,7 +477,7 @@ int32_t qb_ipcs_smq_create(struct qb_ipcs_service * s)
 	s->funcs.request_recv = qb_ipcs_smq_request_recv;
 	s->needs_sock_for_poll = QB_TRUE;
 
-	s->max_msg_size = MSGMAX;
+	//s->max_msg_size = MSGMAX;
 
 	return sysv_mq_create(s);
 }
-- 
1.7.2.3

[PATCH 7/9] TEST: make bmc's output comma delimited

[Angus Salkeld]

Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com&gt;
---
 tests/bmc.c |    4 ++--
 1 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/tests/bmc.c b/tests/bmc.c
index 340f9c9..9a14a10 100644
--- a/tests/bmc.c
+++ b/tests/bmc.c
@@ -75,8 +75,8 @@ static void bm_finish(const char *operation, int32_t 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);
+	printf("write size, %d, OPs/sec, %9.3f, ", size, ops_per_sec);
+	printf("MB/sec, %9.3f\n", mbs_per_sec);
 }
 
 
-- 
1.7.2.3

[PATCH 8/9] TEST: add IPC_TYPE cli option to bms

[Angus Salkeld]

Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com&gt;
---
 tests/bms.c |   19 +++++++++++++++----
 1 files changed, 15 insertions(+), 4 deletions(-)

diff --git a/tests/bms.c b/tests/bms.c
index b95868e..550d003 100644
--- a/tests/bms.c
+++ b/tests/bms.c
@@ -126,13 +126,17 @@ static void show_usage(const char *name)
 	printf("  -n             non-blocking ipc (default blocking)\n");
 	printf("  -v             verbose\n");
 	printf("  -h             show this help text\n");
+	printf("  -m             use shared memory\n");
+	printf("  -p             use posix message queues\n");
+	printf("  -s             use sysv message queues\n");
 	printf("\n");
 }
 
 int32_t main(int32_t argc, char *argv[])
 {
-	const char *options = "nvh";
+	const char *options = "nvhmps";
 	int32_t opt;
+	enum qb_ipc_type ipc_type = QB_IPC_SHM;
 	struct qb_ipcs_service_handlers sh = {
 		.connection_authenticate = s1_connection_authenticate_fn,
 		.connection_created = s1_connection_created_fn,
@@ -142,6 +146,15 @@ int32_t main(int32_t argc, char *argv[])
 
 	while ((opt = getopt(argc, argv, options)) != -1) {
 		switch (opt) {
+		case 'm':
+			ipc_type = QB_IPC_SHM;
+			break;
+		case 's':
+			ipc_type = QB_IPC_SYSV_MQ;
+			break;
+		case 'p':
+			ipc_type = QB_IPC_POSIX_MQ;
+			break;
 		case 'n':	/* non-blocking */
 			blocking = 0;
 			break;
@@ -163,9 +176,7 @@ int32_t main(int32_t argc, char *argv[])
 
 	bms_poll_handle = qb_poll_create();
 
-	s1 = qb_ipcs_create("bm1", QB_IPC_SHM, 8192*64);
-//	s1 = qb_ipcs_create("bm1", QB_IPC_POSIX_MQ, 8192*64);
-//	s1 = qb_ipcs_create("bm1", QB_IPC_SYSV_MQ, 8192*64);
+	s1 = qb_ipcs_create("bm1", ipc_type, 8192*64);
 	if (s1 == 0) {
 		perror("qb_ipcs_create");
 		exit(1);
-- 
1.7.2.3

[PATCH 9/9] IPC_SYSV: get sysv ipc to work with non-root processes.

[Angus Salkeld]

Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com&gt;
---
 lib/ipc_sysv_mq.c |   18 +++++++++++-------
 tests/bms.c       |    4 ++++
 tests/check_ipc.c |   10 +++++-----
 3 files changed, 20 insertions(+), 12 deletions(-)

diff --git a/lib/ipc_sysv_mq.c b/lib/ipc_sysv_mq.c
index 2eb8189..6b2b716 100644
--- a/lib/ipc_sysv_mq.c
+++ b/lib/ipc_sysv_mq.c
@@ -48,7 +48,8 @@ static int32_t sysv_mq_create(struct qb_ipcs_service *s)
 	struct msqid_ds info;
 	int32_t res = 0;
 
-	s->u.smq.q = msgget(s->u.smq.key, IPC_CREAT | O_EXCL | IPC_NOWAIT);
+	s->u.smq.q = msgget(s->u.smq.key, IPC_CREAT | O_EXCL | IPC_NOWAIT |
+			S_IWUSR | S_IRUSR | S_IWGRP);
 	if (s->u.smq.q == -1) {
 		res = -errno;
 		perror("msgget:REQUEST");
@@ -58,16 +59,20 @@ static int32_t sysv_mq_create(struct qb_ipcs_service *s)
 	res = msgctl(s->u.smq.q, IPC_STAT, &info);
 	if (res != 0) {
 		res = -errno;
-		perror("msgctl");
+		perror("msgctl:STAT");
 		qb_util_log(LOG_ERR, "error getting mq info");
 		return res;
 	}
 
-	info.msg_qbytes = 10 * s->max_msg_size;
+	if (info.msg_perm.uid != 0) {
+		qb_util_log(LOG_WARNING, "not enough privileges to increase msg_qbytes");
+		return res;
+	}
+	info.msg_qbytes = 2 * s->max_msg_size;
 	res = msgctl(s->u.smq.q, IPC_SET, &info);
 	if (res != 0) {
 		res = -errno;
-		perror("msgctl");
+		perror("msgctl:SET");
 		qb_util_log(LOG_ERR, "error changing msg_qbytes to %zu",
 			    10 * s->max_msg_size);
 	}
@@ -78,7 +83,8 @@ static int32_t sysv_mq_unnamed_create(struct qb_ipcc_smq_one_way
*queue)
 {
 retry_creating_the_q:
 	queue->key = random();
-	queue->q = msgget(queue->key, IPC_CREAT | IPC_EXCL | IPC_NOWAIT);
+	queue->q = msgget(queue->key, IPC_CREAT | IPC_EXCL | IPC_NOWAIT |
+			S_IWUSR | S_IRUSR | S_IRGRP);
 	if (queue->q == -1 && errno == EEXIST) {
 		goto retry_creating_the_q;
 	} else if (queue->q == -1) {
@@ -477,7 +483,5 @@ int32_t qb_ipcs_smq_create(struct qb_ipcs_service * s)
 	s->funcs.request_recv = qb_ipcs_smq_request_recv;
 	s->needs_sock_for_poll = QB_TRUE;
 
-	//s->max_msg_size = MSGMAX;
-
 	return sysv_mq_create(s);
 }
diff --git a/tests/bms.c b/tests/bms.c
index 550d003..8486526 100644
--- a/tests/bms.c
+++ b/tests/bms.c
@@ -54,6 +54,7 @@ static qb_ipcs_service_pt s1;
 
 static int32_t s1_connection_authenticate_fn(qb_ipcs_connection_handle_t conn, uid_t uid,
gid_t gid)
 {
+#if 0
 	if (uid == 0 && gid == 0) {
 		if (verbose) {
 			printf("%s:%d %s authenticated connection\n",
@@ -63,6 +64,9 @@ static int32_t s1_connection_authenticate_fn(qb_ipcs_connection_handle_t
conn, u
 	}
 	printf("%s:%d %s() BAD user!\n", __FILE__, __LINE__, __func__);
 	return 0;
+#else
+	return 1;
+#endif
 }
 
 
diff --git a/tests/check_ipc.c b/tests/check_ipc.c
index 8dbc85d..59ffa21 100644
--- a/tests/check_ipc.c
+++ b/tests/check_ipc.c
@@ -38,7 +38,6 @@
 #define MAX_MSG_SIZE (8192*16)
 static qb_ipcc_connection_t *conn;
 static enum qb_ipc_type ipc_type;
-static qb_handle_t poll_handle;
 
 #define IPC_MSG_REQ_TX_RX	(QB_IPC_MSG_USER_START + 3)
 #define IPC_MSG_RES_TX_RX	(QB_IPC_MSG_USER_START + 13)
@@ -296,7 +295,7 @@ repeat_send:
 		}
 	}
 
-	res = qb_ipcc_event_recv(conn, &res_header, 0);
+	res = qb_ipcc_event_recv(conn, (void**)&res_header, 0);
 	ck_assert_int_eq(res, sizeof(struct qb_ipc_response_header));
 	ck_assert_int_eq(res_header->id, IPC_MSG_RES_DISPATCH);
 	qb_ipcc_event_release(conn);
@@ -327,9 +326,10 @@ static Suite *ipc_suite(void)
 	tcase_set_timeout(tc, 10);
 	suite_add_tcase(s, tc);
 
-//	tc = tcase_create("ipc_txrx_sysv_mq");
-//	tcase_add_test(tc, test_ipc_txrx_smq);
-//	suite_add_tcase(s, tc);
+	tc = tcase_create("ipc_txrx_sysv_mq");
+	tcase_add_test(tc, test_ipc_txrx_smq);
+	tcase_set_timeout(tc, 10);
+	suite_add_tcase(s, tc);
 
 	tc = tcase_create("ipc_dispatch_shm");
 	tcase_add_test(tc, test_ipc_disp_shm);
-- 
1.7.2.3