Merged in sandervrijders/ouroboros/be-fix-conn (pull request #291)0.3

ipcpd: normal: Fix bugs in connection establishment

4 files changed, 58 insertions, 46 deletions

diff --git a/src/ipcpd/normal/fmgr.c b/src/ipcpd/normal/fmgr.c
index b5ae40dd..607308c0 100644
--- a/src/ipcpd/normal/fmgr.c
+++ b/src/ipcpd/normal/fmgr.c
@@ -44,7 +44,7 @@
 #include "flow_alloc.pb-c.h"
 typedef FlowAllocMsg flow_alloc_msg_t;
 
-#define FD_UPDATE_TIMEOUT 100 /* microseconds */
+#define FD_UPDATE_TIMEOUT 100000 /* nanoseconds */
 
 struct np1_flow {
         int           fd;
@@ -54,7 +54,6 @@ struct np1_flow {
 
 struct nm1_flow {
         int           fd;
-        char *        ae_name;
         enum qos_cube qos;
 };
 
@@ -63,6 +62,7 @@ struct {
         struct nm1_flow ** nm1_flows;
         pthread_rwlock_t   nm1_flows_lock;
         flow_set_t *       nm1_set;
+        pthread_t          nm1_sdu_reader;
 
         struct np1_flow ** np1_flows;
         struct np1_flow ** np1_flows_cep;
@@ -75,26 +75,23 @@ struct {
 } fmgr;
 
 static int add_nm1_fd(int fd,
-                      char * ae_name,
                       enum qos_cube qos)
 {
         struct nm1_flow * tmp;
 
-        if (ae_name == NULL)
-                return -1;
-
         tmp = malloc(sizeof(*tmp));
         if (tmp == NULL)
                 return -1;
 
         tmp->fd = fd;
-        tmp->ae_name = ae_name;
         tmp->qos = qos;
 
         pthread_rwlock_wrlock(&fmgr.nm1_flows_lock);
         fmgr.nm1_flows[fd] = tmp;
         pthread_rwlock_unlock(&fmgr.nm1_flows_lock);
 
+        flow_set_add(fmgr.nm1_set, fd);
+
         /* FIXME: Temporary, until we have a PFF */
         fmgr.fd = fd;
 
@@ -116,7 +113,7 @@ static int add_np1_fd(int           fd,
         flow->fd = fd;
 
         fmgr.np1_flows[fd] = flow;
-        fmgr.np1_flows_cep[fd] = flow;
+        fmgr.np1_flows_cep[cep_id] = flow;
 
         return 0;
 }
@@ -170,14 +167,16 @@ static void * fmgr_nm1_acceptor(void * o)
                                 flow_dealloc(fd);
                                 continue;
                         }
+                } else {
+                        /* FIXME: Pass correct QoS cube */
+                        if (add_nm1_fd(fd, QOS_CUBE_BE)) {
+                                LOG_ERR("Failed to add fd to list.");
+                                flow_dealloc(fd);
+                                continue;
+                        }
                 }
 
-                /* FIXME: Pass correct QoS cube */
-                if (add_nm1_fd(fd, ae_name, QOS_CUBE_BE)) {
-                        LOG_ERR("Failed to add file descriptor to list.");
-                        flow_dealloc(fd);
-                        continue;
-                }
+                free(ae_name);
         }
 
         return (void *) 0;
@@ -365,6 +364,7 @@ int fmgr_init()
 
         pthread_create(&fmgr.nm1_flow_acceptor, NULL, fmgr_nm1_acceptor, NULL);
         pthread_create(&fmgr.np1_sdu_reader, NULL, fmgr_np1_sdu_reader, NULL);
+        pthread_create(&fmgr.nm1_sdu_reader, NULL, fmgr_nm1_sdu_reader, NULL);
 
         return 0;
 }
@@ -375,15 +375,15 @@ int fmgr_fini()
 
         pthread_cancel(fmgr.nm1_flow_acceptor);
         pthread_cancel(fmgr.np1_sdu_reader);
+        pthread_cancel(fmgr.nm1_sdu_reader);
 
         pthread_join(fmgr.nm1_flow_acceptor, NULL);
         pthread_join(fmgr.np1_sdu_reader, NULL);
+        pthread_join(fmgr.nm1_sdu_reader, NULL);
 
         for (i = 0; i < IRMD_MAX_FLOWS; i++) {
                 if (fmgr.nm1_flows[i] == NULL)
                         continue;
-                if (fmgr.nm1_flows[i]->ae_name != NULL)
-                        free(fmgr.nm1_flows[i]->ae_name);
                 if (ribmgr_remove_flow(fmgr.nm1_flows[i]->fd))
                     LOG_ERR("Failed to remove management flow.");
         }
@@ -447,8 +447,6 @@ int fmgr_np1_alloc(int           fd,
                 return -1;
         }
 
-        free(buf.data);
-
         if (add_np1_fd(fd, cep_id, qos)) {
                 pthread_rwlock_unlock(&fmgr.np1_flows_lock);
                 return -1;
@@ -671,17 +669,11 @@ int fmgr_nm1_mgmt_flow(char * dst_name)
 {
         int fd;
         int result;
-        char * ae_name;
-
-        ae_name = strdup(MGMT_AE);
-        if (ae_name == NULL)
-                return -1;
 
         /* FIXME: Request retransmission. */
         fd = flow_alloc(dst_name, MGMT_AE, NULL);
         if (fd < 0) {
                 LOG_ERR("Failed to allocate flow to %s", dst_name);
-                free(ae_name);
                 return -1;
         }
 
@@ -689,21 +681,12 @@ int fmgr_nm1_mgmt_flow(char * dst_name)
         if (result < 0) {
                 LOG_ERR("Result of flow allocation to %s is %d",
                         dst_name, result);
-                free(ae_name);
                 return -1;
         }
 
         if (ribmgr_add_flow(fd)) {
                 LOG_ERR("Failed to hand file descriptor to RIB manager");
                 flow_dealloc(fd);
-                free(ae_name);
-                return -1;
-        }
-
-        /* FIXME: Pass correct QoS cube */
-        if (add_nm1_fd(fd, ae_name, QOS_CUBE_BE)) {
-                LOG_ERR("Failed to add file descriptor to list.");
-                flow_dealloc(fd);
                 return -1;
         }
 
@@ -715,17 +698,11 @@ int fmgr_nm1_dt_flow(char * dst_name,
 {
         int fd;
         int result;
-        char * ae_name;
-
-        ae_name = strdup(DT_AE);
-        if (ae_name == NULL)
-                return -1;
 
         /* FIXME: Map qos cube on correct QoS. */
         fd = flow_alloc(dst_name, DT_AE, NULL);
         if (fd < 0) {
                 LOG_ERR("Failed to allocate flow to %s", dst_name);
-                free(ae_name);
                 return -1;
         }
 
@@ -733,14 +710,12 @@ int fmgr_nm1_dt_flow(char * dst_name,
         if (result < 0) {
                 LOG_ERR("Result of flow allocation to %s is %d",
                         dst_name, result);
-                free(ae_name);
                 return -1;
         }
 
-        if (add_nm1_fd(fd, ae_name, qos)) {
+        if (add_nm1_fd(fd, qos)) {
                 LOG_ERR("Failed to add file descriptor to list.");
                 flow_dealloc(fd);
-                free(ae_name);
                 return -1;
         }
 
diff --git a/src/ipcpd/normal/frct.c b/src/ipcpd/normal/frct.c
index 3f80f91e..ff980772 100644
--- a/src/ipcpd/normal/frct.c
+++ b/src/ipcpd/normal/frct.c
@@ -156,7 +156,8 @@ int frct_nm1_post_sdu(struct pci * pci,
         if (pci == NULL || sdb == NULL)
                 return -1;
 
-        if (pci->dst_cep_id == INVALID_CEP_ID) {
+        if (pci->dst_cep_id == INVALID_CEP_ID &&
+            pci->pdu_type == PDU_TYPE_MGMT) {
                 pthread_mutex_lock(&frct.instances_lock);
                 instance = create_frct_i(pci->src_addr,
                                          pci->src_cep_id);
@@ -176,6 +177,25 @@ int frct_nm1_post_sdu(struct pci * pci,
                         free(pci);
                         return -1;
                 }
+        } else if (pci->pdu_type == PDU_TYPE_MGMT) {
+                pthread_mutex_lock(&frct.instances_lock);
+                instance = frct.instances[pci->dst_cep_id];
+                if (instance == NULL) {
+                        pthread_mutex_unlock(&frct.instances_lock);
+                        return -1;
+                }
+                instance->r_cep_id = pci->src_cep_id;
+                instance->state = CONN_ESTABLISHED;
+                pthread_mutex_unlock(&frct.instances_lock);
+
+                buf.len = shm_du_buff_tail(sdb) - shm_du_buff_head(sdb);
+                buf.data = shm_du_buff_head(sdb);
+
+                if (fmgr_np1_post_buf(pci->dst_cep_id, &buf)) {
+                        LOG_ERR("Failed to hand buffer to Flow Manager.");
+                        free(pci);
+                        return -1;
+                }
         } else {
                 /* FIXME: Known cep-ids are delivered to FMGR (minimal DTP) */
                 if (fmgr_np1_post_sdu(pci->dst_cep_id, sdb)) {
@@ -357,5 +377,7 @@ int frct_i_write_sdu(cep_id_t             id,
                 return -1;
         }
 
+        pthread_mutex_unlock(&frct.instances_lock);
+
         return 0;
 }
diff --git a/src/ipcpd/normal/main.c b/src/ipcpd/normal/main.c
index 2402972f..a5161718 100644
--- a/src/ipcpd/normal/main.c
+++ b/src/ipcpd/normal/main.c
@@ -108,6 +108,10 @@ static int normal_ipcp_name_query(char * name)
          * for certain names.
          */
 
+        /* FIXME: Here for testing purposes */
+        if (strcmp(name, "normal.app") == 0)
+                return 0;
+
         return -1;
 }
 
@@ -145,6 +149,12 @@ static int normal_ipcp_enroll(char * dif_name)
 
         pthread_rwlock_unlock(&ipcpi.state_lock);
 
+        /* FIXME: Remove once we obtain neighbors during enrollment */
+        if (fmgr_nm1_dt_flow(dif_name, QOS_CUBE_BE)) {
+                LOG_ERR("Failed to establish data transfer flow.");
+                return -1;
+        }
+
         return 0;
 }
 
diff --git a/src/ipcpd/normal/shm_pci.c b/src/ipcpd/normal/shm_pci.c
index aa18fa38..3c55d081 100644
--- a/src/ipcpd/normal/shm_pci.c
+++ b/src/ipcpd/normal/shm_pci.c
@@ -34,6 +34,7 @@
 #include "crc32.h"
 #include "ribmgr.h"
 
+#define PDU_TYPE_SIZE 1
 #define QOS_ID_SIZE 1
 #define DEFAULT_TTL 60
 #define TTL_SIZE 1
@@ -43,7 +44,7 @@ static size_t shm_pci_head_size(struct dt_const * dtc)
 {
         size_t len = 0;
 
-        len = dtc->addr_size * 2 + dtc->cep_id_size * 2
+        len = PDU_TYPE_SIZE + dtc->addr_size * 2 + dtc->cep_id_size * 2
                 + dtc->pdu_length_size + dtc->seqno_size + QOS_ID_SIZE;
 
         if (dtc->has_ttl)
@@ -64,7 +65,9 @@ static void ser_pci_head(uint8_t * head,
         int offset = 0;
         uint8_t ttl = DEFAULT_TTL;
 
-        memcpy(head, &pci->dst_addr, dtc->addr_size);
+        memcpy(head, &pci->pdu_type, PDU_TYPE_SIZE);
+        offset += PDU_TYPE_SIZE;
+        memcpy(head + offset, &pci->dst_addr, dtc->addr_size);
         offset += dtc->addr_size;
         memcpy(head + offset, &pci->src_addr, dtc->addr_size);
         offset += dtc->addr_size;
@@ -174,7 +177,9 @@ struct pci * shm_pci_des(struct shm_du_buff * sdb)
         if (pci == NULL)
                 return NULL;
 
-        memcpy(&pci->dst_addr, head, dtc->addr_size);
+        memcpy(&pci->pdu_type, head, PDU_TYPE_SIZE);
+        offset += PDU_TYPE_SIZE;
+        memcpy(&pci->dst_addr, head + offset, dtc->addr_size);
         offset += dtc->addr_size;
         memcpy(&pci->src_addr, head + offset, dtc->addr_size);
         offset += dtc->addr_size;