Change in libosmocore[master]: WIP: osmo_it_msgq: Osmocom Inter-thread Message Queue

Harald Welte gerrit-no-reply at lists.osmocom.org
Sat Apr 6 16:18:38 UTC 2019


Harald Welte has uploaded this change for review. ( https://gerrit.osmocom.org/13538


Change subject: WIP: osmo_it_msgq: Osmocom Inter-thread Message Queue
......................................................................

WIP: osmo_it_msgq: Osmocom Inter-thread Message Queue

This implements an inter-thread message queue based on top of
osmocom msgb, linked lists, eventfd and pthread mutex.

Change-Id: I30a7233cb0b51a883ad85c8c4165270b32c4be61
---
M include/Makefile.am
A include/osmocom/core/it_msgq.h
M src/Makefile.am
A src/it_msgq.c
M tests/Makefile.am
A tests/it_msgq/it_msgq_test.c
A tests/it_msgq/it_msgq_test.ok
M tests/testsuite.at
8 files changed, 399 insertions(+), 3 deletions(-)



  git pull ssh://gerrit.osmocom.org:29418/libosmocore refs/changes/38/13538/1

diff --git a/include/Makefile.am b/include/Makefile.am
index 17f7d1c..8b18530 100644
--- a/include/Makefile.am
+++ b/include/Makefile.am
@@ -27,6 +27,7 @@
                        osmocom/core/gsmtap.h \
                        osmocom/core/gsmtap_util.h \
                        osmocom/core/isdnhdlc.h \
+                       osmocom/core/it_msgq.h \
                        osmocom/core/linuxlist.h \
                        osmocom/core/linuxrbtree.h \
                        osmocom/core/logging.h \
diff --git a/include/osmocom/core/it_msgq.h b/include/osmocom/core/it_msgq.h
new file mode 100644
index 0000000..8e81a9a
--- /dev/null
+++ b/include/osmocom/core/it_msgq.h
@@ -0,0 +1,43 @@
+#pragma once
+
+#include <osmocom/core/linuxlist.h>
+#include <osmocom/core/msgb.h>
+#include <osmocom/core/select.h>
+#include <pthread.h>
+
+/*! \defgroup it_msgq Inter-ThreadMessage Queue
+ *  @{
+ *  \file it_msgq.h */
+
+struct osmo_it_msgq {
+	/* entry in global list of message queues */
+	struct llist_head entry;
+
+	/* the actual list of msgb's. HEAD: first in queue; TAIL: last in queue */
+	struct llist_head list;
+	/* A pthread mutex to safeguard accesses to the queue. No rwlock as we always write. */
+	pthread_mutex_t mutex;
+	/* Current count of messages in the queue */
+	unsigned int current_length;
+	/* osmo-fd wrapped eventfd */
+	struct osmo_fd event_ofd;
+
+	/* a user-defined name for this queue */
+	const char *name;
+	/* maximum permitted length of queue */
+	unsigned int max_length;
+	/* read call-back, called for each de-queued message */
+	void (*read_cb)(struct osmo_it_msgq *q, struct msgb *msg);
+	/* opaque data pointer passed through to call-back function */
+	void *data;
+};
+
+struct osmo_it_msgq *osmo_it_msgq_by_name(const char *name);
+int osmo_it_msgq_enqueue(struct osmo_it_msgq *queue, struct msgb *msg);
+struct msgb *osmo_it_msgq_dequeue(struct osmo_it_msgq *queue);
+struct osmo_it_msgq *osmo_it_msgq_alloc(void *ctx, const char *name, unsigned int max_length,
+					void (*read_cb)(struct osmo_it_msgq *q, struct msgb *msg));
+void osmo_it_msgq_destroy(struct osmo_it_msgq *q);
+void osmo_it_msgq_flush(struct osmo_it_msgq *q);
+
+/*! @} */
diff --git a/src/Makefile.am b/src/Makefile.am
index 54e9280..fb2e125 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -12,7 +12,7 @@
 
 lib_LTLIBRARIES = libosmocore.la
 
-libosmocore_la_LIBADD = $(BACKTRACE_LIB) $(TALLOC_LIBS) $(LIBRARY_RT)
+libosmocore_la_LIBADD = $(BACKTRACE_LIB) $(TALLOC_LIBS) $(LIBRARY_RT) -lpthread
 libosmocore_la_SOURCES = context.c timer.c timer_gettimeofday.c timer_clockgettime.c \
 			 select.c signal.c msgb.c bits.c \
 			 bitvec.c bitcomp.c counter.c fsm.c \
@@ -23,7 +23,7 @@
 			 loggingrb.c crc8gen.c crc16gen.c crc32gen.c crc64gen.c \
 			 macaddr.c stat_item.c stats.c stats_statsd.c prim.c \
 			 conv_acc.c conv_acc_generic.c sercomm.c prbs.c \
-			 isdnhdlc.c \
+			 isdnhdlc.c it_msgq.c \
 			 tdef.c
 
 if HAVE_SSSE3
diff --git a/src/it_msgq.c b/src/it_msgq.c
new file mode 100644
index 0000000..3086d01
--- /dev/null
+++ b/src/it_msgq.c
@@ -0,0 +1,220 @@
+/*! \file it_msgq.c
+ * Osmocom Inter-Thread message queue implementation */
+/* (C) 2019 by Harald Welte <laforge at gnumonks.org>
+ * All Rights Reserved.
+ *
+ * SPDX-License-Identifier: GPL-2.0+
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 2 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program; if not, write to the Free Software
+ *  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
+ *  MA  02110-1301, USA.
+ */
+
+/*! \addtogroup it_msgq
+ *  @{
+ *  Inter-Thread Message Queue.
+ *
+ * This implements a general-purpose message queue between threads. It
+ * uses 'struct msgb' as elements in the queue and an eventfd-based notification
+ * mechanism.  Hence, it can be used for pretty much anything that can be stored
+ * inside msgbs, including msgb-wrapped osmo_prim.
+ *
+ * The idea is that the sending thread simply calls osmo_it_msgq_enqueue().
+ * The receiving thread is woken up from its osmo_select_main() loop by eventfd,
+ * and a general osmo_fd callback function for the eventfd will dequeue each msgb
+ * and call a queue-specific callback function.
+ */
+
+#include <pthread.h>
+#include <unistd.h>
+#include <string.h>
+#include <errno.h>
+#include <sys/eventfd.h>
+
+#include <osmocom/core/linuxlist.h>
+#include <osmocom/core/msgb.h>
+#include <osmocom/core/it_msgq.h>
+
+static int eventfd_increment(int fd, uint64_t inc)
+{
+	int rc;
+
+	rc = write(fd, &inc, sizeof(inc));
+	if (rc != sizeof(inc))
+		return -1;
+
+	return 0;
+}
+
+/* global (for all threads) list of message queues in a program + associated lock */
+static LLIST_HEAD(msg_queues);
+static pthread_rwlock_t msg_queues_rwlock;
+
+static struct osmo_it_msgq *_osmo_it_msgq_by_name(const char *name)
+{
+	struct osmo_it_msgq *q;
+	llist_for_each_entry(q, &msg_queues, entry) {
+		if (!strcmp(q->name, name))
+			return q;
+	}
+	return NULL;
+}
+
+struct osmo_it_msgq *osmo_it_msgq_by_name(const char *name)
+{
+	struct osmo_it_msgq *q;
+	pthread_rwlock_rdlock(&msg_queues_rwlock);
+	q = _osmo_it_msgq_by_name(name);
+	pthread_rwlock_unlock(&msg_queues_rwlock);
+	return q;
+}
+
+/* osmo_fd call-back when eventfd is readable */
+static int osmo_it_msgq_fd_cb(struct osmo_fd *ofd, unsigned int what)
+{
+	struct osmo_it_msgq *q = (struct osmo_it_msgq *) ofd->data;
+	uint64_t val;
+	int i, rc;
+
+	if (!(what & OSMO_FD_READ))
+		return 0;
+
+	rc = read(ofd->fd, &val, sizeof(val));
+	if (rc < sizeof(val))
+		return rc;
+
+	for (i = 0; i < val; i++) {
+		struct msgb *msg = osmo_it_msgq_dequeue(q);
+		/* in case the user might have called osmo_it_msgq_flush() we may
+		 * end up in the eventfd-dispatch but witout any messages left in the queue,
+		 * otherwise I'd have loved to OSMO_ASSERT(msg) here. */
+		if (!msg)
+			break;
+		q->read_cb(q, msg);
+	}
+
+	return 0;
+}
+
+/*! Allocate a new inter-thread message queue.
+ *  \param[in] ctx talloc context from which to allocate the queue
+ *  \param[in] name human-readable string name of the queue; function creates a copy.
+ *  \param[in] read_cb call-back function to be called for each de-queued message
+ *  \returns a newly-allocated inter-thread message queue; NULL in case of error */
+struct osmo_it_msgq *osmo_it_msgq_alloc(void *ctx, const char *name, unsigned int max_length,
+					void (*read_cb)(struct osmo_it_msgq *q, struct msgb *msg))
+{
+	struct osmo_it_msgq *q;
+	int fd;
+
+	q = talloc_zero(ctx, struct osmo_it_msgq);
+	if (!q)
+		return NULL;
+	q->name = talloc_strdup(q, name);
+	q->current_length = 0;
+	q->max_length = max_length;
+	q->read_cb = read_cb;
+	INIT_LLIST_HEAD(&q->list);
+
+	/* create eventfd */
+	fd = eventfd(0, 0);
+	if (fd < 0) {
+		talloc_free(q);
+		return NULL;
+	}
+
+	/* initialize BUT NOT REGISTER the osmo_fd. The receiving thread must
+	 * take are to select/poll/read/... on ot */
+	osmo_fd_setup(&q->event_ofd, fd, OSMO_FD_READ, osmo_it_msgq_fd_cb, q, 0);
+
+	/* add to global list of queues, checking for duplicate names */
+	pthread_rwlock_wrlock(&msg_queues_rwlock);
+	if (_osmo_it_msgq_by_name(q->name)) {
+		pthread_rwlock_unlock(&msg_queues_rwlock);
+		osmo_fd_close(&q->event_ofd);
+		talloc_free(q);
+		return NULL;
+	}
+	llist_add_tail(&q->entry, &msg_queues);
+	pthread_rwlock_unlock(&msg_queues_rwlock);
+
+	return q;
+}
+
+/*! Flush all messages currently present in queue */
+static void _osmo_it_msgq_flush(struct osmo_it_msgq *q)
+{
+	struct msgb *msg;
+	while ((msg = msgb_dequeue_count(&q->list, &q->current_length))) {
+		msgb_free(msg);
+	}
+}
+
+/*! Flush all messages currently present in queue */
+void osmo_it_msgq_flush(struct osmo_it_msgq *q)
+{
+	pthread_mutex_lock(&q->mutex);
+	_osmo_it_msgq_flush(q);
+	pthread_mutex_unlock(&q->mutex);
+}
+
+/*! Destroy a message queue */
+void osmo_it_msgq_destroy(struct osmo_it_msgq *q)
+{
+	/* first remove from global list of queues */
+	pthread_rwlock_wrlock(&msg_queues_rwlock);
+	llist_del(&q->entry);
+	pthread_rwlock_unlock(&msg_queues_rwlock);
+	/* next, close the eventfd */
+	osmo_fd_close(&q->event_ofd);
+	/* flush all messages still present */
+	osmo_it_msgq_flush(q);
+	/* and finally release memory */
+	talloc_free(q);
+}
+
+/*! Thread-safe en-queue to an inter-thread message queue.
+ *  \param[in] queue Inter-thread queue on which to enqueue
+ *  \param[in] msgb Message buffer to enqueue
+ *  \returns 0 on success; negative on error */
+int osmo_it_msgq_enqueue(struct osmo_it_msgq *queue, struct msgb *msg)
+{
+	pthread_mutex_lock(&queue->mutex);
+	if (queue->current_length+1 > queue->max_length) {
+		pthread_mutex_unlock(&queue->mutex);
+		return -ENOSPC;
+	}
+	msgb_enqueue_count(&queue->list, msg, &queue->current_length);
+	pthread_mutex_unlock(&queue->mutex);
+	/* increment eventfd counter by one */
+	eventfd_increment(queue->event_ofd.fd, 1);
+	return 0;
+}
+
+/*! Thread-safe de-queue from an inter-thread message queue.
+ *  \param[in] queue Inter-thread queue from which to dequeue
+ *  \returns dequeued message buffer; NULL if none available
+ */
+struct msgb *osmo_it_msgq_dequeue(struct osmo_it_msgq *queue)
+{
+	struct msgb *msg;
+
+	pthread_mutex_lock(&queue->mutex);
+	msg = msgb_dequeue_count(&queue->list, &queue->current_length);
+	pthread_mutex_unlock(&queue->mutex);
+
+	return msg;
+}
+
+/*! @} */
diff --git a/tests/Makefile.am b/tests/Makefile.am
index 6aa1279..c51c039 100644
--- a/tests/Makefile.am
+++ b/tests/Makefile.am
@@ -30,7 +30,7 @@
 		 tdef/tdef_test tdef/tdef_vty_test_config_root		\
 		 tdef/tdef_vty_test_config_subnode			\
 		 tdef/tdef_vty_test_dynamic				\
-		 context/context_test					\
+		 context/context_test it_msgq/it_msgq_test		\
 		 $(NULL)
 
 if ENABLE_MSGFILE
@@ -240,6 +240,10 @@
 context_context_test_SOURCES = context/context_test.c
 context_context_test_LDADD = $(LDADD)
 
+
+it_msgq_it_msgq_test_SOURCES = it_msgq/it_msgq_test.c
+it_msgq_it_msgq_test_LDADD = $(LDADD)
+
 # The `:;' works around a Bash 3.2 bug when the output is not writeable.
 $(srcdir)/package.m4: $(top_srcdir)/configure.ac
 	:;{ \
@@ -309,6 +313,7 @@
 	     tdef/tdef_vty_test_config_subnode.vty \
 	     tdef/tdef_vty_test_dynamic.vty \
 	     context/context_test.ok \
+	     it_msgq/it_msgq_test.ok \
 	     $(NULL)
 
 DISTCLEANFILES = atconfig atlocal conv/gsm0503_test_vectors.c
diff --git a/tests/it_msgq/it_msgq_test.c b/tests/it_msgq/it_msgq_test.c
new file mode 100644
index 0000000..c01b613
--- /dev/null
+++ b/tests/it_msgq/it_msgq_test.c
@@ -0,0 +1,106 @@
+#include <stdio.h>
+#include <errno.h>
+
+#include <osmocom/core/it_msgq.h>
+#include <osmocom/core/msgb.h>
+
+#define ENTER_TC	printf("\n== Entering test case %s\n", __func__)
+
+static void tc_alloc(void)
+{
+	struct osmo_it_msgq *q1, *q2;
+
+	ENTER_TC;
+
+	printf("allocating q1\n");
+	q1 = osmo_it_msgq_alloc(OTC_GLOBAL, "q1", 3, NULL);
+	OSMO_ASSERT(q1);
+
+	/* ensure that no duplicate allocation for the */
+	printf("attempting duplicate allocation of qa\n");
+	q2 = osmo_it_msgq_alloc(OTC_GLOBAL, "q1", 3, NULL);
+	OSMO_ASSERT(!q2);
+
+	/* ensure that same name can be re-created after destroying old one */
+	osmo_it_msgq_destroy(q1);
+	printf("re-allocating q1\n");
+	q1 = osmo_it_msgq_alloc(OTC_GLOBAL, "q1", 3, NULL);
+	OSMO_ASSERT(q1);
+
+	osmo_it_msgq_destroy(q1);
+}
+
+static void tc_queue_length(void)
+{
+	struct osmo_it_msgq *q1;
+	unsigned int qlen = 3;
+	struct msgb *msg;
+	int i, rc;
+
+	ENTER_TC;
+
+	printf("allocating q1\n");
+	q1 = osmo_it_msgq_alloc(OTC_GLOBAL, "q1", qlen, NULL);
+	OSMO_ASSERT(q1);
+
+	printf("adding queue entries up to the limit\n");
+	for (i = 0; i < qlen; i++) {
+		msg = msgb_alloc(23, __func__);
+		rc = osmo_it_msgq_enqueue(q1, msg);
+		OSMO_ASSERT(rc == 0);
+	}
+	printf("attempting to add more than the limit\n");
+	msg = msgb_alloc(23, __func__);
+	rc = osmo_it_msgq_enqueue(q1, msg);
+	OSMO_ASSERT(rc == -ENOSPC);
+
+	osmo_it_msgq_destroy(q1);
+}
+
+static int g_read_cb_count;
+
+static void q_read_cb(struct osmo_it_msgq *q, struct msgb *msg)
+{
+	g_read_cb_count++;
+	talloc_free(msg);
+}
+
+static void tc_eventfd(void)
+{
+	struct osmo_it_msgq *q1;
+	unsigned int qlen = 30;
+	struct msgb *msg;
+	int i, rc;
+
+	ENTER_TC;
+
+	printf("allocating q1\n");
+	q1 = osmo_it_msgq_alloc(OTC_GLOBAL, "q1", qlen, q_read_cb);
+	OSMO_ASSERT(q1);
+	osmo_fd_register(&q1->event_ofd);
+
+	/* ensure read-cb isn't called unless we enqueue something */
+	osmo_select_main(1);
+	OSMO_ASSERT(g_read_cb_count == 0);
+
+	/* ensure read-cb is called for each enqueued msg once */
+	printf("adding %u queue entries up to the limit\n", qlen);
+	for (i = 0; i < qlen; i++) {
+		msg = msgb_alloc(23, __func__);
+		rc = osmo_it_msgq_enqueue(q1, msg);
+		OSMO_ASSERT(rc == 0);
+	}
+
+	osmo_select_main(1);
+	printf("%u entries were dequeued\n", qlen);
+	OSMO_ASSERT(g_read_cb_count == qlen);
+
+	osmo_it_msgq_destroy(q1);
+}
+
+int main(int argc, char **argv)
+{
+	tc_alloc();
+	tc_queue_length();
+	tc_eventfd();
+}
diff --git a/tests/it_msgq/it_msgq_test.ok b/tests/it_msgq/it_msgq_test.ok
new file mode 100644
index 0000000..7f102c6
--- /dev/null
+++ b/tests/it_msgq/it_msgq_test.ok
@@ -0,0 +1,15 @@
+
+== Entering test case tc_alloc
+allocating q1
+attempting duplicate allocation of qa
+re-allocating q1
+
+== Entering test case tc_queue_length
+allocating q1
+adding queue entries up to the limit
+attempting to add more than the limit
+
+== Entering test case tc_eventfd
+allocating q1
+adding 30 queue entries up to the limit
+30 entries were dequeued
diff --git a/tests/testsuite.at b/tests/testsuite.at
index f1534e4..2c48485 100644
--- a/tests/testsuite.at
+++ b/tests/testsuite.at
@@ -337,3 +337,9 @@
 cat $abs_srcdir/context/context_test.ok > expout
 AT_CHECK([$abs_top_builddir/tests/context/context_test], [0], [expout], [ignore])
 AT_CLEANUP
+
+AT_SETUP([it_msgq])
+AT_KEYWORDS([it_msgq])
+cat $abs_srcdir/it_msgq/it_msgq_test.ok > expout
+AT_CHECK([$abs_top_builddir/tests/it_msgq/it_msgq_test], [0], [expout], [ignore])
+AT_CLEANUP

-- 
To view, visit https://gerrit.osmocom.org/13538
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings

Gerrit-Project: libosmocore
Gerrit-Branch: master
Gerrit-MessageType: newchange
Gerrit-Change-Id: I30a7233cb0b51a883ad85c8c4165270b32c4be61
Gerrit-Change-Number: 13538
Gerrit-PatchSet: 1
Gerrit-Owner: Harald Welte <laforge at gnumonks.org>
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://lists.osmocom.org/pipermail/gerrit-log/attachments/20190406/fbed2e79/attachment.html>


More information about the gerrit-log mailing list