Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com>
---
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