commit de5ab3029c796e51d246bab9a83c66bbb5602e86 Author: Chrissie Caulfield Date: Wed Jan 5 10:53:09 2022 +0000 ipcc: Add an async connect API (#450) diff --git a/include/qb/qbipcc.h b/include/qb/qbipcc.h index de96c72..867ba04 100644 --- a/include/qb/qbipcc.h +++ b/include/qb/qbipcc.h @@ -80,6 +80,36 @@ typedef struct qb_ipcc_connection qb_ipcc_connection_t; qb_ipcc_connection_t* qb_ipcc_connect(const char *name, size_t max_msg_size); +/** + * Asynchronously connect to an IPC service + * @param name name of the service. + * @param max_msg_size biggest msg size. + * @param connect_fd return FD to continue connection with + * @return NULL (error: see errno) or a connection object. + * + * qb_ipcc_connect_async() returns a connection FD which + * should be used added to the application's mainloop - when it is + * active, qb_ipcc_connect_continue() should be called for the + * connection to be finalised. + * NOTE: This is NOT the same FD that is used for normal applicaion + * polling. qb_ipcc_fd_get() must still be called once the connection + * is established. + */ +qb_ipcc_connection_t * +qb_ipcc_connect_async(const char *name, size_t max_msg_size, int *connect_fd); + +/** + * Finish up an asynchonous IPC connection + * @param c connection handle as returned from qb_ipcc_connect_async() + * @return 0 or -errno. + * + * Finishes up a connection that was initiated by qb_ipcc_connect_async(), + * this should only be called when the fd returned by qb_ipcc_connect_async() + * becomes active, usually as a callback in the application's main loop. + */ +int +qb_ipcc_connect_continue(struct qb_ipcc_connection * c); + /** * Test kernel dgram socket buffers to verify the largest size up * to the max_msg_size value a single msg can be. Rounds down to the diff --git a/lib/ipc_int.h b/lib/ipc_int.h index 03c5dab..87f1de1 100644 --- a/lib/ipc_int.h +++ b/lib/ipc_int.h @@ -106,7 +106,8 @@ struct qb_ipcc_connection { }; int32_t qb_ipcc_us_setup_connect(struct qb_ipcc_connection *c, - struct qb_ipc_connection_response *r); + struct qb_ipc_connection_response *r); +int qb_ipcc_setup_connect_continue(struct qb_ipcc_connection *c, struct qb_ipc_connection_response *response); ssize_t qb_ipc_us_send(struct qb_ipc_one_way *one_way, const void *msg, size_t len); ssize_t qb_ipc_us_recv(struct qb_ipc_one_way *one_way, void *msg, size_t len, int32_t timeout); int32_t qb_ipc_us_ready(struct qb_ipc_one_way *ow_data, struct qb_ipc_one_way *ow_conn, diff --git a/lib/ipc_setup.c b/lib/ipc_setup.c index c144a5e..0ef9bb6 100644 --- a/lib/ipc_setup.c +++ b/lib/ipc_setup.c @@ -446,9 +446,7 @@ qb_ipcc_us_setup_connect(struct qb_ipcc_connection *c, { int32_t res; struct qb_ipc_connection_request request; - struct ipc_auth_data *data; #ifdef QB_LINUX - int off = 0; int on = 1; #endif @@ -471,13 +469,24 @@ qb_ipcc_us_setup_connect(struct qb_ipcc_connection *c, return res; } + /* ... To be continued ... (when the FD is active) */ + return 0; +} + +/* Called from ipcc_connect_continue() when async connect socket is active */ +int qb_ipcc_setup_connect_continue(struct qb_ipcc_connection *c, struct qb_ipc_connection_response *r) +{ + struct ipc_auth_data *data; + int32_t res; +#ifdef QB_LINUX + int off = 0; +#endif data = init_ipc_auth_data(c->setup.u.us.sock, sizeof(struct qb_ipc_connection_response)); if (data == NULL) { qb_ipcc_us_sock_close(c->setup.u.us.sock); return -ENOMEM; } - qb_ipc_us_ready(&c->setup, NULL, -1, POLLIN); res = qb_ipc_us_recv_msghdr(data); #ifdef QB_LINUX @@ -498,6 +507,7 @@ qb_ipcc_us_setup_connect(struct qb_ipcc_connection *c, c->server_pid = data->ugp.pid; destroy_ipc_auth_data(data); + return r->hdr.error; } diff --git a/lib/ipcc.c b/lib/ipcc.c index a6cf409..c744ea1 100644 --- a/lib/ipcc.c +++ b/lib/ipcc.c @@ -45,6 +45,70 @@ qb_ipcc_connect(const char *name, size_t max_msg_size) if (res < 0) { goto disconnect_and_cleanup; } + qb_ipc_us_ready(&c->setup, NULL, -1, POLLIN); + res = qb_ipcc_connect_continue(c); + if (res != 0) { + /* qb_ipcc_connect_continue() has cleaned up for us */ + errno = -res; + return NULL; + } + + return c; + +disconnect_and_cleanup: + if (c->setup.u.us.sock >= 0) { + qb_ipcc_us_sock_close(c->setup.u.us.sock); + } + free(c->receive_buf); + free(c); + errno = -res; + return NULL; +} + +qb_ipcc_connection_t * +qb_ipcc_connect_async(const char *name, size_t max_msg_size, int *connect_fd) +{ + int32_t res; + qb_ipcc_connection_t *c = NULL; + struct qb_ipc_connection_response response; + + c = calloc(1, sizeof(struct qb_ipcc_connection)); + if (c == NULL) { + return NULL; + } + + c->setup.max_msg_size = QB_MAX(max_msg_size, + sizeof(struct qb_ipc_connection_response)); + (void)strlcpy(c->name, name, NAME_MAX); + res = qb_ipcc_us_setup_connect(c, &response); + if (res < 0) { + goto disconnect_and_cleanup; + } + + *connect_fd = c->setup.u.us.sock; + return c; + +disconnect_and_cleanup: + if (c->setup.u.us.sock >= 0) { + qb_ipcc_us_sock_close(c->setup.u.us.sock); + } + free(c->receive_buf); + free(c); + errno = -res; + return NULL; +} + +int qb_ipcc_connect_continue(struct qb_ipcc_connection * c) +{ + struct qb_ipc_connection_response response; + int32_t res; + + /* Finish up the authentication part */ + res = qb_ipcc_setup_connect_continue(c, &response); + if (res != 0) { + goto disconnect_and_cleanup; + } + c->response.type = response.connection_type; c->request.type = response.connection_type; c->event.type = response.connection_type; @@ -79,7 +143,7 @@ qb_ipcc_connect(const char *name, size_t max_msg_size) goto disconnect_and_cleanup; } c->is_connected = QB_TRUE; - return c; + return 0; disconnect_and_cleanup: if (c->setup.u.us.sock >= 0) { @@ -88,7 +152,8 @@ disconnect_and_cleanup: free(c->receive_buf); free(c); errno = -res; - return NULL; + return -res; + } static int32_t diff --git a/tests/check_ipc.c b/tests/check_ipc.c index e8f81f3..6090354 100644 --- a/tests/check_ipc.c +++ b/tests/check_ipc.c @@ -1007,6 +1007,62 @@ repeat_send: return res; } + +static int32_t +process_async_connect(int32_t fd, int32_t revents, void *data) +{ + qb_loop_t *cl = (qb_loop_t *)data; + int res; + + res = qb_ipcc_connect_continue(conn); + ck_assert_int_eq(res, 0); + qb_loop_stop(cl); + return 0; +} +static void test_ipc_connect_async(void) +{ + struct qb_ipc_request_header req_header; + struct qb_ipc_response_header res_header; + int32_t res; + pid_t pid; + uint32_t max_size = MAX_MSG_SIZE; + int connect_fd; + struct iovec iov[1]; + static qb_loop_t *cl; + + pid = run_function_in_new_process("server", run_ipc_server, NULL); + ck_assert(pid != -1); + + conn = qb_ipcc_connect_async(ipc_name, max_size, &connect_fd); + ck_assert(conn != NULL); + + cl = qb_loop_create(); + res = qb_loop_poll_add(cl, QB_LOOP_MED, + connect_fd, POLLIN, + cl, process_async_connect); + ck_assert_int_eq(res, 0); + qb_loop_run(cl); + + /* Send some data */ + req_header.id = IPC_MSG_REQ_TX_RX; + 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), 5000); + + ck_assert_int_ge(res, 0); + + request_server_exit(); + verify_graceful_stop(pid); + + + qb_ipcc_disconnect(conn); +} + static void test_ipc_txrx_timeout(void) { @@ -1226,6 +1282,7 @@ START_TEST(test_ipc_txrx_shm_timeout) } END_TEST + START_TEST(test_ipc_txrx_us_timeout) { qb_enter(); @@ -1236,6 +1293,25 @@ START_TEST(test_ipc_txrx_us_timeout) } END_TEST +START_TEST(test_ipc_shm_connect_async) +{ + qb_enter(); + ipc_type = QB_IPC_SHM; + set_ipc_name(__func__); + test_ipc_connect_async(); + qb_leave(); +} +END_TEST + +START_TEST(test_ipc_us_connect_async) +{ + qb_enter(); + ipc_type = QB_IPC_SHM; + set_ipc_name(__func__); + test_ipc_connect_async(); + qb_leave(); +} +END_TEST START_TEST(test_ipc_txrx_shm_getauth) { @@ -2277,6 +2353,8 @@ make_shm_suite(void) TCase *tc; Suite *s = suite_create("shm"); + add_tcase(s, tc, test_ipc_shm_connect_async, 7); + add_tcase(s, tc, test_ipc_txrx_shm_getauth, 7); add_tcase(s, tc, test_ipc_txrx_shm_timeout, 28); add_tcase(s, tc, test_ipc_server_fail_shm, 7); @@ -2308,6 +2386,8 @@ make_soc_suite(void) Suite *s = suite_create("socket"); TCase *tc; + add_tcase(s, tc, test_ipc_us_connect_async, 7); + add_tcase(s, tc, test_ipc_txrx_us_getauth, 7); add_tcase(s, tc, test_ipc_txrx_us_timeout, 28); /* Commented out for the moment as space in /dev/shm on the CI machines