IPC: make each connection ref the owning service

This is because the connection functions use the c->service pointer
and this needs to be mirrored in the reference counting.

The service can only be free'd when all connections are destroyed
and the user as unreferenced all previously referenced connections
and the service.

Fixes #62
Thanks to Jan Friesse for the reproducer
Signed-off-by: Angus Salkeld <asalkeld@redhat.com>

examples/ipcserver.c

@@ -53,10 +53,41 @@ s1_connection_accept_fn(qb_ipcs_connection_t * c, uid_t uid, gid_t gid)
 #endif
 }
 
+static void
+outq_flush (void *data)
+{
+	static int i = 0;
+	struct cs_ipcs_conn_context *cnx;
+	cnx = qb_ipcs_context_get(data);
+
+	fprintf(stderr,"iter %u\n", i);
+	i++;
+	if (i == 20) {
+		qb_ipcs_destroy(s1);
+		s1 = NULL;
+	}
+	if (i == 21) {
+		qb_ipcs_event_send(data, "test", 4);
+	}
+	assert(memcmp(cnx, "test", 4) == 0);
+	if (i < 25) {
+		qb_loop_job_add(bms_loop, QB_LOOP_HIGH, data, outq_flush);
+	}
+}
+
+
 static void
 s1_connection_created_fn(qb_ipcs_connection_t * c)
 {
 	struct qb_ipcs_stats srv_stats;
+	struct cs_ipcs_conn_context *context;
+
+	qb_ipcs_connection_ref(c);
+	qb_loop_job_add(bms_loop, QB_LOOP_HIGH, c, outq_flush);
+
+	context = calloc(1, 20);
+	memcpy(context, "test", 4);
+	qb_ipcs_context_set(c, context);
 
 	qb_ipcs_stats_get(s1, &srv_stats, QB_FALSE);
 	qb_log(LOG_INFO, "Connection created (active:%d, closed:%d)",
@@ -66,7 +97,12 @@ s1_connection_created_fn(qb_ipcs_connection_t * c)
 static void
 s1_connection_destroyed_fn(qb_ipcs_connection_t * c)
 {
+	struct cs_ipcs_conn_context *cnx;
+
 	qb_log(LOG_INFO, "Connection about to be freed");
+
+	cnx = qb_ipcs_context_get(c);
+	free(cnx);
 }
 
 static int32_t

lib/ipcs.c

@@ -201,22 +201,11 @@ void
 qb_ipcs_unref(struct qb_ipcs_service *s)
 {
 	int32_t free_it;
-	struct qb_ipcs_connection *c = NULL;
-	struct qb_list_head *pos;
-	struct qb_list_head *n;
 
 	assert(s->ref_count > 0);
 	free_it = qb_atomic_int_dec_and_test(&s->ref_count);
 	if (free_it) {
 		qb_util_log(LOG_DEBUG, "%s() - destroying", __func__);
-		qb_list_for_each_safe(pos, n, &s->connections) {
-			c = qb_list_entry(pos, struct qb_ipcs_connection, list);
-			if (c == NULL) {
-				continue;
-			}
-			qb_ipcs_disconnect(c);
-		}
-		(void)qb_ipcs_us_withdraw(s);
 		free(s);
 	}
 }
@@ -224,6 +213,22 @@ qb_ipcs_unref(struct qb_ipcs_service *s)
 void
 qb_ipcs_destroy(struct qb_ipcs_service *s)
 {
+	struct qb_ipcs_connection *c = NULL;
+	struct qb_list_head *pos;
+	struct qb_list_head *n;
+
+	if (s == NULL) {
+		return;
+	}
+	qb_list_for_each_safe(pos, n, &s->connections) {
+		c = qb_list_entry(pos, struct qb_ipcs_connection, list);
+		if (c == NULL) {
+			continue;
+		}
+		qb_ipcs_disconnect(c);
+	}
+	(void)qb_ipcs_us_withdraw(s);
+
 	qb_ipcs_unref(s);
 }
 
@@ -489,11 +494,9 @@ qb_ipcs_connection_alloc(struct qb_ipcs_service *s)
 	}
 
 	c->refcount = 1;
-	c->service = s;
 	c->pid = 0;
 	c->euid = -1;
 	c->egid = -1;
-	qb_list_init(&c->list);
 	c->receive_buf = NULL;
 	c->context = NULL;
 	c->fc_enabled = QB_FALSE;
@@ -506,6 +509,12 @@ qb_ipcs_connection_alloc(struct qb_ipcs_service *s)
 	c->event.type = s->type;
 	(void)strlcpy(c->description, "not set yet", CONNECTION_DESCRIPTION);
 
+	/* the connection references the containing service so, make a reference.
+	 */
+	qb_ipcs_ref(s);
+	c->service = s;
+	qb_list_init(&c->list);
+
 	return c;
 }
 
@@ -537,6 +546,7 @@ qb_ipcs_connection_unref(struct qb_ipcs_connection *c)
 			c->service->serv_fns.connection_destroyed(c);
 		}
 		c->service->funcs.disconnect(c);
+		qb_ipcs_unref(c->service);
 		free(c->receive_buf);
 		free(c);
 	}

tests/check_ipc.c

@@ -75,6 +75,8 @@ static int32_t fc_enabled = 89;
 static int32_t send_event_on_created = QB_FALSE;
 static int32_t disconnect_after_created = QB_FALSE;
 static int32_t num_bulk_events = 10;
+static int32_t reference_count_test = QB_FALSE;
+
 
 static int32_t
 exit_handler(int32_t rsignal, void *data)
@@ -190,11 +192,47 @@ s1_connection_closed(qb_ipcs_connection_t *c)
 	return 0;
 }
 
+static void
+outq_flush (void *data)
+{
+	static int i = 0;
+	struct cs_ipcs_conn_context *cnx;
+	cnx = qb_ipcs_context_get(data);
+
+	qb_log(LOG_DEBUG,"iter %u\n", i);
+	i++;
+	if (i == 2) {
+		qb_ipcs_destroy(s1);
+		s1 = NULL;
+	}
+	/* is the reference counting is not working, this should fail
+	 * for i > 1.
+	 */
+	qb_ipcs_event_send(data, "test", 4);
+	assert(memcmp(cnx, "test", 4) == 0);
+	if (i < 5) {
+		qb_loop_job_add(my_loop, QB_LOOP_HIGH, data, outq_flush);
+	} else {
+		/* this single unref should clean everything up.
+		 */
+		qb_ipcs_connection_unref(data);
+		qb_log(LOG_INFO, "end of test, stopping loop");
+		qb_loop_stop(my_loop);
+	}
+}
+
+
 static void
 s1_connection_destroyed(qb_ipcs_connection_t *c)
 {
 	qb_enter();
-	qb_loop_stop(my_loop);
+	if (reference_count_test) {
+		struct cs_ipcs_conn_context *cnx;
+		cnx = qb_ipcs_context_get(c);
+		free(cnx);
+	} else {
+		qb_loop_stop(my_loop);
+	}
 	qb_leave();
 }
 
@@ -212,6 +250,16 @@ s1_connection_created(qb_ipcs_connection_t *c)
 					 sizeof(response));
 		ck_assert_int_eq(res, response.size);
 	}
+	if (reference_count_test) {
+		struct cs_ipcs_conn_context *context;
+
+		qb_ipcs_connection_ref(c);
+		qb_loop_job_add(my_loop, QB_LOOP_HIGH, c, outq_flush);
+
+		context = calloc(1, 20);
+		memcpy(context, "test", 4);
+		qb_ipcs_context_set(c, context);
+	}
 }
 
 static void
@@ -251,6 +299,7 @@ run_ipc_server(void)
 	ck_assert_int_eq(res, 0);
 
 	qb_loop_run(my_loop);
+	qb_log(LOG_DEBUG, "loop finished - done ...");
 }
 
 static int32_t
@@ -894,6 +943,56 @@ START_TEST(test_ipc_server_fail_shm)
 }
 END_TEST
 
+static void
+test_ipc_service_ref_count(void)
+{
+	int32_t c = 0;
+	int32_t j = 0;
+	pid_t pid;
+
+	reference_count_test = QB_TRUE;
+
+	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);
+
+	sleep(5);
+
+	qb_ipcc_disconnect(conn);
+	stop_process(pid);
+}
+
+
+START_TEST(test_ipc_service_ref_count_shm)
+{
+	qb_enter();
+	ipc_type = QB_IPC_SHM;
+	ipc_name = __func__;
+	test_ipc_service_ref_count();
+	qb_leave();
+}
+END_TEST
+
+START_TEST(test_ipc_service_ref_count_us)
+{
+	qb_enter();
+	ipc_type = QB_IPC_SOCKET;
+	ipc_name = __func__;
+	test_ipc_service_ref_count();
+	qb_leave();
+}
+END_TEST
 
 static Suite *
 make_shm_suite(void)
@@ -940,6 +1039,11 @@ make_shm_suite(void)
 	tcase_add_test(tc, test_ipc_event_on_created_shm);
 	suite_add_tcase(s, tc);
 
+	tc = tcase_create("ipc_service_ref_count_shm");
+	tcase_add_test(tc, test_ipc_service_ref_count_shm);
+	tcase_set_timeout(tc, 10);
+	suite_add_tcase(s, tc);
+
 	return s;
 }
 
@@ -992,6 +1096,11 @@ make_soc_suite(void)
 	tcase_add_test(tc, test_ipc_disconnect_after_created_us);
 	suite_add_tcase(s, tc);
 
+	tc = tcase_create("ipc_service_ref_count_us");
+	tcase_add_test(tc, test_ipc_service_ref_count_us);
+	tcase_set_timeout(tc, 10);
+	suite_add_tcase(s, tc);
+
 	return s;
 }