Merged in sandervrijders/ouroboros/be-sched-api (pull request #491)

ipcpd: normal: Extract flow sets from components

4 files changed, 58 insertions, 65 deletions

diff --git a/src/ipcpd/normal/dt.c b/src/ipcpd/normal/dt.c
index 72e0195e..593064f4 100644
--- a/src/ipcpd/normal/dt.c
+++ b/src/ipcpd/normal/dt.c
@@ -48,7 +48,6 @@
 #include <assert.h>
 
 struct {
-        flow_set_t *       set[QOS_CUBE_MAX];
         struct sdu_sched * sdu_sched;
 
         struct pff *       pff[QOS_CUBE_MAX];
@@ -64,19 +63,15 @@ struct {
 static int dt_neighbor_event(enum nb_event event,
                              struct conn   conn)
 {
-        qoscube_t cube;
-
         /* We are only interested in neighbors being added and removed. */
         switch (event) {
         case NEIGHBOR_ADDED:
-                ipcp_flow_get_qoscube(conn.flow_info.fd, &cube);
-                flow_set_add(dt.set[cube], conn.flow_info.fd);
-                log_dbg("Added fd %d to flow set.", conn.flow_info.fd);
+                sdu_sched_add(dt.sdu_sched, conn.flow_info.fd);
+                log_dbg("Added fd %d to SDU scheduler.", conn.flow_info.fd);
                 break;
         case NEIGHBOR_REMOVED:
-                ipcp_flow_get_qoscube(conn.flow_info.fd, &cube);
-                flow_set_del(dt.set[cube], conn.flow_info.fd);
-                log_dbg("Removed fd %d from flow set.", conn.flow_info.fd);
+                sdu_sched_del(dt.sdu_sched, conn.flow_info.fd);
+                log_dbg("Removed fd %d from SDU scheduler.", conn.flow_info.fd);
                 break;
         default:
                 break;
@@ -137,17 +132,8 @@ int dt_init(void)
         int              j;
         struct conn_info info;
 
-        for (i = 0; i < QOS_CUBE_MAX; ++i) {
-                dt.set[i] = flow_set_create();
-                if (dt.set[i] == NULL) {
-                        goto fail_flows;
-                        return -1;
-                }
-        }
-
         if (shm_pci_init()) {
                 log_err("Failed to init shm pci.");
-                goto fail_flows;
                 return -1;
         }
 
@@ -162,7 +148,7 @@ int dt_init(void)
         dt.ae = connmgr_ae_create(info);
         if (dt.ae == NULL) {
                 log_err("Failed to create AE struct.");
-                goto fail_flows;
+                return -1;
         }
 
         dt.nbs = nbs_create();
@@ -212,10 +198,6 @@ int dt_init(void)
         nbs_destroy(dt.nbs);
  fail_connmgr:
         connmgr_ae_destroy(dt.ae);
- fail_flows:
-        for (i = 0; i < QOS_CUBE_MAX; ++i)
-                flow_set_destroy(dt.set[i]);
-
         return -1;
 }
 
@@ -236,9 +218,6 @@ void dt_fini(void)
         nbs_destroy(dt.nbs);
 
         connmgr_ae_destroy(dt.ae);
-
-        for (i = 0; i < QOS_CUBE_MAX; ++i)
-                flow_set_destroy(dt.set[i]);
 }
 
 int dt_start(void)
@@ -257,7 +236,7 @@ int dt_start(void)
                 return -1;
         }
 
-        dt.sdu_sched = sdu_sched_create(dt.set, sdu_handler);
+        dt.sdu_sched = sdu_sched_create(sdu_handler);
         if (dt.sdu_sched == NULL) {
                 log_err("Failed to create N-1 SDU scheduler.");
                 gam_destroy(dt.gam);
diff --git a/src/ipcpd/normal/fa.c b/src/ipcpd/normal/fa.c
index be1080b1..6c3df7a1 100644
--- a/src/ipcpd/normal/fa.c
+++ b/src/ipcpd/normal/fa.c
@@ -48,7 +48,6 @@ struct {
         cep_id_t           fd_to_cep_id[AP_MAX_FLOWS];
         int                cep_id_to_fd[IPCPD_MAX_CONNS];
 
-        flow_set_t *       set[QOS_CUBE_MAX];
         struct sdu_sched * sdu_sched;
 } fa;
 
@@ -82,36 +81,20 @@ int fa_init(void)
         for (i = 0; i < IPCPD_MAX_CONNS; ++i)
                 fa.cep_id_to_fd[i] = -1;
 
-        for (i = 0; i < QOS_CUBE_MAX; ++i) {
-                fa.set[i] = flow_set_create();
-                if (fa.set[i] == NULL)
-                        goto fail_flows;
-        }
-
         if (pthread_rwlock_init(&fa.flows_lock, NULL))
-                goto fail_flows;
+                return -1;
 
         return 0;
-fail_flows:
-        for (i = 0; i < QOS_CUBE_MAX; ++i)
-                flow_set_destroy(fa.set[i]);
-
-        return -1;
 }
 
 void fa_fini(void)
 {
-        int i;
-
-        for (i = 0; i < QOS_CUBE_MAX; ++i)
-                flow_set_destroy(fa.set[i]);
-
         pthread_rwlock_destroy(&fa.flows_lock);
 }
 
 int fa_start(void)
 {
-        fa.sdu_sched = sdu_sched_create(fa.set, sdu_handler);
+        fa.sdu_sched = sdu_sched_create(sdu_handler);
         if (fa.sdu_sched == NULL) {
                 log_err("Failed to create SDU scheduler.");
                 return -1;
@@ -215,10 +198,8 @@ static int fa_flow_dealloc(int fd)
         flow_alloc_msg_t msg = FLOW_ALLOC_MSG__INIT;
         buffer_t         buf;
         int              ret;
-        qoscube_t        qc;
 
-        ipcp_flow_get_qoscube(fd, &qc);
-        flow_set_del(fa.set[qc], fd);
+        sdu_sched_del(fa.sdu_sched, fd);
 
         msg.code = FLOW_ALLOC_CODE__FLOW_DEALLOC;
 
@@ -296,7 +277,7 @@ int fa_alloc_resp(int fd,
                         free(buf.data);
                         return -1;
                 }
-                flow_set_add(fa.set[qc], fd);
+                sdu_sched_add(fa.sdu_sched, fd);
         }
 
         pthread_rwlock_unlock(&fa.flows_lock);
@@ -326,7 +307,6 @@ int fa_post_buf(cep_id_t   cep_id,
         int                ret = 0;
         int                fd;
         flow_alloc_msg_t * msg;
-        qoscube_t          qc;
 
         /* Depending on the message call the function in ipcp-dev.h */
 
@@ -393,9 +373,7 @@ int fa_post_buf(cep_id_t   cep_id,
                         fa.fd_to_cep_id[fd] = INVALID_CEP_ID;
                         fa.cep_id_to_fd[cep_id] = -1;
                 } else {
-                        ipcp_flow_get_qoscube(fd, &qc);
-                        flow_set_add(fa.set[qc],
-                                     fa.cep_id_to_fd[cep_id]);
+                        sdu_sched_add(fa.sdu_sched, fa.cep_id_to_fd[cep_id]);
                 }
 
                 pthread_rwlock_unlock(&fa.flows_lock);
@@ -403,8 +381,7 @@ int fa_post_buf(cep_id_t   cep_id,
                 break;
         case FLOW_ALLOC_CODE__FLOW_DEALLOC:
                 fd = fa.cep_id_to_fd[cep_id];
-                ipcp_flow_get_qoscube(fd, &qc);
-                flow_set_del(fa.set[qc], fd);
+                sdu_sched_del(fa.sdu_sched, fd);
                 ret = flow_dealloc(fd);
                 break;
         default:
diff --git a/src/ipcpd/normal/sdu_sched.c b/src/ipcpd/normal/sdu_sched.c
index 14fc672f..63259430 100644
--- a/src/ipcpd/normal/sdu_sched.c
+++ b/src/ipcpd/normal/sdu_sched.c
@@ -81,8 +81,7 @@ static void * sdu_reader(void * o)
         return (void *) 0;
 }
 
-struct sdu_sched * sdu_sched_create(flow_set_t * set[QOS_CUBE_MAX],
-                                    next_sdu_t   callback)
+struct sdu_sched * sdu_sched_create(next_sdu_t callback)
 {
         struct sdu_sched * sdu_sched;
         int                i;
@@ -95,14 +94,20 @@ struct sdu_sched * sdu_sched_create(flow_set_t * set[QOS_CUBE_MAX],
         sdu_sched->callback = callback;
 
         for (i = 0; i < QOS_CUBE_MAX; ++i) {
-                sdu_sched->set[i] = set[i];
+                sdu_sched->set[i] = flow_set_create();
+                if (sdu_sched->set[i] == NULL) {
+                        for (j = 0; j < i; ++j)
+                                flow_set_destroy(sdu_sched->set[j]);
+                        goto fail_sdu_sched;
+                }
+        }
 
+        for (i = 0; i < QOS_CUBE_MAX; ++i) {
                 sdu_sched->fqs[i] = fqueue_create();
                 if (sdu_sched->fqs[i] == NULL) {
-                        for (j = i; j >= 0; --j)
+                        for (j = 0; j < i; ++j)
                                 fqueue_destroy(sdu_sched->fqs[j]);
-                        free(sdu_sched);
-                        return NULL;
+                        goto fail_flow_set;
                 }
         }
 
@@ -112,6 +117,13 @@ struct sdu_sched * sdu_sched_create(flow_set_t * set[QOS_CUBE_MAX],
                        (void *) sdu_sched);
 
         return sdu_sched;
+
+ fail_flow_set:
+        for (i = 0; i < QOS_CUBE_MAX; ++i)
+                flow_set_destroy(sdu_sched->set[i]);
+ fail_sdu_sched:
+         free(sdu_sched);
+         return NULL;
 }
 
 void sdu_sched_destroy(struct sdu_sched * sdu_sched)
@@ -124,8 +136,28 @@ void sdu_sched_destroy(struct sdu_sched * sdu_sched)
 
         pthread_join(sdu_sched->sdu_reader, NULL);
 
-        for (i = 0; i < QOS_CUBE_MAX; ++i)
+        for (i = 0; i < QOS_CUBE_MAX; ++i) {
                 fqueue_destroy(sdu_sched->fqs[i]);
+                flow_set_destroy(sdu_sched->set[i]);
+        }
 
         free(sdu_sched);
 }
+
+void sdu_sched_add(struct sdu_sched * sdu_sched,
+                   int                fd)
+{
+        qoscube_t qc;
+
+        ipcp_flow_get_qoscube(fd, &qc);
+        flow_set_add(sdu_sched->set[qc], fd);
+}
+
+void sdu_sched_del(struct sdu_sched * sdu_sched,
+                   int                fd)
+{
+        qoscube_t qc;
+
+        ipcp_flow_get_qoscube(fd, &qc);
+        flow_set_del(sdu_sched->set[qc], fd);
+}
diff --git a/src/ipcpd/normal/sdu_sched.h b/src/ipcpd/normal/sdu_sched.h
index 1a22b041..3c95b683 100644
--- a/src/ipcpd/normal/sdu_sched.h
+++ b/src/ipcpd/normal/sdu_sched.h
@@ -30,9 +30,14 @@ typedef int (* next_sdu_t)(int                  fd,
                            qoscube_t            qc,
                            struct shm_du_buff * sdb);
 
-struct sdu_sched * sdu_sched_create(flow_set_t * set[QOS_CUBE_MAX],
-                                    next_sdu_t   callback);
+struct sdu_sched * sdu_sched_create(next_sdu_t callback);
 
 void               sdu_sched_destroy(struct sdu_sched * sdu_sched);
 
+void               sdu_sched_add(struct sdu_sched * sdu_sched,
+                                 int                fd);
+
+void               sdu_sched_del(struct sdu_sched * sdu_sched,
+                                 int                fd);
+
 #endif /* OUROBOROS_IPCPD_NORMAL_SDU_SCHED_H */