Move to send() instead of sendmsg() as it's easier
to track resending partial messages.
Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com>
---
lib/ipc_us.c | 120 +++++++++++++++++++++++------------------------------
tests/check_ipc.c | 48 +++++++++++++++------
2 files changed, 86 insertions(+), 82 deletions(-)
diff --git a/lib/ipc_us.c b/lib/ipc_us.c
index 9c8b5be..2f36b9a 100644
--- a/lib/ipc_us.c
+++ b/lib/ipc_us.c
@@ -85,33 +85,21 @@ ssize_t
qb_ipc_us_send(struct qb_ipc_one_way *one_way, const void *msg, size_t len)
{
int32_t result;
- struct msghdr msg_send;
- struct iovec iov_send;
- char *rbuf = (char *)msg;
int32_t processed = 0;
- struct ipc_us_control *ctl = NULL;
-
- msg_send.msg_iov = &iov_send;
- msg_send.msg_iovlen = 1;
- msg_send.msg_name = 0;
- msg_send.msg_namelen = 0;
-
-#if !defined(QB_SOLARIS)
- msg_send.msg_control = 0;
- msg_send.msg_controllen = 0;
- msg_send.msg_flags = 0;
-#else
- msg_send.msg_accrights = NULL;
- msg_send.msg_accrightslen = 0;
-#endif
+ char *rbuf = (char *)msg;
retry_send:
- iov_send.iov_base = &rbuf[processed];
- iov_send.iov_len = len - processed;
+ result = send(one_way->u.us.sock,
+ &rbuf[processed],
+ len - processed,
+ MSG_NOSIGNAL);
- result = sendmsg(one_way->u.us.sock, &msg_send, MSG_NOSIGNAL);
if (result == -1) {
- return -errno;
+ if (errno == EAGAIN && processed > 0) {
+ goto retry_send;
+ } else {
+ return -errno;
+ }
}
processed += result;
@@ -119,6 +107,7 @@ retry_send:
goto retry_send;
}
if (one_way->type == QB_IPC_SOCKET) {
+ struct ipc_us_control *ctl = NULL;
ctl = (struct ipc_us_control *)one_way->u.us.shared_data;
if (ctl) {
qb_atomic_int_inc(&ctl->sent);
@@ -132,46 +121,46 @@ qb_ipc_us_sendv(struct qb_ipc_one_way *one_way, const struct iovec
*iov,
size_t iov_len)
{
int32_t result;
- struct msghdr msg_send;
int32_t processed = 0;
- size_t len = 0;
- int32_t i;
- struct ipc_us_control *ctl = NULL;
-
- for (i = 0; i < iov_len; i++) {
- len += iov[i].iov_len;
- }
- msg_send.msg_iov = (struct iovec *)iov;
- msg_send.msg_iovlen = iov_len;
- msg_send.msg_name = 0;
- msg_send.msg_namelen = 0;
-
-#if !defined(QB_SOLARIS)
- msg_send.msg_control = 0;
- msg_send.msg_controllen = 0;
- msg_send.msg_flags = 0;
-#else
- msg_send.msg_accrights = NULL;
- msg_send.msg_accrightslen = 0;
-#endif
+ int32_t total_processed = 0;
+ int32_t iov_p = 0;
+ char *rbuf = (char *)iov[iov_p].iov_base;
retry_send:
- result = sendmsg(one_way->u.us.sock, &msg_send, MSG_NOSIGNAL);
+ result = send(one_way->u.us.sock,
+ &rbuf[processed],
+ iov[iov_p].iov_len - processed,
+ MSG_NOSIGNAL);
+
if (result == -1) {
- return -errno;
+ if (errno == EAGAIN &&
+ (processed > 0 || iov_p > 0)) {
+ goto retry_send;
+ } else {
+ return -errno;
+ }
}
processed += result;
- if (processed != len) {
+ if (processed == iov[iov_p].iov_len) {
+ iov_p++;
+ total_processed += processed;
+ if (iov_p < iov_len) {
+ processed = 0;
+ rbuf = (char *)iov[iov_p].iov_base;
+ goto retry_send;
+ }
+ } else {
goto retry_send;
}
if (one_way->type == QB_IPC_SOCKET) {
+ struct ipc_us_control *ctl;
ctl = (struct ipc_us_control *)one_way->u.us.shared_data;
if (ctl) {
qb_atomic_int_inc(&ctl->sent);
}
}
- return processed;
+ return total_processed;
}
static ssize_t
@@ -245,34 +234,33 @@ qb_ipc_us_recv(struct qb_ipc_one_way * one_way,
int32_t processed = 0;
int32_t to_recv = len;
char *data = msg;
- struct ipc_us_control *ctl = NULL;
retry_recv:
result = recv(one_way->u.us.sock, &data[processed], to_recv,
MSG_NOSIGNAL | MSG_WAITALL);
- if (timeout == -1) {
- if (result == -1 && errno == EAGAIN) {
+
+ if (result == -1) {
+ if (errno == EAGAIN &&
+ (processed > 0 || timeout == -1)) {
goto retry_recv;
+ } else {
+ return -errno;
}
}
+
if (result == 0) {
qb_util_log(LOG_DEBUG,
"recv(fd %d) got 0 bytes assuming ENOTCONN",
one_way->u.us.sock);
return -ENOTCONN;
}
- if (result == -1) {
- if (errno != EAGAIN) {
- return -errno;
- }
- } else {
- processed += result;
- to_recv -= result;
- }
+ processed += result;
+ to_recv -= result;
if (processed != len) {
goto retry_recv;
}
if (one_way->type == QB_IPC_SOCKET) {
+ struct ipc_us_control *ctl = NULL;
ctl = (struct ipc_us_control *)one_way->u.us.shared_data;
if (ctl) {
(void)qb_atomic_int_dec_and_test(&ctl->sent);
@@ -282,7 +270,7 @@ retry_recv:
}
/*
- * recv a message of unknow size.
+ * recv a message of unknown size.
*/
static ssize_t
qb_ipc_us_recv_at_most(struct qb_ipc_one_way * one_way,
@@ -299,17 +287,11 @@ retry_recv:
result = recv(one_way->u.us.sock, &data[processed], to_recv,
MSG_NOSIGNAL | MSG_WAITALL);
if (result == -1) {
- if (errno != EAGAIN) {
- return -errno;
+ if (errno == EAGAIN &&
+ (processed > 0 || timeout == -1)) {
+ goto retry_recv;
} else {
- if (processed > 0 || timeout == -1) {
- /* if we are part way into receiving the
- * message or the call is blocking, retry.
- */
- goto retry_recv;
- } else {
- return -errno;
- }
+ return -errno;
}
} else if (result == 0) {
qb_util_log(LOG_DEBUG,
diff --git a/tests/check_ipc.c b/tests/check_ipc.c
index 4205ff6..1ea59ed 100644
--- a/tests/check_ipc.c
+++ b/tests/check_ipc.c
@@ -117,23 +117,27 @@ s1_msg_process_fn(qb_ipcs_connection_t *c,
}
} else if (req_pt->id == IPC_MSG_REQ_BULK_EVENTS) {
int32_t m;
+ int32_t num;
struct qb_ipcs_connection_stats stats;
response.size = sizeof(struct qb_ipc_response_header);
- response.id = IPC_MSG_RES_BULK_EVENTS;
response.error = 0;
+ qb_ipcs_connection_stats_get(c, &stats, QB_FALSE);
+ num = stats.event_q_length;
+
for (m = 0; m < num_bulk_events; m++) {
res = qb_ipcs_event_send(c, &response,
sizeof(response));
- if (res < 0) {
- qb_perror(LOG_INFO, "qb_ipcs_event_send");
- }
+ ck_assert_int_eq(res, sizeof(response));
+ response.id++;
}
qb_ipcs_connection_stats_get(c, &stats, QB_FALSE);
- ck_assert_int_eq(stats.event_q_length, num_bulk_events);
- qb_ipcs_response_send(c, &response, response.size);
+ ck_assert_int_eq(stats.event_q_length - num, num_bulk_events);
+ response.id = IPC_MSG_RES_BULK_EVENTS;
+ res = qb_ipcs_response_send(c, &response, response.size);
+ ck_assert_int_eq(res, sizeof(response));
} else if (req_pt->id == IPC_MSG_REQ_SERVER_FAIL) {
exit(0);
@@ -293,14 +297,15 @@ repeat_send:
return res;
}
}
- if (req_id == IPC_MSG_REQ_TX_RX) {
- res = qb_ipcc_recv(conn, &res_header,
- sizeof(struct qb_ipc_response_header),
- ms_timeout);
- } else {
+
+ if (req_id == IPC_MSG_REQ_DISPATCH) {
res = qb_ipcc_event_recv(conn, &res_header,
sizeof(struct qb_ipc_response_header),
ms_timeout);
+ } else {
+ res = qb_ipcc_recv(conn, &res_header,
+ sizeof(struct qb_ipc_response_header),
+ ms_timeout);
}
if (res == -EINTR) {
return -1;
@@ -601,6 +606,9 @@ count_bulk_events(int32_t fd, int32_t revents, void *data)
static void
test_ipc_bulk_events(void)
{
+ struct qb_ipc_request_header req_header;
+ struct qb_ipc_response_header res_header;
+ struct iovec iov[1];
int32_t c = 0;
int32_t j = 0;
pid_t pid;
@@ -633,13 +641,26 @@ test_ipc_bulk_events(void)
ck_assert_int_eq(res, 0);
res = send_and_check(IPC_MSG_REQ_BULK_EVENTS,
- sizeof(struct qb_ipc_request_header),
- recv_timeout, QB_TRUE);
+ 0,
+ recv_timeout, QB_TRUE);
ck_assert_int_eq(res, sizeof(struct qb_ipc_response_header));
qb_loop_run(cl);
ck_assert_int_eq(events_received, num_bulk_events);
+ req_header.id = IPC_MSG_REQ_SERVER_FAIL;
+ req_header.size = sizeof(struct qb_ipc_request_header);
+
+ iov[0].iov_len = req_header.size;
+ iov[0].iov_base = &req_header;
+ res = qb_ipcc_sendv_recv(conn, iov, 1,
+ &res_header,
+ sizeof(struct qb_ipc_response_header), -1);
+ if (res != -ECONNRESET && res != -ENOTCONN) {
+ qb_log(LOG_ERR, "id:%d size:%d", res_header.id, res_header.size);
+ ck_assert_int_eq(res, -ENOTCONN);
+ }
+
qb_ipcc_disconnect(conn);
stop_process(pid);
}
@@ -647,6 +668,7 @@ test_ipc_bulk_events(void)
START_TEST(test_ipc_bulk_events_us)
{
qb_enter();
+ send_event_on_created = QB_FALSE;
ipc_type = QB_IPC_SOCKET;
ipc_name = __func__;
test_ipc_bulk_events();
--
1.7.10