[mirror_frr.git] / pimd / pim_zpthread.c

// SPDX-License-Identifier: GPL-2.0-or-later
/*
 * PIM for Quagga
 * Copyright (C) 2008  Everton da Silva Marques
 */

#include <zebra.h>
#include <lib/log.h>
#include <lib/lib_errors.h>

#include "pimd.h"
#include "pim_instance.h"
#include "pim_mlag.h"
#include "pim_zebra.h"

extern struct zclient *zclient;

#define PIM_MLAG_POST_LIMIT 100

int32_t mlag_bulk_cnt;

static void pim_mlag_zebra_fill_header(enum mlag_msg_type msg_type)
{
	uint32_t fill_msg_type = msg_type;
	uint16_t data_len = 0;
	uint16_t msg_cnt = 1;

	switch (msg_type) {
	case MLAG_REGISTER:
	case MLAG_DEREGISTER:
		data_len = sizeof(struct mlag_msg);
		break;
	case MLAG_MROUTE_ADD:
		data_len = sizeof(struct mlag_mroute_add);
		fill_msg_type = MLAG_MROUTE_ADD_BULK;
		break;
	case MLAG_MROUTE_DEL:
		data_len = sizeof(struct mlag_mroute_del);
		fill_msg_type = MLAG_MROUTE_DEL_BULK;
		break;
	case MLAG_MSG_NONE:
		return;
	case MLAG_STATUS_UPDATE:
	case MLAG_DUMP:
	case MLAG_MROUTE_ADD_BULK:
	case MLAG_MROUTE_DEL_BULK:
	case MLAG_PIM_CFG_DUMP:
	case MLAG_VXLAN_UPDATE:
	case MLAG_PEER_FRR_STATUS:
		data_len = 0;
		break;
	}

	stream_reset(router->mlag_stream);
	/* ADD Hedaer */
	stream_putl(router->mlag_stream, fill_msg_type);
	/*
	 * In case of Bulk actual size & msg_cnt will be updated
	 * just before writing onto zebra
	 */
	stream_putw(router->mlag_stream, data_len);
	stream_putw(router->mlag_stream, msg_cnt);

	if (PIM_DEBUG_MLAG)
		zlog_debug(":%s: msg_type: %d/%d len %d",
			   __func__, msg_type, fill_msg_type, data_len);
}

static void pim_mlag_zebra_flush_buffer(void)
{
	uint32_t msg_type;

	/* Stream had bulk messages update the Hedaer */
	if (mlag_bulk_cnt > 1) {
		/*
		 * No need to reset the pointer, below api reads from data[0]
		 */
		STREAM_GETL(router->mlag_stream, msg_type);
		if (msg_type == MLAG_MROUTE_ADD_BULK) {
			stream_putw_at(
				router->mlag_stream, 4,
				(mlag_bulk_cnt * sizeof(struct mlag_mroute_add)));
			stream_putw_at(router->mlag_stream, 6, mlag_bulk_cnt);
		} else if (msg_type == MLAG_MROUTE_DEL_BULK) {
			stream_putw_at(
				router->mlag_stream, 4,
				(mlag_bulk_cnt * sizeof(struct mlag_mroute_del)));
			stream_putw_at(router->mlag_stream, 6, mlag_bulk_cnt);
		} else {
			flog_err(EC_LIB_ZAPI_ENCODE,
				"unknown bulk message type %d bulk_count %d",
				msg_type, mlag_bulk_cnt);
			stream_reset(router->mlag_stream);
			mlag_bulk_cnt = 0;
			return;
		}
	}

	zclient_send_mlag_data(zclient, router->mlag_stream);
stream_failure:
	stream_reset(router->mlag_stream);
	mlag_bulk_cnt = 0;
}

/*
 * Only ROUTE add & Delete will be bulked.
 * Buffer will be flushed, when
 * 1) there were no messages in the queue
 * 2) Curr_msg_type != prev_msg_type
 */

static void pim_mlag_zebra_check_for_buffer_flush(uint32_t curr_msg_type,
						  uint32_t prev_msg_type)
{
	/* First Message, keep bulking */
	if (prev_msg_type == MLAG_MSG_NONE) {
		mlag_bulk_cnt = 1;
		return;
	}

	/*msg type is route add & delete, keep bulking */
	if (curr_msg_type == prev_msg_type
	    && (curr_msg_type == MLAG_MROUTE_ADD
		|| curr_msg_type == MLAG_MROUTE_DEL)) {
		mlag_bulk_cnt++;
		return;
	}

	pim_mlag_zebra_flush_buffer();
}

/*
 * Thsi thread reads the clients data from the Gloabl queue and encodes with
 * protobuf and pass on to the MLAG socket.
 */
static void pim_mlag_zthread_handler(struct thread *event)
{
	struct stream *read_s;
	uint32_t wr_count = 0;
	uint32_t prev_msg_type = MLAG_MSG_NONE;
	uint32_t curr_msg_type = MLAG_MSG_NONE;

	router->zpthread_mlag_write = NULL;
	wr_count = stream_fifo_count_safe(router->mlag_fifo);

	if (PIM_DEBUG_MLAG)
		zlog_debug(":%s: Processing MLAG write, %d messages in queue",
			   __func__, wr_count);

	if (wr_count == 0)
		return;

	for (wr_count = 0; wr_count < PIM_MLAG_POST_LIMIT; wr_count++) {
		/* FIFO is empty,wait for teh message to be add */
		if (stream_fifo_count_safe(router->mlag_fifo) == 0)
			break;

		read_s = stream_fifo_pop_safe(router->mlag_fifo);
		if (!read_s) {
			zlog_debug(":%s: Got a NULL Messages, some thing wrong",
				   __func__);
			break;
		}
		STREAM_GETL(read_s, curr_msg_type);
		/*
		 * Check for Buffer Overflow,
		 * MLAG Can't process more than 'PIM_MLAG_BUF_LIMIT' bytes
		 */
		if (router->mlag_stream->endp + read_s->endp + ZEBRA_HEADER_SIZE
		    > MLAG_BUF_LIMIT)
			pim_mlag_zebra_flush_buffer();

		pim_mlag_zebra_check_for_buffer_flush(curr_msg_type,
						      prev_msg_type);

		/*
		 * First message to Buffer, fill the Header
		 */
		if (router->mlag_stream->endp == 0)
			pim_mlag_zebra_fill_header(curr_msg_type);

		/*
		 * add the data now
		 */
		stream_put(router->mlag_stream, read_s->data + read_s->getp,
			   read_s->endp - read_s->getp);

		stream_free(read_s);
		prev_msg_type = curr_msg_type;
	}

stream_failure:
	/*
	 * we are here , because
	 * 1. Queue might be empty
	 * 2. we crossed the max Q Read limit
	 * In any acse flush the buffer towards zebra
	 */
	pim_mlag_zebra_flush_buffer();

	if (wr_count >= PIM_MLAG_POST_LIMIT)
		pim_mlag_signal_zpthread();
}


int pim_mlag_signal_zpthread(void)
{
	if (router->master) {
		if (PIM_DEBUG_MLAG)
			zlog_debug(":%s: Scheduling PIM MLAG write Thread",
				   __func__);
		thread_add_event(router->master, pim_mlag_zthread_handler, NULL,
				 0, &router->zpthread_mlag_write);
	}
	return (0);
}
Commit	Line	Data
acddc0ed	1	// SPDX-License-Identifier: GPL-2.0-or-later
17823cdd DS	2	/*
	3	* PIM for Quagga
	4	* Copyright (C) 2008 Everton da Silva Marques
17823cdd DS	5	*/
	6
	7	#include <zebra.h>
	8	#include <lib/log.h>
	9	#include <lib/lib_errors.h>
	10
	11	#include "pimd.h"
993e3d8e	12	#include "pim_instance.h"
17823cdd DS	13	#include "pim_mlag.h"
	14	#include "pim_zebra.h"
	15
	16	extern struct zclient *zclient;
	17
	18	#define PIM_MLAG_POST_LIMIT 100
	19
	20	int32_t mlag_bulk_cnt;
	21
	22	static void pim_mlag_zebra_fill_header(enum mlag_msg_type msg_type)
	23	{
	24	uint32_t fill_msg_type = msg_type;
da21ae9d	25	uint16_t data_len = 0;
17823cdd DS	26	uint16_t msg_cnt = 1;
17823cdd DS	27
17823cdd DS	28	switch (msg_type) {
	29	case MLAG_REGISTER:
	30	case MLAG_DEREGISTER:
	31	data_len = sizeof(struct mlag_msg);
	32	break;
	33	case MLAG_MROUTE_ADD:
	34	data_len = sizeof(struct mlag_mroute_add);
	35	fill_msg_type = MLAG_MROUTE_ADD_BULK;
	36	break;
	37	case MLAG_MROUTE_DEL:
	38	data_len = sizeof(struct mlag_mroute_del);
	39	fill_msg_type = MLAG_MROUTE_DEL_BULK;
	40	break;
da21ae9d	41	case MLAG_MSG_NONE:
88c1f613	42	return;
da21ae9d DS	43	case MLAG_STATUS_UPDATE:
	44	case MLAG_DUMP:
	45	case MLAG_MROUTE_ADD_BULK:
	46	case MLAG_MROUTE_DEL_BULK:
	47	case MLAG_PIM_CFG_DUMP:
	48	case MLAG_VXLAN_UPDATE:
	49	case MLAG_PEER_FRR_STATUS:
17823cdd DS	50	data_len = 0;
	51	break;
	52	}
	53
	54	stream_reset(router->mlag_stream);
	55	/* ADD Hedaer */
	56	stream_putl(router->mlag_stream, fill_msg_type);
	57	/*
	58	* In case of Bulk actual size & msg_cnt will be updated
	59	* just before writing onto zebra
	60	*/
	61	stream_putw(router->mlag_stream, data_len);
	62	stream_putw(router->mlag_stream, msg_cnt);
	63
	64	if (PIM_DEBUG_MLAG)
	65	zlog_debug(":%s: msg_type: %d/%d len %d",
	66	__func__, msg_type, fill_msg_type, data_len);
	67	}
	68
	69	static void pim_mlag_zebra_flush_buffer(void)
	70	{
	71	uint32_t msg_type;
	72
	73	/* Stream had bulk messages update the Hedaer */
	74	if (mlag_bulk_cnt > 1) {
	75	/*
	76	* No need to reset the pointer, below api reads from data[0]
	77	*/
	78	STREAM_GETL(router->mlag_stream, msg_type);
	79	if (msg_type == MLAG_MROUTE_ADD_BULK) {
	80	stream_putw_at(
	81	router->mlag_stream, 4,
	82	(mlag_bulk_cnt * sizeof(struct mlag_mroute_add)));
	83	stream_putw_at(router->mlag_stream, 6, mlag_bulk_cnt);
	84	} else if (msg_type == MLAG_MROUTE_DEL_BULK) {
	85	stream_putw_at(
	86	router->mlag_stream, 4,
	87	(mlag_bulk_cnt * sizeof(struct mlag_mroute_del)));
	88	stream_putw_at(router->mlag_stream, 6, mlag_bulk_cnt);
	89	} else {
	90	flog_err(EC_LIB_ZAPI_ENCODE,
	91	"unknown bulk message type %d bulk_count %d",
	92	msg_type, mlag_bulk_cnt);
	93	stream_reset(router->mlag_stream);
	94	mlag_bulk_cnt = 0;
	95	return;
	96	}
	97	}
	98
	99	zclient_send_mlag_data(zclient, router->mlag_stream);
	100	stream_failure:
	101	stream_reset(router->mlag_stream);
	102	mlag_bulk_cnt = 0;
	103	}
	104
	105	/*
	106	* Only ROUTE add & Delete will be bulked.
	107	* Buffer will be flushed, when
	108	* 1) there were no messages in the queue
	109	* 2) Curr_msg_type != prev_msg_type
	110	*/
	111
	112	static void pim_mlag_zebra_check_for_buffer_flush(uint32_t curr_msg_type,
	113	uint32_t prev_msg_type)
114	{
115	/* First Message, keep bulking */
116	if (prev_msg_type == MLAG_MSG_NONE) {
117	mlag_bulk_cnt = 1;
118	return;
119	}
120
121	/msg type is route add & delete, keep bulking /
122	if (curr_msg_type == prev_msg_type
123	&& (curr_msg_type == MLAG_MROUTE_ADD
124	\|\| curr_msg_type == MLAG_MROUTE_DEL)) {
125	mlag_bulk_cnt++;
126	return;
127	}
128
129	pim_mlag_zebra_flush_buffer();
130	}
131
132	/*
133	* Thsi thread reads the clients data from the Gloabl queue and encodes with
134	* protobuf and pass on to the MLAG socket.
135	*/
cc9f21da	136	static void pim_mlag_zthread_handler(struct thread *event)
17823cdd DS	137	{
	138	struct stream *read_s;
	139	uint32_t wr_count = 0;
	140	uint32_t prev_msg_type = MLAG_MSG_NONE;
	141	uint32_t curr_msg_type = MLAG_MSG_NONE;
	142
	143	router->zpthread_mlag_write = NULL;
	144	wr_count = stream_fifo_count_safe(router->mlag_fifo);
	145
	146	if (PIM_DEBUG_MLAG)
	147	zlog_debug(":%s: Processing MLAG write, %d messages in queue",
	148	__func__, wr_count);
	149
	150	if (wr_count == 0)
cc9f21da	151	return;
17823cdd DS	152
	153	for (wr_count = 0; wr_count < PIM_MLAG_POST_LIMIT; wr_count++) {
	154	/* FIFO is empty,wait for teh message to be add */
	155	if (stream_fifo_count_safe(router->mlag_fifo) == 0)
	156	break;
	157
	158	read_s = stream_fifo_pop_safe(router->mlag_fifo);
	159	if (!read_s) {
	160	zlog_debug(":%s: Got a NULL Messages, some thing wrong",
	161	__func__);
	162	break;
	163	}
	164	STREAM_GETL(read_s, curr_msg_type);
	165	/*
	166	* Check for Buffer Overflow,
	167	* MLAG Can't process more than 'PIM_MLAG_BUF_LIMIT' bytes
	168	*/
	169	if (router->mlag_stream->endp + read_s->endp + ZEBRA_HEADER_SIZE
	170	> MLAG_BUF_LIMIT)
	171	pim_mlag_zebra_flush_buffer();
	172
	173	pim_mlag_zebra_check_for_buffer_flush(curr_msg_type,
	174	prev_msg_type);
	175
	176	/*
	177	* First message to Buffer, fill the Header
	178	*/
	179	if (router->mlag_stream->endp == 0)
	180	pim_mlag_zebra_fill_header(curr_msg_type);
	181
	182	/*
	183	* add the data now
	184	*/
	185	stream_put(router->mlag_stream, read_s->data + read_s->getp,
	186	read_s->endp - read_s->getp);
	187
	188	stream_free(read_s);
	189	prev_msg_type = curr_msg_type;
	190	}
	191
	192	stream_failure:
	193	/*
	194	* we are here , because
	195	* 1. Queue might be empty
	196	* 2. we crossed the max Q Read limit
	197	* In any acse flush the buffer towards zebra
	198	*/
	199	pim_mlag_zebra_flush_buffer();
	200
	201	if (wr_count >= PIM_MLAG_POST_LIMIT)
	202	pim_mlag_signal_zpthread();
17823cdd DS	203	}
	204
	205
	206	int pim_mlag_signal_zpthread(void)
	207	{
	208	if (router->master) {
	209	if (PIM_DEBUG_MLAG)
	210	zlog_debug(":%s: Scheduling PIM MLAG write Thread",
	211	__func__);
	212	thread_add_event(router->master, pim_mlag_zthread_handler, NULL,
	213	0, &router->zpthread_mlag_write);
	214	}
	215	return (0);
	216	}