ipcpd: normal: Split connection establishment

Connection establishment was done at the same time as flow
allocation. This splits it more cleanly, and allows to re-use the DT
AE for other purposes.

1 files changed, 100 insertions, 205 deletions

diff --git a/src/ipcpd/normal/frct.c b/src/ipcpd/normal/frct.c
index 71e32bd1..1adcb299 100644
--- a/src/ipcpd/normal/frct.c
+++ b/src/ipcpd/normal/frct.c
@@ -90,309 +90,200 @@ static int release_cep_id(cep_id_t id)
         return ret;
 }
 
-static int init_cep_ids(void)
+int frct_init()
 {
+        int i;
+
+        if (frct_pci_init())
+                return -1;
+
         if (pthread_mutex_init(&frct.cep_ids_lock, NULL))
                 return -1;
 
         frct.cep_ids = bmp_create(IRMD_MAX_FLOWS, (INVALID_CEP_ID + 1));
-        if (frct.cep_ids == NULL) {
-                pthread_mutex_destroy(&frct.cep_ids_lock);
-                return -1;
-        }
+        if (frct.cep_ids == NULL)
+                goto fail_cep_ids_lock;
+
+        if (pthread_mutex_init(&frct.instances_lock, NULL))
+                goto fail_bmp;
+
+        frct.instances = malloc(sizeof(*(frct.instances)) * IRMD_MAX_FLOWS);
+        if (frct.instances == NULL)
+                goto fail_instance_lock;
+
+        for (i = 0; i < IRMD_MAX_FLOWS; i++)
+                frct.instances[i] = NULL;
 
         return 0;
+
+ fail_instance_lock:
+        pthread_mutex_destroy(&frct.instances_lock);
+ fail_bmp:
+        bmp_destroy(frct.cep_ids);
+ fail_cep_ids_lock:
+        pthread_mutex_destroy(&frct.cep_ids_lock);
+
+        return -1;
 }
 
-static int init_instances(void)
+int frct_fini()
 {
-        int i;
+        pthread_mutex_destroy(&frct.instances_lock);
 
-        if (pthread_mutex_init(&frct.instances_lock, NULL))
-                return -1;
+        free(frct.instances);
 
-        frct.instances = malloc(sizeof(*(frct.instances)) * IRMD_MAX_FLOWS);
-        if (frct.instances == NULL) {
-                pthread_mutex_destroy(&frct.instances_lock);
-                return -1;
-        }
+        bmp_destroy(frct.cep_ids);
 
-        for (i = 0; i < IRMD_MAX_FLOWS; i++)
-                frct.instances[i] = NULL;
+        pthread_mutex_destroy(&frct.cep_ids_lock);
 
         return 0;
 }
 
-static struct frct_i * create_frct_i(uint64_t address,
-                                     cep_id_t r_cep_id)
+cep_id_t frct_i_create(uint64_t   address,
+                       qoscube_t  cube)
 {
         struct frct_i * instance;
         cep_id_t        id;
 
+        pthread_mutex_lock(&frct.instances_lock);
+
         instance = malloc(sizeof(*instance));
         if (instance == NULL)
-                return NULL;
+                return INVALID_CEP_ID;
 
         id = next_cep_id();
         if (id == INVALID_CEP_ID) {
                 free(instance);
-                return NULL;
+                return INVALID_CEP_ID;
         }
 
         instance->r_address = address;
         instance->cep_id = id;
-        instance->r_cep_id = r_cep_id;
         instance->state = CONN_PENDING;
         instance->seqno = 0;
+        instance->cube = cube;
 
         frct.instances[id] = instance;
 
-        return instance;
-}
-
-static void destroy_frct_i(cep_id_t id)
-{
-        struct frct_i * instance;
-
-        instance = frct.instances[id];
-        frct.instances[id] = NULL;
-
-        release_cep_id(instance->cep_id);
-
-        free(instance);
-}
-
-static void fini_cep_ids(void)
-{
-        pthread_mutex_lock(&frct.cep_ids_lock);
-
-        bmp_destroy(frct.cep_ids);
-
-        pthread_mutex_unlock(&frct.cep_ids_lock);
+        pthread_mutex_unlock(&frct.instances_lock);
 
-        pthread_mutex_destroy(&frct.cep_ids_lock);
+        return id;
 }
 
-static void fini_instances(void)
+int frct_i_destroy(cep_id_t   cep_id)
 {
-        int i;
+        struct frct_i * instance;
 
         pthread_mutex_lock(&frct.instances_lock);
 
-        for (i = 0; i < IRMD_MAX_FLOWS; i++)
-                if (frct.instances[i] != NULL)
-                        destroy_frct_i(i);
-
-        pthread_mutex_unlock(&frct.instances_lock);
-
-        pthread_mutex_destroy(&frct.instances_lock);
-
-        free(frct.instances);
-}
-
-int frct_init()
-{
-        if (init_cep_ids())
-                return -1;
-
-        if (init_instances()) {
-                fini_cep_ids();
+        instance = frct.instances[cep_id];
+        if (instance == NULL) {
+                pthread_mutex_unlock(&frct.instances_lock);
+                log_err("Invalid instance.");
                 return -1;
         }
 
-        return 0;
-}
+        frct.instances[cep_id] = NULL;
 
-int frct_fini()
-{
-        fini_instances();
-        fini_cep_ids();
+        release_cep_id(instance->cep_id);
+        free(instance);
+
+        pthread_mutex_unlock(&frct.instances_lock);
 
         return 0;
 }
 
-int frct_post_sdu(struct pci *         pci,
-                  struct shm_du_buff * sdb)
+int frct_i_set_id(cep_id_t cep_id,
+                  cep_id_t r_cep_id)
 {
         struct frct_i * instance;
-        buffer_t        buf;
-        cep_id_t        id;
-
-        assert(pci);
-        assert(sdb);
 
-        if (pci->pdu_type == PDU_TYPE_MGMT) {
-                pthread_mutex_lock(&frct.instances_lock);
-
-                if (pci->dst_cep_id == INVALID_CEP_ID) {
-                        instance = create_frct_i(pci->src_addr,
-                                                 pci->src_cep_id);
-                        if (instance == NULL) {
-                                pthread_mutex_unlock(&frct.instances_lock);
-                                ipcp_sdb_release(sdb);
-                                return -ENOMEM;
-                        }
-                        id = instance->cep_id;
-                } else {
-                        instance = frct.instances[pci->dst_cep_id];
-                        if (instance == NULL) {
-                                pthread_mutex_unlock(&frct.instances_lock);
-                                ipcp_sdb_release(sdb);
-                                return -1;
-                        }
-                        id = pci->dst_cep_id;
-                        instance->state = CONN_ESTABLISHED;
-                }
-
-                instance->r_cep_id = pci->src_cep_id;
+        pthread_mutex_lock(&frct.instances_lock);
 
+        instance = frct.instances[cep_id];
+        if (instance == NULL) {
                 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 (fa_post_buf(id, &buf)) {
-                        log_err("Failed to hand buffer to FA.");
-                        ipcp_sdb_release(sdb);
-                        return -1;
-                }
-
-                ipcp_sdb_release(sdb);
-        } else {
-                /* FIXME: Known cep-ids are delivered to FA (minimal DTP) */
-                if (fa_post_sdu(pci->dst_cep_id, sdb)) {
-                        log_err("Failed to hand SDU to FA.");
-                        ipcp_sdb_release(sdb);
-                        return -1;
-                }
+                log_err("Invalid instance.");
+                return -1;
         }
 
+        instance->r_cep_id = r_cep_id;
+        instance->state = CONN_ESTABLISHED;
+
+        pthread_mutex_unlock(&frct.instances_lock);
+
         return 0;
 }
 
-cep_id_t frct_i_create(uint64_t   address,
-                       buffer_t * buf,
-                       qoscube_t  cube)
+cep_id_t frct_i_get_id(cep_id_t cep_id)
 {
         struct frct_i * instance;
-        struct pci      pci;
-        cep_id_t        id;
-
-        assert(buf);
-        assert(buf->data);
+        cep_id_t        r_cep_id;
 
         pthread_mutex_lock(&frct.instances_lock);
 
-        instance = create_frct_i(address, INVALID_CEP_ID);
+        instance = frct.instances[cep_id];
         if (instance == NULL) {
                 pthread_mutex_unlock(&frct.instances_lock);
                 return INVALID_CEP_ID;
         }
 
-        id = instance->cep_id;
-        instance->cube = cube;
+        r_cep_id = instance->r_cep_id;
 
         pthread_mutex_unlock(&frct.instances_lock);
 
-        pci.pdu_type = PDU_TYPE_MGMT;
-        pci.dst_addr = address;
-        pci.src_addr = ipcpi.dt_addr;
-        pci.dst_cep_id = 0;
-        pci.src_cep_id = id;
-        pci.seqno = 0;
-        pci.qos_id = cube;
-
-        if (dt_write_buf(&pci, buf)) {
-                pthread_mutex_lock(&frct.instances_lock);
-                destroy_frct_i(id);
-                pthread_mutex_unlock(&frct.instances_lock);
-                log_err("Failed to hand PDU to DT.");
-                return INVALID_CEP_ID;
-        }
-
-        return id;
+        return r_cep_id;
 }
 
-int frct_i_accept(cep_id_t   id,
-                  buffer_t * buf,
-                  qoscube_t  cube)
+uint64_t frct_i_get_addr(cep_id_t cep_id)
 {
-        struct pci      pci;
         struct frct_i * instance;
-
-        assert(buf);
-        assert(buf->data);
+        uint64_t        r_addr;
 
         pthread_mutex_lock(&frct.instances_lock);
 
-        instance = frct.instances[id];
+        instance = frct.instances[cep_id];
         if (instance == NULL) {
                 pthread_mutex_unlock(&frct.instances_lock);
-                log_err("Invalid instance.");
-                return -1;
-        }
-
-        if (instance->state != CONN_PENDING) {
-                pthread_mutex_unlock(&frct.instances_lock);
-                return -1;
+                return INVALID_ADDR;
         }
 
-        instance->state = CONN_ESTABLISHED;
-        instance->cube = cube;
-        instance->seqno = 0;
-
-        pci.pdu_type = PDU_TYPE_MGMT;
-        pci.dst_addr = instance->r_address;
-        pci.src_addr = ipcpi.dt_addr;
-        pci.dst_cep_id = instance->r_cep_id;
-        pci.src_cep_id = instance->cep_id;
-        pci.seqno = 0;
-        pci.qos_id = cube;
+        r_addr = instance->r_address;
 
         pthread_mutex_unlock(&frct.instances_lock);
 
-        if (dt_write_buf(&pci, buf))
-                return -1;
-
-        return 0;
+        return r_addr;
 }
 
-int frct_i_destroy(cep_id_t   id,
-                   buffer_t * buf)
+int frct_post_sdu(struct shm_du_buff * sdb)
 {
-        struct pci      pci;
+        struct frct_pci frct_pci;
         struct frct_i * instance;
 
+        assert(sdb);
+
+        frct_pci_des(sdb, &frct_pci);
+
+        /* Known cep-ids are delivered to FA (minimal DTP) */
         pthread_mutex_lock(&frct.instances_lock);
 
-        instance = frct.instances[id];
+        instance = frct.instances[frct_pci.dst_cep_id];
         if (instance == NULL) {
                 pthread_mutex_unlock(&frct.instances_lock);
                 log_err("Invalid instance.");
                 return -1;
         }
 
-        if (!(instance->state == CONN_PENDING ||
-              instance->state == CONN_ESTABLISHED)) {
+        if (instance->state != CONN_ESTABLISHED) {
                 pthread_mutex_unlock(&frct.instances_lock);
+                log_err("Connection is not established.");
                 return -1;
         }
 
-        pci.pdu_type = PDU_TYPE_MGMT;
-        pci.dst_addr = instance->r_address;
-        pci.src_addr = ipcpi.dt_addr;
-        pci.dst_cep_id = instance->r_cep_id;
-        pci.src_cep_id = instance->cep_id;
-        pci.seqno = 0;
-        pci.qos_id = instance->cube;
-
-        destroy_frct_i(id);
-
         pthread_mutex_unlock(&frct.instances_lock);
 
-        if (buf != NULL && buf->data != NULL)
-                if (dt_write_buf(&pci, buf))
-                        return -1;
+        if (fa_post_sdu_user(frct_pci.dst_cep_id, sdb))
+                return -1;
 
         return 0;
 }
@@ -400,8 +291,8 @@ int frct_i_destroy(cep_id_t   id,
 int frct_i_write_sdu(cep_id_t             id,
                      struct shm_du_buff * sdb)
 {
-        struct pci      pci;
         struct frct_i * instance;
+        struct frct_pci frct_pci;
 
         assert(sdb);
 
@@ -420,15 +311,19 @@ int frct_i_write_sdu(cep_id_t             id,
                 return -1;
         }
 
-        pci.pdu_type = PDU_TYPE_DTP;
-        pci.dst_addr = instance->r_address;
-        pci.src_addr = ipcpi.dt_addr;
-        pci.dst_cep_id = instance->r_cep_id;
-        pci.src_cep_id = instance->cep_id;
-        pci.seqno = (instance->seqno)++;
-        pci.qos_id = instance->cube;
+        frct_pci.dst_cep_id = instance->r_cep_id;
+        frct_pci.seqno = (instance->seqno)++;
+
+        if (frct_pci_ser(sdb, &frct_pci)) {
+                pthread_mutex_unlock(&frct.instances_lock);
+                log_err("Failed to serialize.");
+                return -1;
+        }
 
-        if (dt_write_sdu(&pci, sdb)) {
+        if (dt_write_sdu(instance->r_address,
+                         instance->cube,
+                         PDU_TYPE_FRCT,
+                         sdb)) {
                 pthread_mutex_unlock(&frct.instances_lock);
                 log_err("Failed to hand SDU to DT.");
                 return -1;