From 3dc5bb627ffc01a0bf03901f893ae981b5815386 Mon Sep 17 00:00:00 2001 From: David Lamparter Date: Wed, 17 May 2017 18:27:54 +0200 Subject: [PATCH] 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 --- configure.ac | 17 +++++ lib/frr_zmq.c | 191 ++++++++++++++++++++++++++++++++++++++++++++++++++ lib/frr_zmq.h | 50 +++++++++++++ lib/subdir.am | 15 ++++ 4 files changed, 273 insertions(+) create mode 100644 lib/frr_zmq.c create mode 100644 lib/frr_zmq.h diff --git a/configure.ac b/configure.ac index bf66e00f9fbc..ec0419048558 100755 --- a/configure.ac +++ b/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 ---------- diff --git a/lib/frr_zmq.c b/lib/frr_zmq.c new file mode 100644 index 000000000000..861f7a5f0c2e --- /dev/null +++ b/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 +#include + +#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); +} diff --git a/lib/frr_zmq.h b/lib/frr_zmq.h new file mode 100644 index 000000000000..d37824522b79 --- /dev/null +++ b/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 + +/* 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 */ diff --git a/lib/subdir.am b/lib/subdir.am index d6349ba22df2..5f38cc7a14f1 100644 --- a/lib/subdir.am +++ b/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 #