Merge pull request #5925 from volta-networks/synchronous_client

zebra: synchronous client queues accumulate messages from zebra

ldpd/lde.c

@@ -1698,8 +1698,11 @@ lde_address_list_free(struct lde_nbr *ln)
 
 static void zclient_sync_init(unsigned short instance)
 {
+	struct zclient_options options = zclient_options_default;
+	options.synchronous = true;
+
 	/* Initialize special zclient for synchronous message exchanges. */
-	zclient_sync = zclient_new(master, &zclient_options_default);
+	zclient_sync = zclient_new(master, &options);
 	zclient_sync->sock = -1;
 	zclient_sync->redist_default = ZEBRA_ROUTE_LDP;
 	zclient_sync->instance = instance;
@@ -1712,6 +1715,12 @@ static void zclient_sync_init(unsigned short instance)
 	/* make socket non-blocking */
 	sock_set_nonblock(zclient_sync->sock);
 
+	/* Send hello to notify zebra this is a synchronous client */
+	while (zclient_send_hello(zclient_sync) < 0) {
+		log_warnx("Error sending hello for synchronous zclient!");
+		sleep(1);
+	}
+
 	/* Connect to label manager */
 	while (lm_label_manager_connect(zclient_sync, 0) != 0) {
 		log_warnx("Error connecting to label manager!");

lib/zclient.c

@@ -52,7 +52,8 @@ static void zclient_event(enum event, struct zclient *);
 static void zebra_interface_if_set_value(struct stream *s,
 					 struct interface *ifp);
 
-struct zclient_options zclient_options_default = {.receive_notify = false};
+struct zclient_options zclient_options_default = {.receive_notify = false,
+						  .synchronous = false};
 
 struct sockaddr_storage zclient_addr;
 socklen_t zclient_addr_len;
@@ -76,6 +77,7 @@ struct zclient *zclient_new(struct thread_master *master,
 	zclient->master = master;
 
 	zclient->receive_notify = opt->receive_notify;
+	zclient->synchronous = opt->synchronous;
 
 	return zclient;
 }
@@ -374,11 +376,11 @@ static int zebra_message_send(struct zclient *zclient, int command,
 	return zclient_send_message(zclient);
 }
 
-static int zebra_hello_send(struct zclient *zclient)
+int zclient_send_hello(struct zclient *zclient)
 {
 	struct stream *s;
 
-	if (zclient->redist_default) {
+	if (zclient->redist_default || zclient->synchronous) {
 		s = zclient->obuf;
 		stream_reset(s);
 
@@ -390,6 +392,10 @@ static int zebra_hello_send(struct zclient *zclient)
 			stream_putc(s, 1);
 		else
 			stream_putc(s, 0);
+		if (zclient->synchronous)
+			stream_putc(s, 1);
+		else
+			stream_putc(s, 0);
 
 		stream_putw_at(s, 0, stream_get_endp(s));
 		return zclient_send_message(zclient);
@@ -629,7 +635,7 @@ int zclient_start(struct zclient *zclient)
 	/* Create read thread. */
 	zclient_event(ZCLIENT_READ, zclient);
 
-	zebra_hello_send(zclient);
+	zclient_send_hello(zclient);
 
 	zebra_message_send(zclient, ZEBRA_INTERFACE_ADD, VRF_DEFAULT);
 

lib/zclient.h

@@ -255,6 +255,9 @@ struct zclient {
 	/* Do we care about failure events for route install? */
 	bool receive_notify;
 
+	/* Is this a synchronous client? */
+	bool synchronous;
+
 	/* Socket to zebra daemon. */
 	int sock;
 
@@ -569,6 +572,7 @@ enum zebra_neigh_state { ZEBRA_NEIGH_INACTIVE = 0, ZEBRA_NEIGH_ACTIVE = 1 };
 
 struct zclient_options {
 	bool receive_notify;
+	bool synchronous;
 };
 
 extern struct zclient_options zclient_options_default;
@@ -796,4 +800,9 @@ extern void zclient_send_mlag_deregister(struct zclient *client);
 extern void zclient_send_mlag_data(struct zclient *client,
 				   struct stream *client_s);
 
+/* Send the hello message.
+ * Returns 0 for success or -1 on an I/O error.
+ */
+extern int zclient_send_hello(struct zclient *client);
+
 #endif /* _ZEBRA_ZCLIENT_H */

pimd/pim_zlookup.c

@@ -61,6 +61,14 @@ static int zclient_lookup_connect(struct thread *t)
 		zlookup->fail = 0; /* reset counter on connection */
 	}
 
+	if (zclient_send_hello(zlookup) < 0) {
+		if (close(zlookup->sock)) {
+			zlog_warn("%s: closing fd=%d: errno=%d %s", __func__,
+				  zlookup->sock, errno, safe_strerror(errno));
+		}
+		zlookup->sock = -1;
+	}
+
 	if (zlookup->sock < 0) {
 		/* Since last connect failed, retry within 10 secs */
 		zclient_lookup_sched(zlookup, 10);
@@ -125,7 +133,10 @@ void zclient_lookup_free(void)
 
 void zclient_lookup_new(void)
 {
-	zlookup = zclient_new(router->master, &zclient_options_default);
+	struct zclient_options options = zclient_options_default;
+	options.synchronous = true;
+
+	zlookup = zclient_new(router->master, &options);
 	if (!zlookup) {
 		flog_err(EC_LIB_ZAPI_SOCKET, "%s: zclient_new() failure",
 			 __func__);
@@ -161,6 +172,7 @@ static int zclient_read_nexthop(struct pim_instance *pim,
 
 	if (PIM_DEBUG_PIM_NHT_DETAIL) {
 		char addr_str[INET_ADDRSTRLEN];
+
 		pim_inet4_dump("<addr?>", addr, addr_str, sizeof(addr_str));
 		zlog_debug("%s: addr=%s(%s)", __func__, addr_str,
 			   pim->vrf->name);

zebra/redistribute.c

@@ -291,6 +291,10 @@ void redistribute_delete(const struct prefix *p, const struct prefix *src_p,
 	}
 
 	for (ALL_LIST_ELEMENTS(zrouter.client_list, node, nnode, client)) {
+		/* Do not send unsolicited messages to synchronous clients. */
+		if (client->synchronous)
+			continue;
+
 		if (new_re) {
 			/* Skip this client if it will receive an update for the
 			 * 'new' re
@@ -472,6 +476,12 @@ void zebra_interface_up_update(struct interface *ifp)
 	if (ifp->ptm_status || !ifp->ptm_enable) {
 		for (ALL_LIST_ELEMENTS(zrouter.client_list, node, nnode,
 				       client)) {
+			/* Do not send unsolicited messages to synchronous
+			 * clients.
+			 */
+			if (client->synchronous)
+				continue;
+
 			zsend_interface_update(ZEBRA_INTERFACE_UP,
 					       client, ifp);
 			zsend_interface_link_params(client, ifp);
@@ -490,6 +500,10 @@ void zebra_interface_down_update(struct interface *ifp)
 			   ifp->name, ifp->vrf_id);
 
 	for (ALL_LIST_ELEMENTS(zrouter.client_list, node, nnode, client)) {
+		/* Do not send unsolicited messages to synchronous clients. */
+		if (client->synchronous)
+			continue;
+
 		zsend_interface_update(ZEBRA_INTERFACE_DOWN, client, ifp);
 	}
 }
@@ -505,6 +519,10 @@ void zebra_interface_add_update(struct interface *ifp)
 			   ifp->vrf_id);
 
 	for (ALL_LIST_ELEMENTS(zrouter.client_list, node, nnode, client)) {
+		/* Do not send unsolicited messages to synchronous clients. */
+		if (client->synchronous)
+			continue;
+
 		client->ifadd_cnt++;
 		zsend_interface_add(client, ifp);
 		zsend_interface_link_params(client, ifp);
@@ -521,6 +539,10 @@ void zebra_interface_delete_update(struct interface *ifp)
 			   ifp->name, ifp->vrf_id);
 
 	for (ALL_LIST_ELEMENTS(zrouter.client_list, node, nnode, client)) {
+		/* Do not send unsolicited messages to synchronous clients. */
+		if (client->synchronous)
+			continue;
+
 		client->ifdel_cnt++;
 		zsend_interface_delete(client, ifp);
 	}
@@ -552,12 +574,17 @@ void zebra_interface_address_add_update(struct interface *ifp,
 
 	router_id_add_address(ifc);
 
-	for (ALL_LIST_ELEMENTS(zrouter.client_list, node, nnode, client))
+	for (ALL_LIST_ELEMENTS(zrouter.client_list, node, nnode, client)) {
+		/* Do not send unsolicited messages to synchronous clients. */
+		if (client->synchronous)
+			continue;
+
 		if (CHECK_FLAG(ifc->conf, ZEBRA_IFC_REAL)) {
 			client->connected_rt_add_cnt++;
 			zsend_interface_address(ZEBRA_INTERFACE_ADDRESS_ADD,
 						client, ifp, ifc);
 		}
+	}
 }
 
 /* Interface address deletion. */
@@ -581,12 +608,17 @@ void zebra_interface_address_delete_update(struct interface *ifp,
 
 	router_id_del_address(ifc);
 
-	for (ALL_LIST_ELEMENTS(zrouter.client_list, node, nnode, client))
+	for (ALL_LIST_ELEMENTS(zrouter.client_list, node, nnode, client)) {
+		/* Do not send unsolicited messages to synchronous clients. */
+		if (client->synchronous)
+			continue;
+
 		if (CHECK_FLAG(ifc->conf, ZEBRA_IFC_REAL)) {
 			client->connected_rt_del_cnt++;
 			zsend_interface_address(ZEBRA_INTERFACE_ADDRESS_DELETE,
 						client, ifp, ifc);
 		}
+	}
 }
 
 /* Interface VRF change. May need to delete from clients not interested in
@@ -603,6 +635,10 @@ void zebra_interface_vrf_update_del(struct interface *ifp, vrf_id_t new_vrf_id)
 			ifp->name, ifp->vrf_id, new_vrf_id);
 
 	for (ALL_LIST_ELEMENTS(zrouter.client_list, node, nnode, client)) {
+		/* Do not send unsolicited messages to synchronous clients. */
+		if (client->synchronous)
+			continue;
+
 		/* Need to delete if the client is not interested in the new
 		 * VRF. */
 		zsend_interface_update(ZEBRA_INTERFACE_DOWN, client, ifp);
@@ -626,6 +662,10 @@ void zebra_interface_vrf_update_add(struct interface *ifp, vrf_id_t old_vrf_id)
 			ifp->name, old_vrf_id, ifp->vrf_id);
 
 	for (ALL_LIST_ELEMENTS(zrouter.client_list, node, nnode, client)) {
+		/* Do not send unsolicited messages to synchronous clients. */
+		if (client->synchronous)
+			continue;
+
 		/* Need to add if the client is interested in the new VRF. */
 		client->ifadd_cnt++;
 		zsend_interface_add(client, ifp);
@@ -913,6 +953,11 @@ void zebra_interface_parameters_update(struct interface *ifp)
 		zlog_debug("MESSAGE: ZEBRA_INTERFACE_LINK_PARAMS %s(%u)",
 			   ifp->name, ifp->vrf_id);
 
-	for (ALL_LIST_ELEMENTS(zrouter.client_list, node, nnode, client))
+	for (ALL_LIST_ELEMENTS(zrouter.client_list, node, nnode, client)) {
+		/* Do not send unsolicited messages to synchronous clients. */
+		if (client->synchronous)
+			continue;
+
 		zsend_interface_link_params(client, ifp);
+	}
 }

zebra/zapi_msg.c

@@ -383,9 +383,14 @@ static void zebra_interface_nbr_address_add_update(struct interface *ifp,
 			p->prefixlen, ifc->ifp->name);
 	}
 
-	for (ALL_LIST_ELEMENTS(zrouter.client_list, node, nnode, client))
+	for (ALL_LIST_ELEMENTS(zrouter.client_list, node, nnode, client)) {
+		/* Do not send unsolicited messages to synchronous clients. */
+		if (client->synchronous)
+			continue;
+
 		zsend_interface_nbr_address(ZEBRA_INTERFACE_NBR_ADDRESS_ADD,
 					    client, ifp, ifc);
+	}
 }
 
 /* Interface address deletion. */
@@ -407,9 +412,14 @@ static void zebra_interface_nbr_address_delete_update(struct interface *ifp,
 			p->prefixlen, ifc->ifp->name);
 	}
 
-	for (ALL_LIST_ELEMENTS(zrouter.client_list, node, nnode, client))
+	for (ALL_LIST_ELEMENTS(zrouter.client_list, node, nnode, client)) {
+		/* Do not send unsolicited messages to synchronous clients. */
+		if (client->synchronous)
+			continue;
+
 		zsend_interface_nbr_address(ZEBRA_INTERFACE_NBR_ADDRESS_DELETE,
 					    client, ifp, ifc);
+	}
 }
 
 /* Send addresses on interface to client */
@@ -1740,6 +1750,10 @@ void zsend_capabilities_all_clients(void)
 
 	zvrf = vrf_info_lookup(VRF_DEFAULT);
 	for (ALL_LIST_ELEMENTS(zrouter.client_list, node, nnode, client)) {
+		/* Do not send unsolicited messages to synchronous clients. */
+		if (client->synchronous)
+			continue;
+
 		zsend_capabilities(client, zvrf);
 	}
 }
@@ -1751,13 +1765,18 @@ static void zread_hello(ZAPI_HANDLER_ARGS)
 	uint8_t proto;
 	unsigned short instance;
 	uint8_t notify;
+	uint8_t synchronous;
 
 	STREAM_GETC(msg, proto);
 	STREAM_GETW(msg, instance);
 	STREAM_GETC(msg, notify);
+	STREAM_GETC(msg, synchronous);
 	if (notify)
 		client->notify_owner = true;
 
+	if (synchronous)
+		client->synchronous = true;
+
 	/* accept only dynamic routing protocols */
 	if ((proto < ZEBRA_ROUTE_MAX) && (proto > ZEBRA_ROUTE_CONNECT)) {
 		zlog_notice(
@@ -1774,8 +1793,10 @@ static void zread_hello(ZAPI_HANDLER_ARGS)
 		zebra_gr_client_reconnect(client);
 	}
 
-	zsend_capabilities(client, zvrf);
-	zebra_vrf_update_all(client);
+	if (!client->synchronous) {
+		zsend_capabilities(client, zvrf);
+		zebra_vrf_update_all(client);
+	}
 stream_failure:
 	return;
 }

zebra/zebra_vrf.c

@@ -60,8 +60,13 @@ static void zebra_vrf_add_update(struct zebra_vrf *zvrf)
 	if (IS_ZEBRA_DEBUG_EVENT)
 		zlog_debug("MESSAGE: ZEBRA_VRF_ADD %s", zvrf_name(zvrf));
 
-	for (ALL_LIST_ELEMENTS(zrouter.client_list, node, nnode, client))
+	for (ALL_LIST_ELEMENTS(zrouter.client_list, node, nnode, client)) {
+		/* Do not send unsolicited messages to synchronous clients. */
+		if (client->synchronous)
+			continue;
+
 		zsend_vrf_add(client, zvrf);
+	}
 }
 
 static void zebra_vrf_delete_update(struct zebra_vrf *zvrf)
@@ -72,8 +77,13 @@ static void zebra_vrf_delete_update(struct zebra_vrf *zvrf)
 	if (IS_ZEBRA_DEBUG_EVENT)
 		zlog_debug("MESSAGE: ZEBRA_VRF_DELETE %s", zvrf_name(zvrf));
 
-	for (ALL_LIST_ELEMENTS(zrouter.client_list, node, nnode, client))
+	for (ALL_LIST_ELEMENTS(zrouter.client_list, node, nnode, client)) {
+		/* Do not send unsolicited messages to synchronous clients. */
+		if (client->synchronous)
+			continue;
+
 		zsend_vrf_delete(client, zvrf);
+	}
 }
 
 void zebra_vrf_update_all(struct zserv *client)

zebra/zserv.h

@@ -131,6 +131,9 @@ struct zserv {
 
 	bool notify_owner;
 
+	/* Indicates if client is synchronous. */
+	bool synchronous;
+
 	/* client's protocol */
 	uint8_t proto;
 	uint16_t instance;