timers: Add some locking

include/tlist.h

@@ -36,6 +36,7 @@ static int64_t timerlist_hertz;
 
 struct timerlist {
 	struct qb_list_head timer_head;
+	pthread_mutex_t list_mutex;
 };
 
 struct timerlist_timer {
@@ -50,16 +51,20 @@ struct timerlist_timer {
 static inline void timerlist_init(struct timerlist *timerlist)
 {
 	qb_list_init(&timerlist->timer_head);
+	pthread_mutex_init(&timerlist->list_mutex, NULL);
 	timerlist_hertz = qb_util_nano_monotonic_hz();
 }
 
-static inline void timerlist_add(struct timerlist *timerlist,
+static inline int32_t timerlist_add(struct timerlist *timerlist,
 				 struct timerlist_timer *timer)
 {
 	struct qb_list_head *timer_list = 0;
 	struct timerlist_timer *timer_from_list;
 	int32_t found = QB_FALSE;
 
+	if (pthread_mutex_lock(&timerlist->list_mutex)) {
+		return -errno;
+	}
 	qb_list_for_each(timer_list, &timerlist->timer_head) {
 
 		timer_from_list = qb_list_entry(timer_list,
@@ -74,6 +79,8 @@ static inline void timerlist_add(struct timerlist *timerlist,
 	if (found == QB_FALSE) {
 		qb_list_add_tail(&timer->list, &timerlist->timer_head);
 	}
+	pthread_mutex_unlock(&timerlist->list_mutex);
+	return 0;
 }
 
 static inline int32_t timerlist_add_duration(struct timerlist *timerlist,
@@ -82,6 +89,7 @@ static inline int32_t timerlist_add_duration(struct timerlist *timerlist,
 					 uint64_t nano_duration,
 					 timer_handle * handle)
 {
+	int res;
 	struct timerlist_timer *timer;
 
 	timer =
@@ -95,7 +103,11 @@ static inline int32_t timerlist_add_duration(struct timerlist *timerlist,
 	timer->data = data;
 	timer->timer_fn = timer_fn;
 	timer->handle_addr = handle;
-	timerlist_add(timerlist, timer);
+	res = timerlist_add(timerlist, timer);
+	if (res) {
+		free(timer);
+		return res;
+	}
 
 	*handle = timer;
 	return (0);

lib/loop_timerlist.c

@@ -44,6 +44,7 @@ struct qb_timer_source {
 	struct timerlist timerlist;
 	qb_array_t *timers;
 	size_t timer_entry_count;
+	pthread_mutex_t lock;
 };
 
 static void
@@ -104,6 +105,7 @@ qb_loop_timer_create(struct qb_loop *l)
 	timerlist_init(&my_src->timerlist);
 	my_src->timers = qb_array_create_2(16, sizeof(struct qb_loop_timer), 16);
 	my_src->timer_entry_count = 0;
+	pthread_mutex_init(&my_src->lock, NULL);
 
 	return (struct qb_loop_source *)my_src;
 }
@@ -192,6 +194,9 @@ qb_loop_timer_add(struct qb_loop * lp,
 	}
 	my_src = (struct qb_timer_source *)l->timer_source;
 
+	if (pthread_mutex_lock(&my_src->lock)) {
+		return -errno;
+	}
 	i = _get_empty_array_position_(my_src);
 	assert(qb_array_index(my_src->timers, i, (void **)&t) >= 0);
 	t->state = QB_POLL_ENTRY_ACTIVE;
@@ -202,6 +207,9 @@ qb_loop_timer_add(struct qb_loop * lp,
 	t->p = p;
 	qb_list_init(&t->item.list);
 
+	/* Unlock here to stop anyone else changing the state while we're initializing */
+	pthread_mutex_unlock(&my_src->lock);
+
 	/*
 	 * Make sure just positive integers are used for the integrity(?)
 	 * checks within 2^32 address space, if we miss 200 times in a row

tests/check_loop.c

@@ -439,6 +439,56 @@ START_TEST(test_loop_timer_basic)
 }
 END_TEST
 
+static void *loop_timer_thread(void *arg)
+{
+	int res;
+	qb_loop_t *l = (qb_loop_t *)arg;
+	qb_loop_timer_handle test_tht;
+
+	res = qb_loop_timer_add(l, QB_LOOP_LOW, 5*QB_TIME_NS_IN_MSEC, l, one_shot_tmo, &test_tht);
+	ck_assert_int_eq(res, 0);
+
+	res = qb_loop_timer_is_running(l, test_th);
+	ck_assert_int_eq(res, QB_TRUE);
+
+	sleep(5);
+
+	return (void *)0;
+}
+
+/* This test will probably never fail (unless something
+   really bad happens) but is useful for running under
+   helgrind to find threading issues */
+START_TEST(test_loop_timer_threads)
+{
+	int32_t res;
+	pthread_t thr;
+	qb_loop_t *l = qb_loop_create();
+	ck_assert(l != NULL);
+
+	res = pthread_create(&thr, NULL, loop_timer_thread, l);
+
+	res = qb_loop_timer_add(l, QB_LOOP_LOW, 7*QB_TIME_NS_IN_MSEC, l, reset_one_shot_tmo, &reset_th);
+	ck_assert_int_eq(res, 0);
+
+	res = qb_loop_timer_add(l, QB_LOOP_HIGH, 20*QB_TIME_NS_IN_MSEC, l, check_time_left, &test_th2);
+	ck_assert_int_eq(res, 0);
+
+	res = qb_loop_timer_add(l, QB_LOOP_LOW, 60*QB_TIME_NS_IN_MSEC, l, job_stop, &test_th);
+	ck_assert_int_eq(res, 0);
+
+	qb_loop_run(l);
+
+	ck_assert_int_eq(reset_timer_step, 2);
+
+	pthread_join(thr, NULL);
+	qb_loop_destroy(l);
+}
+END_TEST
+
+
+
+
 struct qb_stop_watch {
 	uint64_t start;
 	uint64_t end;
@@ -742,6 +792,7 @@ loop_timer_suite(void)
 	add_tcase(s, tc, test_loop_timer_basic, 30);
 	add_tcase(s, tc, test_loop_timer_precision, 30);
 	add_tcase(s, tc, test_loop_timer_expire_leak, 30);
+	add_tcase(s, tc, test_loop_timer_threads, 30);
 
 	return s;
 }