zebra: reorganize zserv, batch i/o

Group send and receive functions together, change handlers to take a
message instead of looking at ->ibuf and ->obuf, allow zebra to read
multiple packets off the wire at a time.

Signed-off-by: Quentin Young <qlyoung@cumulusnetworks.com>

zebra/label_manager.c

@@ -50,6 +50,8 @@ DEFINE_MTYPE_STATIC(LBL_MGR, LM_CHUNK, "Label Manager Chunk");
  * it will be a proxy to relay messages to external label manager
  * This zclient thus is to connect to it
  */
+static struct stream *ibuf;
+static struct stream *obuf;
 static struct zclient *zclient;
 bool lm_is_external;
 
@@ -69,7 +71,7 @@ static int relay_response_back(struct zserv *zserv)
 	u_int16_t resp_cmd;
 
 	src = zclient->ibuf;
-	dst = zserv->obuf;
+	dst = obuf;
 
 	stream_reset(src);
 
@@ -87,7 +89,7 @@ static int relay_response_back(struct zserv *zserv)
 
 	/* send response back */
 	stream_copy(dst, src);
-	ret = writen(zserv->sock, dst->data, stream_get_endp(dst));
+	ret = writen(zserv->sock, src->data, stream_get_endp(src));
 	if (ret <= 0) {
 		zlog_err("%s: Error sending Label Manager response back: %s",
 			 __func__, strerror(errno));
@@ -116,10 +118,10 @@ static int lm_zclient_read(struct thread *t)
 
 static int reply_error(int cmd, struct zserv *zserv, vrf_id_t vrf_id)
 {
+	int ret;
 	struct stream *s;
 
-	s = zserv->obuf;
-	stream_reset(s);
+	s = stream_new(ZEBRA_MAX_PACKET_SIZ);
 
 	zclient_create_header(s, cmd, vrf_id);
 
@@ -129,7 +131,10 @@ static int reply_error(int cmd, struct zserv *zserv, vrf_id_t vrf_id)
 	/* Write packet size. */
 	stream_putw_at(s, 0, stream_get_endp(s));
 
-	return writen(zserv->sock, s->data, stream_get_endp(s));
+	ret = writen(zserv->sock, s->data, stream_get_endp(s));
+
+	stream_free(s);
+	return ret;
 }
 /**
  * Receive a request to get or release a label chunk and forward it to external
@@ -161,7 +166,7 @@ int zread_relay_label_manager_request(int cmd, struct zserv *zserv,
 		ret = relay_response_back(zserv);
 
 	/* Send request to external label manager */
-	src = zserv->ibuf;
+	src = ibuf;
 	dst = zclient->obuf;
 
 	stream_copy(dst, src);
@@ -247,6 +252,9 @@ void label_manager_init(char *lm_zserv_path)
 		lm_is_external = true;
 		lm_zclient_init(lm_zserv_path);
 	}
+
+	ibuf = stream_new(ZEBRA_MAX_PACKET_SIZ);
+	obuf = stream_new(ZEBRA_MAX_PACKET_SIZ);
 }
 
 /**
@@ -379,4 +387,6 @@ int release_daemon_chunks(u_char proto, u_short instance)
 void label_manager_close()
 {
 	list_delete_and_null(&lbl_mgr.lc_list);
+	stream_free(ibuf);
+	stream_free(obuf);
 }

zebra/redistribute.c

@@ -249,9 +249,9 @@ void zebra_redistribute_add(ZAPI_HANDLER_ARGS)
 	int type = 0;
 	u_short instance;
 
-	STREAM_GETC(client->ibuf, afi);
-	STREAM_GETC(client->ibuf, type);
-	STREAM_GETW(client->ibuf, instance);
+	STREAM_GETC(msg, afi);
+	STREAM_GETC(msg, type);
+	STREAM_GETW(msg, instance);
 
 	if (afi == 0 || afi > AFI_MAX) {
 		zlog_warn("%s: Specified afi %d does not exist",
@@ -292,9 +292,9 @@ void zebra_redistribute_delete(ZAPI_HANDLER_ARGS)
 	int type = 0;
 	u_short instance;
 
-	STREAM_GETC(client->ibuf, afi);
-	STREAM_GETC(client->ibuf, type);
-	STREAM_GETW(client->ibuf, instance);
+	STREAM_GETC(msg, afi);
+	STREAM_GETC(msg, type);
+	STREAM_GETW(msg, instance);
 
 	if (afi == 0 || afi > AFI_MAX) {
 		zlog_warn("%s: Specified afi %d does not exist",

zebra/redistribute.h

@@ -27,11 +27,12 @@
 #include "vty.h"
 #include "vrf.h"
 
-/* zapi handlers */
+/* ZAPI command handlers */
 extern void zebra_redistribute_add(ZAPI_HANDLER_ARGS);
 extern void zebra_redistribute_delete(ZAPI_HANDLER_ARGS);
 extern void zebra_redistribute_default_add(ZAPI_HANDLER_ARGS);
 extern void zebra_redistribute_default_delete(ZAPI_HANDLER_ARGS);
+/* ----------------- */
 
 extern void redistribute_update(struct prefix *, struct prefix *,
 				struct route_entry *, struct route_entry *);

zebra/rtadv.c

@@ -43,7 +43,6 @@
 #include "zebra/zserv.h"
 #include "zebra/zebra_ns.h"
 #include "zebra/zebra_vrf.h"
-#include "zebra/zserv.h"
 
 extern struct zebra_privs_t zserv_privs;
 
@@ -802,42 +801,43 @@ static void ipv6_nd_suppress_ra_set(struct interface *ifp,
  * if the operator has explicitly enabled RA. The enable request can also
  * specify a RA interval (in seconds).
  */
-void zebra_interface_radv_set(ZAPI_HANDLER_ARGS, int set)
+static void zebra_interface_radv_set(ZAPI_HANDLER_ARGS, int enable)
 {
 	struct stream *s;
 	ifindex_t ifindex;
 	struct interface *ifp;
 	struct zebra_if *zif;
 	int ra_interval;
 
-	s = client->ibuf;
+	s = msg;
 
 	/* Get interface index and RA interval. */
 	STREAM_GETL(s, ifindex);
 	STREAM_GETL(s, ra_interval);
 
 	if (IS_ZEBRA_DEBUG_EVENT)
 		zlog_debug("%u: IF %u RA %s from client %s, interval %ds",
-			   zvrf_id(zvrf), ifindex, set ? "enable" : "disable",
+			   zvrf_id(zvrf), ifindex,
+			   enable ? "enable" : "disable",
 			   zebra_route_string(client->proto), ra_interval);
 
 	/* Locate interface and check VRF match. */
 	ifp = if_lookup_by_index_per_ns(zebra_ns_lookup(NS_DEFAULT), ifindex);
 	if (!ifp) {
 		zlog_warn("%u: IF %u RA %s client %s - interface unknown",
-			  zvrf_id(zvrf), ifindex, set ? "enable" : "disable",
+			  zvrf_id(zvrf), ifindex, enable ? "enable" : "disable",
 			  zebra_route_string(client->proto));
 		return;
 	}
 	if (ifp->vrf_id != zvrf_id(zvrf)) {
 		zlog_warn("%u: IF %u RA %s client %s - VRF mismatch, IF VRF %u",
-			  zvrf_id(zvrf), ifindex, set ? "enable" : "disable",
+			  zvrf_id(zvrf), ifindex, enable ? "enable" : "disable",
 			  zebra_route_string(client->proto), ifp->vrf_id);
 		return;
 	}
 
 	zif = ifp->info;
-	if (set) {
+	if (enable) {
 		SET_FLAG(zif->rtadv.ra_configured, BGP_RA_CONFIGURED);
 		ipv6_nd_suppress_ra_set(ifp, RA_ENABLE);
 		if (ra_interval
@@ -860,12 +860,11 @@ void zebra_interface_radv_set(ZAPI_HANDLER_ARGS, int set)
 
 void zebra_interface_radv_disable(ZAPI_HANDLER_ARGS)
 {
-	zebra_interface_radv_set(client, hdr, zvrf, 0);
+	zebra_interface_radv_set(client, hdr, msg, zvrf, 0);
 }
-
 void zebra_interface_radv_enable(ZAPI_HANDLER_ARGS)
 {
-	zebra_interface_radv_set(client, hdr, zvrf, 1);
+	zebra_interface_radv_set(client, hdr, msg, zvrf, 1);
 }
 
 DEFUN (ipv6_nd_suppress_ra,

zebra/rtadv.h

@@ -103,8 +103,8 @@ typedef enum {
 extern void rtadv_init(struct zebra_ns *);
 extern void rtadv_terminate(struct zebra_ns *);
 extern void rtadv_cmd_init(void);
-extern void zebra_interface_radv_set(ZAPI_HANDLER_ARGS, int enable);
 extern void zebra_interface_radv_disable(ZAPI_HANDLER_ARGS);
 extern void zebra_interface_radv_enable(ZAPI_HANDLER_ARGS);
 
+
 #endif /* _ZEBRA_RTADV_H */

zebra/zebra_mpls.c

@@ -455,16 +455,15 @@ static int fec_send(zebra_fec_t *fec, struct zserv *client)
 	rn = fec->rn;
 
 	/* Get output stream. */
-	s = client->obuf;
-	stream_reset(s);
+	s = stream_new(ZEBRA_MAX_PACKET_SIZ);
 
 	zclient_create_header(s, ZEBRA_FEC_UPDATE, VRF_DEFAULT);
 
 	stream_putw(s, rn->p.family);
 	stream_put_prefix(s, &rn->p);
 	stream_putl(s, fec->label);
 	stream_putw_at(s, 0, stream_get_endp(s));
-	return zebra_server_send_message(client);
+	return zebra_server_send_message(client, s);
 }
 
 /*

zebra/zebra_mroute.c

@@ -31,7 +31,6 @@
 #include "zebra/zebra_mroute.h"
 #include "zebra/rt.h"
 #include "zebra/debug.h"
-#include "zebra/zserv.h"
 
 void zebra_ipmr_route_stats(ZAPI_HANDLER_ARGS)
 {
@@ -40,9 +39,9 @@ void zebra_ipmr_route_stats(ZAPI_HANDLER_ARGS)
 	int suc = -1;
 
 	memset(&mroute, 0, sizeof(mroute));
-	STREAM_GET(&mroute.sg.src, client->ibuf, 4);
-	STREAM_GET(&mroute.sg.grp, client->ibuf, 4);
-	STREAM_GETL(client->ibuf, mroute.ifindex);
+	STREAM_GET(&mroute.sg.src, msg, 4);
+	STREAM_GET(&mroute.sg.grp, msg, 4);
+	STREAM_GETL(msg, mroute.ifindex);
 
 	if (IS_ZEBRA_DEBUG_KERNEL) {
 		char sbuf[40];
@@ -57,7 +56,7 @@ void zebra_ipmr_route_stats(ZAPI_HANDLER_ARGS)
 	suc = kernel_get_ipmr_sg_stats(zvrf, &mroute);
 
 stream_failure:
-	s = client->obuf;
+	s = stream_new(ZEBRA_MAX_PACKET_SIZ);
 
 	stream_reset(s);
 
@@ -68,5 +67,5 @@ void zebra_ipmr_route_stats(ZAPI_HANDLER_ARGS)
 	stream_putl(s, suc);
 
 	stream_putw_at(s, 0, stream_get_endp(s));
-	zebra_server_send_message(client);
+	zebra_server_send_message(client, s);
 }

zebra/zebra_ptm.c

@@ -702,7 +702,7 @@ void zebra_ptm_bfd_dst_register(ZAPI_HANDLER_ARGS)
 	ptm_lib_append_msg(ptm_hdl, out_ctxt, ZEBRA_PTM_BFD_CLIENT_FIELD,
 			   tmp_buf);
 
-	s = client->ibuf;
+	s = msg;
 
 	STREAM_GETL(s, pid);
 	sprintf(tmp_buf, "%d", pid);
@@ -819,7 +819,6 @@ void zebra_ptm_bfd_dst_register(ZAPI_HANDLER_ARGS)
 
 stream_failure:
 	ptm_lib_cleanup_msg(ptm_hdl, out_ctxt);
-	return;
 }
 
 /* BFD peer/dst deregister */
@@ -859,7 +858,7 @@ void zebra_ptm_bfd_dst_deregister(ZAPI_HANDLER_ARGS)
 	ptm_lib_append_msg(ptm_hdl, out_ctxt, ZEBRA_PTM_BFD_CLIENT_FIELD,
 			   tmp_buf);
 
-	s = client->ibuf;
+	s = msg;
 
 	STREAM_GETL(s, pid);
 	sprintf(tmp_buf, "%d", pid);
@@ -950,7 +949,6 @@ void zebra_ptm_bfd_dst_deregister(ZAPI_HANDLER_ARGS)
 
 stream_failure:
 	ptm_lib_cleanup_msg(ptm_hdl, out_ctxt);
-	return;
 }
 
 /* BFD client register */
@@ -968,7 +966,7 @@ void zebra_ptm_bfd_client_register(ZAPI_HANDLER_ARGS)
 		zlog_debug("bfd_client_register msg from client %s: length=%d",
 			   zebra_route_string(client->proto), hdr->length);
 
-	s = client->ibuf;
+	s = msg;
 	STREAM_GETL(s, pid);
 
 	if (ptm_cb.ptm_sock == -1) {

zebra/zebra_ptm.h

@@ -22,14 +22,14 @@
 #ifndef _ZEBRA_PTM_H
 #define _ZEBRA_PTM_H
 
-#include "zebra/zserv.h"
-
 extern const char ZEBRA_PTM_SOCK_NAME[];
 #define ZEBRA_PTM_MAX_SOCKBUF 3200 /* 25B *128 ports */
 #define ZEBRA_PTM_SEND_MAX_SOCKBUF 512
 
 #define ZEBRA_PTM_BFD_CLIENT_FLAG_REG   (1 << 1) /* client registered with BFD */
 
+#include "zebra/zserv.h"
+
 /* Zebra ptm context block */
 struct zebra_ptm_cb {
 	int ptm_sock; /* ptm file descriptor. */
@@ -64,6 +64,7 @@ int zebra_ptm_connect(struct thread *t);
 void zebra_ptm_write(struct vty *vty);
 int zebra_ptm_get_enable_state(void);
 
+/* ZAPI message handlers */
 void zebra_ptm_bfd_dst_register(ZAPI_HANDLER_ARGS);
 void zebra_ptm_bfd_dst_deregister(ZAPI_HANDLER_ARGS);
 void zebra_ptm_bfd_client_register(ZAPI_HANDLER_ARGS);

zebra/zebra_ptm_redistribute.c

@@ -38,8 +38,7 @@ static int zsend_interface_bfd_update(int cmd, struct zserv *client,
 	if (!client->ifinfo)
 		return 0;
 
-	s = client->obuf;
-	stream_reset(s);
+	s = stream_new(ZEBRA_MAX_PACKET_SIZ);
 
 	zclient_create_header(s, cmd, vrf_id);
 	if (ifp)
@@ -66,7 +65,7 @@ static int zsend_interface_bfd_update(int cmd, struct zserv *client,
 	stream_putw_at(s, 0, stream_get_endp(s));
 
 	client->if_bfd_cnt++;
-	return zebra_server_send_message(client);
+	return zebra_server_send_message(client, s);
 }
 
 void zebra_interface_bfd_update(struct interface *ifp, struct prefix *dp,
@@ -93,16 +92,15 @@ static int zsend_bfd_peer_replay(int cmd, struct zserv *client)
 {
 	struct stream *s;
 
-	s = client->obuf;
-	stream_reset(s);
+	s = stream_new(ZEBRA_MAX_PACKET_SIZ);
 
 	zclient_create_header(s, cmd, VRF_DEFAULT);
 
 	/* Write packet size. */
 	stream_putw_at(s, 0, stream_get_endp(s));
 
 	client->bfd_peer_replay_cnt++;
-	return zebra_server_send_message(client);
+	return zebra_server_send_message(client, s);
 }
 
 void zebra_bfd_peer_replay_req(void)

zebra/zebra_rnh.c

@@ -998,8 +998,7 @@ static int send_client(struct rnh *rnh, struct zserv *client, rnh_type_t type,
 	re = rnh->state;
 
 	/* Get output stream. */
-	s = client->obuf;
-	stream_reset(s);
+	s = stream_new(ZEBRA_MAX_PACKET_SIZ);
 
 	zclient_create_header(s, cmd, vrf_id);
 
@@ -1066,7 +1065,7 @@ static int send_client(struct rnh *rnh, struct zserv *client, rnh_type_t type,
 
 	client->nh_last_upd_time = monotime(NULL);
 	client->last_write_cmd = cmd;
-	return zebra_server_send_message(client);
+	return zebra_server_send_message(client, s);
 }
 
 static void print_nh(struct nexthop *nexthop, struct vty *vty)