ipcpd: normal: React to flow events in link state policy

This will let the link state policy react to flow up and down events
by notifying the PFFs of the routing instances of this event so they
can take an appropriate action.

3 files changed, 56 insertions, 4 deletions

diff --git a/src/ipcpd/normal/connmgr.h b/src/ipcpd/normal/connmgr.h
index a8edee7d..2ad5316d 100644
--- a/src/ipcpd/normal/connmgr.h
+++ b/src/ipcpd/normal/connmgr.h
@@ -31,11 +31,11 @@
 #define NOTIFY_DT_CONN_ADD    0x00D0
 #define NOTIFY_DT_CONN_DEL    0x00D1
 #define NOTIFY_DT_CONN_QOS    0x00D2
-#define NOTIFY_DT_CONN_DOWN   0x00D3
+#define NOTIFY_DT_CONN_UP     0x00D3
+#define NOTIFY_DT_CONN_DOWN   0x00D4
 
 #define NOTIFY_MGMT_CONN_ADD  0x00F0
 #define NOTIFY_MGMT_CONN_DEL  0x00F1
-#define NOTIFY_MGMT_CONN_DOWN 0x00F2
 
 int         connmgr_init(void);
 
diff --git a/src/ipcpd/normal/enroll.c b/src/ipcpd/normal/enroll.c
index d245d0bd..ee87aa23 100644
--- a/src/ipcpd/normal/enroll.c
+++ b/src/ipcpd/normal/enroll.c
@@ -131,6 +131,7 @@ static int send_rcv_enroll_msg(int fd)
         enroll.conf.has_ttl        = reply->conf->has_ttl;
         enroll.conf.addr_auth_type = reply->conf->addr_auth_type;
         enroll.conf.routing_type   = reply->conf->routing_type;
+        enroll.conf.pff_type       = reply->conf->pff_type;
         enroll.conf.dif_info.dir_hash_algo
                 = reply->conf->dif_info->dir_hash_algo;
 
diff --git a/src/ipcpd/normal/pol/link_state.c b/src/ipcpd/normal/pol/link_state.c
index 6330cf81..f3af2771 100644
--- a/src/ipcpd/normal/pol/link_state.c
+++ b/src/ipcpd/normal/pol/link_state.c
@@ -62,8 +62,10 @@ typedef LinkStateMsg link_state_msg_t;
 #endif
 
 struct routing_i {
-        struct pff * pff;
-        pthread_t    calculator;
+        struct list_head next;
+
+        struct pff *     pff;
+        pthread_t        calculator;
 };
 
 /* TODO: link weight support. */
@@ -108,6 +110,9 @@ struct {
         pthread_t        lsreader;
         pthread_t        listener;
 
+        struct list_head routing_instances;
+        pthread_mutex_t  routing_i_lock;
+
         rtable_fn_t      rtable;
 } ls;
 
@@ -590,6 +595,24 @@ static void * lsreader(void * o)
         return (void *) 0;
 }
 
+static void flow_event(int  fd,
+                       bool up)
+{
+
+        struct list_head * p;
+
+        log_dbg("Notifying routing instances of flow event.");
+
+        pthread_mutex_lock(&ls.routing_i_lock);
+
+        list_for_each(p, &ls.routing_instances) {
+                struct routing_i * ri = list_entry(p, struct routing_i, next);
+                pff_flow_state_change(ri->pff, fd, up);
+        }
+
+        pthread_mutex_unlock(&ls.routing_i_lock);
+}
+
 static void handle_event(void *       self,
                          int          event,
                          const void * o)
@@ -614,6 +637,8 @@ static void handle_event(void *       self,
                 send_lsm(ipcpi.dt_addr, c->conn_info.addr);
                 break;
         case NOTIFY_DT_CONN_DEL:
+                flow_event(c->flow_info.fd, false);
+
                 if (lsdb_del_nb(c->conn_info.addr, c->flow_info.fd))
                         log_dbg("Failed to delete neighbor from LSDB.");
 
@@ -623,6 +648,12 @@ static void handle_event(void *       self,
         case NOTIFY_DT_CONN_QOS:
                 log_dbg("QoS changes currently unsupported.");
                 break;
+        case NOTIFY_DT_CONN_UP:
+                flow_event(c->flow_info.fd, true);
+                break;
+        case NOTIFY_DT_CONN_DOWN:
+                flow_event(c->flow_info.fd, false);
+                break;
         case NOTIFY_MGMT_CONN_ADD:
                 fset_add(ls.mgmt_set, c->flow_info.fd);
                 if (lsdb_add_nb(c->conn_info.addr, c->flow_info.fd, NB_MGMT))
@@ -656,6 +687,12 @@ struct routing_i * link_state_routing_i_create(struct pff * pff)
                 return NULL;
         }
 
+        pthread_mutex_lock(&ls.routing_i_lock);
+
+        list_add(&tmp->next, &ls.routing_instances);
+
+        pthread_mutex_unlock(&ls.routing_i_lock);
+
         return tmp;
 }
 
@@ -663,6 +700,12 @@ void link_state_routing_i_destroy(struct routing_i * instance)
 {
         assert(instance);
 
+        pthread_mutex_lock(&ls.routing_i_lock);
+
+        list_del(&instance->next);
+
+        pthread_mutex_unlock(&ls.routing_i_lock);
+
         pthread_cancel(instance->calculator);
 
         pthread_join(instance->calculator, NULL);
@@ -703,6 +746,9 @@ int link_state_init(enum pol_routing pr)
         if (pthread_rwlock_init(&ls.db_lock, NULL))
                 goto fail_db_lock_init;
 
+        if (pthread_mutex_init(&ls.routing_i_lock, NULL))
+                goto fail_routing_i_lock_init;
+
         if (connmgr_ae_init(AEID_MGMT, &info))
                 goto fail_connmgr_ae_init;
 
@@ -712,6 +758,7 @@ int link_state_init(enum pol_routing pr)
 
         list_head_init(&ls.db);
         list_head_init(&ls.nbs);
+        list_head_init(&ls.routing_instances);
 
         if (pthread_create(&ls.lsupdate, NULL, lsupdate, NULL))
                 goto fail_pthread_create_lsupdate;
@@ -739,6 +786,8 @@ int link_state_init(enum pol_routing pr)
  fail_fset_create:
         connmgr_ae_fini(AEID_MGMT);
  fail_connmgr_ae_init:
+        pthread_mutex_destroy(&ls.routing_i_lock);
+ fail_routing_i_lock_init:
         pthread_rwlock_destroy(&ls.db_lock);
  fail_db_lock_init:
         notifier_unreg(handle_event);
@@ -782,5 +831,7 @@ void link_state_fini(void)
 
         pthread_rwlock_destroy(&ls.db_lock);
 
+        pthread_mutex_destroy(&ls.routing_i_lock);
+
         notifier_unreg(handle_event);
 }