ipcpd: Change connection down to flow down

The DT component was flagging a connection as down and passing the fd
that was down. Of course the other components expect a connection
instead of just a fd. Now the connection manager will listen to flow
up and down events, and flag the connection up or down if needed.

Signed-off-by: Sander Vrijders <sander.vrijders@ugent.be>
Signed-off-by: Dimitri Staessens <dimitri.staessens@ugent.be>

7 files changed, 82 insertions, 19 deletions

diff --git a/src/ipcpd/normal/connmgr.c b/src/ipcpd/normal/connmgr.c
index 6301baed..9530633b 100644
--- a/src/ipcpd/normal/connmgr.c
+++ b/src/ipcpd/normal/connmgr.c
@@ -54,7 +54,6 @@ struct conn_el {
 };
 
 struct comp {
-        struct nbs *     nbs;
         struct conn_info info;
 
         struct list_head conns;
@@ -82,6 +81,29 @@ static int get_id_by_name(const char * name)
         return -1;
 }
 
+static int get_conn_by_fd(int           fd,
+                          enum comp_id  id,
+                          struct conn * conn)
+{
+        struct list_head * p;
+
+        pthread_mutex_lock(&connmgr.comps[id].lock);
+
+        list_for_each(p, &connmgr.comps[id].conns) {
+                struct conn_el * c =
+                        list_entry(p, struct conn_el, next);
+                if (c->conn.flow_info.fd == fd) {
+                        *conn = c->conn;
+                        pthread_mutex_unlock(&connmgr.comps[id].lock);
+                        return 0;
+                }
+        }
+
+        pthread_mutex_unlock(&connmgr.comps[id].lock);
+
+        return -1;
+}
+
 static int add_comp_conn(enum comp_id       id,
                          int                fd,
                          qosspec_t          qs,
@@ -163,10 +185,39 @@ static void * flow_acceptor(void * o)
         return (void *) 0;
 }
 
+static void handle_event(void *       self,
+                         int          event,
+                         const void * o)
+{
+        struct conn conn;
+
+        (void) self;
+
+        if (!(event == NOTIFY_DT_FLOW_UP || event == NOTIFY_DT_FLOW_DOWN))
+                return;
+
+        if (get_conn_by_fd(*((int *) o), COMPID_DT, &conn))
+                return;
+
+        switch (event) {
+        case NOTIFY_DT_FLOW_UP:
+                notifier_event(NOTIFY_DT_CONN_UP, &conn);
+                break;
+        case NOTIFY_DT_FLOW_DOWN:
+                notifier_event(NOTIFY_DT_CONN_DOWN, &conn);
+                break;
+        default:
+                break;
+        }
+}
+
 int connmgr_init(void)
 {
         connmgr.state = CONNMGR_INIT;
 
+        if (notifier_reg(handle_event, NULL))
+                return -1;
+
         return 0;
 }
 
@@ -174,6 +225,8 @@ void connmgr_fini(void)
 {
         int i;
 
+        notifier_unreg(handle_event);
+
         if (connmgr.state == CONNMGR_RUNNING)
                 pthread_join(connmgr.acceptor, NULL);
 
@@ -455,7 +508,7 @@ int connmgr_wait(enum comp_id  id,
         *conn = el->conn;
 
         list_del(&el->next);
-        free(el);
+        list_add(&el->next, &connmgr.comps[id].conns);
 
         pthread_mutex_unlock(&comp->lock);
 
diff --git a/src/ipcpd/normal/connmgr.h b/src/ipcpd/normal/connmgr.h
index cf627d60..510b8e4e 100644
--- a/src/ipcpd/normal/connmgr.h
+++ b/src/ipcpd/normal/connmgr.h
@@ -33,6 +33,8 @@
 #define NOTIFY_DT_CONN_QOS    0x00D2
 #define NOTIFY_DT_CONN_UP     0x00D3
 #define NOTIFY_DT_CONN_DOWN   0x00D4
+#define NOTIFY_DT_FLOW_UP     0x00D5
+#define NOTIFY_DT_FLOW_DOWN   0x00D6
 
 #define NOTIFY_MGMT_CONN_ADD  0x00F0
 #define NOTIFY_MGMT_CONN_DEL  0x00F1
diff --git a/src/ipcpd/normal/dt.c b/src/ipcpd/normal/dt.c
index 397169f3..b9d8934e 100644
--- a/src/ipcpd/normal/dt.c
+++ b/src/ipcpd/normal/dt.c
@@ -409,7 +409,7 @@ static void sdu_handler(int                  fd,
                 if (ret < 0) {
                         log_dbg("Failed to write SDU to fd %d.", ofd);
                         if (ret == -EFLOWDOWN)
-                                notifier_event(NOTIFY_DT_CONN_DOWN, &ofd);
+                                notifier_event(NOTIFY_DT_FLOW_DOWN, &ofd);
                         ipcp_sdb_release(sdb);
 #ifdef IPCP_FLOW_STATS
                         pthread_mutex_lock(&dt.stat[fd].lock);
@@ -781,7 +781,7 @@ int dt_write_sdu(uint64_t             dst_addr,
         if (ret < 0) {
                 log_dbg("Failed to write SDU to fd %d.", fd);
                 if (ret == -EFLOWDOWN)
-                        notifier_event(NOTIFY_DT_CONN_DOWN, &fd);
+                        notifier_event(NOTIFY_DT_FLOW_DOWN, &fd);
                 goto fail_write;
         }
 #ifdef IPCP_FLOW_STATS
diff --git a/src/ipcpd/normal/pff.c b/src/ipcpd/normal/pff.c
index a335bc04..08a50f4a 100644
--- a/src/ipcpd/normal/pff.c
+++ b/src/ipcpd/normal/pff.c
@@ -20,7 +20,10 @@
  * Foundation, Inc., http://www.fsf.org/about/contact/.
  */
 
+#define OUROBOROS_PREFIX "pff"
+
 #include <ouroboros/errno.h>
+#include <ouroboros/logs.h>
 
 #include "pff.h"
 #include "pol-pff-ops.h"
@@ -42,9 +45,11 @@ struct pff * pff_create(enum pol_pff pol)
 
         switch (pol) {
         case PFF_ALTERNATE:
+                log_dbg("Using alternate PFF policy.");
                 pff->ops = &alternate_pff_ops;
                 break;
         case PFF_SIMPLE:
+                log_dbg("Using simple PFF policy.");
                 pff->ops = &simple_pff_ops;
                 break;
         default:
diff --git a/src/ipcpd/normal/pol/alternate_pff.c b/src/ipcpd/normal/pol/alternate_pff.c
index 3cb99d9c..cfe00923 100644
--- a/src/ipcpd/normal/pol/alternate_pff.c
+++ b/src/ipcpd/normal/pol/alternate_pff.c
@@ -384,7 +384,7 @@ int alternate_flow_state_change(struct pff_i * pff_i,
                 } else {
                         /* Need to switch to a (different) alternate */
                         if (fds[0] == fd) {
-                                for (i = 0 ; i < len; i++) {
+                                for (i = 1; i < len; i++) {
                                         /* Usable alternate */
                                         if (!nhops_down_has(pff_i, fds[i])) {
                                                 tmp = fds[0];
diff --git a/src/ipcpd/normal/pol/link_state.c b/src/ipcpd/normal/pol/link_state.c
index 91c18a9d..c55bff70 100644
--- a/src/ipcpd/normal/pol/link_state.c
+++ b/src/ipcpd/normal/pol/link_state.c
@@ -785,7 +785,6 @@ static void handle_event(void *       self,
                         log_warn("Failed to add mgmt neighbor to LSDB.");
                 break;
         default:
-                log_info("Unknown routing event.");
                 break;
         }
 }
@@ -848,9 +847,11 @@ int link_state_init(enum pol_routing pr)
 
         switch (pr) {
         case ROUTING_LINK_STATE:
+                log_dbg("Using link state routing policy.");
                 ls.rtable = graph_routing_table;
                 break;
         case ROUTING_LINK_STATE_LFA:
+                log_dbg("Using Loop-Free Alternates policy.");
                 ls.rtable = graph_routing_table_lfa;
                 break;
         default:
diff --git a/src/lib/notifier.c b/src/lib/notifier.c
index 2c098f6f..593a0e70 100644
--- a/src/lib/notifier.c
+++ b/src/lib/notifier.c
@@ -20,6 +20,8 @@
  * Foundation, Inc., http://www.fsf.org/about/contact/.
  */
 
+#define _POSIX_C_SOURCE 200112L
+
 #include <ouroboros/errno.h>
 #include <ouroboros/notifier.h>
 #include <ouroboros/list.h>
@@ -35,12 +37,12 @@ struct listener {
 
 struct {
         struct list_head listeners;
-        pthread_mutex_t  lock;
+        pthread_rwlock_t lock;
 } notifier;
 
 int notifier_init(void)
 {
-        if (pthread_mutex_init(&notifier.lock, NULL))
+        if (pthread_rwlock_init(&notifier.lock, NULL))
                 return -1;
 
         list_head_init(&notifier.listeners);
@@ -53,7 +55,7 @@ void notifier_fini(void)
         struct list_head * p;
         struct list_head * h;
 
-        pthread_mutex_lock(&notifier.lock);
+        pthread_rwlock_wrlock(&notifier.lock);
 
         list_for_each_safe(p, h, &notifier.listeners) {
                 struct listener * l = list_entry(p, struct listener, next);
@@ -61,9 +63,9 @@ void notifier_fini(void)
                 free(l);
         }
 
-        pthread_mutex_unlock(&notifier.lock);
+        pthread_rwlock_unlock(&notifier.lock);
 
-        pthread_mutex_destroy(&notifier.lock);
+        pthread_rwlock_destroy(&notifier.lock);
 }
 
 void notifier_event(int          event,
@@ -71,14 +73,14 @@ void notifier_event(int          event,
 {
         struct list_head * p;
 
-        pthread_mutex_lock(&notifier.lock);
+        pthread_rwlock_rdlock(&notifier.lock);
 
         list_for_each(p, &notifier.listeners) {
                 struct listener * l = list_entry(p, struct listener, next);
                 l->callback(l->obj, event, o);
         }
 
-        pthread_mutex_unlock(&notifier.lock);
+        pthread_rwlock_unlock(&notifier.lock);
 }
 
 int notifier_reg(notifier_fn_t callback,
@@ -87,19 +89,19 @@ int notifier_reg(notifier_fn_t callback,
         struct listener *  l;
         struct list_head * p;
 
-        pthread_mutex_lock(&notifier.lock);
+        pthread_rwlock_wrlock(&notifier.lock);
 
         list_for_each(p, &notifier.listeners) {
                 struct listener * l = list_entry(p, struct listener, next);
                 if (l->callback == callback) {
-                        pthread_mutex_unlock(&notifier.lock);
+                        pthread_rwlock_unlock(&notifier.lock);
                         return -EPERM;
                 }
         }
 
         l = malloc(sizeof(*l));
         if (l == NULL) {
-                pthread_mutex_unlock(&notifier.lock);
+                pthread_rwlock_unlock(&notifier.lock);
                 return -ENOMEM;
         }
 
@@ -108,7 +110,7 @@ int notifier_reg(notifier_fn_t callback,
 
         list_add_tail(&l->next, &notifier.listeners);
 
-        pthread_mutex_unlock(&notifier.lock);
+        pthread_rwlock_unlock(&notifier.lock);
 
         return 0;
 }
@@ -118,7 +120,7 @@ void notifier_unreg(notifier_fn_t callback)
         struct list_head * p;
         struct list_head * h;
 
-        pthread_mutex_lock(&notifier.lock);
+        pthread_rwlock_wrlock(&notifier.lock);
 
         list_for_each_safe(p, h, &notifier.listeners) {
                 struct listener * l = list_entry(p, struct listener, next);
@@ -129,5 +131,5 @@ void notifier_unreg(notifier_fn_t callback)
                 }
         }
 
-        pthread_mutex_unlock(&notifier.lock);
+        pthread_rwlock_unlock(&notifier.lock);
 }