[mirror_frr.git] / lib / frr_zmq.c

/*
 * 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
 */

/*
 * IF YOU MODIFY THIS FILE PLEASE RUN `make check` and ensure that
 * the test_zmq.c unit test is still working.  There are dependencies
 * between the two that are extremely fragile.  My understanding
 * is that there is specialized ownership of the cb pointer based
 * upon what is happening.  Those assumptions are supposed to be
 * tested in the test_zmq.c
 */
#include <zebra.h>
#include <zmq.h>

#include "thread.h"
#include "memory.h"
#include "frr_zmq.h"
#include "log.h"
#include "lib_errors.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;
	}
}

static int frrzmq_read_msg(struct thread *t)
{
	struct frrzmq_cb **cbp = THREAD_ARG(t);
	struct frrzmq_cb *cb;
	zmq_msg_t msg;
	unsigned partno;
	unsigned char read = 0;
	int ret, more;
	size_t moresz;

	if (!cbp)
		return 1;
	cb = (*cbp);
	if (!cb || !cb->zmqsock)
		return 1;

	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->read.cb_msg) {
			cb->in_cb = true;
			cb->read.cb_msg(cb->read.arg, cb->zmqsock);
			cb->in_cb = false;

			read = 1;

			if (cb->read.cancelled) {
				frrzmq_check_events(cbp, &cb->write,
						    ZMQ_POLLOUT);
				cb->read.thread = NULL;
				if (cb->write.cancelled && !cb->write.thread)
					XFREE(MTYPE_ZEROMQ_CB, *cbp);

				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;
			}
			read = 1;

			cb->in_cb = true;
			cb->read.cb_part(cb->read.arg, cb->zmqsock, &msg,
					 partno);
			cb->in_cb = false;

			if (cb->read.cancelled) {
				zmq_msg_close(&msg);
				frrzmq_check_events(cbp, &cb->write,
						    ZMQ_POLLOUT);
				cb->read.thread = NULL;
				if (cb->write.cancelled && !cb->write.thread)
					XFREE(MTYPE_ZEROMQ_CB, *cbp);

				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);
	}

	if (read)
		frrzmq_check_events(cbp, &cb->write, ZMQ_POLLOUT);

	thread_add_read(t->master, frrzmq_read_msg, cbp,
			cb->fd, &cb->read.thread);
	return 0;

out_err:
	flog_err(EC_LIB_ZMQ, "ZeroMQ read error: %s(%d)", strerror(errno),
		 errno);
	if (cb->read.cb_error)
		cb->read.cb_error(cb->read.arg, cb->zmqsock);
	return 1;
}

int _frrzmq_thread_add_read(const struct xref_threadsched *xref,
			    struct thread_master *master,
			    void (*msgfunc)(void *arg, void *zmqsock),
			    void (*partfunc)(void *arg, void *zmqsock,
					     zmq_msg_t *msg, unsigned partnum),
			    void (*errfunc)(void *arg, void *zmqsock),
			    void *arg, void *zmqsock,
			    struct frrzmq_cb **cbp)
{
	int fd, events;
	size_t len;
	struct frrzmq_cb *cb;

	if (!cbp)
		return -1;
	if (!(msgfunc || partfunc) || (msgfunc && partfunc))
		return -1;
	len = sizeof(fd);
	if (zmq_getsockopt(zmqsock, ZMQ_FD, &fd, &len))
		return -1;
	len = sizeof(events);
	if (zmq_getsockopt(zmqsock, ZMQ_EVENTS, &events, &len))
		return -1;

	if (*cbp)
		cb = *cbp;
	else {
		cb = XCALLOC(MTYPE_ZEROMQ_CB, sizeof(struct frrzmq_cb));
		cb->write.cancelled = true;
		*cbp = cb;
	}

	cb->zmqsock = zmqsock;
	cb->fd = fd;
	cb->read.arg = arg;
	cb->read.cb_msg = msgfunc;
	cb->read.cb_part = partfunc;
	cb->read.cb_error = errfunc;
	cb->read.cancelled = false;
	cb->in_cb = false;

	if (events & ZMQ_POLLIN) {
		thread_cancel(&cb->read.thread);

		thread_add_event(master, frrzmq_read_msg, cbp, fd,
				  &cb->read.thread);
	} else
		thread_add_read(master, frrzmq_read_msg, cbp, fd,
				&cb->read.thread);
	return 0;
}

static int frrzmq_write_msg(struct thread *t)
{
	struct frrzmq_cb **cbp = THREAD_ARG(t);
	struct frrzmq_cb *cb;
	unsigned char written = 0;
	int ret;

	if (!cbp)
		return 1;
	cb = (*cbp);
	if (!cb || !cb->zmqsock)
		return 1;

	while (1) {
		zmq_pollitem_t polli = {.socket = cb->zmqsock,
					.events = ZMQ_POLLOUT};
		ret = zmq_poll(&polli, 1, 0);

		if (ret < 0)
			goto out_err;

		if (!(polli.revents & ZMQ_POLLOUT))
			break;

		if (cb->write.cb_msg) {
			cb->in_cb = true;
			cb->write.cb_msg(cb->write.arg, cb->zmqsock);
			cb->in_cb = false;

			written = 1;

			if (cb->write.cancelled) {
				frrzmq_check_events(cbp, &cb->read, ZMQ_POLLIN);
				cb->write.thread = NULL;
				if (cb->read.cancelled && !cb->read.thread)
					XFREE(MTYPE_ZEROMQ_CB, *cbp);

				return 0;
			}
			continue;
		}
	}

	if (written)
		frrzmq_check_events(cbp, &cb->read, ZMQ_POLLIN);

	thread_add_write(t->master, frrzmq_write_msg, cbp,
			 cb->fd, &cb->write.thread);
	return 0;

out_err:
	flog_err(EC_LIB_ZMQ, "ZeroMQ write error: %s(%d)", strerror(errno),
		 errno);
	if (cb->write.cb_error)
		cb->write.cb_error(cb->write.arg, cb->zmqsock);
	return 1;
}

int _frrzmq_thread_add_write(const struct xref_threadsched *xref,
			     struct thread_master *master,
			     void (*msgfunc)(void *arg, void *zmqsock),
			     void (*errfunc)(void *arg, void *zmqsock),
			     void *arg, void *zmqsock, struct frrzmq_cb **cbp)
{
	int fd, events;
	size_t len;
	struct frrzmq_cb *cb;

	if (!cbp)
		return -1;
	if (!msgfunc)
		return -1;
	len = sizeof(fd);
	if (zmq_getsockopt(zmqsock, ZMQ_FD, &fd, &len))
		return -1;
	len = sizeof(events);
	if (zmq_getsockopt(zmqsock, ZMQ_EVENTS, &events, &len))
		return -1;

	if (*cbp)
		cb = *cbp;
	else {
		cb = XCALLOC(MTYPE_ZEROMQ_CB, sizeof(struct frrzmq_cb));
		cb->read.cancelled = true;
		*cbp = cb;
	}

	cb->zmqsock = zmqsock;
	cb->fd = fd;
	cb->write.arg = arg;
	cb->write.cb_msg = msgfunc;
	cb->write.cb_part = NULL;
	cb->write.cb_error = errfunc;
	cb->write.cancelled = false;
	cb->in_cb = false;

	if (events & ZMQ_POLLOUT) {
		thread_cancel(&cb->write.thread);

		_thread_add_event(xref, master, frrzmq_write_msg, cbp, fd,
				  &cb->write.thread);
	} else
		thread_add_write(master, frrzmq_write_msg, cbp, fd,
				 &cb->write.thread);
	return 0;
}

void frrzmq_thread_cancel(struct frrzmq_cb **cb, struct cb_core *core)
{
	if (!cb || !*cb)
		return;
	core->cancelled = true;
	thread_cancel(&core->thread);

	/* If cancelled from within a callback, don't try to free memory
	 * in this path.
	 */
	if ((*cb)->in_cb)
		return;

	/* Ok to free the callback context if no more ... context. */
	if ((*cb)->read.cancelled && !(*cb)->read.thread
	    && (*cb)->write.cancelled && ((*cb)->write.thread == NULL))
		XFREE(MTYPE_ZEROMQ_CB, *cb);
}

void frrzmq_check_events(struct frrzmq_cb **cbp, struct cb_core *core,
			 int event)
{
	struct frrzmq_cb *cb;
	int events;
	size_t len;

	if (!cbp)
		return;
	cb = (*cbp);
	if (!cb || !cb->zmqsock)
		return;

	len = sizeof(events);
	if (zmq_getsockopt(cb->zmqsock, ZMQ_EVENTS, &events, &len))
		return;
	if ((events & event) && core->thread && !core->cancelled) {
		struct thread_master *tm = core->thread->master;

		thread_cancel(&core->thread);

		if (event == ZMQ_POLLIN)
			thread_add_event(tm, frrzmq_read_msg,
					 cbp, cb->fd, &core->thread);
		else
			thread_add_event(tm, frrzmq_write_msg,
					 cbp, cb->fd, &core->thread);
	}
}
Commit	Line	Data
b6116506 DL	1	/*
	2	* libzebra ZeroMQ bindings
	3	* Copyright (C) 2015 David Lamparter
	4	*
	5	* This program is free software; you can redistribute it and/or modify it
	6	* under the terms of the GNU General Public License as published by the Free
	7	* Software Foundation; either version 2 of the License, or (at your option)
	8	* any later version.
	9	*
	10	* This program is distributed in the hope that it will be useful, but WITHOUT
	11	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
	12	* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
	13	* more details.
	14	*
	15	* You should have received a copy of the GNU General Public License along
	16	* with this program; see the file COPYING; if not, write to the Free Software
	17	* Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
	18	*/
	19
fb1df4cd DS	20	/*
fb1df4cd DS	21	* IF YOU MODIFY THIS FILE PLEASE RUN `make check` and ensure that
214d8a60	22	* the test_zmq.c unit test is still working. There are dependencies
fb1df4cd DS	23	* between the two that are extremely fragile. My understanding
	24	* is that there is specialized ownership of the cb pointer based
	25	* upon what is happening. Those assumptions are supposed to be
	26	* tested in the test_zmq.c
	27	*/
b6116506 DL	28	#include <zebra.h>
	29	#include <zmq.h>
	30
	31	#include "thread.h"
	32	#include "memory.h"
	33	#include "frr_zmq.h"
	34	#include "log.h"
35774357	35	#include "lib_errors.h"
b6116506	36
bf8d3d6a	37	DEFINE_MTYPE_STATIC(LIB, ZEROMQ_CB, "ZeroMQ callback");
b6116506 DL	38
	39	/* libzmq's context */
	40	void *frrzmq_context = NULL;
	41	static unsigned frrzmq_initcount = 0;
	42
	43	void frrzmq_init(void)
	44	{
	45	if (frrzmq_initcount++ == 0) {
	46	frrzmq_context = zmq_ctx_new();
	47	zmq_ctx_set(frrzmq_context, ZMQ_IPV6, 1);
	48	}
	49	}
	50
	51	void frrzmq_finish(void)
	52	{
	53	if (--frrzmq_initcount == 0) {
	54	zmq_ctx_term(frrzmq_context);
	55	frrzmq_context = NULL;
	56	}
	57	}
	58
b6116506 DL	59	static int frrzmq_read_msg(struct thread *t)
b6116506 DL	60	{
afd0f10d	61	struct frrzmq_cb **cbp = THREAD_ARG(t);
afd0f10d	62	struct frrzmq_cb *cb;
b6116506 DL	63	zmq_msg_t msg;
b6116506 DL	64	unsigned partno;
afd0f10d	65	unsigned char read = 0;
b6116506 DL	66	int ret, more;
	67	size_t moresz;
	68
afd0f10d	69	if (!cbp)
	70	return 1;
	71	cb = (*cbp);
	72	if (!cb \|\| !cb->zmqsock)
	73	return 1;
	74
b6116506	75	while (1) {
afd0f10d	76	zmq_pollitem_t polli = {.socket = cb->zmqsock,
afd0f10d	77	.events = ZMQ_POLLIN};
b6116506 DL	78	ret = zmq_poll(&polli, 1, 0);
	79
	80	if (ret < 0)
	81	goto out_err;
afd0f10d	82
b6116506 DL	83	if (!(polli.revents & ZMQ_POLLIN))
	84	break;
	85
afd0f10d	86	if (cb->read.cb_msg) {
8fd5502b	87	cb->in_cb = true;
afd0f10d	88	cb->read.cb_msg(cb->read.arg, cb->zmqsock);
8fd5502b MS	89	cb->in_cb = false;
8fd5502b MS	90
afd0f10d	91	read = 1;
b6116506	92
afd0f10d	93	if (cb->read.cancelled) {
	94	frrzmq_check_events(cbp, &cb->write,
	95	ZMQ_POLLOUT);
	96	cb->read.thread = NULL;
	97	if (cb->write.cancelled && !cb->write.thread)
cf2182c0 MS	98	XFREE(MTYPE_ZEROMQ_CB, *cbp);
cf2182c0 MS	99
b6116506 DL	100	return 0;
	101	}
	102	continue;
	103	}
	104
	105	partno = 0;
	106	if (zmq_msg_init(&msg))
	107	goto out_err;
	108	do {
	109	ret = zmq_msg_recv(&msg, cb->zmqsock, ZMQ_NOBLOCK);
	110	if (ret < 0) {
	111	if (errno == EAGAIN)
	112	break;
	113
	114	zmq_msg_close(&msg);
	115	goto out_err;
	116	}
afd0f10d	117	read = 1;
b6116506	118
8fd5502b	119	cb->in_cb = true;
afd0f10d	120	cb->read.cb_part(cb->read.arg, cb->zmqsock, &msg,
afd0f10d	121	partno);
8fd5502b MS	122	cb->in_cb = false;
8fd5502b MS	123
afd0f10d	124	if (cb->read.cancelled) {
b6116506	125	zmq_msg_close(&msg);
afd0f10d	126	frrzmq_check_events(cbp, &cb->write,
	127	ZMQ_POLLOUT);
	128	cb->read.thread = NULL;
	129	if (cb->write.cancelled && !cb->write.thread)
cf2182c0 MS	130	XFREE(MTYPE_ZEROMQ_CB, *cbp);
cf2182c0 MS	131
b6116506 DL	132	return 0;
	133	}
	134
	135	/* cb_part may have read additional parts of the
	136	* message; don't use zmq_msg_more here */
	137	moresz = sizeof(more);
	138	more = 0;
afd0f10d	139	ret = zmq_getsockopt(cb->zmqsock, ZMQ_RCVMORE, &more,
afd0f10d	140	&moresz);
b6116506 DL	141	if (ret < 0) {
	142	zmq_msg_close(&msg);
	143	goto out_err;
	144	}
	145
	146	partno++;
	147	} while (more);
	148	zmq_msg_close(&msg);
	149	}
	150
afd0f10d	151	if (read)
	152	frrzmq_check_events(cbp, &cb->write, ZMQ_POLLOUT);
	153
1dffe357 MS	154	thread_add_read(t->master, frrzmq_read_msg, cbp,
1dffe357 MS	155	cb->fd, &cb->read.thread);
b6116506 DL	156	return 0;
	157
	158	out_err:
450971aa	159	flog_err(EC_LIB_ZMQ, "ZeroMQ read error: %s(%d)", strerror(errno),
1c50c1c0	160	errno);
afd0f10d	161	if (cb->read.cb_error)
	162	cb->read.cb_error(cb->read.arg, cb->zmqsock);
	163	return 1;
b6116506 DL	164	}
b6116506 DL	165
60a3efec DL	166	int _frrzmq_thread_add_read(const struct xref_threadsched *xref,
	167	struct thread_master *master,
	168	void (msgfunc)(void arg, void *zmqsock),
	169	void (partfunc)(void arg, void *zmqsock,
	170	zmq_msg_t *msg, unsigned partnum),
	171	void (errfunc)(void arg, void *zmqsock),
	172	void arg, void zmqsock,
	173	struct frrzmq_cb **cbp)
b6116506 DL	174	{
	175	int fd, events;
	176	size_t len;
	177	struct frrzmq_cb *cb;
	178
afd0f10d	179	if (!cbp)
afd0f10d	180	return -1;
b6116506	181	if (!(msgfunc \|\| partfunc) \|\| (msgfunc && partfunc))
afd0f10d	182	return -1;
	183	len = sizeof(fd);
	184	if (zmq_getsockopt(zmqsock, ZMQ_FD, &fd, &len))
	185	return -1;
	186	len = sizeof(events);
	187	if (zmq_getsockopt(zmqsock, ZMQ_EVENTS, &events, &len))
	188	return -1;
	189
	190	if (*cbp)
	191	cb = *cbp;
	192	else {
	193	cb = XCALLOC(MTYPE_ZEROMQ_CB, sizeof(struct frrzmq_cb));
08c2d52a	194	cb->write.cancelled = true;
afd0f10d	195	*cbp = cb;
	196	}
	197
	198	cb->zmqsock = zmqsock;
	199	cb->fd = fd;
	200	cb->read.arg = arg;
	201	cb->read.cb_msg = msgfunc;
	202	cb->read.cb_part = partfunc;
	203	cb->read.cb_error = errfunc;
08c2d52a	204	cb->read.cancelled = false;
8fd5502b	205	cb->in_cb = false;
afd0f10d	206
afd0f10d	207	if (events & ZMQ_POLLIN) {
b3d6bc6e MS	208	thread_cancel(&cb->read.thread);
b3d6bc6e MS	209
1dffe357	210	thread_add_event(master, frrzmq_read_msg, cbp, fd,
60a3efec	211	&cb->read.thread);
afd0f10d	212	} else
1dffe357 MS	213	thread_add_read(master, frrzmq_read_msg, cbp, fd,
1dffe357 MS	214	&cb->read.thread);
afd0f10d	215	return 0;
	216	}
	217
	218	static int frrzmq_write_msg(struct thread *t)
	219	{
	220	struct frrzmq_cb **cbp = THREAD_ARG(t);
	221	struct frrzmq_cb *cb;
	222	unsigned char written = 0;
	223	int ret;
	224
	225	if (!cbp)
	226	return 1;
	227	cb = (*cbp);
	228	if (!cb \|\| !cb->zmqsock)
	229	return 1;
	230
	231	while (1) {
	232	zmq_pollitem_t polli = {.socket = cb->zmqsock,
	233	.events = ZMQ_POLLOUT};
	234	ret = zmq_poll(&polli, 1, 0);
	235
	236	if (ret < 0)
	237	goto out_err;
	238
	239	if (!(polli.revents & ZMQ_POLLOUT))
	240	break;
	241
	242	if (cb->write.cb_msg) {
8fd5502b	243	cb->in_cb = true;
afd0f10d	244	cb->write.cb_msg(cb->write.arg, cb->zmqsock);
8fd5502b MS	245	cb->in_cb = false;
8fd5502b MS	246
afd0f10d	247	written = 1;
	248
	249	if (cb->write.cancelled) {
	250	frrzmq_check_events(cbp, &cb->read, ZMQ_POLLIN);
	251	cb->write.thread = NULL;
	252	if (cb->read.cancelled && !cb->read.thread)
cf2182c0 MS	253	XFREE(MTYPE_ZEROMQ_CB, *cbp);
cf2182c0 MS	254
afd0f10d	255	return 0;
	256	}
	257	continue;
	258	}
	259	}
	260
	261	if (written)
	262	frrzmq_check_events(cbp, &cb->read, ZMQ_POLLIN);
	263
1dffe357 MS	264	thread_add_write(t->master, frrzmq_write_msg, cbp,
1dffe357 MS	265	cb->fd, &cb->write.thread);
afd0f10d	266	return 0;
	267
	268	out_err:
450971aa	269	flog_err(EC_LIB_ZMQ, "ZeroMQ write error: %s(%d)", strerror(errno),
1c50c1c0	270	errno);
afd0f10d	271	if (cb->write.cb_error)
	272	cb->write.cb_error(cb->write.arg, cb->zmqsock);
	273	return 1;
	274	}
60a3efec DL	275
	276	int _frrzmq_thread_add_write(const struct xref_threadsched *xref,
	277	struct thread_master *master,
	278	void (msgfunc)(void arg, void *zmqsock),
	279	void (errfunc)(void arg, void *zmqsock),
	280	void arg, void zmqsock, struct frrzmq_cb **cbp)
afd0f10d	281	{
	282	int fd, events;
	283	size_t len;
	284	struct frrzmq_cb *cb;
	285
	286	if (!cbp)
	287	return -1;
	288	if (!msgfunc)
	289	return -1;
b6116506 DL	290	len = sizeof(fd);
b6116506 DL	291	if (zmq_getsockopt(zmqsock, ZMQ_FD, &fd, &len))
afd0f10d	292	return -1;
b6116506 DL	293	len = sizeof(events);
b6116506 DL	294	if (zmq_getsockopt(zmqsock, ZMQ_EVENTS, &events, &len))
afd0f10d	295	return -1;
b6116506	296
afd0f10d	297	if (*cbp)
	298	cb = *cbp;
	299	else {
	300	cb = XCALLOC(MTYPE_ZEROMQ_CB, sizeof(struct frrzmq_cb));
08c2d52a	301	cb->read.cancelled = true;
afd0f10d	302	*cbp = cb;
afd0f10d	303	}
b6116506	304
b6116506	305	cb->zmqsock = zmqsock;
b6116506	306	cb->fd = fd;
afd0f10d	307	cb->write.arg = arg;
	308	cb->write.cb_msg = msgfunc;
	309	cb->write.cb_part = NULL;
	310	cb->write.cb_error = errfunc;
08c2d52a	311	cb->write.cancelled = false;
8fd5502b	312	cb->in_cb = false;
afd0f10d	313
afd0f10d	314	if (events & ZMQ_POLLOUT) {
b3d6bc6e MS	315	thread_cancel(&cb->write.thread);
b3d6bc6e MS	316
60a3efec DL	317	_thread_add_event(xref, master, frrzmq_write_msg, cbp, fd,
60a3efec DL	318	&cb->write.thread);
afd0f10d	319	} else
1dffe357 MS	320	thread_add_write(master, frrzmq_write_msg, cbp, fd,
1dffe357 MS	321	&cb->write.thread);
afd0f10d	322	return 0;
afd0f10d	323	}
b6116506	324
afd0f10d	325	void frrzmq_thread_cancel(struct frrzmq_cb *cb, struct cb_core core)
	326	{
	327	if (!cb \|\| !*cb)
	328	return;
08c2d52a	329	core->cancelled = true;
b3d6bc6e MS	330	thread_cancel(&core->thread);
b3d6bc6e MS	331
8fd5502b MS	332	/* If cancelled from within a callback, don't try to free memory
8fd5502b MS	333	* in this path.
fb1df4cd	334	*/
8fd5502b MS	335	if ((*cb)->in_cb)
	336	return;
	337
	338	/* Ok to free the callback context if no more ... context. */
afd0f10d	339	if ((cb)->read.cancelled && !(cb)->read.thread
8fd5502b	340	&& (cb)->write.cancelled && ((cb)->write.thread == NULL))
afd0f10d	341	XFREE(MTYPE_ZEROMQ_CB, *cb);
b6116506 DL	342	}
b6116506 DL	343
afd0f10d	344	void frrzmq_check_events(struct frrzmq_cb *cbp, struct cb_core core,
afd0f10d	345	int event)
b6116506	346	{
afd0f10d	347	struct frrzmq_cb *cb;
	348	int events;
	349	size_t len;
	350
	351	if (!cbp)
	352	return;
	353	cb = (*cbp);
	354	if (!cb \|\| !cb->zmqsock)
	355	return;
	356
81b8afcf	357	len = sizeof(events);
afd0f10d	358	if (zmq_getsockopt(cb->zmqsock, ZMQ_EVENTS, &events, &len))
b6116506	359	return;
e08165de	360	if ((events & event) && core->thread && !core->cancelled) {
afd0f10d	361	struct thread_master *tm = core->thread->master;
e08165de	362
b3d6bc6e MS	363	thread_cancel(&core->thread);
b3d6bc6e MS	364
e08165de MS	365	if (event == ZMQ_POLLIN)
	366	thread_add_event(tm, frrzmq_read_msg,
	367	cbp, cb->fd, &core->thread);
	368	else
	369	thread_add_event(tm, frrzmq_write_msg,
	370	cbp, cb->fd, &core->thread);
b6116506	371	}
b6116506	372	}