pimd: Join/Prune Aggregation

Add the ability for PIM to send Join/Prunes as an
aggregated message instead of individual messages
for each S,G.

Signed-off-by: Donald Sharp <sharpd@cumulusnetworks.com>

20 files changed, 698 insertions, 215 deletions

diff --git a/pimd/Makefile.am b/pimd/Makefile.am
index 5c40d2ac64..59abd1aa30 100644
--- a/pimd/Makefile.am
+++ b/pimd/Makefile.am
@@ -52,7 +52,8 @@ libpim_a_SOURCES = \
 	pim_msg.c pim_upstream.c pim_rpf.c pim_macro.c \
 	pim_ssmpingd.c pim_int.c pim_rp.c \
 	pim_static.c pim_br.c pim_register.c pim_routemap.c \
-	pim_msdp.c pim_msdp_socket.c pim_msdp_packet.c
+	pim_msdp.c pim_msdp_socket.c pim_msdp_packet.c \
+	pim_jp_agg.c
 
 noinst_HEADERS = \
 	pim_memory.h \
@@ -64,7 +65,8 @@ noinst_HEADERS = \
 	pim_msg.h pim_upstream.h pim_rpf.h pim_macro.h \
 	pim_igmp_join.h pim_ssmpingd.h pim_int.h pim_rp.h \
 	pim_static.h pim_br.h pim_register.h \
-	pim_msdp.h pim_msdp_socket.h pim_msdp_packet.h
+	pim_msdp.h pim_msdp_socket.h pim_msdp_packet.h \
+	pim_jp_agg.h
 
 pimd_SOURCES = \
 	pim_main.c $(libpim_a_SOURCES)
diff --git a/pimd/pim_cmd.c b/pimd/pim_cmd.c
index 62d8ad8e07..6449b27e14 100644
--- a/pimd/pim_cmd.c
+++ b/pimd/pim_cmd.c
@@ -1672,6 +1672,20 @@ static void pim_show_upstream(struct vty *vty, u_char uj)
     pim_inet4_dump("<grp?>", up->sg.grp, grp_str, sizeof(grp_str));
     pim_time_uptime(uptime, sizeof(uptime), now - up->state_transition);
     pim_time_timer_to_hhmmss (join_timer, sizeof(join_timer), up->t_join_timer);
+
+    /*
+     * If we have a J/P timer for the neighbor display that
+     */
+    if (!up->t_join_timer)
+      {
+        struct pim_neighbor *nbr;
+
+        nbr = pim_neighbor_find (up->rpf.source_nexthop.interface,
+                                 up->rpf.rpf_addr.u.prefix4);
+        if (nbr)
+          pim_time_timer_to_hhmmss (join_timer, sizeof(join_timer), nbr->jp_timer);
+      }
+
     pim_time_timer_to_hhmmss (rs_timer, sizeof (rs_timer), up->t_rs_timer);
     pim_time_timer_to_hhmmss (ka_timer, sizeof (ka_timer), up->t_ka_timer);
     pim_time_timer_to_hhmmss (msdp_reg_timer, sizeof (msdp_reg_timer), up->t_msdp_reg_timer);
diff --git a/pimd/pim_iface.c b/pimd/pim_iface.c
index 8f7d40bb36..3d416409ed 100644
--- a/pimd/pim_iface.c
+++ b/pimd/pim_iface.c
@@ -83,12 +83,15 @@ static void *if_list_clean(struct pim_interface *pim_ifp)
     list_delete(pim_ifp->pim_neighbor_list);
   }
 
+  if (pim_ifp->upstream_switch_list)
+    list_delete(pim_ifp->upstream_switch_list);
+
   if (pim_ifp->pim_ifchannel_list) {
     list_delete(pim_ifp->pim_ifchannel_list);
   }
 
   if (pim_ifp->pim_ifchannel_hash)
-    hash_free (pim_ifp->pim_ifchannel_hash);
+    hash_free(pim_ifp->pim_ifchannel_hash);
 
   XFREE(MTYPE_PIM_INTERFACE, pim_ifp);
 
@@ -134,6 +137,7 @@ struct pim_interface *pim_if_new(struct interface *ifp, int igmp, int pim)
   pim_ifp->igmp_join_list = NULL;
   pim_ifp->igmp_socket_list = NULL;
   pim_ifp->pim_neighbor_list = NULL;
+  pim_ifp->upstream_switch_list = NULL;
   pim_ifp->pim_ifchannel_list = NULL;
   pim_ifp->pim_ifchannel_hash = NULL;
   pim_ifp->pim_generation_id = 0;
@@ -156,6 +160,13 @@ struct pim_interface *pim_if_new(struct interface *ifp, int igmp, int pim)
   }
   pim_ifp->pim_neighbor_list->del = (void (*)(void *)) pim_neighbor_free;
 
+  pim_ifp->upstream_switch_list = list_new();
+  if (!pim_ifp->upstream_switch_list) {
+    zlog_err("%s %s: failure: upstream_switch_list=list_new()",
+             __FILE__, __PRETTY_FUNCTION__);
+    return if_list_clean(pim_ifp);
+  }
+
   /* list of struct pim_ifchannel */
   pim_ifp->pim_ifchannel_list = list_new();
   if (!pim_ifp->pim_ifchannel_list) {
@@ -203,6 +214,7 @@ void pim_if_delete(struct interface *ifp)
 
   list_delete(pim_ifp->igmp_socket_list);
   list_delete(pim_ifp->pim_neighbor_list);
+  list_delete(pim_ifp->upstream_switch_list);
   list_delete(pim_ifp->pim_ifchannel_list);
 
   hash_free (pim_ifp->pim_ifchannel_hash);
diff --git a/pimd/pim_iface.h b/pimd/pim_iface.h
index 7c0d57a5d1..8d332c70b0 100644
--- a/pimd/pim_iface.h
+++ b/pimd/pim_iface.h
@@ -53,6 +53,11 @@
 
 #define PIM_I_am_DR(pim_ifp) (pim_ifp)->pim_dr_addr.s_addr == (pim_ifp)->primary_address.s_addr
 
+struct pim_iface_upstream_switch {
+  struct in_addr address;
+  struct list *us;
+};
+
 enum pim_interface_type {
   PIM_INTERFACE_SSM,
   PIM_INTERFACE_SM
@@ -97,6 +102,7 @@ struct pim_interface {
   uint16_t       pim_propagation_delay_msec; /* config */
   uint16_t       pim_override_interval_msec; /* config */
   struct list   *pim_neighbor_list; /* list of struct pim_neighbor */
+  struct list   *upstream_switch_list;
   struct list   *pim_ifchannel_list; /* list of struct pim_ifchannel */
   struct hash   *pim_ifchannel_hash;
 
diff --git a/pimd/pim_ifchannel.c b/pimd/pim_ifchannel.c
index 891bdc448d..ee75793024 100644
--- a/pimd/pim_ifchannel.c
+++ b/pimd/pim_ifchannel.c
@@ -602,7 +602,7 @@ static int on_ifjoin_prune_pending_timer(struct thread *t)
 
           rpf.source_nexthop.interface = ifp;
           rpf.rpf_addr.u.prefix4 = pim_ifp->primary_address;
-          pim_joinprune_send (&rpf, ch->upstream, 0);
+          pim_jp_agg_single_upstream_send(&rpf, ch->upstream, 0);
         }
     }
   else
diff --git a/pimd/pim_join.c b/pimd/pim_join.c
index 783dd75075..c19468da6f 100644
--- a/pimd/pim_join.c
+++ b/pimd/pim_join.c
@@ -38,6 +38,7 @@
 #include "pim_ifchannel.h"
 #include "pim_rpf.h"
 #include "pim_rp.h"
+#include "pim_jp_agg.h"
 
 static void
 on_trace (const char *label,
@@ -303,13 +304,86 @@ int pim_joinprune_recv(struct interface *ifp,
   return 0;
 }
 
+/*
+ * J/P Message Format
+ *
+ * While the RFC clearly states that this is 32 bits wide, it
+ * is cheating.  These fields:
+ * Encoded-Unicast format   (6 bytes MIN)
+ * Encoded-Group format     (8 bytes MIN)
+ * Encoded-Source format    (8 bytes MIN)
+ * are *not* 32 bits wide.
+ *
+ * Nor does the RFC explicitly call out the size for:
+ * Reserved                 (1 byte)
+ * Num Groups               (1 byte)
+ * Holdtime                 (2 bytes)
+ * Number of Joined Sources (2 bytes)
+ * Number of Pruned Sources (2 bytes)
+ *
+ * This leads to a missleading representation from casual
+ * reading and making assumptions.  Be careful!
+ *
+ *   0                   1                   2                   3
+ *   0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+ *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ *  |PIM Ver| Type  |   Reserved    |           Checksum            |
+ *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ *  |        Upstream Neighbor Address (Encoded-Unicast format)     |
+ *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ *  |  Reserved     | Num groups    |          Holdtime             |
+ *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ *  |         Multicast Group Address 1 (Encoded-Group format)      |
+ *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ *  |   Number of Joined Sources    |   Number of Pruned Sources    |
+ *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ *  |        Joined Source Address 1 (Encoded-Source format)        |
+ *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ *  |                             .                                 |
+ *  |                             .                                 |
+ *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ *  |        Joined Source Address n (Encoded-Source format)        |
+ *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ *  |        Pruned Source Address 1 (Encoded-Source format)        |
+ *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ *  |                             .                                 |
+ *  |                             .                                 |
+ *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ *  |        Pruned Source Address n (Encoded-Source format)        |
+ *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ *  |         Multicast Group Address m (Encoded-Group format)      |
+ *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ *  |   Number of Joined Sources    |   Number of Pruned Sources    |
+ *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ *  |        Joined Source Address 1 (Encoded-Source format)        |
+ *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ *  |                             .                                 |
+ *  |                             .                                 |
+ *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ *  |        Joined Source Address n (Encoded-Source format)        |
+ *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ *  |        Pruned Source Address 1 (Encoded-Source format)        |
+ *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ *  |                             .                                 |
+ *  |                             .                                 |
+ *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ *  |        Pruned Source Address n (Encoded-Source format)        |
+ *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ */
 int pim_joinprune_send(struct pim_rpf *rpf,
-                       struct pim_upstream *up,
-                       int send_join)
+                       struct list *groups)
 {
+  struct pim_jp_agg_group *group;
   struct pim_interface *pim_ifp;
-  uint8_t pim_msg[9000];
-  int pim_msg_size;
+  struct pim_jp_groups *grp = NULL;
+  struct pim_jp *msg;
+  struct listnode *node, *nnode;
+  uint8_t pim_msg[10000];
+  uint8_t *curr_ptr = pim_msg;
+  bool new_packet = true;
+  size_t packet_left = 0;
+  size_t packet_size = 0;
+  size_t group_size = 0;
 
   on_trace (__PRETTY_FUNCTION__, rpf->source_nexthop.interface, rpf->rpf_addr.u.prefix4);
 
@@ -322,17 +396,17 @@ int pim_joinprune_send(struct pim_rpf *rpf,
     return -1;
   }
 
-  if (PIM_INADDR_IS_ANY(rpf->rpf_addr.u.prefix4)) {
-    if (PIM_DEBUG_PIM_J_P) {
-      char dst_str[INET_ADDRSTRLEN];
-      pim_inet4_dump("<dst?>", rpf->rpf_addr.u.prefix4, dst_str, sizeof(dst_str));
-      zlog_debug("%s: %s(S,G)=%s: upstream=%s is myself on interface %s",
-		 __PRETTY_FUNCTION__,
-		 send_join ? "Join" : "Prune",
-		 up->sg_str, dst_str, rpf->source_nexthop.interface->name);
+  if (PIM_INADDR_IS_ANY(rpf->rpf_addr.u.prefix4))
+    {
+      if (PIM_DEBUG_PIM_J_P) {
+        char dst_str[INET_ADDRSTRLEN];
+        pim_inet4_dump("<dst?>", rpf->rpf_addr.u.prefix4, dst_str, sizeof(dst_str));
+        zlog_debug("%s: upstream=%s is myself on interface %s",
+                   __PRETTY_FUNCTION__,
+                   dst_str, rpf->source_nexthop.interface->name);
+      }
+      return 0;
     }
-    return 0;
-  }
 
   /*
     RFC 4601: 4.3.1.  Sending Hello Messages
@@ -345,34 +419,115 @@ int pim_joinprune_send(struct pim_rpf *rpf,
   */
   pim_hello_require(rpf->source_nexthop.interface);
 
-  /*
-    Build PIM message
-  */
-  pim_msg_size = pim_msg_join_prune_encode (pim_msg, 9000, send_join,
-					    up, rpf->rpf_addr.u.prefix4, PIM_JP_HOLDTIME);
+  for (ALL_LIST_ELEMENTS(groups, node, nnode, group))
+    {
+      if (new_packet)
+        {
+          msg = (struct pim_jp *)pim_msg;
 
-  if (pim_msg_size < 0)
-    return pim_msg_size;
+          memset(msg, 0, sizeof (*msg));
 
-  if (PIM_DEBUG_PIM_J_P) {
-    char dst_str[INET_ADDRSTRLEN];
-    pim_inet4_dump("<dst?>", rpf->rpf_addr.u.prefix4, dst_str, sizeof(dst_str));
-    zlog_debug("%s: sending %s(S,G)=%s to upstream=%s on interface %s",
-	       __PRETTY_FUNCTION__,
-	       send_join ? "Join" : "Prune",
-	       up->sg_str, dst_str, rpf->source_nexthop.interface->name);
-  }
+          pim_msg_addr_encode_ipv4_ucast ((uint8_t *)&msg->addr, rpf->rpf_addr.u.prefix4);
+          msg->reserved   = 0;
+          msg->holdtime   = htons(PIM_JP_HOLDTIME);
 
-  if (pim_msg_send(pim_ifp->pim_sock_fd,
-		   pim_ifp->primary_address,
-		   qpim_all_pim_routers_addr,
-		   pim_msg,
-		   pim_msg_size,
-		   rpf->source_nexthop.interface->name)) {
-    zlog_warn("%s: could not send PIM message on interface %s",
-	      __PRETTY_FUNCTION__, rpf->source_nexthop.interface->name);
-    return -8;
-  }
+          new_packet = false;
+
+          grp = &msg->groups[0];
+          curr_ptr = (uint8_t *)grp;
+          packet_size  = sizeof (struct pim_msg_header);
+          packet_size += sizeof (struct pim_encoded_ipv4_unicast);
+          packet_size += 4;  // reserved (1) + groups (1) + holdtime (2)
+
+          packet_left = rpf->source_nexthop.interface->mtu - 24;
+          packet_left -= packet_size;
+        }
+      if (PIM_DEBUG_PIM_J_P) {
+        char dst_str[INET_ADDRSTRLEN];
+        char grp_str[INET_ADDRSTRLEN];
+        pim_inet4_dump("<dst?>", rpf->rpf_addr.u.prefix4, dst_str, sizeof(dst_str));
+        pim_inet4_dump("<grp?>", group->group, grp_str, sizeof(grp_str));
+        zlog_debug("%s: sending (G)=%s to upstream=%s on interface %s",
+                   __PRETTY_FUNCTION__,
+                   grp_str, dst_str, rpf->source_nexthop.interface->name);
+      }
+
+      group_size = pim_msg_get_jp_group_size (group->sources);
+      if (group_size > packet_left)
+        {
+          pim_msg_build_header (pim_msg, packet_size, PIM_MSG_TYPE_JOIN_PRUNE);
+          if (pim_msg_send(pim_ifp->pim_sock_fd,
+                           pim_ifp->primary_address,
+                           qpim_all_pim_routers_addr,
+                           pim_msg,
+                           packet_size,
+                           rpf->source_nexthop.interface->name)) {
+            zlog_warn("%s: could not send PIM message on interface %s",
+                      __PRETTY_FUNCTION__, rpf->source_nexthop.interface->name);
+          }
+
+          msg = (struct pim_jp *)pim_msg;
+          memset(msg, 0, sizeof (*msg));
+
+          pim_msg_addr_encode_ipv4_ucast ((uint8_t *)&msg->addr, rpf->rpf_addr.u.prefix4);
+          msg->reserved   = 0;
+          msg->holdtime   = htons(PIM_JP_HOLDTIME);
+
+          new_packet = false;
+
+          grp = &msg->groups[0];
+          curr_ptr = (uint8_t *)grp;
+          packet_size  = sizeof (struct pim_msg_header);
+          packet_size += sizeof (struct pim_encoded_ipv4_unicast);
+          packet_size += 4;  // reserved (1) + groups (1) + holdtime (2)
+
+          packet_left = rpf->source_nexthop.interface->mtu - 24;
+          packet_left -= packet_size;
+        }
 
+      msg->num_groups++;
+      /*
+        Build PIM message
+      */
+
+      curr_ptr += group_size;
+      packet_left -= group_size;
+      packet_size += group_size;
+      zlog_debug ("\tpl: %zd ps: %zd", packet_left, packet_size);
+      pim_msg_build_jp_groups (grp, group);
+
+      grp = (struct pim_jp_groups *)curr_ptr;
+      if (packet_left < sizeof (struct pim_jp_groups) || msg->num_groups == 255)
+        {
+          pim_msg_build_header (pim_msg, packet_size, PIM_MSG_TYPE_JOIN_PRUNE);
+          if (pim_msg_send(pim_ifp->pim_sock_fd,
+                           pim_ifp->primary_address,
+                           qpim_all_pim_routers_addr,
+                           pim_msg,
+                           packet_size,
+                           rpf->source_nexthop.interface->name)) {
+            zlog_warn("%s: could not send PIM message on interface %s",
+                      __PRETTY_FUNCTION__, rpf->source_nexthop.interface->name);
+          }
+
+          new_packet = true;
+        }
+    }
+
+
+  if (!new_packet)
+    {
+      //msg->num_groups = htons (msg->num_groups);
+      pim_msg_build_header (pim_msg, packet_size, PIM_MSG_TYPE_JOIN_PRUNE);
+      if (pim_msg_send(pim_ifp->pim_sock_fd,
+                       pim_ifp->primary_address,
+                       qpim_all_pim_routers_addr,
+                       pim_msg,
+                       packet_size,
+                       rpf->source_nexthop.interface->name)) {
+        zlog_warn("%s: could not send PIM message on interface %s",
+                  __PRETTY_FUNCTION__, rpf->source_nexthop.interface->name);
+      }
+    }
   return 0;
 }
diff --git a/pimd/pim_join.h b/pimd/pim_join.h
index 4b76166328..adedde3cf8 100644
--- a/pimd/pim_join.h
+++ b/pimd/pim_join.h
@@ -33,7 +33,6 @@ int pim_joinprune_recv(struct interface *ifp,
 		       uint8_t *tlv_buf, int tlv_buf_size);
 
 int pim_joinprune_send(struct pim_rpf *nexthop,
-                       struct pim_upstream *up,
-                       int send_join);
+                       struct list *groups);
 
 #endif /* PIM_JOIN_H */
diff --git a/pimd/pim_jp_agg.c b/pimd/pim_jp_agg.c
new file mode 100644
index 0000000000..0799d6b4b4
--- /dev/null
+++ b/pimd/pim_jp_agg.c
@@ -0,0 +1,228 @@
+#include <zebra.h>
+
+#include "linklist.h"
+#include "log.h"
+
+#include "pimd.h"
+#include "pim_msg.h"
+#include "pim_jp_agg.h"
+#include "pim_join.h"
+#include "pim_iface.h"
+
+void
+pim_jp_agg_group_list_free (struct pim_jp_agg_group *jag)
+{
+  list_delete(jag->sources);
+
+  XFREE (MTYPE_PIM_JP_AGG_GROUP, jag);
+}
+
+static void
+pim_jp_agg_src_free (struct pim_jp_sources *js)
+{
+  /*
+   * When we are being called here, we know
+   * that the neighbor is going away start
+   * the normal j/p timer so that it can
+   * pick this shit back up when the
+   * nbr comes back alive
+   */
+  join_timer_start(js->up);
+  XFREE (MTYPE_PIM_JP_AGG_SOURCE, js);
+}
+
+int
+pim_jp_agg_group_list_cmp (void *arg1, void *arg2)
+{
+  const struct pim_jp_agg_group *jag1 = (const struct pim_jp_agg_group *)arg1;
+  const struct pim_jp_agg_group *jag2 = (const struct pim_jp_agg_group *)arg2;
+
+  if (jag1->group.s_addr < jag2->group.s_addr)
+    return -1;
+
+  if (jag1->group.s_addr > jag2->group.s_addr)
+    return 1;
+
+  return 0;
+}
+
+static int
+pim_jp_agg_src_cmp (void *arg1, void *arg2)
+{
+  const struct pim_jp_sources *js1 = (const struct pim_jp_sources *)arg1;
+  const struct pim_jp_sources *js2 = (const struct pim_jp_sources *)arg2;
+
+  if (js1->up->sg.src.s_addr < js2->up->sg.src.s_addr)
+    return -1;
+
+  if (js1->up->sg.src.s_addr > js2->up->sg.src.s_addr)
+    return 1;
+
+  return 0;
+}
+
+void
+pim_jp_agg_clear_group (struct list *group)
+{
+  struct listnode *node, *nnode;
+  struct pim_jp_agg_group *jag;
+
+  for (ALL_LIST_ELEMENTS(group, node, nnode, jag))
+    {
+      list_delete(jag->sources);
+      jag->sources = NULL;
+      listnode_delete(group, jag);
+      XFREE(MTYPE_PIM_JP_AGG_GROUP, jag);
+    }
+}
+
+static struct pim_iface_upstream_switch *
+pim_jp_agg_get_interface_upstream_switch_list (struct pim_rpf *rpf)
+{
+  struct pim_interface *pim_ifp = rpf->source_nexthop.interface->info;
+  struct pim_iface_upstream_switch *pius;
+  struct listnode *node, *nnode;
+
+  for (ALL_LIST_ELEMENTS(pim_ifp->upstream_switch_list, node, nnode, pius))
+    {
+      if (pius->address.s_addr == rpf->rpf_addr.u.prefix4.s_addr)
+        break;
+    }
+
+  if (!pius)
+    {
+      pius = XCALLOC(MTYPE_PIM_JP_AGG_GROUP, sizeof (struct pim_iface_upstream_switch));
+      pius->address.s_addr = rpf->rpf_addr.u.prefix4.s_addr;
+      pius->us = list_new();
+      listnode_add (pim_ifp->upstream_switch_list, pius);
+    }
+
+  return pius;
+}
+
+void
+pim_jp_agg_remove_group (struct list *group, struct pim_upstream *up)
+{
+  struct listnode *node, *nnode;
+  struct pim_jp_agg_group *jag = NULL;
+  struct pim_jp_sources *js = NULL;
+
+  for (ALL_LIST_ELEMENTS(group, node, nnode, jag))
+    {
+      if (jag->group.s_addr == up->sg.grp.s_addr)
+        break;
+    }
+
+  if (!jag)
+    return;
+
+  for (ALL_LIST_ELEMENTS(jag->sources, node, nnode, js))
+    {
+      if (js->up == up)
+        break;
+    }
+
+  listnode_delete(jag->sources, js);
+
+  XFREE(MTYPE_PIM_JP_AGG_SOURCE, js);
+
+  if (jag->sources->count == 0)
+    {
+      list_delete(jag->sources);
+      listnode_delete(group, jag);
+    }
+}
+
+void
+pim_jp_agg_add_group (struct list *group, struct pim_upstream *up, bool is_join)
+{
+  struct listnode *node, *nnode;
+  struct pim_jp_agg_group *jag = NULL;
+  struct pim_jp_sources *js;
+
+  for (ALL_LIST_ELEMENTS(group, node, nnode, jag))
+    {
+      if (jag->group.s_addr == up->sg.grp.s_addr)
+        break;
+    }
+
+  if (!jag)
+    {
+      jag = XCALLOC(MTYPE_PIM_JP_AGG_GROUP, sizeof (struct pim_jp_agg_group));
+      jag->group.s_addr = up->sg.grp.s_addr;
+      jag->sources = list_new();
+      jag->sources->cmp = pim_jp_agg_src_cmp;
+      jag->sources->del = (void (*)(void *))pim_jp_agg_src_free;
+      listnode_add (group, jag);
+    }
+
+  js = XCALLOC(MTYPE_PIM_JP_AGG_SOURCE, sizeof (struct pim_jp_sources));
+  js->up = up;
+  js->is_join = is_join;
+
+  listnode_add (jag->sources, js);
+}
+
+void
+pim_jp_agg_switch_interface (struct pim_rpf *orpf,
+                             struct pim_rpf *nrpf,
+                             struct pim_upstream *up)
+{
+  struct pim_iface_upstream_switch *opius;
+  struct pim_iface_upstream_switch *npius;
+
+  opius = pim_jp_agg_get_interface_upstream_switch_list(orpf);
+  npius = pim_jp_agg_get_interface_upstream_switch_list(nrpf);
+
+  /*
+   * 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.
+   */
+
+  /* send Prune(S,G) to the old upstream neighbor */
+  pim_jp_agg_add_group (opius->us, up, false);
+
+  /* send Join(S,G) to the current upstream neighbor */
+  pim_jp_agg_add_group (npius->us, up, true);
+
+}
+
+
+void
+pim_jp_agg_single_upstream_send (struct pim_rpf *rpf,
+                                 struct pim_upstream *up,
+                                 bool is_join)
+{
+  static struct list *groups = NULL;
+  static struct pim_jp_agg_group jag;
+  static struct pim_jp_sources js;
+
+  static bool first = true;
+
+  if (first)
+    {
+      groups = list_new();
+
+      jag.sources = list_new();
+
+      listnode_add(groups, &jag);
+      listnode_add(jag.sources, &js);
+
+      first = false;
+    }
+
+  jag.group.s_addr = up->sg.grp.s_addr;
+  js.up = up;
+  js.is_join = is_join;
+
+  pim_joinprune_send(rpf, groups);
+}
diff --git a/pimd/pim_jp_agg.h b/pimd/pim_jp_agg.h
new file mode 100644
index 0000000000..a50e8282ce
--- /dev/null
+++ b/pimd/pim_jp_agg.h
@@ -0,0 +1,33 @@
+#ifndef __PIM_JP_AGG_H__
+#define __PIM_JP_AGG_H__
+
+struct pim_jp_sources
+{
+  struct pim_upstream *up;
+  int is_join;
+};
+
+struct pim_jp_agg_group
+{
+  struct in_addr group;
+  //int onetime;
+  struct list *sources;
+};
+
+void pim_jp_agg_group_list_free (struct pim_jp_agg_group *jag);
+int pim_jp_agg_group_list_cmp (void *arg1, void *arg2);
+
+void pim_jp_agg_clear_group (struct list *group);
+void pim_jp_agg_remove_group (struct list *group, struct pim_upstream *up);
+
+void pim_jp_agg_add_group (struct list *group,
+                           struct pim_upstream *up, bool is_join);
+
+void pim_jp_agg_switch_interface (struct pim_rpf *orpf,
+                                  struct pim_rpf *nrpf,
+                                  struct pim_upstream *up);
+
+void pim_jp_agg_single_upstream_send (struct pim_rpf *rpf,
+                                      struct pim_upstream *up,
+                                      bool is_join);
+#endif
diff --git a/pimd/pim_memory.c b/pimd/pim_memory.c
index ccd0fa81ac..5af2a8203f 100644
--- a/pimd/pim_memory.c
+++ b/pimd/pim_memory.c
@@ -47,3 +47,5 @@ DEFINE_MTYPE(PIMD, PIM_MSDP_SA,           "PIM MSDP source-active cache")
 DEFINE_MTYPE(PIMD, PIM_MSDP_MG,           "PIM MSDP mesh group")
 DEFINE_MTYPE(PIMD, PIM_MSDP_MG_MBR,       "PIM MSDP mesh group mbr")
 DEFINE_MTYPE(PIMD, PIM_SEC_ADDR,          "PIM secondary address")
+DEFINE_MTYPE(PIMD, PIM_JP_AGG_GROUP,      "PIM JP AGG Group")
+DEFINE_MTYPE(PIMD, PIM_JP_AGG_SOURCE,     "PIM JP AGG Source")
diff --git a/pimd/pim_memory.h b/pimd/pim_memory.h
index b6b9b23239..0d5f131a4f 100644
--- a/pimd/pim_memory.h
+++ b/pimd/pim_memory.h
@@ -46,5 +46,6 @@ DECLARE_MTYPE(PIM_MSDP_SA)
 DECLARE_MTYPE(PIM_MSDP_MG)
 DECLARE_MTYPE(PIM_MSDP_MG_MBR)
 DECLARE_MTYPE(PIM_SEC_ADDR)
-
+DECLARE_MTYPE(PIM_JP_AGG_GROUP)
+DECLARE_MTYPE(PIM_JP_AGG_SOURCE)
 #endif /* _QUAGGA_PIM_MEMORY_H */
diff --git a/pimd/pim_msg.c b/pimd/pim_msg.c
index e6b13f3121..4018fd639e 100644
--- a/pimd/pim_msg.c
+++ b/pimd/pim_msg.c
@@ -36,6 +36,7 @@
 #include "pim_rp.h"
 #include "pim_rpf.h"
 #include "pim_register.h"
+#include "pim_jp_agg.h"
 
 void pim_msg_build_header(uint8_t *pim_msg, size_t pim_msg_size, uint8_t pim_msg_type)
 {
@@ -93,36 +94,62 @@ pim_msg_addr_encode_ipv4_source(uint8_t *buf,
   return buf + PIM_ENCODED_IPV4_SOURCE_SIZE;
 }
 
-static size_t
-pim_msg_build_jp_groups (struct pim_jp_groups *grp, struct pim_upstream *up, int is_join)
+/*
+ * For the given 'struct pim_jp_sources' list
+ * determine the size_t it would take up.
+ */
+size_t
+pim_msg_get_jp_group_size (struct list *sources)
+{
+  size_t size = 0;
+
+  size += sizeof (struct pim_encoded_group_ipv4);
+  size += 4;  // Joined sources (2) + Pruned Sources (2)
+
+  size += sizeof (struct pim_encoded_source_ipv4) * sources->count;
+
+  return size;
+}
+
+size_t
+pim_msg_build_jp_groups (struct pim_jp_groups *grp, struct pim_jp_agg_group *sgs)
 {
+  struct listnode *node, *nnode;
+  struct pim_jp_sources *source;
   struct in_addr stosend;
   uint8_t bits;
+  size_t size = pim_msg_get_jp_group_size (sgs->sources);
+  uint8_t tgroups = 0;
 
-  /* number of joined/pruned sources */
-  grp->joins  = htons(is_join ? 1 : 0);
-  grp->prunes = htons(is_join ? 0 : 1);
+  memset (grp, 0, size);
+  pim_msg_addr_encode_ipv4_group ((uint8_t *)&grp->g, sgs->group);
 
-  if (up->sg.src.s_addr == INADDR_ANY)
-    {
-      struct pim_rpf *rpf = pim_rp_g (up->sg.grp);
-      bits = PIM_ENCODE_SPARSE_BIT | PIM_ENCODE_WC_BIT | PIM_ENCODE_RPT_BIT;
-      stosend = rpf->rpf_addr.u.prefix4;
-    }
-  else
+  for (ALL_LIST_ELEMENTS(sgs->sources, node, nnode, source))
     {
-      bits = PIM_ENCODE_SPARSE_BIT;
-      stosend = up->sg.src;
-    }
+      /* number of joined/pruned sources */
+      if (source->is_join)
+        grp->joins++;
+      else
+        grp->prunes++;
 
-  if (!pim_msg_addr_encode_ipv4_source ((uint8_t *)&grp->s[0], stosend, bits)) {
-    char source_str[INET_ADDRSTRLEN];
-    pim_inet4_dump("<src?>", up->sg.src, source_str, sizeof(source_str));
-    zlog_warn("%s: failure encoding source address %s",
-              __PRETTY_FUNCTION__, source_str);
-    return 0;
-  }
+      if (source->up->sg.src.s_addr == INADDR_ANY)
+        {
+          struct pim_rpf *rpf = pim_rp_g (source->up->sg.grp);
+          bits = PIM_ENCODE_SPARSE_BIT | PIM_ENCODE_WC_BIT | PIM_ENCODE_RPT_BIT;
+          stosend = rpf->rpf_addr.u.prefix4;
+        }
+      else
+        {
+          bits = PIM_ENCODE_SPARSE_BIT;
+          stosend = source->up->sg.src;
+        }
+
+      pim_msg_addr_encode_ipv4_source ((uint8_t *)&grp->s[tgroups], stosend, bits);
+      tgroups++;
+    }
 
+  grp->joins = htons(grp->joins);
+  grp->prunes = htons(grp->prunes);
   /*
    * This is not implemented correctly at this point in time
    * Make it stop.
@@ -190,107 +217,5 @@ pim_msg_build_jp_groups (struct pim_jp_groups *grp, struct pim_upstream *up, int
     }
 #endif
 
-  return sizeof (*grp);
-}
-
-/*
- * J/P Message Format
- *
- * While the RFC clearly states that this is 32 bits wide, it
- * is cheating.  These fields:
- * Encoded-Unicast format   (6 bytes MIN)
- * Encoded-Group format     (8 bytes MIN)
- * Encoded-Source format    (8 bytes MIN)
- * are *not* 32 bits wide.
- *
- * Nor does the RFC explicitly call out the size for:
- * Reserved                 (1 byte)
- * Num Groups               (1 byte)
- * Holdtime                 (2 bytes)
- * Number of Joined Sources (2 bytes)
- * Number of Pruned Sources (2 bytes)
- *
- * This leads to a missleading representation from casual
- * reading and making assumptions.  Be careful!
- *
- *   0                   1                   2                   3
- *   0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
- *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
- *  |PIM Ver| Type  |   Reserved    |           Checksum            |
- *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
- *  |        Upstream Neighbor Address (Encoded-Unicast format)     |
- *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
- *  |  Reserved     | Num groups    |          Holdtime             |
- *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
- *  |         Multicast Group Address 1 (Encoded-Group format)      |
- *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
- *  |   Number of Joined Sources    |   Number of Pruned Sources    |
- *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
- *  |        Joined Source Address 1 (Encoded-Source format)        |
- *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
- *  |                             .                                 |
- *  |                             .                                 |
- *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
- *  |        Joined Source Address n (Encoded-Source format)        |
- *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
- *  |        Pruned Source Address 1 (Encoded-Source format)        |
- *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
- *  |                             .                                 |
- *  |                             .                                 |
- *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
- *  |        Pruned Source Address n (Encoded-Source format)        |
- *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
- *  |         Multicast Group Address m (Encoded-Group format)      |
- *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
- *  |   Number of Joined Sources    |   Number of Pruned Sources    |
- *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
- *  |        Joined Source Address 1 (Encoded-Source format)        |
- *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
- *  |                             .                                 |
- *  |                             .                                 |
- *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
- *  |        Joined Source Address n (Encoded-Source format)        |
- *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
- *  |        Pruned Source Address 1 (Encoded-Source format)        |
- *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
- *  |                             .                                 |
- *  |                             .                                 |
- *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
- *  |        Pruned Source Address n (Encoded-Source format)        |
- *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
- */
-int
-pim_msg_join_prune_encode (uint8_t *buf, size_t buf_size, int is_join,
-                           struct pim_upstream *up,
-                           struct in_addr upstream, int holdtime)
-{
-  struct pim_jp *msg = (struct pim_jp *)buf;
-
-  assert(buf_size > sizeof (struct pim_jp));
-
-  if (!pim_msg_addr_encode_ipv4_ucast ((uint8_t *)&msg->addr, upstream)) {
-    char dst_str[INET_ADDRSTRLEN];
-    pim_inet4_dump("<dst?>", upstream, dst_str, sizeof(dst_str));
-    zlog_warn("%s: failure encoding destination address %s",
-	      __PRETTY_FUNCTION__, dst_str);
-    return -3;
-  }
-
-  msg->reserved   = 0;
-  msg->num_groups = 1;
-  msg->holdtime   = htons(holdtime);
-
-  if (!pim_msg_addr_encode_ipv4_group ((uint8_t *)&msg->groups[0].g, up->sg.grp)) {
-    char group_str[INET_ADDRSTRLEN];
-    pim_inet4_dump("<grp?>", up->sg.grp, group_str, sizeof(group_str));
-    zlog_warn("%s: failure encoding group address %s",
-              __PRETTY_FUNCTION__, group_str);
-    return -5;
-  }
-
-  pim_msg_build_jp_groups (&msg->groups[0], up, is_join);
-
-  pim_msg_build_header (buf, sizeof (struct pim_jp), PIM_MSG_TYPE_JOIN_PRUNE);
-
-  return sizeof (struct pim_jp);
+  return size;
 }
diff --git a/pimd/pim_msg.h b/pimd/pim_msg.h
index 3af8486d5d..9774ef3ed0 100644
--- a/pimd/pim_msg.h
+++ b/pimd/pim_msg.h
@@ -23,6 +23,7 @@
 
 #include <netinet/in.h>
 
+#include "pim_jp_agg.h"
 /*
   Number       Description       
   ----------   ------------------
@@ -94,7 +95,6 @@ uint8_t *pim_msg_addr_encode_ipv4_source(uint8_t *buf,
                                          struct in_addr addr, uint8_t bits);
 
 
-int pim_msg_join_prune_encode (uint8_t *buf, size_t buf_size, int is_join,
-                               struct pim_upstream *up,
-                               struct in_addr upstream, int holdtime);
+size_t pim_msg_get_jp_group_size (struct list *sources);
+size_t pim_msg_build_jp_groups (struct pim_jp_groups *grp, struct pim_jp_agg_group *sgs);
 #endif /* PIM_MSG_H */
diff --git a/pimd/pim_neighbor.c b/pimd/pim_neighbor.c
index 346b911157..c1325df260 100644
--- a/pimd/pim_neighbor.c
+++ b/pimd/pim_neighbor.c
@@ -37,6 +37,8 @@
 #include "pim_ifchannel.h"
 #include "pim_rp.h"
 #include "pim_zebra.h"
+#include "pim_join.h"
+#include "pim_jp_agg.h"
 
 static void dr_election_by_addr(struct interface *ifp)
 {
@@ -265,6 +267,41 @@ void pim_neighbor_timer_reset(struct pim_neighbor *neigh, uint16_t holdtime)
 		  neigh, neigh->holdtime);
 }
 
+static int
+on_neighbor_jp_timer (struct thread *t)
+{
+  struct pim_neighbor *neigh = THREAD_ARG(t);
+  struct pim_rpf rpf;
+
+  if (PIM_DEBUG_PIM_TRACE)
+    {
+      char src_str[INET_ADDRSTRLEN];
+      pim_inet4_dump("<src?>", neigh->source_addr, src_str, sizeof(src_str));
+      zlog_debug("%s:Sending JP Agg to %s on %s with %d groups", __PRETTY_FUNCTION__,
+                 src_str, neigh->interface->name, neigh->upstream_jp_agg->count);
+    }
+  neigh->jp_timer = NULL;
+
+  rpf.source_nexthop.interface = neigh->interface;
+  rpf.rpf_addr.u.prefix4 = neigh->source_addr;
+  pim_joinprune_send(&rpf, neigh->upstream_jp_agg);
+
+  THREAD_TIMER_ON(master, neigh->jp_timer,
+                  on_neighbor_jp_timer,
+                  neigh, qpim_t_periodic);
+
+  return 0;
+}
+
+static void
+pim_neighbor_start_jp_timer (struct pim_neighbor *neigh)
+{
+  THREAD_TIMER_OFF(neigh->jp_timer);
+  THREAD_TIMER_ON(master, neigh->jp_timer,
+                  on_neighbor_jp_timer,
+                  neigh, qpim_t_periodic);
+}
+
 static struct pim_neighbor *pim_neighbor_new(struct interface *ifp,
 					     struct in_addr source_addr,
 					     pim_hello_options hello_options,
@@ -301,6 +338,11 @@ static struct pim_neighbor *pim_neighbor_new(struct interface *ifp,
   neigh->t_expire_timer         = NULL;
   neigh->interface              = ifp;
 
+  neigh->upstream_jp_agg = list_new();
+  neigh->upstream_jp_agg->cmp = pim_jp_agg_group_list_cmp;
+  neigh->upstream_jp_agg->del = (void (*)(void *))pim_jp_agg_group_list_free;
+  pim_neighbor_start_jp_timer(neigh);
+
   pim_neighbor_timer_reset(neigh, holdtime);
   /*
    * The pim_ifstat_hello_sent variable is used to decide if
@@ -375,6 +417,9 @@ void pim_neighbor_free(struct pim_neighbor *neigh)
 
   delete_prefix_list(neigh);
 
+  list_delete(neigh->upstream_jp_agg);
+  THREAD_OFF(neigh->jp_timer);
+
   XFREE(MTYPE_PIM_NEIGHBOR, neigh);
 }
 
diff --git a/pimd/pim_neighbor.h b/pimd/pim_neighbor.h
index 211eda25c7..986721666e 100644
--- a/pimd/pim_neighbor.h
+++ b/pimd/pim_neighbor.h
@@ -41,6 +41,9 @@ struct pim_neighbor {
   struct list       *prefix_list; /* list of struct prefix */
   struct thread     *t_expire_timer;
   struct interface  *interface;
+
+  struct thread     *jp_timer;
+  struct list       *upstream_jp_agg;
 };
 
 void pim_neighbor_timer_reset(struct pim_neighbor *neigh, uint16_t holdtime);
diff --git a/pimd/pim_rpf.c b/pimd/pim_rpf.c
index 40e035c3f4..ff8a6054cf 100644
--- a/pimd/pim_rpf.c
+++ b/pimd/pim_rpf.c
@@ -52,6 +52,7 @@ pim_rpf_set_refresh_time (void)
 int pim_nexthop_lookup(struct pim_nexthop *nexthop, struct in_addr addr, int neighbor_needed)
 {
   struct pim_zlookup_nexthop nexthop_tab[MULTIPATH_NUM];
+  struct pim_neighbor *nbr = NULL;
   int num_ifindex;
   struct interface *ifp = NULL;
   ifindex_t first_ifindex = 0;
@@ -134,8 +135,6 @@ int pim_nexthop_lookup(struct pim_nexthop *nexthop, struct in_addr addr, int nei
         }
       else if (neighbor_needed && !pim_if_connected_to_source (ifp, addr))
         {
-          struct pim_neighbor *nbr;
-
           nbr = pim_neighbor_find (ifp, nexthop_tab[i].nexthop_addr.u.prefix4);
           if (PIM_DEBUG_PIM_TRACE_DETAIL)
             zlog_debug ("ifp name: %s, pim nbr: %p", ifp->name, nbr);
@@ -169,6 +168,7 @@ int pim_nexthop_lookup(struct pim_nexthop *nexthop, struct in_addr addr, int nei
       nexthop->mrib_route_metric        = nexthop_tab[i].route_metric;
       nexthop->last_lookup              = addr;
       nexthop->last_lookup_time         = pim_time_monotonic_usec();
+      nexthop->nbr                      = nbr;
       return 0;
     }
   else
diff --git a/pimd/pim_rpf.h b/pimd/pim_rpf.h
index 51e84b4593..f4a987793e 100644
--- a/pimd/pim_rpf.h
+++ b/pimd/pim_rpf.h
@@ -45,6 +45,7 @@ struct pim_nexthop {
   struct prefix     mrib_nexthop_addr;      /* MRIB.next_hop(S) */
   uint32_t          mrib_metric_preference; /* MRIB.pref(S) */
   uint32_t          mrib_route_metric;      /* MRIB.metric(S) */
+  struct pim_neighbor *nbr;
 };
 
 struct pim_rpf {
diff --git a/pimd/pim_upstream.c b/pimd/pim_upstream.c
index 1712acaa1e..ce567824f2 100644
--- a/pimd/pim_upstream.c
+++ b/pimd/pim_upstream.c
@@ -51,12 +51,13 @@
 #include "pim_br.h"
 #include "pim_register.h"
 #include "pim_msdp.h"
+#include "pim_jp_agg.h"
 
 struct hash *pim_upstream_hash = NULL;
 struct list *pim_upstream_list = NULL;
 struct timer_wheel *pim_upstream_sg_wheel = NULL;
 
-static void join_timer_start(struct pim_upstream *up);
+static void join_timer_stop(struct pim_upstream *up);
 static void pim_upstream_update_assert_tracking_desired(struct pim_upstream *up);
 
 /*
@@ -165,13 +166,14 @@ pim_upstream_del(struct pim_upstream *up, const char *name)
   if (up->ref_count >= 1)
     return;
 
-  THREAD_OFF(up->t_join_timer);
+  join_timer_stop(up);
   THREAD_OFF(up->t_ka_timer);
   THREAD_OFF(up->t_rs_timer);
   THREAD_OFF(up->t_msdp_reg_timer);
 
   if (up->join_state == PIM_UPSTREAM_JOINED) {
-    pim_joinprune_send (&up->rpf, up, 0);
+    pim_jp_agg_single_upstream_send (&up->rpf, up, 0);
+
     if (up->sg.src.s_addr == INADDR_ANY) {
         /* if a (*, G) entry in the joined state is being deleted we
          * need to notify MSDP */
@@ -229,7 +231,7 @@ pim_upstream_send_join (struct pim_upstream *up)
   }
 
   /* send Join(S,G) to the current upstream neighbor */
-  pim_joinprune_send(&up->rpf, up, 1 /* join */);
+  pim_jp_agg_single_upstream_send(&up->rpf, up, 1 /* join */);
 }
 
 static int on_join_timer(struct thread *t)
@@ -258,8 +260,27 @@ static int on_join_timer(struct thread *t)
   return 0;
 }
 
-static void join_timer_start(struct pim_upstream *up)
+static void join_timer_stop(struct pim_upstream *up)
+{
+  struct pim_neighbor *nbr;
+
+  nbr = pim_neighbor_find (up->rpf.source_nexthop.interface,
+                           up->rpf.rpf_addr.u.prefix4);
+
+  if (nbr)
+    pim_jp_agg_remove_group (nbr->upstream_jp_agg, up);
+
+  THREAD_OFF (up->t_join_timer);
+}
+
+void
+join_timer_start(struct pim_upstream *up)
 {
+  struct pim_neighbor *nbr;
+
+  nbr = pim_neighbor_find (up->rpf.source_nexthop.interface,
+                           up->rpf.rpf_addr.u.prefix4);
+
   if (PIM_DEBUG_PIM_EVENTS) {
     zlog_debug("%s: starting %d sec timer for upstream (S,G)=%s",
 	       __PRETTY_FUNCTION__,
@@ -267,15 +288,34 @@ static void join_timer_start(struct pim_upstream *up)
 	       up->sg_str);
   }
 
-  THREAD_OFF (up->t_join_timer);
-  THREAD_TIMER_ON(master, up->t_join_timer,
-		  on_join_timer,
-		  up, qpim_t_periodic);
+  if (nbr)
+    pim_jp_agg_add_group (nbr->upstream_jp_agg, up, 1);
+  else
+    {
+      THREAD_OFF (up->t_join_timer);
+      THREAD_TIMER_ON(master, up->t_join_timer,
+                      on_join_timer,
+                      up, qpim_t_periodic);
+    }
 }
 
-void pim_upstream_join_timer_restart(struct pim_upstream *up)
+/*
+ * This is only called when we are switching the upstream
+ * J/P from one neighbor to another
+ *
+ * As such we need to remove from the old list and
+ * add to the new list.
+ */
+void pim_upstream_join_timer_restart(struct pim_upstream *up, struct pim_rpf *old)
 {
-  THREAD_OFF(up->t_join_timer);
+  struct pim_neighbor *nbr;
+
+  nbr = pim_neighbor_find (old->source_nexthop.interface,
+                           old->rpf_addr.u.prefix4);
+  if (nbr)
+    pim_jp_agg_remove_group (nbr->upstream_jp_agg, up);
+
+  //THREAD_OFF(up->t_join_timer);
   join_timer_start(up);
 }
 
@@ -479,12 +519,13 @@ pim_upstream_switch(struct pim_upstream *up,
       }
   }
   else {
+
     forward_off(up);
     if (old_state == PIM_UPSTREAM_JOINED)
       pim_msdp_up_join_state_changed(up);
-    pim_joinprune_send(&up->rpf, up, 0 /* prune */);
-    if (up->t_join_timer)
-      THREAD_OFF(up->t_join_timer);
+
+    pim_jp_agg_single_upstream_send(&up->rpf, up, 0 /* prune */);
+    join_timer_stop(up);
   }
 }
 
diff --git a/pimd/pim_upstream.h b/pimd/pim_upstream.h
index f36b6fba8b..7cdf73759d 100644
--- a/pimd/pim_upstream.h
+++ b/pimd/pim_upstream.h
@@ -140,7 +140,7 @@ void pim_upstream_join_suppress(struct pim_upstream *up,
 void pim_upstream_join_timer_decrease_to_t_override(const char *debug_label,
                                                     struct pim_upstream *up);
 
-void pim_upstream_join_timer_restart(struct pim_upstream *up);
+void pim_upstream_join_timer_restart(struct pim_upstream *up, struct pim_rpf *old);
 void pim_upstream_rpf_genid_changed(struct in_addr neigh_addr);
 void pim_upstream_rpf_interface_changed(struct pim_upstream *up,
 					struct interface *old_rpf_ifp);
@@ -173,4 +173,6 @@ void pim_upstream_msdp_reg_timer_start(struct pim_upstream *up);
 
 void pim_upstream_init (void);
 void pim_upstream_terminate (void);
+
+void join_timer_start (struct pim_upstream *up);
 #endif /* PIM_UPSTREAM_H */
diff --git a/pimd/pim_zebra.c b/pimd/pim_zebra.c
index 0f92b4a7b0..70b0302806 100644
--- a/pimd/pim_zebra.c
+++ b/pimd/pim_zebra.c
@@ -44,6 +44,7 @@
 #include "pim_ifchannel.h"
 #include "pim_rp.h"
 #include "pim_igmpv3.h"
+#include "pim_jp_agg.h"
 
 #undef PIM_DEBUG_IFADDR_DUMP
 #define PIM_DEBUG_IFADDR_DUMP
@@ -362,20 +363,24 @@ static int pim_zebra_if_address_del(int command, struct zclient *client,
 static void scan_upstream_rpf_cache()
 {
   struct listnode     *up_node;
+  struct listnode     *ifnode;
   struct listnode     *up_nextnode;
+  struct listnode     *node;
   struct pim_upstream *up;
+  struct interface    *ifp;
 
   for (ALL_LIST_ELEMENTS(pim_upstream_list, up_node, up_nextnode, up)) {
     enum pim_rpf_result rpf_result;
     struct pim_rpf      old;
 
     old.source_nexthop.interface = up->rpf.source_nexthop.interface;
+    old.source_nexthop.nbr       = up->rpf.source_nexthop.nbr;
     rpf_result = pim_rpf_update(up, &old);
+    zlog_debug ("Looking at upstream: %s %d", up->sg_str, rpf_result);
     if (rpf_result == PIM_RPF_FAILURE)
       continue;
 
     if (rpf_result == PIM_RPF_CHANGED) {
-
       /*
        * We have detected a case where we might need to rescan
        * the inherited o_list so do it.
@@ -395,28 +400,22 @@ static void scan_upstream_rpf_cache()
 	if (!up->channel_oil->installed)
 	  pim_mroute_add (up->channel_oil, __PRETTY_FUNCTION__);
 
-	/*
-	  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.
-	*/
-
-    
-	/* send Prune(S,G) to the old upstream neighbor */
-	pim_joinprune_send(&old, up, 0 /* prune */);
-	
-	/* send Join(S,G) to the current upstream neighbor */
-	pim_joinprune_send(&up->rpf, up, 1 /* join */);
-
-	pim_upstream_join_timer_restart(up);
+        /*
+         * 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 */
 
       /* FIXME can join_desired actually be changed by pim_rpf_update()
@@ -426,7 +425,22 @@ static void scan_upstream_rpf_cache()
     } /* PIM_RPF_CHANGED */
 
   } /* for (qpim_upstream_list) */
-  
+
+  for (ALL_LIST_ELEMENTS_RO (vrf_iflist (VRF_DEFAULT), ifnode, ifp))
+    if (ifp->info)
+      {
+        struct pim_interface *pim_ifp = ifp->info;
+        struct pim_iface_upstream_switch *us;
+
+        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