lib: ZeroMQ read handler, v2

This uses zmq_getsockopt(ZMQ_FD) to create a libfrr read event, which
then wraps zmq_poll and calls an user-specified ZeroMQ read handler.
It's wrapped in a separate library in order to make ZeroMQ support an
installation-time option instead of build-time.

Extended to support per-message and per-fragment callbacks as discussed
with Bingen in PR #566.

Signed-off-by: David Lamparter <equinox@opensourcerouting.org>

configure.ac

@@ -320,6 +320,8 @@ AC_ARG_WITH(rfp-path,
   AS_HELP_STRING([--with-rfp-path[=DIR]],[path to replaced stub RFP used with BGP VNC]))
 AC_ARG_ENABLE(snmp,
   AS_HELP_STRING([--enable-snmp=ARG], [enable SNMP support (smux or agentx)]))
+AC_ARG_ENABLE(zeromq,
+  AS_HELP_STRING([--enable-zeromq], [enable ZeroMQ handler (libfrrzmq)]))
 AC_ARG_WITH(libpam,
   AS_HELP_STRING([--with-libpam], [use libpam for PAM support in vtysh]))
 AC_ARG_ENABLE(ospfapi,
@@ -1714,6 +1716,21 @@ AC_CHECK_HEADER([malloc.h],
   )
  ], [], FRR_INCLUDES)
 
+dnl ------
+dnl ZeroMQ
+dnl ------
+if test "x$enable_zeromq" != "xno"; then
+  PKG_CHECK_MODULES(ZEROMQ, [libzmq >= 4.0.0], [
+    AC_DEFINE(HAVE_ZEROMQ, 1, [Enable ZeroMQ support])
+    ZEROMQ=true
+  ], [
+    if test "x$enable_zeromq" = "xyes"; then
+      AC_MSG_ERROR([configuration specifies --enable-zeromq but libzmq was not found])
+    fi
+  ])
+fi
+AM_CONDITIONAL([ZEROMQ], test "x$ZEROMQ" = "xtrue")
+
 dnl ----------
 dnl configure date
 dnl ----------

lib/frr_zmq.c

@@ -0,0 +1,191 @@
+/*
+ * libzebra ZeroMQ bindings
+ * Copyright (C) 2015  David Lamparter
+ *
+ * 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; see the file COPYING; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+
+#include <zebra.h>
+#include <zmq.h>
+
+#include "thread.h"
+#include "memory.h"
+#include "frr_zmq.h"
+#include "log.h"
+
+DEFINE_MTYPE_STATIC(LIB, ZEROMQ_CB, "ZeroMQ callback")
+
+/* libzmq's context */
+void *frrzmq_context = NULL;
+static unsigned frrzmq_initcount = 0;
+
+void frrzmq_init(void)
+{
+	if (frrzmq_initcount++ == 0) {
+		frrzmq_context = zmq_ctx_new();
+		zmq_ctx_set(frrzmq_context, ZMQ_IPV6, 1);
+	}
+}
+
+void frrzmq_finish(void)
+{
+	if (--frrzmq_initcount == 0) {
+		zmq_ctx_term(frrzmq_context);
+		frrzmq_context = NULL;
+	}
+}
+
+/* read callback integration */
+struct frrzmq_cb {
+	struct thread *thread;
+	void *zmqsock;
+	void *arg;
+	int fd;
+
+	bool cancelled;
+
+	void (*cb_msg)(void *arg, void *zmqsock);
+	void (*cb_part)(void *arg, void *zmqsock,
+			zmq_msg_t *msg, unsigned partnum);
+};
+
+
+static int frrzmq_read_msg(struct thread *t)
+{
+	struct frrzmq_cb *cb = THREAD_ARG(t);
+	zmq_msg_t msg;
+	unsigned partno;
+	int ret, more;
+	size_t moresz;
+
+	while (1) {
+		zmq_pollitem_t polli = {
+			.socket = cb->zmqsock,
+			.events = ZMQ_POLLIN
+		};
+		ret = zmq_poll(&polli, 1, 0);
+
+		if (ret < 0)
+			goto out_err;
+		if (!(polli.revents & ZMQ_POLLIN))
+			break;
+
+		if (cb->cb_msg) {
+			cb->cb_msg(cb->arg, cb->zmqsock);
+
+			if (cb->cancelled) {
+				XFREE(MTYPE_ZEROMQ_CB, cb);
+				return 0;
+			}
+			continue;
+		}
+
+		partno = 0;
+		if (zmq_msg_init(&msg))
+			goto out_err;
+		do {
+			ret = zmq_msg_recv(&msg, cb->zmqsock, ZMQ_NOBLOCK);
+			if (ret < 0) {
+				if (errno == EAGAIN)
+					break;
+
+				zmq_msg_close(&msg);
+				goto out_err;
+			}
+
+			cb->cb_part(cb->arg, cb->zmqsock, &msg, partno);
+			if (cb->cancelled) {
+				zmq_msg_close(&msg);
+				XFREE(MTYPE_ZEROMQ_CB, cb);
+				return 0;
+			}
+
+			/* cb_part may have read additional parts of the
+			 * message; don't use zmq_msg_more here */
+			moresz = sizeof(more);
+			more = 0;
+			ret = zmq_getsockopt(cb->zmqsock, ZMQ_RCVMORE,
+					     &more, &moresz);
+			if (ret < 0) {
+				zmq_msg_close(&msg);
+				goto out_err;
+			}
+
+			partno++;
+		} while (more);
+		zmq_msg_close(&msg);
+	}
+
+	funcname_thread_add_read_write(THREAD_READ, t->master, frrzmq_read_msg,
+			cb, cb->fd, &cb->thread, t->funcname, t->schedfrom,
+			t->schedfrom_line);
+	return 0;
+
+out_err:
+	zlog_err("ZeroMQ error: %s(%d)", strerror (errno), errno);
+	return 0;
+}
+
+struct frrzmq_cb *funcname_frrzmq_thread_add_read(
+		struct thread_master *master,
+		void (*msgfunc)(void *arg, void *zmqsock),
+		void (*partfunc)(void *arg, void *zmqsock,
+				 zmq_msg_t *msg, unsigned partnum),
+		void *arg, void *zmqsock, debugargdef)
+{
+	int fd, events;
+	size_t len;
+	struct frrzmq_cb *cb;
+
+	if (!(msgfunc || partfunc) || (msgfunc && partfunc))
+		return NULL;
+	len = sizeof(fd);
+	if (zmq_getsockopt(zmqsock, ZMQ_FD, &fd, &len))
+		return NULL;
+	len = sizeof(events);
+	if (zmq_getsockopt(zmqsock, ZMQ_EVENTS, &events, &len))
+		return NULL;
+
+	cb = XCALLOC(MTYPE_ZEROMQ_CB, sizeof(struct frrzmq_cb));
+	if (!cb)
+		return NULL;
+
+	cb->arg = arg;
+	cb->zmqsock = zmqsock;
+	cb->cb_msg = msgfunc;
+	cb->cb_part = partfunc;
+	cb->fd = fd;
+
+	if (events & ZMQ_POLLIN)
+		funcname_thread_add_event(master,
+				frrzmq_read_msg, cb, fd, &cb->thread,
+				funcname, schedfrom, fromln);
+	else
+		funcname_thread_add_read_write(THREAD_READ, master,
+				frrzmq_read_msg, cb, fd, &cb->thread,
+				funcname, schedfrom, fromln);
+	return cb;
+}
+
+void frrzmq_thread_cancel(struct frrzmq_cb *cb)
+{
+	if (!cb->thread) {
+		/* canceling from within callback */
+		cb->cancelled = 1;
+		return;
+	}
+	thread_cancel(cb->thread);
+	XFREE(MTYPE_ZEROMQ_CB, cb);
+}

lib/frr_zmq.h

@@ -0,0 +1,50 @@
+/*
+ * libzebra ZeroMQ bindings
+ * Copyright (C) 2015  David Lamparter
+ *
+ * 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; see the file COPYING; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+
+#ifndef _FRRZMQ_H
+#define _FRRZMQ_H
+
+#include "thread.h"
+#include <zmq.h>
+
+/* libzmq's context */
+extern void *frrzmq_context;
+
+extern void frrzmq_init (void);
+extern void frrzmq_finish (void);
+
+#define debugargdef const char *funcname, const char *schedfrom, int fromln
+
+#define frrzmq_thread_add_read_msg(m,f,a,z) funcname_frrzmq_thread_add_read( \
+				m,f,NULL,a,z,#f,__FILE__,__LINE__)
+#define frrzmq_thread_add_read_part(m,f,a,z) funcname_frrzmq_thread_add_read( \
+				m,NULL,f,a,z,#f,__FILE__,__LINE__)
+
+struct frrzmq_cb;
+
+extern struct frrzmq_cb *funcname_frrzmq_thread_add_read(
+		struct thread_master *master,
+		void (*msgfunc)(void *arg, void *zmqsock),
+		void (*partfunc)(void *arg, void *zmqsock,
+				 zmq_msg_t *msg, unsigned partnum),
+		void *arg, void *zmqsock, debugargdef);
+
+extern void frrzmq_thread_cancel(struct frrzmq_cb *cb);
+
+#endif /* _FRRZMQ_H */

lib/subdir.am

@@ -183,6 +183,21 @@ lib_libfrrsnmp_la_SOURCES = \
 	lib/snmp.c \
 	# end
 
+#
+# ZeroMQ support
+#
+if ZEROMQ
+lib_LTLIBRARIES += lib/libfrrzmq.la
+pkginclude_HEADERS += lib/frr_zmq.h
+endif
+
+lib_libfrrzmq_la_CFLAGS = $(WERROR) $(ZEROMQ_CFLAGS)
+lib_libfrrzmq_la_LDFLAGS = -version-info 0:0:0
+lib_libfrrzmq_la_LIBADD = lib/libfrr.la $(ZEROMQ_LIBS)
+lib_libfrrzmq_la_SOURCES = \
+	lib/frr_zmq.c \
+	#end
+
 #
 # CLI utilities
 #