Merge pull request #10621 from donaldsharp/cov_fun

[mirror_frr.git] / pimd / pim_zebra.c

/*
 * PIM for Quagga
 * Copyright (C) 2008  Everton da Silva Marques
 *
 * 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 "if.h"
#include "log.h"
#include "prefix.h"
#include "zclient.h"
#include "stream.h"
#include "network.h"
#include "vty.h"
#include "plist.h"
#include "lib/bfd.h"

#include "pimd.h"
#include "pim_pim.h"
#include "pim_zebra.h"
#include "pim_iface.h"
#include "pim_str.h"
#include "pim_oil.h"
#include "pim_rpf.h"
#include "pim_time.h"
#include "pim_join.h"
#include "pim_zlookup.h"
#include "pim_ifchannel.h"
#include "pim_rp.h"
#include "pim_igmpv3.h"
#include "pim_jp_agg.h"
#include "pim_nht.h"
#include "pim_ssm.h"
#include "pim_vxlan.h"
#include "pim_mlag.h"

#undef PIM_DEBUG_IFADDR_DUMP
#define PIM_DEBUG_IFADDR_DUMP

struct zclient *zclient;


/* Router-id update message from zebra. */
static int pim_router_id_update_zebra(ZAPI_CALLBACK_ARGS)
{
	struct prefix router_id;

	zebra_router_id_update_read(zclient->ibuf, &router_id);

	return 0;
}

static int pim_zebra_interface_vrf_update(ZAPI_CALLBACK_ARGS)
{
	struct interface *ifp;
	vrf_id_t new_vrf_id;

	ifp = zebra_interface_vrf_update_read(zclient->ibuf, vrf_id,
					      &new_vrf_id);
	if (!ifp)
		return 0;

	if (PIM_DEBUG_ZEBRA)
		zlog_debug("%s: %s updating from %u to %u", __func__, ifp->name,
			   vrf_id, new_vrf_id);

	if_update_to_new_vrf(ifp, new_vrf_id);

	return 0;
}

#ifdef PIM_DEBUG_IFADDR_DUMP
static void dump_if_address(struct interface *ifp)
{
	struct connected *ifc;
	struct listnode *node;

	zlog_debug("%s %s: interface %s addresses:", __FILE__, __func__,
		   ifp->name);

	for (ALL_LIST_ELEMENTS_RO(ifp->connected, node, ifc)) {
		struct prefix *p = ifc->address;

		if (p->family != AF_INET)
			continue;

		zlog_debug("%s %s: interface %s address %pI4 %s", __FILE__,
			   __func__, ifp->name, &p->u.prefix4,
			   CHECK_FLAG(ifc->flags, ZEBRA_IFA_SECONDARY)
				   ? "secondary"
				   : "primary");
	}
}
#endif

static int pim_zebra_if_address_add(ZAPI_CALLBACK_ARGS)
{
	struct connected *c;
	struct prefix *p;
	struct pim_interface *pim_ifp;
	struct pim_instance *pim;

	/*
	  zebra api notifies address adds/dels events by using the same call
	  interface_add_read below, see comments in lib/zclient.c

	  zebra_interface_address_read(ZEBRA_INTERFACE_ADDRESS_ADD, ...)
	  will add address to interface list by calling
	  connected_add_by_prefix()
	*/
	c = zebra_interface_address_read(cmd, zclient->ibuf, vrf_id);
	if (!c)
		return 0;

	pim_ifp = c->ifp->info;
	p = c->address;

	if (PIM_DEBUG_ZEBRA) {
		zlog_debug("%s: %s(%u) connected IP address %pFX flags %u %s",
			   __func__, c->ifp->name, vrf_id, p, c->flags,
			   CHECK_FLAG(c->flags, ZEBRA_IFA_SECONDARY)
				   ? "secondary"
				   : "primary");

#ifdef PIM_DEBUG_IFADDR_DUMP
		dump_if_address(c->ifp);
#endif
	}

	if (p->family != PIM_AF)
		SET_FLAG(c->flags, ZEBRA_IFA_SECONDARY);
	else if (!CHECK_FLAG(c->flags, ZEBRA_IFA_SECONDARY)) {
		/* trying to add primary address? */
		pim_addr primary_addr = pim_find_primary_addr(c->ifp);
		pim_addr addr = pim_addr_from_prefix(p);

		if (pim_addr_cmp(primary_addr, addr)) {
			if (PIM_DEBUG_ZEBRA)
				zlog_warn(
					"%s: %s : forcing secondary flag on %pFX",
					__func__, c->ifp->name, p);
			SET_FLAG(c->flags, ZEBRA_IFA_SECONDARY);
		}
	}

	pim_if_addr_add(c);
	if (pim_ifp) {
		pim = pim_get_pim_instance(vrf_id);
		pim_ifp->pim = pim;

		pim_rp_check_on_if_add(pim_ifp);
	}

	if (if_is_loopback(c->ifp)) {
		struct vrf *vrf = vrf_lookup_by_id(vrf_id);
		struct interface *ifp;

		FOR_ALL_INTERFACES (vrf, ifp) {
			if (!if_is_loopback(ifp) && if_is_operative(ifp))
				pim_if_addr_add_all(ifp);
		}
	}

	return 0;
}

static int pim_zebra_if_address_del(ZAPI_CALLBACK_ARGS)
{
	struct connected *c;
	struct prefix *p;
	struct vrf *vrf = vrf_lookup_by_id(vrf_id);
	struct pim_instance *pim;

	if (!vrf)
		return 0;
	pim = vrf->info;

	/*
	  zebra api notifies address adds/dels events by using the same call
	  interface_add_read below, see comments in lib/zclient.c

	  zebra_interface_address_read(ZEBRA_INTERFACE_ADDRESS_DELETE, ...)
	  will remove address from interface list by calling
	  connected_delete_by_prefix()
	*/
	c = zebra_interface_address_read(cmd, zclient->ibuf, vrf_id);
	if (!c)
		return 0;

	p = c->address;
	if (p->family == AF_INET) {
		if (PIM_DEBUG_ZEBRA) {
			zlog_debug(
				"%s: %s(%u) disconnected IP address %pFX flags %u %s",
				__func__, c->ifp->name, vrf_id, p, c->flags,
				CHECK_FLAG(c->flags, ZEBRA_IFA_SECONDARY)
					? "secondary"
					: "primary");

#ifdef PIM_DEBUG_IFADDR_DUMP
			dump_if_address(c->ifp);
#endif
		}

		pim_if_addr_del(c, 0);
		pim_rp_setup(pim);
		pim_i_am_rp_re_evaluate(pim);
	}

	connected_free(&c);
	return 0;
}

void pim_zebra_update_all_interfaces(struct pim_instance *pim)
{
	struct interface *ifp;

	FOR_ALL_INTERFACES (pim->vrf, ifp) {
		struct pim_interface *pim_ifp = ifp->info;
		struct pim_iface_upstream_switch *us;
		struct listnode *node;

		if (!pim_ifp)
			continue;

		for (ALL_LIST_ELEMENTS_RO(pim_ifp->upstream_switch_list, node,
					  us)) {
			struct pim_rpf rpf;

			rpf.source_nexthop.interface = ifp;
			rpf.rpf_addr.u.prefix4 = us->address;
			pim_joinprune_send(&rpf, us->us);
			pim_jp_agg_clear_group(us->us);
		}
	}
}

void pim_zebra_upstream_rpf_changed(struct pim_instance *pim,
				    struct pim_upstream *up,
				    struct pim_rpf *old)
{
	if (old->source_nexthop.interface) {
		struct pim_neighbor *nbr;

		nbr = pim_neighbor_find_prefix(old->source_nexthop.interface,
					       &old->rpf_addr);
		if (nbr)
			pim_jp_agg_remove_group(nbr->upstream_jp_agg, up, nbr);

		/*
		 * We have detected a case where we might need
		 * to rescan the inherited o_list so do it.
		 */
		if (up->channel_oil->oil_inherited_rescan) {
			pim_upstream_inherited_olist_decide(pim, up);
			up->channel_oil->oil_inherited_rescan = 0;
		}

		if (up->join_state == PIM_UPSTREAM_JOINED) {
			/*
			 * If we come up real fast we can be here
			 * where the mroute has not been installed
			 * so install it.
			 */
			if (!up->channel_oil->installed)
				pim_upstream_mroute_add(up->channel_oil,
							__func__);

			/*
			 * RFC 4601: 4.5.7.  Sending (S,G)
			 * Join/Prune Messages
			 *
			 * Transitions from Joined State
			 *
			 * RPF'(S,G) changes not due to an Assert
			 *
			 * The upstream (S,G) state machine remains
			 * in Joined state. Send Join(S,G) to the new
			 * upstream neighbor, which is the new value
			 * of RPF'(S,G).  Send Prune(S,G) to the old
			 * upstream neighbor, which is the old value
			 * of RPF'(S,G).  Set the Join Timer (JT) to
			 * expire after t_periodic seconds.
			 */
			pim_jp_agg_switch_interface(old, &up->rpf, up);

			pim_upstream_join_timer_restart(up, old);
		} /* up->join_state == PIM_UPSTREAM_JOINED */
	}

	else {
		/*
		 * We have detected a case where we might need
		 * to rescan the inherited o_list so do it.
		 */
		if (up->channel_oil->oil_inherited_rescan) {
			pim_upstream_inherited_olist_decide(pim, up);
			up->channel_oil->oil_inherited_rescan = 0;
		}

		if (up->join_state == PIM_UPSTREAM_JOINED)
			pim_jp_agg_switch_interface(old, &up->rpf, up);

		if (!up->channel_oil->installed)
			pim_upstream_mroute_add(up->channel_oil, __func__);
	}

	/* FIXME can join_desired actually be changed by pim_rpf_update()
	 * returning PIM_RPF_CHANGED ?
	 */
	pim_upstream_update_join_desired(pim, up);
}

static int pim_zebra_vxlan_sg_proc(ZAPI_CALLBACK_ARGS)
{
	struct stream *s;
	struct pim_instance *pim;
	pim_sgaddr sg;
	size_t prefixlen;

	pim = pim_get_pim_instance(vrf_id);
	if (!pim)
		return 0;

	s = zclient->ibuf;

	prefixlen = stream_getl(s);
	stream_get(&sg.src, s, prefixlen);
	stream_get(&sg.grp, s, prefixlen);

	if (PIM_DEBUG_ZEBRA)
		zlog_debug("%u:recv SG %s %pSG", vrf_id,
			   (cmd == ZEBRA_VXLAN_SG_ADD) ? "add" : "del", &sg);

	if (cmd == ZEBRA_VXLAN_SG_ADD)
		pim_vxlan_sg_add(pim, &sg);
	else
		pim_vxlan_sg_del(pim, &sg);

	return 0;
}

static void pim_zebra_vxlan_replay(void)
{
	struct stream *s = NULL;

	/* Check socket. */
	if (!zclient || zclient->sock < 0)
		return;

	s = zclient->obuf;
	stream_reset(s);

	zclient_create_header(s, ZEBRA_VXLAN_SG_REPLAY, VRF_DEFAULT);
	stream_putw_at(s, 0, stream_get_endp(s));

	zclient_send_message(zclient);
}

void pim_scan_oil(struct pim_instance *pim)
{
	struct channel_oil *c_oil;

	pim->scan_oil_last = pim_time_monotonic_sec();
	++pim->scan_oil_events;

	frr_each (rb_pim_oil, &pim->channel_oil_head, c_oil)
		pim_upstream_mroute_iif_update(c_oil, __func__);
}

static int on_rpf_cache_refresh(struct thread *t)
{
	struct pim_instance *pim = THREAD_ARG(t);

	/* update kernel multicast forwarding cache (MFC) */
	pim_scan_oil(pim);

	pim->rpf_cache_refresh_last = pim_time_monotonic_sec();
	++pim->rpf_cache_refresh_events;

	// It is called as part of pim_neighbor_add
	// pim_rp_setup ();
	return 0;
}

void sched_rpf_cache_refresh(struct pim_instance *pim)
{
	++pim->rpf_cache_refresh_requests;

	pim_rpf_set_refresh_time(pim);

	if (pim->rpf_cache_refresher) {
		/* Refresh timer is already running */
		return;
	}

	/* Start refresh timer */

	if (PIM_DEBUG_ZEBRA) {
		zlog_debug("%s: triggering %ld msec timer", __func__,
			   router->rpf_cache_refresh_delay_msec);
	}

	thread_add_timer_msec(router->master, on_rpf_cache_refresh, pim,
			      router->rpf_cache_refresh_delay_msec,
			      &pim->rpf_cache_refresher);
}

static void pim_zebra_connected(struct zclient *zclient)
{
	/* Send the client registration */
	bfd_client_sendmsg(zclient, ZEBRA_BFD_CLIENT_REGISTER, router->vrf_id);

	zclient_send_reg_requests(zclient, router->vrf_id);

	/* request for VxLAN BUM group addresses */
	pim_zebra_vxlan_replay();
}

static void pim_zebra_capabilities(struct zclient_capabilities *cap)
{
	router->mlag_role = cap->role;
}

static zclient_handler *const pim_handlers[] = {
	[ZEBRA_ROUTER_ID_UPDATE] = pim_router_id_update_zebra,
	[ZEBRA_INTERFACE_ADDRESS_ADD] = pim_zebra_if_address_add,
	[ZEBRA_INTERFACE_ADDRESS_DELETE] = pim_zebra_if_address_del,
	[ZEBRA_INTERFACE_VRF_UPDATE] = pim_zebra_interface_vrf_update,
	[ZEBRA_NEXTHOP_UPDATE] = pim_parse_nexthop_update,

	[ZEBRA_VXLAN_SG_ADD] = pim_zebra_vxlan_sg_proc,
	[ZEBRA_VXLAN_SG_DEL] = pim_zebra_vxlan_sg_proc,

	[ZEBRA_MLAG_PROCESS_UP] = pim_zebra_mlag_process_up,
	[ZEBRA_MLAG_PROCESS_DOWN] = pim_zebra_mlag_process_down,
	[ZEBRA_MLAG_FORWARD_MSG] = pim_zebra_mlag_handle_msg,
};

void pim_zebra_init(void)
{
	/* Socket for receiving updates from Zebra daemon */
	zclient = zclient_new(router->master, &zclient_options_default,
			      pim_handlers, array_size(pim_handlers));

	zclient->zebra_capabilities = pim_zebra_capabilities;
	zclient->zebra_connected = pim_zebra_connected;

	zclient_init(zclient, ZEBRA_ROUTE_PIM, 0, &pimd_privs);
	if (PIM_DEBUG_PIM_TRACE) {
		zlog_notice("%s: zclient socket initialized", __func__);
	}

	zclient_lookup_new();
}

void igmp_anysource_forward_start(struct pim_instance *pim,
				  struct gm_group *group)
{
	struct gm_source *source;
	struct in_addr src_addr = {.s_addr = 0};
	/* Any source (*,G) is forwarded only if mode is EXCLUDE {empty} */
	assert(group->group_filtermode_isexcl);
	assert(listcount(group->group_source_list) < 1);

	source = igmp_get_source_by_addr(group, src_addr, NULL);
	if (!source) {
		zlog_warn("%s: Failure to create * source", __func__);
		return;
	}

	igmp_source_forward_start(pim, source);
}

void igmp_anysource_forward_stop(struct gm_group *group)
{
	struct gm_source *source;
	struct in_addr star = {.s_addr = 0};

	source = igmp_find_source_by_addr(group, star);
	if (source)
		igmp_source_forward_stop(source);
}

static void igmp_source_forward_reevaluate_one(struct pim_instance *pim,
					       struct gm_source *source)
{
	pim_sgaddr sg;
	struct gm_group *group = source->source_group;
	struct pim_ifchannel *ch;

	if ((source->source_addr.s_addr != INADDR_ANY)
	    || !IGMP_SOURCE_TEST_FORWARDING(source->source_flags))
		return;

	memset(&sg, 0, sizeof(sg));
	sg.src = source->source_addr;
	sg.grp = group->group_addr;

	ch = pim_ifchannel_find(group->interface, &sg);
	if (pim_is_grp_ssm(pim, group->group_addr)) {
		/* If SSM group withdraw local membership */
		if (ch
		    && (ch->local_ifmembership == PIM_IFMEMBERSHIP_INCLUDE)) {
			if (PIM_DEBUG_PIM_EVENTS)
				zlog_debug("local membership del for %pSG as G is now SSM",
					   &sg);
			pim_ifchannel_local_membership_del(group->interface,
							   &sg);
		}
	} else {
		/* If ASM group add local membership */
		if (!ch
		    || (ch->local_ifmembership == PIM_IFMEMBERSHIP_NOINFO)) {
			if (PIM_DEBUG_PIM_EVENTS)
				zlog_debug("local membership add for %pSG as G is now ASM",
					   &sg);
			pim_ifchannel_local_membership_add(
				group->interface, &sg, false /*is_vxlan*/);
		}
	}
}

void igmp_source_forward_reevaluate_all(struct pim_instance *pim)
{
	struct interface *ifp;

	FOR_ALL_INTERFACES (pim->vrf, ifp) {
		struct pim_interface *pim_ifp = ifp->info;
		struct listnode *grpnode;
		struct gm_group *grp;
		struct pim_ifchannel *ch, *ch_temp;

		if (!pim_ifp)
			continue;

		/* scan igmp groups */
		for (ALL_LIST_ELEMENTS_RO(pim_ifp->gm_group_list, grpnode,
					  grp)) {
			struct listnode *srcnode;
			struct gm_source *src;

			/* scan group sources */
			for (ALL_LIST_ELEMENTS_RO(grp->group_source_list,
						  srcnode, src)) {
				igmp_source_forward_reevaluate_one(pim, src);
			} /* scan group sources */
		}	 /* scan igmp groups */

		RB_FOREACH_SAFE (ch, pim_ifchannel_rb, &pim_ifp->ifchannel_rb,
				 ch_temp) {
			if (pim_is_grp_ssm(pim, ch->sg.grp)) {
				if (pim_addr_is_any(ch->sg.src))
					pim_ifchannel_delete(ch);
			}
		}
	} /* scan interfaces */
}

void igmp_source_forward_start(struct pim_instance *pim,
			       struct gm_source *source)
{
	struct pim_interface *pim_oif;
	struct gm_group *group;
	pim_sgaddr sg;
	int result;
	int input_iface_vif_index = 0;

	memset(&sg, 0, sizeof(sg));
	sg.src = source->source_addr;
	sg.grp = source->source_group->group_addr;

	if (PIM_DEBUG_IGMP_TRACE) {
		zlog_debug("%s: (S,G)=%pSG oif=%s fwd=%d", __func__, &sg,
			   source->source_group->interface->name,
			   IGMP_SOURCE_TEST_FORWARDING(source->source_flags));
	}

	/* Prevent IGMP interface from installing multicast route multiple
	   times */
	if (IGMP_SOURCE_TEST_FORWARDING(source->source_flags)) {
		return;
	}

	group = source->source_group;
	pim_oif = group->interface->info;
	if (!pim_oif) {
		if (PIM_DEBUG_IGMP_TRACE) {
			zlog_debug("%s: multicast not enabled on oif=%s ?",
				   __func__,
				   source->source_group->interface->name);
		}
		return;
	}

	if (!source->source_channel_oil) {
		pim_addr vif_source;
		struct prefix src, grp;
		struct pim_nexthop nexthop;
		struct pim_upstream *up = NULL;

		if (!pim_rp_set_upstream_addr(pim, &vif_source,
					      source->source_addr, sg.grp)) {
			/*Create a dummy channel oil */
			source->source_channel_oil =
				pim_channel_oil_add(pim, &sg, __func__);
		}

		else {
			pim_addr_to_prefix(&src, vif_source); // RP or Src addr
			pim_addr_to_prefix(&grp, sg.grp);

			up = pim_upstream_find(pim, &sg);
			if (up) {
				memcpy(&nexthop, &up->rpf.source_nexthop,
				       sizeof(struct pim_nexthop));
				pim_ecmp_nexthop_lookup(pim, &nexthop, &src,
							&grp, 0);
				if (nexthop.interface)
					input_iface_vif_index =
						pim_if_find_vifindex_by_ifindex(
							pim,
							nexthop.interface->ifindex);
			} else
				input_iface_vif_index =
					pim_ecmp_fib_lookup_if_vif_index(
						pim, &src, &grp);

			if (PIM_DEBUG_ZEBRA)
				zlog_debug(
					"%s: NHT %pSG vif_source %pPAs vif_index:%d ",
					__func__, &sg, &vif_source,
					input_iface_vif_index);

			if (input_iface_vif_index < 1) {
				if (PIM_DEBUG_IGMP_TRACE) {
					char source_str[INET_ADDRSTRLEN];
					pim_inet4_dump("<source?>",
						source->source_addr,
						source_str, sizeof(source_str));
					zlog_debug(
						"%s %s: could not find input interface for source %s",
						__FILE__, __func__, source_str);
				}
				source->source_channel_oil =
					pim_channel_oil_add(pim, &sg, __func__);
			}

			else {
				/*
				 * Protect IGMP against adding looped MFC
				 * entries created by both source and receiver
				 * attached to the same interface. See TODO
				 * T22. Block only when the intf is non DR
				 * DR must create upstream.
				 */
				if ((input_iface_vif_index ==
				    pim_oif->mroute_vif_index) &&
				    !(PIM_I_am_DR(pim_oif))) {
					/* ignore request for looped MFC entry
					 */
					if (PIM_DEBUG_IGMP_TRACE) {
						zlog_debug("%s: ignoring request for looped MFC entry (S,G)=%pSG: oif=%s vif_index=%d",
							   __func__,
							   &sg,
							   source->source_group
							   ->interface->name,
							   input_iface_vif_index);
					}
					return;
				}

				source->source_channel_oil =
					pim_channel_oil_add(pim, &sg, __func__);
				if (!source->source_channel_oil) {
					if (PIM_DEBUG_IGMP_TRACE) {
						zlog_debug("%s %s: could not create OIL for channel (S,G)=%pSG",
							   __FILE__, __func__,
							   &sg);
					}
					return;
				}
			}
		}
	}

	if (PIM_I_am_DR(pim_oif) || PIM_I_am_DualActive(pim_oif)) {
		result = pim_channel_add_oif(source->source_channel_oil,
					     group->interface,
					     PIM_OIF_FLAG_PROTO_IGMP, __func__);
		if (result) {
			if (PIM_DEBUG_MROUTE) {
				zlog_warn("%s: add_oif() failed with return=%d",
					  __func__, result);
			}
			return;
		}
	} else {
		if (PIM_DEBUG_IGMP_TRACE)
			zlog_debug("%s: %pSG was received on %s interface but we are not DR for that interface",
				   __func__, &sg,
				   group->interface->name);

		return;
	}
	/*
	  Feed IGMPv3-gathered local membership information into PIM
	  per-interface (S,G) state.
	 */
	if (!pim_ifchannel_local_membership_add(group->interface, &sg,
						false /*is_vxlan*/)) {
		if (PIM_DEBUG_MROUTE)
			zlog_warn("%s: Failure to add local membership for %pSG",
				  __func__, &sg);

		pim_channel_del_oif(source->source_channel_oil,
				    group->interface, PIM_OIF_FLAG_PROTO_IGMP,
				    __func__);
		return;
	}

	IGMP_SOURCE_DO_FORWARDING(source->source_flags);
}

/*
  igmp_source_forward_stop: stop fowarding, but keep the source
  igmp_source_delete:       stop fowarding, and delete the source
 */
void igmp_source_forward_stop(struct gm_source *source)
{
	struct gm_group *group;
	pim_sgaddr sg;
	int result;

	memset(&sg, 0, sizeof(sg));
	sg.src = source->source_addr;
	sg.grp = source->source_group->group_addr;

	if (PIM_DEBUG_IGMP_TRACE) {
		zlog_debug("%s: (S,G)=%pSG oif=%s fwd=%d", __func__, &sg,
			   source->source_group->interface->name,
			   IGMP_SOURCE_TEST_FORWARDING(source->source_flags));
	}

	/* Prevent IGMP interface from removing multicast route multiple
	   times */
	if (!IGMP_SOURCE_TEST_FORWARDING(source->source_flags)) {
		return;
	}

	group = source->source_group;

	/*
	 It appears that in certain circumstances that
	 igmp_source_forward_stop is called when IGMP forwarding
	 was not enabled in oif_flags for this outgoing interface.
	 Possibly because of multiple calls. When that happens, we
	 enter the below if statement and this function returns early
	 which in turn triggers the calling function to assert.
	 Making the call to pim_channel_del_oif and ignoring the return code
	 fixes the issue without ill effect, similar to
	 pim_forward_stop below.
	*/
	result = pim_channel_del_oif(source->source_channel_oil,
				     group->interface, PIM_OIF_FLAG_PROTO_IGMP,
				     __func__);
	if (result) {
		if (PIM_DEBUG_IGMP_TRACE)
			zlog_debug(
				"%s: pim_channel_del_oif() failed with return=%d",
				__func__, result);
		return;
	}

	/*
	  Feed IGMPv3-gathered local membership information into PIM
	  per-interface (S,G) state.
	 */
	pim_ifchannel_local_membership_del(group->interface, &sg);

	IGMP_SOURCE_DONT_FORWARDING(source->source_flags);
}

void pim_forward_start(struct pim_ifchannel *ch)
{
	struct pim_upstream *up = ch->upstream;
	uint32_t mask = 0;

	if (PIM_DEBUG_PIM_TRACE)
		zlog_debug("%s: (S,G)=%pSG oif=%s (%pI4)", __func__, &ch->sg,
			   ch->interface->name, &up->upstream_addr);

	if (PIM_IF_FLAG_TEST_PROTO_IGMP(ch->flags))
		mask = PIM_OIF_FLAG_PROTO_IGMP;

	if (PIM_IF_FLAG_TEST_PROTO_PIM(ch->flags))
		mask |= PIM_OIF_FLAG_PROTO_PIM;

	pim_channel_add_oif(up->channel_oil, ch->interface,
			mask, __func__);
}

void pim_forward_stop(struct pim_ifchannel *ch)
{
	struct pim_upstream *up = ch->upstream;

	if (PIM_DEBUG_PIM_TRACE) {
		zlog_debug("%s: (S,G)=%s oif=%s installed: %d",
			   __func__, ch->sg_str, ch->interface->name,
			   up->channel_oil->installed);
	}

	/*
	 * If a channel is being removed, check to see if we still need
	 * to inherit the interface.  If so make sure it is added in
	 */
	if (pim_upstream_evaluate_join_desired_interface(up, ch, ch->parent))
		pim_channel_add_oif(up->channel_oil, ch->interface,
				    PIM_OIF_FLAG_PROTO_PIM, __func__);
	else
		pim_channel_del_oif(up->channel_oil, ch->interface,
				    PIM_OIF_FLAG_PROTO_PIM, __func__);
}

void pim_zebra_zclient_update(struct vty *vty)
{
	vty_out(vty, "Zclient update socket: ");

	if (zclient) {
		vty_out(vty, "%d failures=%d\n", zclient->sock, zclient->fail);
	} else {
		vty_out(vty, "<null zclient>\n");
	}
}

struct zclient *pim_zebra_zclient_get(void)
{
	if (zclient)
		return zclient;
	else
		return NULL;
}

void pim_zebra_interface_set_master(struct interface *vrf,
				    struct interface *ifp)
{
	zclient_interface_set_master(zclient, vrf, ifp);
}