[PATCH 1/3] TEST: improve the tracing in the ipc tests.

[Angus Salkeld]

Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com&gt;
---
 tests/check_ipc.c |   34 +++++++++++++++++++++++++++++++++-
 1 file changed, 33 insertions(+), 1 deletion(-)

diff --git a/tests/check_ipc.c b/tests/check_ipc.c
index 70e41ef..f2124b2 100644
--- a/tests/check_ipc.c
+++ b/tests/check_ipc.c
@@ -383,99 +383,119 @@ test_ipc_exit(void)
 
 START_TEST(test_ipc_exit_us)
 {
+	qb_enter();
 	ipc_type = QB_IPC_SOCKET;
 	ipc_name = __func__;
 	recv_timeout = 5000;
 	test_ipc_exit();
+	qb_leave();
 }
 END_TEST
 
 #ifdef HAVE_SEM_TIMEDWAIT
 START_TEST(test_ipc_exit_shm)
 {
+	qb_enter();
 	ipc_type = QB_IPC_SHM;
 	ipc_name = __func__;
 	recv_timeout = 1000;
 	test_ipc_exit();
+	qb_leave();
 }
 END_TEST
 
 START_TEST(test_ipc_txrx_shm_tmo)
 {
+	qb_enter();
 	turn_on_fc = QB_FALSE;
 	ipc_type = QB_IPC_SHM;
 	ipc_name = __func__;
 	recv_timeout = 1000;
 	test_ipc_txrx();
+	qb_leave();
 }
 END_TEST
 
 START_TEST(test_ipc_txrx_shm_block)
 {
+	qb_enter();
 	turn_on_fc = QB_FALSE;
 	ipc_type = QB_IPC_SHM;
 	ipc_name = __func__;
 	recv_timeout = -1;
 	test_ipc_txrx();
+	qb_leave();
 }
 END_TEST
 
 START_TEST(test_ipc_fc_shm)
 {
+	qb_enter();
 	turn_on_fc = QB_TRUE;
 	ipc_type = QB_IPC_SHM;
 	recv_timeout = 500;
 	ipc_name = __func__;
 	test_ipc_txrx();
+	qb_leave();
 }
 END_TEST
 #endif /* HAVE_SEM_TIMEDWAIT */
 
 START_TEST(test_ipc_txrx_us_block)
 {
+	qb_enter();
 	turn_on_fc = QB_FALSE;
 	ipc_type = QB_IPC_SOCKET;
 	ipc_name = __func__;
 	recv_timeout = -1;
 	test_ipc_txrx();
+	qb_leave();
 }
 END_TEST
 
 START_TEST(test_ipc_txrx_us_tmo)
 {
+	qb_enter();
 	turn_on_fc = QB_FALSE;
 	ipc_type = QB_IPC_SOCKET;
 	ipc_name = __func__;
 	recv_timeout = 1000;
 	test_ipc_txrx();
+	qb_leave();
 }
 END_TEST
 
 START_TEST(test_ipc_fc_us)
 {
+	qb_enter();
 	turn_on_fc = QB_TRUE;
 	ipc_type = QB_IPC_SOCKET;
 	recv_timeout = 500;
 	ipc_name = __func__;
 	test_ipc_txrx();
+	qb_leave();
 }
 END_TEST
 
 START_TEST(test_ipc_txrx_pmq)
 {
+	qb_enter();
 	turn_on_fc = QB_FALSE;
 	ipc_type = QB_IPC_POSIX_MQ;
 	ipc_name = __func__;
 	test_ipc_txrx();
+	qb_leave();
 }
 END_TEST
 
 START_TEST(test_ipc_txrx_smq)
 {
+	qb_enter();
 	turn_on_fc = QB_FALSE;
 	ipc_type = QB_IPC_SYSV_MQ;
 	ipc_name = __func__;
 	test_ipc_txrx();
+	qb_leave();
 }
 END_TEST
 
@@ -524,9 +544,11 @@ test_ipc_dispatch(void)
 
 START_TEST(test_ipc_disp_us)
 {
+	qb_enter();
 	ipc_type = QB_IPC_SOCKET;
 	ipc_name = __func__;
 	test_ipc_dispatch();
+	qb_leave();
 }
 END_TEST
 
@@ -594,9 +616,11 @@ test_ipc_bulk_events(void)
 
 START_TEST(test_ipc_bulk_events_us)
 {
+	qb_enter();
 	ipc_type = QB_IPC_SOCKET;
 	ipc_name = __func__;
 	test_ipc_bulk_events();
+	qb_leave();
 }
 END_TEST
 
@@ -649,34 +673,42 @@ test_ipc_server_fail(void)
 
 START_TEST(test_ipc_server_fail_soc)
 {
+	qb_enter();
 	ipc_type = QB_IPC_SOCKET;
 	ipc_name = __func__;
 	test_ipc_server_fail();
+	qb_leave();
 }
 END_TEST
 
 #ifdef HAVE_SEM_TIMEDWAIT
 START_TEST(test_ipc_disp_shm)
 {
+	qb_enter();
 	ipc_type = QB_IPC_SHM;
 	ipc_name = __func__;
 	test_ipc_dispatch();
+	qb_leave();
 }
 END_TEST
 
 START_TEST(test_ipc_bulk_events_shm)
 {
+	qb_enter();
 	ipc_type = QB_IPC_SHM;
 	ipc_name = __func__;
 	test_ipc_bulk_events();
+	qb_leave();
 }
 END_TEST
 
 START_TEST(test_ipc_server_fail_shm)
 {
+	qb_enter();
 	ipc_type = QB_IPC_SHM;
 	ipc_name = __func__;
 	test_ipc_server_fail();
+	qb_leave();
 }
 END_TEST
 #endif /* HAVE_SEM_TIMEDWAIT */
@@ -788,7 +820,7 @@ main(void)
 	qb_log_filter_ctl(QB_LOG_STDERR, QB_LOG_FILTER_ADD,
 			  QB_LOG_FILTER_FILE, "*", LOG_TRACE);
 	qb_log_ctl(QB_LOG_STDERR, QB_LOG_CONF_ENABLED, QB_TRUE);
-	qb_log_format_set(QB_LOG_STDERR, "[%p] %f:%l %b");
+	qb_log_format_set(QB_LOG_STDERR, "lib/%f|%l| %b");
 
 	srunner_run_all(sr, CK_VERBOSE);
 	number_failed = srunner_ntests_failed(sr);
-- 
1.7.10

[PATCH 2/3] Add some special commands into the ipc example

[Angus Salkeld]

type "events" and the server will send 10 events.
type 'kill' into the client and this will simulate the server dying.

Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com&gt;
---
 examples/ipcclient.c |   14 ++++++++++++++
 examples/ipcserver.c |   33 ++++++++++++++++++++++++++-------
 2 files changed, 40 insertions(+), 7 deletions(-)

diff --git a/examples/ipcclient.c b/examples/ipcclient.c
index 412e6a8..8f62527 100644
--- a/examples/ipcclient.c
+++ b/examples/ipcclient.c
@@ -78,6 +78,7 @@ main(int argc, char *argv[])
 			rc = qb_ipcc_send(conn, &req, req.hdr.size);
 			if (rc < 0) {
 				perror("qb_ipcc_send");
+				exit(0);
 			}
 		}
 
@@ -85,6 +86,19 @@ main(int argc, char *argv[])
 			rc = qb_ipcc_recv(conn, &res, sizeof(res), -1);
 			if (rc < 0) {
 				perror("qb_ipcc_recv");
+				exit(0);
+			}
+			if (strcasecmp(req.message, "events") == 0) {
+				int32_t i;
+				printf("waiting for 10 events\n");
+				for (i = 0; i < 10; i++) {
+					rc = qb_ipcc_event_recv(conn, &res, sizeof(res), -1);
+					if (rc < 0) {
+						perror("qb_ipcc_event_recv");
+					} else {
+						printf("got event %d rc:%d\n", i, rc);
+					}
+				}
 			}
 			printf("Response[%d]: %s \n", res.hdr.id, res.message);
 		}
diff --git a/examples/ipcserver.c b/examples/ipcserver.c
index e863690..383157d 100644
--- a/examples/ipcserver.c
+++ b/examples/ipcserver.c
@@ -91,32 +91,50 @@ s1_connection_closed_fn(qb_ipcs_connection_t * c)
 	return 0;
 }
 
+struct my_req {
+	struct qb_ipc_request_header hdr;
+	char message[256];
+};
+
 static int32_t
 s1_msg_process_fn(qb_ipcs_connection_t * c, void *data, size_t size)
 {
-	struct qb_ipc_request_header *req_pt =
-	    (struct qb_ipc_request_header *)data;
+	struct my_req *req_pt = (struct my_req *)data;
 	struct qb_ipc_response_header response;
 	ssize_t res;
 	struct iovec iov[2];
 	char resp[100];
+	int32_t sl;
+
+	qb_log(LOG_DEBUG, "msg received (id:%d, size:%d, data:%s)",
+	       req_pt->hdr.id, req_pt->hdr.size, req_pt->message);
 
-	qb_log(LOG_DEBUG, "msg received (id:%d, size:%d)",
-	       req_pt->id, req_pt->size);
+	if (strcmp(req_pt->message, "kill") == 0) {
+		exit(0);
+	}
 	response.size = sizeof(struct qb_ipc_response_header);
 	response.id = 13;
 	response.error = 0;
 
-	snprintf(resp, 100, "ACK %zd bytes", size);
+	sl = snprintf(resp, 100, "ACK %zd bytes", size) + 1;
 	iov[0].iov_len = sizeof(response);
 	iov[0].iov_base = &response;
-	iov[1].iov_len = strlen(resp) + 1;
+	iov[1].iov_len = sl;
 	iov[1].iov_base = resp;
+	response.size += sl;
 
 	res = qb_ipcs_response_sendv(c, iov, 2);
 	if (res < 0) {
 		qb_perror(LOG_ERR, "qb_ipcs_response_send");
 	}
+	if (strcmp(req_pt->message, "events") == 0) {
+		int32_t i;
+		qb_log(LOG_INFO, "request to send 10 events");
+		for (i = 0; i < 10; i++) {
+			res = qb_ipcs_event_sendv(c, iov, 2);
+			qb_log(LOG_INFO, "sent event %d res:%d", i, res);
+		}
+	}
 	return 0;
 }
 
@@ -142,6 +160,7 @@ show_usage(const char *name)
 	printf("  -s             use sysv message queues\n");
 	printf("  -u             use unix sockets\n");
 	printf("  -g             use glib mainloop\n");
+	printf("  -e             use events\n");
 	printf("\n");
 }
 
@@ -244,7 +263,7 @@ my_dispatch_del(int32_t fd)
 int32_t
 main(int32_t argc, char *argv[])
 {
-	const char *options = "mpsugh";
+	const char *options = "mpseugh";
 	int32_t opt;
 	enum qb_ipc_type ipc_type = QB_IPC_NATIVE;
 	struct qb_ipcs_service_handlers sh = {
-- 
1.7.10

[PATCH 3/3] IPC: change the socket recv function to read the response header.

[Angus Salkeld]

It needs to only recv the size of this message, else we recv more
than one message and effectively drop messages (hidden at the end
of the current message).

Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com&gt;
---
 lib/ipc_us.c |   28 +++++++++++++++++++++++-----
 1 file changed, 23 insertions(+), 5 deletions(-)

diff --git a/lib/ipc_us.c b/lib/ipc_us.c
index 579fa90..15c731f 100644
--- a/lib/ipc_us.c
+++ b/lib/ipc_us.c
@@ -242,7 +242,8 @@ qb_ipc_us_recv(struct qb_ipc_one_way * one_way,
 	struct ipc_us_control *ctl = NULL;
 
 retry_recv:
-	result = recv(one_way->u.us.sock, &data[processed], to_recv, MSG_NOSIGNAL |
MSG_WAITALL);
+	result = recv(one_way->u.us.sock, &data[processed], to_recv,
+		      MSG_NOSIGNAL | MSG_WAITALL);
 	if (timeout == -1) {
 		if (result == -1 && errno == EAGAIN) {
 			goto retry_recv;
@@ -272,35 +273,52 @@ retry_recv:
 
 /*
  * recv a message of unknow size.
- * (could use MSG_PEEK here)
  */
 static ssize_t
 qb_ipc_us_recv_at_most(struct qb_ipc_one_way * one_way,
 	       void *msg, size_t len, int32_t timeout)
 {
 	int32_t result;
+	int32_t processed = 0;
+	int32_t to_recv = sizeof(struct qb_ipc_request_header);
+	char *data = msg;
 	struct ipc_us_control *ctl = NULL;
+	struct qb_ipc_request_header *hdr = NULL;
 
 retry_recv:
-	result = recv(one_way->u.us.sock, msg, len, MSG_NOSIGNAL | MSG_WAITALL);
+	result = recv(one_way->u.us.sock, &data[processed], to_recv,
+		      MSG_NOSIGNAL | MSG_WAITALL);
 	if (timeout == -1) {
 		if (result == -1 && errno == EAGAIN) {
 			goto retry_recv;
 		}
 	}
 	if (result == 0) {
-		qb_util_log(LOG_DEBUG, "recv(fd %d) got 0 bytes assuming ENOTCONN",
+		qb_util_log(LOG_DEBUG,
+			    "recv(fd %d) got 0 bytes assuming ENOTCONN",
 			    one_way->u.us.sock);
 		return -ENOTCONN;
 	}
 	if (result == -1) {
 		return -errno;
 	}
+	processed += result;
+	if (processed >= sizeof(struct qb_ipc_request_header) && hdr == NULL) {
+		hdr = (struct qb_ipc_request_header*)msg;
+	}
+	if (hdr) {
+		to_recv = hdr->size - processed;
+	} else {
+		to_recv = len - processed;
+	}
+	if (to_recv > 0) {
+		goto retry_recv;
+	}
 	ctl = (struct ipc_us_control *)one_way->u.us.shared_data;
 	if (ctl) {
 		(void)qb_atomic_int_dec_and_test(&ctl->sent);
 	}
-	return result;
+	return processed;
 }
 
 
-- 
1.7.10