Re: [libqb] [Openais] question abou qb_ipcs_msg_process_fn()

[Angus Salkeld]

On Thu, Dec 09, 2010 at 10:35:15AM +0000, Dietmar Maurer wrote:
>  Hi all,
>   
Can we move this to the libqb mailing list?
https://fedorahosted.org/mailman/listinfo/quarterback-devel

>  I am playing around with libqb writing my first test server. Normal
> operation work
>  quite good so far. I just wonder how to handle errors on the server side, especially
>  in qb_ipcs_msg_process_fn:
>  
>  typedef int32_t (*qb_ipcs_msg_process_fn) (qb_ipcs_connection_t *c,
>  		void *data, size_t size);
>  
>  I thought the return value is used to indicate errors, but if I simply
>  return a negative number my client hang forever. 
Currently if you return anything but -ENOBUFS or -EAGAIN the server will
remove the socket from the poll loop.

To return an error to the client use qb_ipcs_response_send().
like:
{
  struct qb_ipc_response_header response;

  response.size = sizeof(struct qb_ipc_response_header);
  response.id = MY_MSG_ID;
  response.error = MY_ERROR;
  qb_ipcs_response_send(c, &response, sizeof(response));
}

Regards
-Angus

>  
>  So I always need to send something back? What is the purpose of the return value then?
>  
>  On the client side there seems to be no timeout in qb_ipcc_sendv_recv()?
>  
>  - Dietmar
>  
>  
>  _______________________________________________
>  Openais mailing list
>  Openais(a)lists.linux-foundation.org
>  https://lists.linux-foundation.org/mailman/listinfo/openais

[Openais] question abou qb_ipcs_msg_process_fn()

[Dietmar Maurer]

>  > I thought the return value is used to indicate errors, but if I
> simply
>  > return a negative number my client hang forever.
>  
>  Currently if you return anything but -ENOBUFS or -EAGAIN the server will
>  remove the socket from the poll loop. 
I can't reproduce that behavior. My server continues to work, no matter what I
return.

>  To return an error to the client use qb_ipcs_response_send().
>  like:
>  {
>    struct qb_ipc_response_header response;
>  
>    response.size = sizeof(struct qb_ipc_response_header);
>    response.id = MY_MSG_ID;
>    response.error = MY_ERROR;
>    qb_ipcs_response_send(c, &response, sizeof(response)); } 
OK

>  > On the client side there seems to be no timeout in
> qb_ipcc_sendv_recv()? 
If the server misbehaves, there should be a timeout on the client? Or do I need
to implement that myself?

- Dietmar

[PATCH] TESTS: add a test case for recving from a failed server

[Angus Salkeld]

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

diff --git a/tests/check_ipc.c b/tests/check_ipc.c
index ebb138e..b2988f3 100644
--- a/tests/check_ipc.c
+++ b/tests/check_ipc.c
@@ -46,7 +46,9 @@ enum my_msg_ids {
 	IPC_MSG_REQ_TX_RX,
 	IPC_MSG_RES_TX_RX,
 	IPC_MSG_REQ_DISPATCH,
-	IPC_MSG_RES_DISPATCH
+	IPC_MSG_RES_DISPATCH,
+	IPC_MSG_REQ_SERVER_FAIL,
+	IPC_MSG_RES_SERVER_FAIL,
 };
 
 /* Test Cases
@@ -72,11 +74,12 @@ static qb_ipcs_service_t* s1;
 static int32_t turn_on_fc = QB_FALSE;
 static int32_t fc_enabled = 89;
 
-static void sigterm_handler(int32_t num)
+static int32_t exit_handler(int32_t rsignal, void *data)
 {
 	qb_ipcs_destroy(s1);
 	qb_loop_stop(my_loop);
 	exit(0);
+	return -1;
 }
 
 static int32_t s1_msg_process_fn(qb_ipcs_connection_t *c,
@@ -107,6 +110,9 @@ static int32_t s1_msg_process_fn(qb_ipcs_connection_t *c,
 		if (res < 0) {
 			perror("qb_ipcs_event_send");
 		}
+	} else if (req_pt->id == IPC_MSG_REQ_SERVER_FAIL) {
+		printf("recv'ed server fail - exitting...\n");
+		exit(0);
 	}
 	return 0;
 }
@@ -138,6 +144,7 @@ static int32_t my_dispatch_del(int32_t fd)
 static void run_ipc_server(void)
 {
 	int32_t res;
+	qb_loop_signal_handle handle;
 
 	struct qb_ipcs_service_handlers sh = {
 		.connection_accept = NULL,
@@ -153,7 +160,10 @@ static void run_ipc_server(void)
 		.dispatch_del = my_dispatch_del,
 	};
 
-	signal(SIGTERM, sigterm_handler);
+	qb_loop_signal_add(my_loop, QB_LOOP_HIGH, SIGSTOP,
+			   NULL, exit_handler, &handle);
+	qb_loop_signal_add(my_loop, QB_LOOP_HIGH, SIGTERM,
+			   NULL, exit_handler, &handle);
 
 	my_loop = qb_loop_create();
 
@@ -366,7 +376,7 @@ static void test_ipc_dispatch(void)
 			goto repeat_event_recv;
 		} else {
 			errno = -res;
-			perror("qb_ipcc_send");
+			perror("qb_ipcc_event_recv");
 			goto repeat_send;
 		}
 	}
@@ -391,12 +401,91 @@ START_TEST(test_ipc_disp_us)
 }
 END_TEST
 
+static void test_ipc_server_fail(void)
+{
+	struct qb_ipc_request_header req_header;
+	struct qb_ipc_response_header res_header;
+	size_t size;
+	int32_t res;
+	int32_t try_times = 0;
+	int32_t j;
+	int32_t c = 0;
+	pid_t pid;
+
+	pid = run_function_in_new_process(run_ipc_server);
+	fail_if(pid == -1);
+	sleep(1);
+
+	do {
+		conn = qb_ipcc_connect(IPC_NAME, MAX_MSG_SIZE);
+		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_SERVER_FAIL;
+	req_header.size = sizeof(struct qb_ipc_request_header);
+
+ repeat_send:
+	printf("sending server fail\n");
+	res = qb_ipcc_send(conn, &req_header, req_header.size);
+	try_times++;
+	if (res < 0) {
+		if (res == -EAGAIN && try_times < 10) {
+			goto repeat_send;
+		} else {
+			errno = -res;
+			perror("qb_ipcc_send");
+			return res;
+		}
+	}
+	printf("trying to recv from failed server\n");
+
+ repeat_recv:
+	res = qb_ipcc_recv(conn,
+			&res_header,
+			sizeof(struct qb_ipc_response_header));
+	printf("recv %d\n", res);
+	ck_assert_int_eq(res, 0);
+
+	qb_ipcc_disconnect(conn);
+	stop_process(pid);
+}
+
+START_TEST(test_ipc_server_fail_soc)
+{
+	ipc_type = QB_IPC_SOCKET;
+	test_ipc_server_fail();
+}
+END_TEST
+
+START_TEST(test_ipc_server_fail_shm)
+{
+	ipc_type = QB_IPC_SHM;
+	test_ipc_server_fail();
+}
+END_TEST
+
 static Suite *ipc_suite(void)
 {
 	TCase *tc;
 	uid_t uid;
 	Suite *s = suite_create("ipc");
 
+	tc = tcase_create("ipc_server_fail_shm");
+	tcase_add_test(tc, test_ipc_server_fail_shm);
+	tcase_set_timeout(tc, 6);
+	suite_add_tcase(s, tc);
+
+	tc = tcase_create("ipc_server_fail_soc");
+	tcase_add_test(tc, test_ipc_server_fail_soc);
+	tcase_set_timeout(tc, 6);
+	suite_add_tcase(s, tc);
+
 	tc = tcase_create("ipc_txrx_shm");
 	tcase_add_test(tc, test_ipc_txrx_shm);
 	tcase_set_timeout(tc, 6);
-- 
1.7.3.1

[PATCH] TESTS: add a test case for recving from a failed server

[Angus Salkeld]

The shared memory test fails, I need to add a timeout as Dietmar
susggested above. This test case at least proves the problem.

Thanks for exposing the problem Dietmar.

I'll fix the problem now.

-Angus

On Fri, Dec 10, 2010 at 11:24:11AM +1100, Angus Salkeld wrote:
>  Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com&gt;
>  ---
>   tests/check_ipc.c |   97 ++++++++++++++++++++++++++++++++++++++++++++++++++--
>   1 files changed, 93 insertions(+), 4 deletions(-)
>  
>  diff --git a/tests/check_ipc.c b/tests/check_ipc.c
>  index ebb138e..b2988f3 100644
>  --- a/tests/check_ipc.c
>  +++ b/tests/check_ipc.c
>  @@ -46,7 +46,9 @@ enum my_msg_ids {
>   	IPC_MSG_REQ_TX_RX,
>   	IPC_MSG_RES_TX_RX,
>   	IPC_MSG_REQ_DISPATCH,
>  -	IPC_MSG_RES_DISPATCH
>  +	IPC_MSG_RES_DISPATCH,
>  +	IPC_MSG_REQ_SERVER_FAIL,
>  +	IPC_MSG_RES_SERVER_FAIL,
>   };
>   
>   /* Test Cases
>  @@ -72,11 +74,12 @@ static qb_ipcs_service_t* s1;
>   static int32_t turn_on_fc = QB_FALSE;
>   static int32_t fc_enabled = 89;
>   
>  -static void sigterm_handler(int32_t num)
>  +static int32_t exit_handler(int32_t rsignal, void *data)
>   {
>   	qb_ipcs_destroy(s1);
>   	qb_loop_stop(my_loop);
>   	exit(0);
>  +	return -1;
>   }
>   
>   static int32_t s1_msg_process_fn(qb_ipcs_connection_t *c,
>  @@ -107,6 +110,9 @@ static int32_t s1_msg_process_fn(qb_ipcs_connection_t *c,
>   		if (res < 0) {
>   			perror("qb_ipcs_event_send");
>   		}
>  +	} else if (req_pt->id == IPC_MSG_REQ_SERVER_FAIL) {
>  +		printf("recv'ed server fail - exitting...\n");
>  +		exit(0);
>   	}
>   	return 0;
>   }
>  @@ -138,6 +144,7 @@ static int32_t my_dispatch_del(int32_t fd)
>   static void run_ipc_server(void)
>   {
>   	int32_t res;
>  +	qb_loop_signal_handle handle;
>   
>   	struct qb_ipcs_service_handlers sh = {
>   		.connection_accept = NULL,
>  @@ -153,7 +160,10 @@ static void run_ipc_server(void)
>   		.dispatch_del = my_dispatch_del,
>   	};
>   
>  -	signal(SIGTERM, sigterm_handler);
>  +	qb_loop_signal_add(my_loop, QB_LOOP_HIGH, SIGSTOP,
>  +			   NULL, exit_handler, &handle);
>  +	qb_loop_signal_add(my_loop, QB_LOOP_HIGH, SIGTERM,
>  +			   NULL, exit_handler, &handle);
>   
>   	my_loop = qb_loop_create();
>   
>  @@ -366,7 +376,7 @@ static void test_ipc_dispatch(void)
>   			goto repeat_event_recv;
>   		} else {
>   			errno = -res;
>  -			perror("qb_ipcc_send");
>  +			perror("qb_ipcc_event_recv");
>   			goto repeat_send;
>   		}
>   	}
>  @@ -391,12 +401,91 @@ START_TEST(test_ipc_disp_us)
>   }
>   END_TEST
>   
>  +static void test_ipc_server_fail(void)
>  +{
>  +	struct qb_ipc_request_header req_header;
>  +	struct qb_ipc_response_header res_header;
>  +	size_t size;
>  +	int32_t res;
>  +	int32_t try_times = 0;
>  +	int32_t j;
>  +	int32_t c = 0;
>  +	pid_t pid;
>  +
>  +	pid = run_function_in_new_process(run_ipc_server);
>  +	fail_if(pid == -1);
>  +	sleep(1);
>  +
>  +	do {
>  +		conn = qb_ipcc_connect(IPC_NAME, MAX_MSG_SIZE);
>  +		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_SERVER_FAIL;
>  +	req_header.size = sizeof(struct qb_ipc_request_header);
>  +
>  + repeat_send:
>  +	printf("sending server fail\n");
>  +	res = qb_ipcc_send(conn, &req_header, req_header.size);
>  +	try_times++;
>  +	if (res < 0) {
>  +		if (res == -EAGAIN && try_times < 10) {
>  +			goto repeat_send;
>  +		} else {
>  +			errno = -res;
>  +			perror("qb_ipcc_send");
>  +			return res;
>  +		}
>  +	}
>  +	printf("trying to recv from failed server\n");
>  +
>  + repeat_recv:
>  +	res = qb_ipcc_recv(conn,
>  +			&res_header,
>  +			sizeof(struct qb_ipc_response_header));
>  +	printf("recv %d\n", res);
>  +	ck_assert_int_eq(res, 0);
>  +
>  +	qb_ipcc_disconnect(conn);
>  +	stop_process(pid);
>  +}
>  +
>  +START_TEST(test_ipc_server_fail_soc)
>  +{
>  +	ipc_type = QB_IPC_SOCKET;
>  +	test_ipc_server_fail();
>  +}
>  +END_TEST
>  +
>  +START_TEST(test_ipc_server_fail_shm)
>  +{
>  +	ipc_type = QB_IPC_SHM;
>  +	test_ipc_server_fail();
>  +}
>  +END_TEST
>  +
>   static Suite *ipc_suite(void)
>   {
>   	TCase *tc;
>   	uid_t uid;
>   	Suite *s = suite_create("ipc");
>   
>  +	tc = tcase_create("ipc_server_fail_shm");
>  +	tcase_add_test(tc, test_ipc_server_fail_shm);
>  +	tcase_set_timeout(tc, 6);
>  +	suite_add_tcase(s, tc);
>  +
>  +	tc = tcase_create("ipc_server_fail_soc");
>  +	tcase_add_test(tc, test_ipc_server_fail_soc);
>  +	tcase_set_timeout(tc, 6);
>  +	suite_add_tcase(s, tc);
>  +
>   	tc = tcase_create("ipc_txrx_shm");
>   	tcase_add_test(tc, test_ipc_txrx_shm);
>   	tcase_set_timeout(tc, 6);
>  -- 
>  1.7.3.1

[PATCH] IPC: return -ENOTCONN when the other end exits

[Angus Salkeld]

Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com&gt;
---
 lib/ipc_shm.c     |    8 ++++----
 lib/ipc_us.c      |    9 ++-------
 lib/ipcc.c        |   11 ++++++++++-
 tests/check_ipc.c |   24 ++++++++++++------------
 4 files changed, 28 insertions(+), 24 deletions(-)

diff --git a/lib/ipc_shm.c b/lib/ipc_shm.c
index 024194b..1b8fd6e 100644
--- a/lib/ipc_shm.c
+++ b/lib/ipc_shm.c
@@ -58,7 +58,7 @@ static ssize_t qb_ipc_shm_sendv(struct qb_ipc_one_way *one_way,
 	char *pt = NULL;
 
 	if (one_way->u.shm.rb == NULL) {
-		return -EIDRM;
+		return -ENOTCONN;
 	}
 
 	for (i = 0; i < iov_len; i++) {
@@ -88,7 +88,7 @@ static ssize_t qb_ipc_shm_recv(struct qb_ipc_one_way *one_way,
 {
 	ssize_t res;
 	if (one_way->u.shm.rb == NULL) {
-		return -EIDRM;
+		return -ENOTCONN;
 	}
 	res = qb_rb_chunk_read(one_way->u.shm.rb,
 			       (void *)msg_ptr,
@@ -105,7 +105,7 @@ static ssize_t qb_ipc_shm_peek(struct qb_ipc_one_way *one_way, void
**data_out,
 	ssize_t res;
 
 	if (one_way->u.shm.rb == NULL) {
-		return -EIDRM;
+		return -ENOTCONN;
 	}
 	res = qb_rb_chunk_peek(one_way->u.shm.rb,
 			       data_out,
@@ -141,7 +141,7 @@ static int32_t qb_ipc_shm_fc_get(struct qb_ipc_one_way *one_way)
 static ssize_t qb_ipc_shm_q_len_get(struct qb_ipc_one_way *one_way)
 {
 	if (one_way->u.shm.rb == NULL) {
-		return -EIDRM;
+		return -ENOTCONN;
 	}
 	return qb_rb_chunks_used(one_way->u.shm.rb);
 }
diff --git a/lib/ipc_us.c b/lib/ipc_us.c
index 231b475..09bdc5c 100644
--- a/lib/ipc_us.c
+++ b/lib/ipc_us.c
@@ -214,7 +214,7 @@ int32_t qb_ipc_us_recv_ready(struct qb_ipc_one_way *one_way, int32_t
ms_timeout)
 	} else if (poll_events == -1) {
 		return -errno;
 	} else if (poll_events == 1 && (ufds.revents & (POLLERR|POLLHUP))) {
-		return -ESHUTDOWN;
+		return -ENOTCONN;
 	}
 	return 0;
 }
@@ -233,14 +233,9 @@ ssize_t qb_ipc_us_recv(struct qb_ipc_one_way *one_way,
 	if (result == -1) {
 		return -errno;
 	}
-#if defined(QB_SOLARIS) || defined(QB_BSD) || defined(QB_DARWIN)
-	/* On many OS poll never return POLLHUP or POLLERR.
-	 * EOF is detected when recvmsg return 0.
-	 */
 	if (result == 0) {
-		return -errno;	//ENOTCONN
+		return -ENOTCONN;
 	}
-#endif
 	if (ctl) {
 		(void)qb_atomic_int_dec_and_test(&ctl->sent);
 	}
diff --git a/lib/ipcc.c b/lib/ipcc.c
index cab71af..4119152 100644
--- a/lib/ipcc.c
+++ b/lib/ipcc.c
@@ -129,7 +129,16 @@ ssize_t qb_ipcc_sendv(struct qb_ipcc_connection* c, const struct
iovec* iov,
 ssize_t qb_ipcc_recv(struct qb_ipcc_connection * c, void *msg_ptr,
 		     size_t msg_len)
 {
-	return c->funcs.recv(&c->response, msg_ptr, msg_len, -1);
+	int32_t res = 0;
+
+	res = c->funcs.recv(&c->response, msg_ptr, msg_len, 1000);
+	if (res == -EAGAIN && c->needs_sock_for_poll) {
+		res = qb_ipc_us_recv_ready(&c->setup, 10);
+		if (res < 0) {
+			return res;
+		}
+	}
+	return res;
 }
 
 ssize_t qb_ipcc_sendv_recv (
diff --git a/tests/check_ipc.c b/tests/check_ipc.c
index b2988f3..539e7ee 100644
--- a/tests/check_ipc.c
+++ b/tests/check_ipc.c
@@ -111,7 +111,6 @@ static int32_t s1_msg_process_fn(qb_ipcs_connection_t *c,
 			perror("qb_ipcs_event_send");
 		}
 	} else if (req_pt->id == IPC_MSG_REQ_SERVER_FAIL) {
-		printf("recv'ed server fail - exitting...\n");
 		exit(0);
 	}
 	return 0;
@@ -405,7 +404,6 @@ static void test_ipc_server_fail(void)
 {
 	struct qb_ipc_request_header req_header;
 	struct qb_ipc_response_header res_header;
-	size_t size;
 	int32_t res;
 	int32_t try_times = 0;
 	int32_t j;
@@ -427,30 +425,32 @@ static void test_ipc_server_fail(void)
 	} while (conn == NULL && c < 5);
 	fail_if(conn == NULL);
 
+	/*
+	 * tell the server to exit
+	 */
 	req_header.id = IPC_MSG_REQ_SERVER_FAIL;
 	req_header.size = sizeof(struct qb_ipc_request_header);
 
  repeat_send:
-	printf("sending server fail\n");
 	res = qb_ipcc_send(conn, &req_header, req_header.size);
 	try_times++;
 	if (res < 0) {
 		if (res == -EAGAIN && try_times < 10) {
 			goto repeat_send;
-		} else {
-			errno = -res;
-			perror("qb_ipcc_send");
-			return res;
 		}
+		ck_assert_int_eq(res, 0);
 	}
-	printf("trying to recv from failed server\n");
 
- repeat_recv:
+	/*
+	 * try recv from the exit'ed server
+	 */
 	res = qb_ipcc_recv(conn,
 			&res_header,
 			sizeof(struct qb_ipc_response_header));
-	printf("recv %d\n", res);
-	ck_assert_int_eq(res, 0);
+	/*
+	 * confirm we get -ENOTCONN
+	 */
+	ck_assert_int_eq(res, -ENOTCONN);
 
 	qb_ipcc_disconnect(conn);
 	stop_process(pid);
@@ -541,7 +541,7 @@ int32_t main(void)
 
 	qb_util_set_log_function(ipc_log_fn);
 
-	srunner_run_all(sr, CK_NORMAL);
+	srunner_run_all(sr, CK_VERBOSE);
 	number_failed = srunner_ntests_failed(sr);
 	srunner_free(sr);
 	return (number_failed == 0) ? EXIT_SUCCESS : EXIT_FAILURE;
-- 
1.7.3.1