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

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

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->read.cb_msg(cb->read.arg, cb->zmqsock);
			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, 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;
			}
			read = 1;

			cb->read.cb_part(cb->read.arg, cb->zmqsock, &msg,
					 partno);
			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, 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);
	}

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

	funcname_thread_add_read_write(
		THREAD_READ, t->master, frrzmq_read_msg, cbp, cb->fd,
		&cb->read.thread, t->funcname, t->schedfrom, t->schedfrom_line);
	return 0;

out_err:
	zlog_err("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 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 (*errfunc)(void *arg, void *zmqsock),
				    void *arg, void *zmqsock,
				    struct frrzmq_cb **cbp, debugargdef)
{
	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 = 1;
		if (!cb)
			return -1;
		*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 = 0;

	if (events & ZMQ_POLLIN) {
		if (cb->read.thread) {
			thread_cancel(cb->read.thread);
			cb->read.thread = NULL;
		}
		funcname_thread_add_event(master, frrzmq_read_msg, cbp, fd,
					  &cb->read.thread, funcname, schedfrom,
					  fromln);
	} else
		funcname_thread_add_read_write(
			THREAD_READ, master, frrzmq_read_msg, cbp, fd,
			&cb->read.thread, funcname, schedfrom, fromln);
	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->write.cb_msg(cb->write.arg, cb->zmqsock);
			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, cb);
				return 0;
			}
			continue;
		}
	}

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

	funcname_thread_add_read_write(THREAD_WRITE, t->master,
				       frrzmq_write_msg, cbp, cb->fd,
				       &cb->write.thread, t->funcname,
				       t->schedfrom, t->schedfrom_line);
	return 0;

out_err:
	zlog_err("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 funcname_frrzmq_thread_add_write(struct thread_master *master,
				     void (*msgfunc)(void *arg, void *zmqsock),
				     void (*errfunc)(void *arg, void *zmqsock),
				     void *arg, void *zmqsock,
				     struct frrzmq_cb **cbp, debugargdef)
{
	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 = 1;
		if (!cb)
			return -1;
		*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 = 0;

	if (events & ZMQ_POLLOUT) {
		if (cb->write.thread) {
			thread_cancel(cb->write.thread);
			cb->write.thread = NULL;
		}
		funcname_thread_add_event(master, frrzmq_write_msg, cbp, fd,
					  &cb->write.thread, funcname,
					  schedfrom, fromln);
	} else
		funcname_thread_add_read_write(
			THREAD_WRITE, master, frrzmq_write_msg, cbp, fd,
			&cb->write.thread, funcname, schedfrom, fromln);
	return 0;
}

void frrzmq_thread_cancel(struct frrzmq_cb **cb, struct cb_core *core)
{
	if (!cb || !*cb)
		return;
	core->cancelled = 1;
	if (core->thread) {
		thread_cancel(core->thread);
		core->thread = NULL;
	}
	if ((*cb)->read.cancelled && !(*cb)->read.thread
	    && (*cb)->write.cancelled && (*cb)->write.thread)
		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);
		core->thread = NULL;
		thread_add_event(tm, (event == ZMQ_POLLIN ? frrzmq_read_msg
							  : 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
	20	#include <zebra.h>
	21	#include <zmq.h>
	22
	23	#include "thread.h"
	24	#include "memory.h"
	25	#include "frr_zmq.h"
	26	#include "log.h"
	27
	28	DEFINE_MTYPE_STATIC(LIB, ZEROMQ_CB, "ZeroMQ callback")
	29
	30	/* libzmq's context */
	31	void *frrzmq_context = NULL;
	32	static unsigned frrzmq_initcount = 0;
	33
	34	void frrzmq_init(void)
	35	{
	36	if (frrzmq_initcount++ == 0) {
	37	frrzmq_context = zmq_ctx_new();
	38	zmq_ctx_set(frrzmq_context, ZMQ_IPV6, 1);
	39	}
	40	}
	41
	42	void frrzmq_finish(void)
	43	{
	44	if (--frrzmq_initcount == 0) {
	45	zmq_ctx_term(frrzmq_context);
	46	frrzmq_context = NULL;
	47	}
	48	}
	49
b6116506 DL	50	static int frrzmq_read_msg(struct thread *t)
b6116506 DL	51	{
afd0f10d	52	struct frrzmq_cb **cbp = THREAD_ARG(t);
afd0f10d	53	struct frrzmq_cb *cb;
b6116506 DL	54	zmq_msg_t msg;
b6116506 DL	55	unsigned partno;
afd0f10d	56	unsigned char read = 0;
b6116506 DL	57	int ret, more;
	58	size_t moresz;
	59
afd0f10d	60	if (!cbp)
	61	return 1;
	62	cb = (*cbp);
	63	if (!cb \|\| !cb->zmqsock)
	64	return 1;
	65
b6116506	66	while (1) {
afd0f10d	67	zmq_pollitem_t polli = {.socket = cb->zmqsock,
afd0f10d	68	.events = ZMQ_POLLIN};
b6116506 DL	69	ret = zmq_poll(&polli, 1, 0);
	70
	71	if (ret < 0)
	72	goto out_err;
afd0f10d	73
b6116506 DL	74	if (!(polli.revents & ZMQ_POLLIN))
	75	break;
	76
afd0f10d	77	if (cb->read.cb_msg) {
	78	cb->read.cb_msg(cb->read.arg, cb->zmqsock);
	79	read = 1;
b6116506	80
afd0f10d	81	if (cb->read.cancelled) {
	82	frrzmq_check_events(cbp, &cb->write,
	83	ZMQ_POLLOUT);
	84	cb->read.thread = NULL;
	85	if (cb->write.cancelled && !cb->write.thread)
	86	XFREE(MTYPE_ZEROMQ_CB, cb);
b6116506 DL	87	return 0;
	88	}
	89	continue;
	90	}
	91
	92	partno = 0;
	93	if (zmq_msg_init(&msg))
	94	goto out_err;
	95	do {
	96	ret = zmq_msg_recv(&msg, cb->zmqsock, ZMQ_NOBLOCK);
	97	if (ret < 0) {
	98	if (errno == EAGAIN)
	99	break;
	100
	101	zmq_msg_close(&msg);
	102	goto out_err;
	103	}
afd0f10d	104	read = 1;
b6116506	105
afd0f10d	106	cb->read.cb_part(cb->read.arg, cb->zmqsock, &msg,
	107	partno);
	108	if (cb->read.cancelled) {
b6116506	109	zmq_msg_close(&msg);
afd0f10d	110	frrzmq_check_events(cbp, &cb->write,
	111	ZMQ_POLLOUT);
	112	cb->read.thread = NULL;
	113	if (cb->write.cancelled && !cb->write.thread)
	114	XFREE(MTYPE_ZEROMQ_CB, cb);
b6116506 DL	115	return 0;
	116	}
	117
	118	/* cb_part may have read additional parts of the
	119	* message; don't use zmq_msg_more here */
	120	moresz = sizeof(more);
	121	more = 0;
afd0f10d	122	ret = zmq_getsockopt(cb->zmqsock, ZMQ_RCVMORE, &more,
afd0f10d	123	&moresz);
b6116506 DL	124	if (ret < 0) {
	125	zmq_msg_close(&msg);
	126	goto out_err;
	127	}
	128
	129	partno++;
	130	} while (more);
	131	zmq_msg_close(&msg);
	132	}
	133
afd0f10d	134	if (read)
	135	frrzmq_check_events(cbp, &cb->write, ZMQ_POLLOUT);
	136
	137	funcname_thread_add_read_write(
	138	THREAD_READ, t->master, frrzmq_read_msg, cbp, cb->fd,
	139	&cb->read.thread, t->funcname, t->schedfrom, t->schedfrom_line);
b6116506 DL	140	return 0;
	141
	142	out_err:
afd0f10d	143	zlog_err("ZeroMQ read error: %s(%d)", strerror(errno), errno);
	144	if (cb->read.cb_error)
	145	cb->read.cb_error(cb->read.arg, cb->zmqsock);
	146	return 1;
b6116506 DL	147	}
b6116506 DL	148
afd0f10d	149	int funcname_frrzmq_thread_add_read(struct thread_master *master,
	150	void (msgfunc)(void arg, void *zmqsock),
	151	void (partfunc)(void arg, void *zmqsock,
	152	zmq_msg_t *msg,
	153	unsigned partnum),
	154	void (errfunc)(void arg, void *zmqsock),
	155	void arg, void zmqsock,
	156	struct frrzmq_cb **cbp, debugargdef)
b6116506 DL	157	{
	158	int fd, events;
	159	size_t len;
	160	struct frrzmq_cb *cb;
	161
afd0f10d	162	if (!cbp)
afd0f10d	163	return -1;
b6116506	164	if (!(msgfunc \|\| partfunc) \|\| (msgfunc && partfunc))
afd0f10d	165	return -1;
	166	len = sizeof(fd);
	167	if (zmq_getsockopt(zmqsock, ZMQ_FD, &fd, &len))
	168	return -1;
	169	len = sizeof(events);
	170	if (zmq_getsockopt(zmqsock, ZMQ_EVENTS, &events, &len))
	171	return -1;
	172
	173	if (*cbp)
	174	cb = *cbp;
	175	else {
	176	cb = XCALLOC(MTYPE_ZEROMQ_CB, sizeof(struct frrzmq_cb));
	177	cb->write.cancelled = 1;
	178	if (!cb)
	179	return -1;
	180	*cbp = cb;
	181	}
	182
	183	cb->zmqsock = zmqsock;
	184	cb->fd = fd;
	185	cb->read.arg = arg;
	186	cb->read.cb_msg = msgfunc;
	187	cb->read.cb_part = partfunc;
	188	cb->read.cb_error = errfunc;
	189	cb->read.cancelled = 0;
	190
	191	if (events & ZMQ_POLLIN) {
	192	if (cb->read.thread) {
	193	thread_cancel(cb->read.thread);
	194	cb->read.thread = NULL;
	195	}
	196	funcname_thread_add_event(master, frrzmq_read_msg, cbp, fd,
	197	&cb->read.thread, funcname, schedfrom,
	198	fromln);
	199	} else
	200	funcname_thread_add_read_write(
	201	THREAD_READ, master, frrzmq_read_msg, cbp, fd,
	202	&cb->read.thread, funcname, schedfrom, fromln);
	203	return 0;
	204	}
	205
	206	static int frrzmq_write_msg(struct thread *t)
	207	{
	208	struct frrzmq_cb **cbp = THREAD_ARG(t);
	209	struct frrzmq_cb *cb;
	210	unsigned char written = 0;
	211	int ret;
	212
	213	if (!cbp)
	214	return 1;
	215	cb = (*cbp);
	216	if (!cb \|\| !cb->zmqsock)
	217	return 1;
	218
	219	while (1) {
	220	zmq_pollitem_t polli = {.socket = cb->zmqsock,
	221	.events = ZMQ_POLLOUT};
	222	ret = zmq_poll(&polli, 1, 0);
	223
	224	if (ret < 0)
	225	goto out_err;
	226
	227	if (!(polli.revents & ZMQ_POLLOUT))
	228	break;
229
230	if (cb->write.cb_msg) {
231	cb->write.cb_msg(cb->write.arg, cb->zmqsock);
232	written = 1;
233
234	if (cb->write.cancelled) {
235	frrzmq_check_events(cbp, &cb->read, ZMQ_POLLIN);
236	cb->write.thread = NULL;
237	if (cb->read.cancelled && !cb->read.thread)
238	XFREE(MTYPE_ZEROMQ_CB, cb);
239	return 0;
240	}
241	continue;
242	}
243	}
244
245	if (written)
246	frrzmq_check_events(cbp, &cb->read, ZMQ_POLLIN);
247
248	funcname_thread_add_read_write(THREAD_WRITE, t->master,
249	frrzmq_write_msg, cbp, cb->fd,
250	&cb->write.thread, t->funcname,
251	t->schedfrom, t->schedfrom_line);
252	return 0;
253
254	out_err:
255	zlog_err("ZeroMQ write error: %s(%d)", strerror(errno), errno);
256	if (cb->write.cb_error)
257	cb->write.cb_error(cb->write.arg, cb->zmqsock);
258	return 1;
259	}
260	int funcname_frrzmq_thread_add_write(struct thread_master *master,
261	void (msgfunc)(void arg, void *zmqsock),
262	void (errfunc)(void arg, void *zmqsock),
263	void arg, void zmqsock,
264	struct frrzmq_cb **cbp, debugargdef)
265	{
266	int fd, events;
267	size_t len;
268	struct frrzmq_cb *cb;
269
270	if (!cbp)
271	return -1;
272	if (!msgfunc)
273	return -1;
b6116506 DL	274	len = sizeof(fd);
b6116506 DL	275	if (zmq_getsockopt(zmqsock, ZMQ_FD, &fd, &len))
afd0f10d	276	return -1;
b6116506 DL	277	len = sizeof(events);
b6116506 DL	278	if (zmq_getsockopt(zmqsock, ZMQ_EVENTS, &events, &len))
afd0f10d	279	return -1;
b6116506	280
afd0f10d	281	if (*cbp)
	282	cb = *cbp;
	283	else {
	284	cb = XCALLOC(MTYPE_ZEROMQ_CB, sizeof(struct frrzmq_cb));
	285	cb->read.cancelled = 1;
	286	if (!cb)
	287	return -1;
	288	*cbp = cb;
	289	}
b6116506	290
b6116506	291	cb->zmqsock = zmqsock;
b6116506	292	cb->fd = fd;
afd0f10d	293	cb->write.arg = arg;
	294	cb->write.cb_msg = msgfunc;
	295	cb->write.cb_part = NULL;
	296	cb->write.cb_error = errfunc;
	297	cb->write.cancelled = 0;
	298
	299	if (events & ZMQ_POLLOUT) {
	300	if (cb->write.thread) {
	301	thread_cancel(cb->write.thread);
	302	cb->write.thread = NULL;
	303	}
	304	funcname_thread_add_event(master, frrzmq_write_msg, cbp, fd,
	305	&cb->write.thread, funcname,
	306	schedfrom, fromln);
	307	} else
	308	funcname_thread_add_read_write(
	309	THREAD_WRITE, master, frrzmq_write_msg, cbp, fd,
	310	&cb->write.thread, funcname, schedfrom, fromln);
	311	return 0;
	312	}
b6116506	313
afd0f10d	314	void frrzmq_thread_cancel(struct frrzmq_cb *cb, struct cb_core core)
	315	{
	316	if (!cb \|\| !*cb)
	317	return;
	318	core->cancelled = 1;
	319	if (core->thread) {
	320	thread_cancel(core->thread);
	321	core->thread = NULL;
	322	}
	323	if ((cb)->read.cancelled && !(cb)->read.thread
	324	&& (cb)->write.cancelled && (cb)->write.thread)
	325	XFREE(MTYPE_ZEROMQ_CB, *cb);
b6116506 DL	326	}
b6116506 DL	327
afd0f10d	328	void frrzmq_check_events(struct frrzmq_cb *cbp, struct cb_core core,
afd0f10d	329	int event)
b6116506	330	{
afd0f10d	331	struct frrzmq_cb *cb;
	332	int events;
	333	size_t len;
	334
	335	if (!cbp)
	336	return;
	337	cb = (*cbp);
	338	if (!cb \|\| !cb->zmqsock)
	339	return;
	340
81b8afcf	341	len = sizeof(events);
afd0f10d	342	if (zmq_getsockopt(cb->zmqsock, ZMQ_EVENTS, &events, &len))
b6116506	343	return;
afd0f10d	344	if (events & event && core->thread && !core->cancelled) {
	345	struct thread_master *tm = core->thread->master;
	346	thread_cancel(core->thread);
	347	core->thread = NULL;
	348	thread_add_event(tm, (event == ZMQ_POLLIN ? frrzmq_read_msg
	349	: frrzmq_write_msg),
	350	cbp, cb->fd, &core->thread);
b6116506	351	}
b6116506	352	}