diff --git a/pimd/pim_cmd.c b/pimd/pim_cmd.c
index ebe8b362ba..0690130edf 100644
--- a/pimd/pim_cmd.c
+++ b/pimd/pim_cmd.c
@@ -1632,6 +1632,7 @@ static void pim_show_upstream(struct vty *vty, u_char uj)
     char join_timer[10];
     char rs_timer[10];
     char ka_timer[10];
+    char msdp_reg_timer[10];
 
     pim_inet4_dump("<src?>", up->sg.src, src_str, sizeof(src_str));
     pim_inet4_dump("<grp?>", up->sg.grp, grp_str, sizeof(grp_str));
@@ -1639,6 +1640,7 @@ static void pim_show_upstream(struct vty *vty, u_char uj)
     pim_time_timer_to_hhmmss (join_timer, sizeof(join_timer), up->t_join_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);
 
     if (uj) {
       json_object_object_get_ex(json, grp_str, &json_group);
@@ -1658,6 +1660,7 @@ static void pim_show_upstream(struct vty *vty, u_char uj)
       json_object_string_add(json_row, "joinTimer", join_timer);
       json_object_string_add(json_row, "resetTimer", rs_timer);
       json_object_string_add(json_row, "keepaliveTimer", ka_timer);
+      json_object_string_add(json_row, "msdpRegTimer", msdp_reg_timer);
       json_object_int_add(json_row, "refCount", up->ref_count);
       json_object_int_add(json_row, "sptBit", up->sptbit);
       json_object_object_add(json_group, src_str, json_row);
@@ -5431,6 +5434,15 @@ DEFUN (no_ip_msdp_mesh_group,
   return ip_no_msdp_mesh_group_cmd_worker(vty, argv[4]->arg);
 }
 
+static void
+print_empty_json_obj(struct vty *vty)
+{
+  json_object *json;
+  json = json_object_new_object();
+  vty_out (vty, "%s%s", json_object_to_json_string_ext(json, JSON_C_TO_STRING_PRETTY), VTY_NEWLINE);
+  json_object_free(json);
+}
+
 static void
 ip_msdp_show_mesh_group(struct vty *vty, u_char uj)
 {
@@ -5446,6 +5458,8 @@ ip_msdp_show_mesh_group(struct vty *vty, u_char uj)
   json_object *json_row = NULL;
 
   if (!mg) {
+    if (uj)
+      print_empty_json_obj(vty);
     return;
   }
 
@@ -5670,6 +5684,7 @@ ip_msdp_show_sa(struct vty *vty, u_char uj)
   char rp_str[INET_ADDRSTRLEN];
   char timebuf[PIM_MSDP_UPTIME_STRLEN];
   char spt_str[8];
+  char local_str[8];
   int64_t now;
   json_object *json = NULL;
   json_object *json_group = NULL;
@@ -5678,7 +5693,7 @@ ip_msdp_show_sa(struct vty *vty, u_char uj)
   if (uj) {
     json = json_object_new_object();
   } else {
-    vty_out(vty, "Source                     Group               RP  SPT    Uptime%s", VTY_NEWLINE);
+    vty_out(vty, "Source                     Group               RP  Local  SPT    Uptime%s", VTY_NEWLINE);
   }
 
   for (ALL_LIST_ELEMENTS_RO(msdp->sa_list, sanode, sa)) {
@@ -5686,19 +5701,23 @@ ip_msdp_show_sa(struct vty *vty, u_char uj)
     pim_time_uptime(timebuf, sizeof(timebuf), now - sa->uptime);
     pim_inet4_dump("<src?>", sa->sg.src, src_str, sizeof(src_str));
     pim_inet4_dump("<grp?>", sa->sg.grp, grp_str, sizeof(grp_str));
-    if (sa->flags & PIM_MSDP_SAF_LOCAL) {
-      strcpy(rp_str, "local");
-      strcpy(spt_str, "-");
-    } else {
+    if (sa->flags & PIM_MSDP_SAF_PEER) {
       pim_inet4_dump("<rp?>", sa->rp, rp_str, sizeof(rp_str));
-    }
-
-    if (uj) {
       if (sa->up) {
         strcpy(spt_str, "yes");
       } else {
         strcpy(spt_str, "no");
       }
+    } else {
+      strcpy(rp_str, "-");
+      strcpy(spt_str, "-");
+    }
+    if (sa->flags & PIM_MSDP_SAF_LOCAL) {
+      strcpy(local_str, "yes");
+    } else {
+      strcpy(local_str, "no");
+    }
+    if (uj) {
       json_object_object_get_ex(json, grp_str, &json_group);
 
       if (!json_group) {
@@ -5710,17 +5729,13 @@ ip_msdp_show_sa(struct vty *vty, u_char uj)
       json_object_string_add(json_row, "source", src_str);
       json_object_string_add(json_row, "group", grp_str);
       json_object_string_add(json_row, "rp", rp_str);
+      json_object_string_add(json_row, "local", local_str);
       json_object_string_add(json_row, "sptSetup", spt_str);
       json_object_string_add(json_row, "upTime", timebuf);
       json_object_object_add(json_group, src_str, json_row);
     } else {
-      if (sa->up) {
-        strcpy(spt_str, "y");
-      } else {
-        strcpy(spt_str, "n");
-      }
-      vty_out(vty, "%-15s  %15s  %15s  %3s  %8s%s",
-          src_str, grp_str, rp_str, spt_str, timebuf, VTY_NEWLINE);
+      vty_out(vty, "%-15s  %15s  %15s  %5c  %3c  %8s%s",
+          src_str, grp_str, rp_str, local_str[0], spt_str[0], timebuf, VTY_NEWLINE);
     }
   }
 
@@ -5740,6 +5755,7 @@ ip_msdp_show_sa_entry_detail(struct pim_msdp_sa *sa, const char *src_str,
   char peer_str[INET_ADDRSTRLEN];
   char timebuf[PIM_MSDP_UPTIME_STRLEN];
   char spt_str[8];
+  char local_str[8];
   char statetimer[PIM_MSDP_TIMER_STRLEN];
   int64_t now;
   json_object *json_group = NULL;
@@ -5747,11 +5763,7 @@ ip_msdp_show_sa_entry_detail(struct pim_msdp_sa *sa, const char *src_str,
 
   now = pim_time_monotonic_sec();
   pim_time_uptime(timebuf, sizeof(timebuf), now - sa->uptime);
-  if (sa->flags & PIM_MSDP_SAF_LOCAL) {
-    strcpy(rp_str, "local");
-    strcpy(peer_str, "-");
-    strcpy(spt_str, "-");
-  } else {
+  if (sa->flags & PIM_MSDP_SAF_PEER) {
     pim_inet4_dump("<rp?>", sa->rp, rp_str, sizeof(rp_str));
     pim_inet4_dump("<peer?>", sa->peer, peer_str, sizeof(peer_str));
     if (sa->up) {
@@ -5759,6 +5771,15 @@ ip_msdp_show_sa_entry_detail(struct pim_msdp_sa *sa, const char *src_str,
     } else {
       strcpy(spt_str, "no");
     }
+  } else {
+    strcpy(rp_str, "-");
+    strcpy(peer_str, "-");
+    strcpy(spt_str, "-");
+  }
+  if (sa->flags & PIM_MSDP_SAF_LOCAL) {
+      strcpy(local_str, "yes");
+  } else {
+      strcpy(local_str, "no");
   }
   pim_time_timer_to_hhmmss(statetimer, sizeof(statetimer), sa->sa_state_timer);
   if (uj) {
@@ -5773,6 +5794,7 @@ ip_msdp_show_sa_entry_detail(struct pim_msdp_sa *sa, const char *src_str,
     json_object_string_add(json_row, "source", src_str);
     json_object_string_add(json_row, "group", grp_str);
     json_object_string_add(json_row, "rp", rp_str);
+    json_object_string_add(json_row, "local", local_str);
     json_object_string_add(json_row, "sptSetup", spt_str);
     json_object_string_add(json_row, "upTime", timebuf);
     json_object_string_add(json_row, "stateTime", statetimer);
@@ -5781,6 +5803,7 @@ ip_msdp_show_sa_entry_detail(struct pim_msdp_sa *sa, const char *src_str,
     vty_out(vty, "SA : %s%s", pim_str_sg_dump(&sa->sg), VTY_NEWLINE);
     vty_out(vty, "  RP         : %s%s", rp_str, VTY_NEWLINE);
     vty_out(vty, "  Peer       : %s%s", peer_str, VTY_NEWLINE);
+    vty_out(vty, "  Local      : %s%s", local_str, VTY_NEWLINE);
     vty_out(vty, "  SPT Setup  : %s%s", spt_str, VTY_NEWLINE);
     vty_out(vty, "  Uptime     : %s%s", timebuf, VTY_NEWLINE);
     vty_out(vty, "  State Time : %s%s", statetimer, VTY_NEWLINE);
diff --git a/pimd/pim_mroute.c b/pimd/pim_mroute.c
index 5fba681a73..3f8ec35587 100644
--- a/pimd/pim_mroute.c
+++ b/pimd/pim_mroute.c
@@ -160,8 +160,6 @@ pim_mroute_msg_nocache (int fd, struct interface *ifp, const struct igmpmsg *msg
     return 0;
   }
   PIM_UPSTREAM_FLAG_SET_SRC_STREAM(up->flags);
-  pim_upstream_set_created_by_upstream(up);
-
   pim_upstream_keep_alive_timer_start (up, qpim_keep_alive_time);
 
   up->channel_oil = oil;
@@ -422,8 +420,6 @@ pim_mroute_msg_wrvifwhole (int fd, struct interface *ifp, const char *buf)
 	  return -2;
 	}
       PIM_UPSTREAM_FLAG_SET_SRC_STREAM(up->flags);
-      pim_upstream_set_created_by_upstream(up);
-
       pim_upstream_keep_alive_timer_start (up, qpim_keep_alive_time);
       up->channel_oil = oil;
       up->channel_oil->cc.pktcnt++;
diff --git a/pimd/pim_msdp.c b/pimd/pim_msdp.c
index 76127245ce..12f363e6fe 100644
--- a/pimd/pim_msdp.c
+++ b/pimd/pim_msdp.c
@@ -36,6 +36,7 @@
 #include "pim_rp.h"
 #include "pim_str.h"
 #include "pim_time.h"
+#include "pim_upstream.h"
 
 #include "pim_msdp.h"
 #include "pim_msdp_packet.h"
@@ -139,10 +140,6 @@ pim_msdp_sa_upstream_del(struct pim_msdp_sa *sa)
   }
 
   sa->up = NULL;
-  /* XXX: we can't pull the plug on an active flow even if the SA entry is
-   * removed. so ideally we want to start the kat in parallel and let the
-   * entry age out; but running the kat has fatal consequences. need to
-   * check with Donald on the best way to go abt this */
   if (PIM_UPSTREAM_FLAG_TEST_SRC_MSDP(up->flags)) {
     PIM_UPSTREAM_FLAG_UNSET_SRC_MSDP(up->flags);
     pim_upstream_del(up, __PRETTY_FUNCTION__);
@@ -158,12 +155,6 @@ pim_msdp_sa_upstream_del(struct pim_msdp_sa *sa)
 static bool
 pim_msdp_sa_upstream_add_ok(struct pim_msdp_sa *sa, struct pim_upstream *xg_up)
 {
-  if (sa->flags & PIM_MSDP_SAF_LOCAL) {
-    /* if there is a local reference we should NEVER use it for setting up
-     * SPTs otherwise we will get stuck in a simple circular deadlock */
-    return false;
-  }
-
   if (!(sa->flags & PIM_MSDP_SAF_PEER)) {
     /* SA should have been rxed from a peer */
     return false;
@@ -194,10 +185,9 @@ pim_msdp_sa_upstream_add_ok(struct pim_msdp_sa *sa, struct pim_upstream *xg_up)
 
 /* Upstream add evaluation needs to happen everytime -
  * 1. Peer reference is added or removed.
- * 2. Local reference is added or removed.
- * 3. The RP for a group changes.
- * 4. joinDesired for the associated (*, G) changes
- * 5. associated (*, G) is removed - this seems like a bit redundant
+ * 2. The RP for a group changes.
+ * 3. joinDesired for the associated (*, G) changes
+ * 4. associated (*, G) is removed - this seems like a bit redundant
  *    (considering #4); but just in case an entry gets nuked without
  *    upstream state transition
  *    */
@@ -359,12 +349,17 @@ static void
 pim_msdp_sa_deref(struct pim_msdp_sa *sa, enum pim_msdp_sa_flags flags)
 {
   char key_str[PIM_MSDP_SA_KEY_STRLEN];
+  bool update_up = false;
 
-  pim_msdp_sa_key_dump(sa, key_str, sizeof(key_str), true);
+  if (PIM_DEBUG_MSDP_EVENTS) {
+    pim_msdp_sa_key_dump(sa, key_str, sizeof(key_str), true);
+  }
 
   if ((sa->flags &PIM_MSDP_SAF_LOCAL)) {
     if (flags & PIM_MSDP_SAF_LOCAL) {
-      zlog_debug("%s local reference removed", key_str);
+      if (PIM_DEBUG_MSDP_EVENTS) {
+        zlog_debug("%s local reference removed", key_str);
+      }
       if (msdp->local_cnt)
         --msdp->local_cnt;
     }
@@ -372,13 +367,21 @@ pim_msdp_sa_deref(struct pim_msdp_sa *sa, enum pim_msdp_sa_flags flags)
 
   if ((sa->flags &PIM_MSDP_SAF_PEER)) {
     if (flags & PIM_MSDP_SAF_PEER) {
-      zlog_debug("%s peer reference removed", key_str);
+      if (PIM_DEBUG_MSDP_EVENTS) {
+        zlog_debug("%s peer reference removed", key_str);
+      }
       pim_msdp_sa_state_timer_setup(sa, false /* start */);
+      /* if peer ref was removed we need to remove the msdp reference on the
+       * msdp entry */
+      update_up = true;
     }
   }
 
   sa->flags &= ~flags;
-  pim_msdp_sa_upstream_update(sa, NULL /* xg_up */, "sa-deref");
+  if (update_up) {
+    pim_msdp_sa_upstream_update(sa, NULL /* xg_up */, "sa-deref");
+  }
+
   if (!(sa->flags & PIM_MSDP_SAF_REF)) {
     pim_msdp_sa_del(sa);
   }
@@ -424,27 +427,53 @@ pim_msdp_sa_ref(struct pim_msdp_peer *mp, struct prefix_sg *sg,
       }
       /* send an immediate SA update to peers */
       pim_msdp_pkt_sa_tx_one(sa);
-      pim_msdp_sa_upstream_update(sa, NULL /* xg_up */, "local-ref");
     }
     sa->flags &= ~PIM_MSDP_SAF_STALE;
   }
 }
 
-void
+/* The following criteria must be met to originate an SA from the MSDP
+ * speaker -
+ * 1. KAT must be running i.e. source is active.
+ * 2. We must be RP for the group.
+ * 3. Source must be registrable to the RP (this is where the RFC is vague
+ *    and especially ambiguous in CLOS networks; with anycast RP all sources
+ *    are potentially registrable to all RPs in the domain). We assume #3 is
+ *    satisfied if -
+ *    a. We are also the FHR-DR for the source (OR)
+ *    b. We rxed a pim register (null or data encapsulated) within the last
+ *       (3 * (1.5 * register_suppression_timer))).
+ */
+static bool
+pim_msdp_sa_local_add_ok(struct pim_upstream *up)
+{
+  if (!(msdp->flags & PIM_MSDPF_ENABLE)) {
+    return false;
+  }
+
+  if (!up->t_ka_timer) {
+    /* stream is not active */
+    return false;
+  }
+
+  if (!I_am_RP(up->sg.grp)) {
+    /* we are not RP for the group */
+    return false;
+  }
+
+  /* we are the FHR-DR for this stream  or we are RP and have seen registers
+   * from a FHR for this source */
+  if (PIM_UPSTREAM_FLAG_TEST_FHR(up->flags) || up->t_msdp_reg_timer) {
+    return true;
+  }
+
+  return false;
+}
+
+static void
 pim_msdp_sa_local_add(struct prefix_sg *sg)
 {
   struct in_addr rp;
-
-  if (!(msdp->flags & PIM_MSDPF_ENABLE)) {
-    /* if the feature is not enabled do nothing; we will collect all local
-     * sources whenever it is */
-    return;
-  }
-
-  /* check if I am RP for this group. XXX: is this check really needed? */
-  if (!I_am_RP(sg->grp)) {
-    return;
-  }
   rp.s_addr = 0;
   pim_msdp_sa_ref(NULL /* mp */, sg, rp);
 }
@@ -454,18 +483,31 @@ pim_msdp_sa_local_del(struct prefix_sg *sg)
 {
   struct pim_msdp_sa *sa;
 
-  if (!(msdp->flags & PIM_MSDPF_ENABLE)) {
-    /* if the feature is not enabled do nothing; we will collect all local
-     * sources whenever it is */
-    return;
-  }
-
   sa = pim_msdp_sa_find(sg);
   if (sa) {
     pim_msdp_sa_deref(sa, PIM_MSDP_SAF_LOCAL);
   }
 }
 
+/* Local SA qualification needs to be re-evaluated when -
+ * 1. KAT is started or stopped
+ * 2. on RP changes
+ * 3. Whenever FHR status changes for a (S,G) - XXX - currently there
+ *    is no clear path to transition an entry out of "MASK_FHR" need
+ *    to discuss this with Donald. May result in some strangeness if the
+ *    FHR is also the RP.
+ * 4. When msdp_reg timer is started or stopped
+ */
+void
+pim_msdp_sa_local_update(struct pim_upstream *up)
+{
+  if (pim_msdp_sa_local_add_ok(up)) {
+    pim_msdp_sa_local_add(&up->sg);
+  } else {
+    pim_msdp_sa_local_del(&up->sg);
+  }
+}
+
 static void
 pim_msdp_sa_local_setup(void)
 {
@@ -473,9 +515,7 @@ pim_msdp_sa_local_setup(void)
   struct listnode *up_node;
 
   for (ALL_LIST_ELEMENTS_RO(pim_upstream_list, up_node, up)) {
-    if (PIM_UPSTREAM_FLAG_TEST_CREATED_BY_UPSTREAM(up->flags)) {
-      pim_msdp_sa_local_add(&up->sg);
-    }
+    pim_msdp_sa_local_update(up);
   }
 }
 
@@ -548,8 +588,6 @@ pim_msdp_up_join_state_changed(struct pim_upstream *xg_up)
   }
 }
 
-/* XXX: Need to maintain SAs per-group to avoid all this unnecessary
- * walking */
 void
 pim_msdp_up_xg_del(struct prefix_sg *sg)
 {
diff --git a/pimd/pim_msdp.h b/pimd/pim_msdp.h
index 4809749b94..38ac4aa337 100644
--- a/pimd/pim_msdp.h
+++ b/pimd/pim_msdp.h
@@ -218,7 +218,7 @@ int pim_msdp_config_write(struct vty *vty);
 void pim_msdp_peer_pkt_txed(struct pim_msdp_peer *mp);
 char *pim_msdp_sa_key_dump(struct pim_msdp_sa *sa, char *buf, int buf_size, bool long_format);
 void pim_msdp_sa_ref(struct pim_msdp_peer *mp, struct prefix_sg *sg, struct in_addr rp);
-void pim_msdp_sa_local_add(struct prefix_sg *sg);
+void pim_msdp_sa_local_update(struct pim_upstream *up);
 void pim_msdp_sa_local_del(struct prefix_sg *sg);
 void pim_msdp_i_am_rp_changed(void);
 bool pim_msdp_peer_rpf_check(struct pim_msdp_peer *mp, struct in_addr rp);
diff --git a/pimd/pim_register.c b/pimd/pim_register.c
index 914d196339..8f3edc7580 100644
--- a/pimd/pim_register.c
+++ b/pimd/pim_register.c
@@ -342,7 +342,7 @@ pim_register_recv (struct interface *ifp,
             zlog_warn ("Failure to create upstream state");
             return 1;
           }
-        pim_upstream_set_created_by_upstream(upstream);
+        PIM_UPSTREAM_FLAG_SET_SRC_STREAM(upstream->flags);
 
         upstream->upstream_register = src_addr;
 	pim_rp_set_upstream_addr (&upstream->upstream_addr, sg.src, sg.grp);
@@ -353,7 +353,7 @@ pim_register_recv (struct interface *ifp,
               {
                 zlog_debug ("Received Register(%s), for which I have no path back", pim_str_sg_dump (&upstream->sg));
               }
-            pim_upstream_unset_created_by_upstream(upstream);
+            PIM_UPSTREAM_FLAG_UNSET_SRC_STREAM(upstream->flags);
             pim_upstream_del (upstream, __PRETTY_FUNCTION__);
             return 1;
           }
@@ -390,6 +390,7 @@ pim_register_recv (struct interface *ifp,
 	//inherited_olist(S,G,rpt)
 	// This is taken care of by the kernel for us
       }
+     pim_upstream_msdp_reg_timer_start(upstream);
   } else {
     if (PIM_DEBUG_PIM_REG)
       {
diff --git a/pimd/pim_upstream.c b/pimd/pim_upstream.c
index a6d7f6e749..01b0b3ac69 100644
--- a/pimd/pim_upstream.c
+++ b/pimd/pim_upstream.c
@@ -112,20 +112,6 @@ pim_upstream_find_new_children (struct pim_upstream *up)
     }
 }
 
-void
-pim_upstream_set_created_by_upstream(struct pim_upstream *up)
-{
-  PIM_UPSTREAM_FLAG_SET_CREATED_BY_UPSTREAM(up->flags);
-  pim_msdp_sa_local_add(&up->sg);
-}
-
-void
-pim_upstream_unset_created_by_upstream(struct pim_upstream *up)
-{
-  PIM_UPSTREAM_FLAG_UNSET_CREATED_BY_UPSTREAM(up->flags);
-  pim_msdp_sa_local_del(&up->sg);
-}
-
 /*
  * If we have a (*,*) || (S,*) there is no parent
  * If we have a (S,G), find the (*,G)
@@ -188,6 +174,7 @@ pim_upstream_del(struct pim_upstream *up, const char *name)
   THREAD_OFF(up->t_join_timer);
   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.source_nexthop.interface,
@@ -576,6 +563,7 @@ static struct pim_upstream *pim_upstream_new(struct prefix_sg *sg,
   up->t_join_timer               = NULL;
   up->t_ka_timer                 = NULL;
   up->t_rs_timer                 = NULL;
+  up->t_msdp_reg_timer           = NULL;
   up->join_state                 = 0;
   up->state_transition           = pim_time_monotonic_sec();
   up->channel_oil                = NULL;
@@ -634,6 +622,12 @@ struct pim_upstream *pim_upstream_find(struct prefix_sg *sg)
   return up;
 }
 
+static void pim_upstream_ref(struct pim_upstream *up, int flags)
+{
+  up->flags |= flags;
+  ++up->ref_count;
+}
+
 struct pim_upstream *pim_upstream_add(struct prefix_sg *sg,
 				      struct interface *incoming,
 				      int flags, const char *name)
@@ -642,8 +636,7 @@ struct pim_upstream *pim_upstream_add(struct prefix_sg *sg,
   int found = 0;
   up = pim_upstream_find(sg);
   if (up) {
-    ++up->ref_count;
-    up->flags |= flags;
+    pim_upstream_ref(up, flags);
     found = 1;
   }
   else {
@@ -908,9 +901,48 @@ static void pim_upstream_update_assert_tracking_desired(struct pim_upstream *up)
   } /* scan iface channel list */
 }
 
+/* When kat is stopped CouldRegister goes to false so we need to
+ * transition  the (S, G) on FHR to NI state and remove reg tunnel
+ * from the OIL */
+static void pim_upstream_fhr_kat_expiry(struct pim_upstream *up)
+{
+  if (!PIM_UPSTREAM_FLAG_TEST_FHR(up->flags))
+    return;
+
+  if (PIM_DEBUG_TRACE)
+    zlog_debug ("kat expired on %s; clear fhr reg state",
+        pim_str_sg_dump (&up->sg));
+  /* stop reg-stop timer */
+  THREAD_OFF(up->t_rs_timer);
+  /* remove regiface from the OIL if it is there*/
+  pim_channel_del_oif (up->channel_oil, pim_regiface, PIM_OIF_FLAG_PROTO_PIM);
+  /* move to "not-joined" */
+  up->join_state = PIM_UPSTREAM_NOTJOINED;
+  PIM_UPSTREAM_FLAG_UNSET_FHR(up->flags);
+}
+
+/* When kat is started CouldRegister can go to true. And if it does we
+ * need to transition  the (S, G) on FHR to JOINED state and add reg tunnel
+ * to the OIL */
+static void pim_upstream_fhr_kat_start(struct pim_upstream *up)
+{
+  if (pim_upstream_could_register(up)) {
+    if (PIM_DEBUG_TRACE)
+      zlog_debug ("kat started on %s; set fhr reg state to joined",
+        pim_str_sg_dump (&up->sg));
+    PIM_UPSTREAM_FLAG_SET_FHR(up->flags);
+    if (up->join_state == PIM_UPSTREAM_NOTJOINED) {
+      pim_channel_add_oif (up->channel_oil, pim_regiface, PIM_OIF_FLAG_PROTO_PIM);
+      up->join_state = PIM_UPSTREAM_JOINED;
+    }
+  }
+}
+
 /*
  * On an RP, the PMBR value must be cleared when the
  * Keepalive Timer expires
+ * KAT expiry indicates that flow is inactive. If the flow was created or
+ * maintained by activity now is the time to deref it.
  */
 static int
 pim_upstream_keep_alive_timer (struct thread *t)
@@ -918,56 +950,75 @@ pim_upstream_keep_alive_timer (struct thread *t)
   struct pim_upstream *up;
 
   up = THREAD_ARG(t);
+  up->t_ka_timer = NULL;
 
   if (I_am_RP (up->sg.grp))
-    {
-      pim_br_clear_pmbr (&up->sg);
-      /*
-       * We need to do more here :)
-       * But this is the start.
-       */
-    }
+  {
+    pim_br_clear_pmbr (&up->sg);
+    /*
+     * We need to do more here :)
+     * But this is the start.
+     */
+  }
 
-  pim_mroute_update_counters (up->channel_oil);
+  /* source is no longer active - pull the SA from MSDP's cache */
+  pim_msdp_sa_local_del(&up->sg);
 
-  if (PIM_DEBUG_MROUTE)
-    {
-      zlog_debug ("New: %llu %lu %lu %lu", up->channel_oil->cc.lastused, up->channel_oil->cc.pktcnt,
-                  up->channel_oil->cc.bytecnt, up->channel_oil->cc.wrong_if);
-      zlog_debug ("old: %llu %lu %lu %lu", up->channel_oil->cc.lastused, up->channel_oil->cc.oldpktcnt,
-                  up->channel_oil->cc.oldbytecnt, up->channel_oil->cc.oldwrong_if);
-    }
-  if ((up->channel_oil->cc.oldpktcnt >= up->channel_oil->cc.pktcnt) &&
-      (up->channel_oil->cc.lastused/100 >= qpim_keep_alive_time))
-    {
-      pim_mroute_del (up->channel_oil);
-      THREAD_OFF (up->t_ka_timer);
-      THREAD_OFF (up->t_rs_timer);
-      PIM_UPSTREAM_FLAG_UNSET_SRC_STREAM (up->flags);
-      if (PIM_UPSTREAM_FLAG_TEST_CREATED_BY_UPSTREAM(up->flags))
-	{
-    pim_upstream_unset_created_by_upstream(up);
-	  pim_upstream_del (up, __PRETTY_FUNCTION__);
-	}
-    }
-  else
-    {
-      up->t_ka_timer = NULL;
-      pim_upstream_keep_alive_timer_start (up, qpim_keep_alive_time);
-    }
+  /* if entry was created because of activity we need to deref it */
+  if (PIM_UPSTREAM_FLAG_TEST_SRC_STREAM(up->flags))
+  {
+    pim_upstream_fhr_kat_expiry(up);
+    if (PIM_DEBUG_TRACE)
+      zlog_debug ("kat expired on %s; remove stream reference",
+          pim_str_sg_dump (&up->sg));
+    PIM_UPSTREAM_FLAG_UNSET_SRC_STREAM(up->flags);
+    pim_upstream_del(up, __PRETTY_FUNCTION__);
+  }
 
-  return 1;
+  return 0;
 }
 
 void
 pim_upstream_keep_alive_timer_start (struct pim_upstream *up,
 				     uint32_t time)
 {
+  if (!PIM_UPSTREAM_FLAG_TEST_SRC_STREAM(up->flags)) {
+    if (PIM_DEBUG_TRACE)
+      zlog_debug ("kat start on %s with no stream reference",
+          pim_str_sg_dump (&up->sg));
+  }
   THREAD_OFF (up->t_ka_timer);
   THREAD_TIMER_ON (master,
 		   up->t_ka_timer,
 		   pim_upstream_keep_alive_timer,
 		   up, time);
+
+  /* any time keepalive is started against a SG we will have to
+   * re-evaluate our active source database */
+  pim_msdp_sa_local_update(up);
+}
+
+/* MSDP on RP needs to know if a source is registerable to this RP */
+static int
+pim_upstream_msdp_reg_timer(struct thread *t)
+{
+  struct pim_upstream *up;
+
+  up = THREAD_ARG(t);
+  up->t_msdp_reg_timer = NULL;
+
+  /* source is no longer active - pull the SA from MSDP's cache */
+  pim_msdp_sa_local_del(&up->sg);
+  return 1;
+}
+void
+pim_upstream_msdp_reg_timer_start(struct pim_upstream *up)
+{
+  THREAD_OFF(up->t_msdp_reg_timer);
+  THREAD_TIMER_ON(master, up->t_msdp_reg_timer,
+      pim_upstream_msdp_reg_timer, up, PIM_MSDP_REG_RXED_PERIOD);
+
+  pim_msdp_sa_local_update(up);
 }
 
 /*
@@ -1130,7 +1181,6 @@ pim_upstream_register_stop_timer (struct thread *t)
   struct ip ip_hdr;
   up = THREAD_ARG (t);
 
-  THREAD_TIMER_OFF (up->t_rs_timer);
   up->t_rs_timer = NULL;
 
   if (PIM_DEBUG_TRACE)
@@ -1330,6 +1380,43 @@ pim_upstream_equal (const void *arg1, const void *arg2)
   return 0;
 }
 
+/* rfc4601:section-4.2:"Data Packet Forwarding Rules" defines
+ * the cases where kat has to be restarted on rxing traffic -
+ *
+ * if( DirectlyConnected(S) == TRUE AND iif == RPF_interface(S) ) {
+ * set KeepaliveTimer(S,G) to Keepalive_Period
+ * # Note: a register state transition or UpstreamJPState(S,G)
+ * # transition may happen as a result of restarting
+ * # KeepaliveTimer, and must be dealt with here.
+ * }
+ * if( iif == RPF_interface(S) AND UpstreamJPState(S,G) == Joined AND
+ * inherited_olist(S,G) != NULL ) {
+ * set KeepaliveTimer(S,G) to Keepalive_Period
+ * }
+ */
+static bool pim_upstream_kat_start_ok(struct pim_upstream *up)
+{
+  /* "iif == RPF_interface(S)" check has to be done by the kernel or hw
+   * so we will skip that here */
+  if (pim_if_connected_to_source(up->rpf.source_nexthop.interface,
+        up->sg.src)) {
+    return true;
+  }
+
+  if ((up->join_state == PIM_UPSTREAM_JOINED) &&
+            !pim_upstream_empty_inherited_olist(up)) {
+    /* XXX: I have added this RP check just for 3.2 and it's a digression from
+     * what rfc-4601 says. Till now we were only running KAT on FHR and RP and
+     * there is some angst around making the change to run it all routers that
+     * maintain the (S, G) state. This is tracked via CM-13601 and MUST be
+     * removed to handle spt turn-arounds correctly in a 3-tier clos */
+    if (I_am_RP (up->sg.grp))
+      return true;
+  }
+
+  return false;
+}
+
 /*
  * Code to check and see if we've received packets on a S,G mroute
  * and if so to set the SPT bit appropriately
@@ -1338,16 +1425,6 @@ static void
 pim_upstream_sg_running (void *arg)
 {
   struct pim_upstream *up = (struct pim_upstream *)arg;
-  long long now;
-
-  // If we are TRUE already no need to do more work
-  if (up->sptbit == PIM_UPSTREAM_SPTBIT_TRUE)
-    {
-      if (PIM_DEBUG_TRACE)
-	zlog_debug ("%s: %s sptbit is true", __PRETTY_FUNCTION__,
-		    pim_str_sg_dump(&up->sg));
-      return;
-    }
 
   // No packet can have arrived here if this is the case
   if (!up->channel_oil || !up->channel_oil->installed)
@@ -1358,21 +1435,11 @@ pim_upstream_sg_running (void *arg)
       return;
     }
 
-  // We need at least 30 seconds to see if we are getting packets
-  now = pim_time_monotonic_sec();
-  if (now - up->state_transition <= 30)
-    {
-      if (PIM_DEBUG_TRACE)
-	zlog_debug ("%s: %s uptime is %lld",
-		    __PRETTY_FUNCTION__, pim_str_sg_dump (&up->sg),
-		    now - up->state_transition);
-      return;
-    }
 
   pim_mroute_update_counters (up->channel_oil);
 
   // Have we seen packets?
-  if ((up->channel_oil->cc.oldpktcnt <= up->channel_oil->cc.pktcnt) &&
+  if ((up->channel_oil->cc.oldpktcnt >= up->channel_oil->cc.pktcnt) &&
       (up->channel_oil->cc.lastused/100 > 30))
     {
       if (PIM_DEBUG_TRACE)
@@ -1384,7 +1451,25 @@ pim_upstream_sg_running (void *arg)
       return;
     }
 
-  pim_upstream_set_sptbit (up, up->rpf.source_nexthop.interface);
+  if (pim_upstream_kat_start_ok(up)) {
+    /* Add a source reference to the stream if
+     * one doesn't already exist */
+    if (!PIM_UPSTREAM_FLAG_TEST_SRC_STREAM(up->flags))
+    {
+      if (PIM_DEBUG_TRACE)
+        zlog_debug ("source reference created on kat restart %s",
+            pim_str_sg_dump (&up->sg));
+      pim_upstream_ref(up, PIM_UPSTREAM_FLAG_MASK_SRC_STREAM);
+      PIM_UPSTREAM_FLAG_SET_SRC_STREAM(up->flags);
+      pim_upstream_fhr_kat_start(up);
+    }
+    pim_upstream_keep_alive_timer_start(up, qpim_keep_alive_time);
+  }
+
+  if (up->sptbit != PIM_UPSTREAM_SPTBIT_TRUE)
+  {
+    pim_upstream_set_sptbit(up, up->rpf.source_nexthop.interface);
+  }
   return;
 }
 
diff --git a/pimd/pim_upstream.h b/pimd/pim_upstream.h
index a073125021..b3823604c1 100644
--- a/pimd/pim_upstream.h
+++ b/pimd/pim_upstream.h
@@ -33,8 +33,7 @@
 #define PIM_UPSTREAM_FLAG_MASK_SRC_IGMP                (1 << 3)
 #define PIM_UPSTREAM_FLAG_MASK_SRC_PIM                 (1 << 4)
 #define PIM_UPSTREAM_FLAG_MASK_SRC_STREAM              (1 << 5)
-#define PIM_UPSTREAM_FLAG_MASK_CREATED_BY_UPSTREAM     (1 << 6)
-#define PIM_UPSTREAM_FLAG_MASK_SRC_MSDP                (1 << 7)
+#define PIM_UPSTREAM_FLAG_MASK_SRC_MSDP                (1 << 6)
 
 #define PIM_UPSTREAM_FLAG_TEST_DR_JOIN_DESIRED(flags) ((flags) & PIM_UPSTREAM_FLAG_MASK_DR_JOIN_DESIRED)
 #define PIM_UPSTREAM_FLAG_TEST_DR_JOIN_DESIRED_UPDATED(flags) ((flags) & PIM_UPSTREAM_FLAG_MASK_DR_JOIN_DESIRED_UPDATED)
@@ -42,7 +41,6 @@
 #define PIM_UPSTREAM_FLAG_TEST_SRC_IGMP(flags) ((flags) & PIM_UPSTREAM_FLAG_MASK_SRC_IGMP)
 #define PIM_UPSTREAM_FLAG_TEST_SRC_PIM(flags) ((flags) & PIM_UPSTREAM_FLAG_MASK_SRC_PIM)
 #define PIM_UPSTREAM_FLAG_TEST_SRC_STREAM(flags) ((flags) & PIM_UPSTREAM_FLAG_MASK_SRC_STREAM)
-#define PIM_UPSTREAM_FLAG_TEST_CREATED_BY_UPSTREAM(flags) ((flags) & PIM_UPSTREAM_FLAG_MASK_CREATED_BY_UPSTREAM)
 #define PIM_UPSTREAM_FLAG_TEST_SRC_MSDP(flags) ((flags) & PIM_UPSTREAM_FLAG_MASK_SRC_MSDP)
 
 #define PIM_UPSTREAM_FLAG_SET_DR_JOIN_DESIRED(flags) ((flags) |= PIM_UPSTREAM_FLAG_MASK_DR_JOIN_DESIRED)
@@ -51,7 +49,6 @@
 #define PIM_UPSTREAM_FLAG_SET_SRC_IGMP(flags) ((flags) |= PIM_UPSTREAM_FLAG_MASK_SRC_IGMP)
 #define PIM_UPSTREAM_FLAG_SET_SRC_PIM(flags) ((flags) |= PIM_UPSTREAM_FLAG_MASK_SRC_PIM)
 #define PIM_UPSTREAM_FLAG_SET_SRC_STREAM(flags) ((flags) |= PIM_UPSTREAM_FLAG_MASK_SRC_STREAM)
-#define PIM_UPSTREAM_FLAG_SET_CREATED_BY_UPSTREAM(flags) ((flags) |= PIM_UPSTREAM_FLAG_MASK_CREATED_BY_UPSTREAM)
 #define PIM_UPSTREAM_FLAG_SET_SRC_MSDP(flags) ((flags) |= PIM_UPSTREAM_FLAG_MASK_SRC_MSDP)
 
 #define PIM_UPSTREAM_FLAG_UNSET_DR_JOIN_DESIRED(flags) ((flags) &= ~PIM_UPSTREAM_FLAG_MASK_DR_JOIN_DESIRED)
@@ -60,7 +57,6 @@
 #define PIM_UPSTREAM_FLAG_UNSET_SRC_IGMP(flags) ((flags) &= ~PIM_UPSTREAM_FLAG_MASK_SRC_IGMP)
 #define PIM_UPSTREAM_FLAG_UNSET_SRC_PIM(flags) ((flags) &= ~PIM_UPSTREAM_FLAG_MASK_SRC_PIM)
 #define PIM_UPSTREAM_FLAG_UNSET_SRC_STREAM(flags) ((flags) &= ~PIM_UPSTREAM_FLAG_MASK_SRC_STREAM)
-#define PIM_UPSTREAM_FLAG_UNSET_CREATED_BY_UPSTREAM(flags) ((flags) &= ~PIM_UPSTREAM_FLAG_MASK_CREATED_BY_UPSTREAM)
 #define PIM_UPSTREAM_FLAG_UNSET_SRC_MSDP(flags) ((flags) &= ~PIM_UPSTREAM_FLAG_MASK_SRC_MSDP)
 
 enum pim_upstream_state {
@@ -114,6 +110,11 @@ struct pim_upstream {
 #define PIM_KEEPALIVE_PERIOD  (210)
 #define PIM_RP_KEEPALIVE_PERIOD ( 3 * qpim_register_suppress_time + qpim_register_probe_time )
 
+  /* on the RP we restart a timer to indicate if registers are being rxed for
+   * SG. This is needed by MSDP to determine its local SA cache */
+  struct thread           *t_msdp_reg_timer;
+#define PIM_MSDP_REG_RXED_PERIOD (3 * (1.5 * qpim_register_suppress_time))
+
   int64_t                  state_transition; /* Record current state uptime */
 };
 
@@ -164,9 +165,8 @@ int pim_upstream_inherited_olist (struct pim_upstream *up);
 int pim_upstream_empty_inherited_olist (struct pim_upstream *up);
 
 void pim_upstream_find_new_rpf (void);
+void pim_upstream_msdp_reg_timer_start(struct pim_upstream *up);
 
 void pim_upstream_init (void);
 void pim_upstream_terminate (void);
-void pim_upstream_set_created_by_upstream(struct pim_upstream *up);
-void pim_upstream_unset_created_by_upstream(struct pim_upstream *up);
 #endif /* PIM_UPSTREAM_H */