Merged in sandervrijders/ouroboros/be-connection (pull request #262)

ipcpd: normal: First version of the fast path bootstrap

1 files changed, 428 insertions, 210 deletions

diff --git a/src/ipcpd/normal/fmgr.c b/src/ipcpd/normal/fmgr.c
index b6ec1984..8c627641 100644
--- a/src/ipcpd/normal/fmgr.c
+++ b/src/ipcpd/normal/fmgr.c
@@ -27,6 +27,8 @@
 #include <ouroboros/dev.h>
 #include <ouroboros/list.h>
 #include <ouroboros/ipcp-dev.h>
+#include <ouroboros/select.h>
+#include <ouroboros/errno.h>
 
 #include <stdlib.h>
 #include <stdbool.h>
@@ -37,38 +39,46 @@
 #include "ribmgr.h"
 #include "frct.h"
 #include "ipcp.h"
+#include "shm_pci.h"
 
 #include "flow_alloc.pb-c.h"
 typedef FlowAllocMsg flow_alloc_msg_t;
 
-struct n_flow {
-        int fd;
-        struct frct_i * frct_i;
-        enum qos_cube qos;
+#define FD_UPDATE_TIMEOUT 100 /* microseconds */
 
-        struct list_head next;
+struct np1_flow {
+        int           fd;
+        cep_id_t      cep_id;
+        enum qos_cube qos;
 };
 
-struct n_1_flow {
-        int fd;
-        char * ae_name;
-        struct list_head next;
+struct nm1_flow {
+        int           fd;
+        char *        ae_name;
+        enum qos_cube qos;
 };
 
 struct {
-        pthread_t listen_thread;
-
-        struct list_head n_1_flows;
-        pthread_mutex_t n_1_flows_lock;
-
-        struct list_head n_flows;
-        /* FIXME: Make this a read/write lock */
-        pthread_mutex_t n_flows_lock;
+        pthread_t          nm1_flow_acceptor;
+        struct nm1_flow ** nm1_flows;
+        pthread_rwlock_t   nm1_flows_lock;
+        flow_set_t *       nm1_set;
+
+        struct np1_flow ** np1_flows;
+        struct np1_flow ** np1_flows_cep;
+        pthread_rwlock_t   np1_flows_lock;
+        flow_set_t *       np1_set;
+        pthread_t          np1_sdu_reader;
+
+        /* FIXME: Replace with PFF */
+        int fd;
 } fmgr;
 
-static int add_n_1_fd(int fd, char * ae_name)
+static int add_nm1_fd(int fd,
+                      char * ae_name,
+                      enum qos_cube qos)
 {
-        struct n_1_flow * tmp;
+        struct nm1_flow * tmp;
 
         if (ae_name == NULL)
                 return -1;
@@ -79,19 +89,41 @@ static int add_n_1_fd(int fd, char * ae_name)
 
         tmp->fd = fd;
         tmp->ae_name = ae_name;
+        tmp->qos = qos;
 
-        INIT_LIST_HEAD(&tmp->next);
+        pthread_rwlock_wrlock(&fmgr.nm1_flows_lock);
+        fmgr.nm1_flows[fd] = tmp;
+        pthread_rwlock_unlock(&fmgr.nm1_flows_lock);
 
-        pthread_mutex_lock(&fmgr.n_1_flows_lock);
-        list_add(&tmp->next, &fmgr.n_1_flows);
-        pthread_mutex_unlock(&fmgr.n_1_flows_lock);
+        /* FIXME: Temporary, until we have a PFF */
+        fmgr.fd = fd;
 
         return 0;
 }
 
-static void * fmgr_listen(void * o)
+static int add_np1_fd(int           fd,
+                      cep_id_t      cep_id,
+                      enum qos_cube qos)
 {
-        int fd;
+        struct np1_flow * flow;
+
+        flow = malloc(sizeof(*flow));
+        if (flow == NULL)
+                return -1;
+
+        flow->cep_id = cep_id;
+        flow->qos = qos;
+        flow->fd = fd;
+
+        fmgr.np1_flows[fd] = flow;
+        fmgr.np1_flows_cep[fd] = flow;
+
+        return 0;
+}
+
+static void * fmgr_nm1_acceptor(void * o)
+{
+        int    fd;
         char * ae_name;
 
         while (true) {
@@ -137,16 +169,8 @@ static void * fmgr_listen(void * o)
                         }
                 }
 
-                if (strcmp(ae_name, DT_AE) == 0) {
-                        /* FIXME: Pass correct QoS cube */
-                        if (frct_dt_flow(fd, 0)) {
-                                LOG_ERR("Failed to hand fd to FRCT.");
-                                flow_dealloc(fd);
-                                continue;
-                        }
-                }
-
-                if (add_n_1_fd(fd, ae_name)) {
+                /* 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;
@@ -156,169 +180,215 @@ static void * fmgr_listen(void * o)
         return (void *) 0;
 }
 
-int fmgr_init()
+static void * fmgr_np1_sdu_reader(void * o)
 {
-        INIT_LIST_HEAD(&fmgr.n_1_flows);
-        INIT_LIST_HEAD(&fmgr.n_flows);
+        struct shm_du_buff * sdb;
+        struct timespec timeout = {0, FD_UPDATE_TIMEOUT};
+        struct np1_flow * flow;
 
-        pthread_mutex_init(&fmgr.n_1_flows_lock, NULL);
-        pthread_mutex_init(&fmgr.n_flows_lock, NULL);
-
-        pthread_create(&fmgr.listen_thread, NULL, fmgr_listen, NULL);
+        while (true) {
+                int fd = flow_select(fmgr.np1_set, &timeout);
+                if (fd == -ETIMEDOUT)
+                        continue;
 
-        return 0;
-}
+                if (fd < 0) {
+                        LOG_ERR("Failed to get active fd.");
+                        continue;
+                }
 
-int fmgr_fini()
-{
-        struct list_head * pos = NULL;
+                if (ipcp_flow_read(fd, &sdb)) {
+                        LOG_ERR("Failed to read SDU from fd %d.", fd);
+                        continue;
+                }
 
-        pthread_cancel(fmgr.listen_thread);
+                pthread_rwlock_rdlock(&fmgr.np1_flows_lock);
+                flow = fmgr.np1_flows[fd];
+                if (flow == NULL) {
+                        pthread_rwlock_unlock(&fmgr.np1_flows_lock);
+                        ipcp_flow_del(sdb);
+                        LOG_ERR("Failed to retrieve flow.");
+                        continue;
+                }
 
-        pthread_join(fmgr.listen_thread, NULL);
+                if (frct_i_write_sdu(flow->cep_id, sdb)) {
+                        pthread_rwlock_unlock(&fmgr.np1_flows_lock);
+                        ipcp_flow_del(sdb);
+                        LOG_ERR("Failed to hand SDU to FRCT.");
+                        continue;
+                }
 
-        list_for_each(pos, &fmgr.n_1_flows) {
-                struct n_1_flow * e = list_entry(pos, struct n_1_flow, next);
-                if (e->ae_name != NULL)
-                        free(e->ae_name);
-                if (ribmgr_remove_flow(e->fd))
-                    LOG_ERR("Failed to remove management flow.");
+                pthread_rwlock_unlock(&fmgr.np1_flows_lock);
         }
 
-        pthread_mutex_destroy(&fmgr.n_1_flows_lock);
-        pthread_mutex_destroy(&fmgr.n_flows_lock);
-
-        return 0;
+        return (void *) 0;
 }
 
-int fmgr_mgmt_flow(char * dst_name)
+void * fmgr_nm1_sdu_reader(void * o)
 {
-        int fd;
-        int result;
-        char * ae_name;
+        struct timespec timeout = {0, FD_UPDATE_TIMEOUT};
+        struct shm_du_buff * sdb;
+        struct pci * pci;
 
-        ae_name = strdup(MGMT_AE);
-        if (ae_name == NULL)
-                return -1;
+        while (true) {
+                int fd = flow_select(fmgr.nm1_set, &timeout);
+                if (fd == -ETIMEDOUT)
+                        continue;
 
-        /* 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;
-        }
+                if (fd < 0) {
+                        LOG_ERR("Failed to get active fd.");
+                        continue;
+                }
 
-        result = flow_alloc_res(fd);
-        if (result < 0) {
-                LOG_ERR("Result of flow allocation to %s is %d",
-                        dst_name, result);
-                free(ae_name);
-                return -1;
-        }
+                if (ipcp_flow_read(fd, &sdb)) {
+                        LOG_ERR("Failed to read SDU from fd %d.", fd);
+                        continue;
+                }
 
-        if (ribmgr_add_flow(fd)) {
-                LOG_ERR("Failed to hand file descriptor to RIB manager");
-                flow_dealloc(fd);
-                free(ae_name);
-                return -1;
-        }
+                pci = shm_pci_des(sdb);
+                if (pci == NULL) {
+                        LOG_ERR("Failed to get PCI.");
+                        ipcp_flow_del(sdb);
+                        continue;
+                }
 
-        if (add_n_1_fd(fd, ae_name)) {
-                LOG_ERR("Failed to add file descriptor to list.");
-                flow_dealloc(fd);
-                return -1;
+                if (pci->dst_addr != ribmgr_address()) {
+                        LOG_DBG("PDU needs to be forwarded.");
+
+                        if (pci->ttl == 0) {
+                                LOG_DBG("TTL was zero.");
+                                ipcp_flow_del(sdb);
+                                free(pci);
+                                continue;
+                        }
+
+                        if (shm_pci_dec_ttl(sdb)) {
+                                LOG_ERR("Failed to decrease TTL.");
+                                ipcp_flow_del(sdb);
+                                free(pci);
+                                continue;
+                        }
+                        /*
+                         * FIXME: Dropping for now, since
+                         * we don't have a PFF yet
+                         */
+                        ipcp_flow_del(sdb);
+                        free(pci);
+                        continue;
+                }
+
+                if (shm_pci_shrink(sdb)) {
+                        LOG_ERR("Failed to shrink PDU.");
+                        ipcp_flow_del(sdb);
+                        free(pci);
+                        continue;
+                }
+
+                if (frct_nm1_post_sdu(pci, sdb)) {
+                        LOG_ERR("Failed to hand PDU to FRCT.");
+                        ipcp_flow_del(sdb);
+                        free(pci);
+                        continue;
+                }
         }
 
-        return 0;
+        return (void *) 0;
 }
 
-int fmgr_dt_flow(char * dst_name, enum qos_cube qos)
+int fmgr_init()
 {
-        int fd;
-        int result;
-        char * ae_name;
+        int i;
 
-        ae_name = strdup(DT_AE);
-        if (ae_name == NULL)
+        fmgr.nm1_flows = malloc(sizeof(*(fmgr.nm1_flows)) * IRMD_MAX_FLOWS);
+        if (fmgr.nm1_flows == 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);
+        fmgr.np1_flows = malloc(sizeof(*(fmgr.np1_flows)) * IRMD_MAX_FLOWS);
+        if (fmgr.np1_flows == NULL) {
+                free(fmgr.nm1_flows);
                 return -1;
         }
 
-        result = flow_alloc_res(fd);
-        if (result < 0) {
-                LOG_ERR("Result of flow allocation to %s is %d",
-                        dst_name, result);
-                free(ae_name);
+        fmgr.np1_flows_cep =
+                malloc(sizeof(*(fmgr.np1_flows_cep)) * IRMD_MAX_FLOWS);
+        if (fmgr.np1_flows_cep == NULL) {
+                free(fmgr.np1_flows);
+                free(fmgr.nm1_flows);
                 return -1;
         }
 
-        if (frct_dt_flow(fd, qos)) {
-                LOG_ERR("Failed to hand file descriptor to FRCT");
-                flow_dealloc(fd);
-                free(ae_name);
+        for (i = 0; i < IRMD_MAX_FLOWS; i++) {
+                fmgr.nm1_flows[i] = NULL;
+                fmgr.np1_flows[i] = NULL;
+                fmgr.np1_flows_cep[i] = NULL;
+        }
+
+        pthread_rwlock_init(&fmgr.nm1_flows_lock, NULL);
+        pthread_rwlock_init(&fmgr.np1_flows_lock, NULL);
+
+        fmgr.np1_set = flow_set_create();
+        if (fmgr.np1_set == NULL) {
+                free(fmgr.np1_flows_cep);
+                free(fmgr.np1_flows);
+                free(fmgr.nm1_flows);
                 return -1;
         }
 
-        if (add_n_1_fd(fd, ae_name)) {
-                LOG_ERR("Failed to add file descriptor to list.");
-                flow_dealloc(fd);
-                free(ae_name);
+        fmgr.nm1_set = flow_set_create();
+        if (fmgr.nm1_set == NULL) {
+                flow_set_destroy(fmgr.np1_set);
+                free(fmgr.np1_flows_cep);
+                free(fmgr.np1_flows);
+                free(fmgr.nm1_flows);
                 return -1;
         }
 
+        pthread_create(&fmgr.nm1_flow_acceptor, NULL, fmgr_nm1_acceptor, NULL);
+        pthread_create(&fmgr.np1_sdu_reader, NULL, fmgr_np1_sdu_reader, NULL);
+
         return 0;
 }
 
-/* Call under n_flows lock */
-static struct n_flow * get_n_flow_by_fd(int fd)
+int fmgr_fini()
 {
-        struct list_head * pos = NULL;
+        int i;
 
-        list_for_each(pos, &fmgr.n_flows) {
-                struct n_flow * e = list_entry(pos, struct n_flow, next);
-                if (e->fd == fd)
-                        return e;
-        }
+        pthread_cancel(fmgr.nm1_flow_acceptor);
+        pthread_cancel(fmgr.np1_sdu_reader);
 
-        return NULL;
-}
+        pthread_join(fmgr.nm1_flow_acceptor, NULL);
+        pthread_join(fmgr.np1_sdu_reader, NULL);
 
-/* Call under n_flows lock */
-static struct n_flow * get_n_flow_by_frct_i(struct frct_i * frct_i)
-{
-        struct list_head * pos = NULL;
-
-        list_for_each(pos, &fmgr.n_flows) {
-                struct n_flow * e = list_entry(pos, struct n_flow, next);
-                if (e->frct_i == frct_i)
-                        return e;
+        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.");
         }
 
-        return NULL;
+        pthread_rwlock_destroy(&fmgr.nm1_flows_lock);
+        pthread_rwlock_destroy(&fmgr.np1_flows_lock);
+
+        flow_set_destroy(fmgr.nm1_set);
+        flow_set_destroy(fmgr.np1_set);
+        free(fmgr.np1_flows_cep);
+        free(fmgr.np1_flows);
+        free(fmgr.nm1_flows);
+
+        return 0;
 }
 
-int fmgr_flow_alloc(int           fd,
+int fmgr_np1_alloc(int           fd,
                     char *        dst_ap_name,
                     char *        src_ae_name,
                     enum qos_cube qos)
 {
-        struct n_flow * flow;
-        struct frct_i * frct_i;
+        cep_id_t cep_id;
         uint32_t address = 0;
         buffer_t buf;
         flow_alloc_msg_t msg = FLOW_ALLOC_MSG__INIT;
 
-        flow = malloc(sizeof(*flow));
-        if (flow == NULL)
-                return -1;
-
         /* FIXME: Obtain correct address here from DIF NSM */
 
         msg.code = FLOW_ALLOC_CODE__FLOW_REQ;
@@ -328,53 +398,47 @@ int fmgr_flow_alloc(int           fd,
         msg.has_qos_cube = true;
 
         buf.len = flow_alloc_msg__get_packed_size(&msg);
-        if (buf.len == 0) {
-                free(flow);
+        if (buf.len == 0)
                 return -1;
-        }
 
         buf.data = malloc(buf.len);
-        if (buf.data == NULL) {
-                free(flow);
+        if (buf.data == NULL)
                 return -1;
-        }
 
         flow_alloc_msg__pack(&msg, buf.data);
 
-        pthread_mutex_lock(&fmgr.n_flows_lock);
+        pthread_rwlock_wrlock(&fmgr.np1_flows_lock);
 
-        frct_i = frct_i_create(address, &buf, qos);
-        if (frct_i == NULL) {
+        cep_id = frct_i_create(address, &buf, qos);
+        if (cep_id == INVALID_CEP_ID) {
                 free(buf.data);
-                free(flow);
-                pthread_mutex_unlock(&fmgr.n_flows_lock);
+                pthread_rwlock_unlock(&fmgr.np1_flows_lock);
                 return -1;
         }
 
         free(buf.data);
 
-        flow->fd     = fd;
-        flow->frct_i = frct_i;
-        flow->qos    = qos;
-
-        INIT_LIST_HEAD(&flow->next);
-
-        list_add(&flow->next, &fmgr.n_flows);
+        if (add_np1_fd(fd, cep_id, qos)) {
+                pthread_rwlock_unlock(&fmgr.np1_flows_lock);
+                return -1;
+        }
 
-        pthread_mutex_unlock(&fmgr.n_flows_lock);
+        pthread_rwlock_unlock(&fmgr.np1_flows_lock);
 
         return 0;
 }
 
-/* Call under n_flows lock */
-static int n_flow_dealloc(int fd)
+/* Call under np1_flows lock */
+static int np1_flow_dealloc(int fd)
 {
-        struct n_flow * flow;
+        struct np1_flow * flow;
         flow_alloc_msg_t msg = FLOW_ALLOC_MSG__INIT;
         buffer_t buf;
         int ret;
 
-        flow = get_n_flow_by_fd(fd);
+        flow_set_del(fmgr.np1_set, fd);
+
+        flow = fmgr.np1_flows[fd];
         if (flow == NULL)
                 return -1;
 
@@ -390,8 +454,10 @@ static int n_flow_dealloc(int fd)
 
         flow_alloc_msg__pack(&msg, buf.data);
 
-        ret = frct_i_destroy(flow->frct_i, &buf);
-        list_del(&flow->next);
+        ret = frct_i_destroy(flow->cep_id, &buf);
+
+        fmgr.np1_flows[fd] = NULL;
+        fmgr.np1_flows_cep[flow->cep_id] = NULL;
 
         free(flow);
         free(buf.data);
@@ -399,17 +465,17 @@ static int n_flow_dealloc(int fd)
         return ret;
 }
 
-int fmgr_flow_alloc_resp(int fd, int response)
+int fmgr_np1_alloc_resp(int fd, int response)
 {
-        struct n_flow * flow;
+        struct np1_flow * flow;
         flow_alloc_msg_t msg = FLOW_ALLOC_MSG__INIT;
         buffer_t buf;
 
-        pthread_mutex_lock(&fmgr.n_flows_lock);
+        pthread_rwlock_wrlock(&fmgr.np1_flows_lock);
 
-        flow = get_n_flow_by_fd(fd);
+        flow = fmgr.np1_flows[fd];
         if (flow == NULL) {
-                pthread_mutex_unlock(&fmgr.n_flows_lock);
+                pthread_rwlock_unlock(&fmgr.np1_flows_lock);
                 return -1;
         }
 
@@ -419,95 +485,91 @@ int fmgr_flow_alloc_resp(int fd, int response)
 
         buf.len = flow_alloc_msg__get_packed_size(&msg);
         if (buf.len == 0) {
-                pthread_mutex_unlock(&fmgr.n_flows_lock);
+                pthread_rwlock_unlock(&fmgr.np1_flows_lock);
                 return -1;
         }
 
         buf.data = malloc(buf.len);
         if (buf.data == NULL) {
-                pthread_mutex_unlock(&fmgr.n_flows_lock);
+                pthread_rwlock_unlock(&fmgr.np1_flows_lock);
                 return -1;
         }
 
         flow_alloc_msg__pack(&msg, buf.data);
 
         if (response < 0) {
-                frct_i_destroy(flow->frct_i, &buf);
+                frct_i_destroy(flow->cep_id, &buf);
                 free(buf.data);
-                list_del(&flow->next);
+                fmgr.np1_flows[fd] = NULL;
+                fmgr.np1_flows_cep[flow->cep_id] = NULL;
                 free(flow);
-        } else if (frct_i_accept(flow->frct_i, &buf)) {
-                pthread_mutex_unlock(&fmgr.n_flows_lock);
-                return -1;
+        } else {
+                if (frct_i_accept(flow->cep_id, &buf, flow->qos)) {
+                        pthread_rwlock_unlock(&fmgr.np1_flows_lock);
+                        return -1;
+                }
+                flow_set_add(fmgr.np1_set, fd);
         }
 
-        pthread_mutex_unlock(&fmgr.n_flows_lock);
+        pthread_rwlock_unlock(&fmgr.np1_flows_lock);
 
         return 0;
 }
 
-int fmgr_flow_dealloc(int fd)
+int fmgr_np1_dealloc(int fd)
 {
         int ret;
 
-        pthread_mutex_lock(&fmgr.n_flows_lock);
-        ret = n_flow_dealloc(fd);
-        pthread_mutex_unlock(&fmgr.n_flows_lock);
+        pthread_rwlock_wrlock(&fmgr.np1_flows_lock);
+        ret = np1_flow_dealloc(fd);
+        pthread_rwlock_unlock(&fmgr.np1_flows_lock);
 
         return ret;
 }
 
-int fmgr_flow_alloc_msg(struct frct_i * frct_i, buffer_t * buf)
+int fmgr_np1_post_buf(cep_id_t   cep_id,
+                      buffer_t * buf)
 {
-        struct n_flow * flow;
+        struct np1_flow * flow;
         int ret = 0;
         int fd;
         flow_alloc_msg_t * msg;
 
-        pthread_mutex_lock(&fmgr.n_flows_lock);
+        pthread_rwlock_wrlock(&fmgr.np1_flows_lock);
 
         /* Depending on the message call the function in ipcp-dev.h */
 
         msg = flow_alloc_msg__unpack(NULL, buf->len, buf->data);
         if (msg == NULL) {
-                pthread_mutex_unlock(&fmgr.n_flows_lock);
+                pthread_rwlock_unlock(&fmgr.np1_flows_lock);
                 LOG_ERR("Failed to unpack flow alloc message");
                 return -1;
         }
 
         switch (msg->code) {
         case FLOW_ALLOC_CODE__FLOW_REQ:
-                flow = malloc(sizeof(*flow));
-                if (flow == NULL) {
-                        pthread_mutex_unlock(&fmgr.n_flows_lock);
-                        flow_alloc_msg__free_unpacked(msg, NULL);
-                        return -1;
-                }
-
-                flow->frct_i = frct_i;
-                flow->qos = msg->qos_cube;
-
                 fd = ipcp_flow_req_arr(getpid(),
                                        msg->dst_name,
                                        msg->src_ae_name);
                 if (fd < 0) {
-                        pthread_mutex_unlock(&fmgr.n_flows_lock);
-                        free(flow);
+                        pthread_rwlock_unlock(&fmgr.np1_flows_lock);
                         flow_alloc_msg__free_unpacked(msg, NULL);
                         LOG_ERR("Failed to get fd for flow.");
                         return -1;
                 }
 
-                flow->fd = fd;
-
-                INIT_LIST_HEAD(&flow->next);
+                if (add_np1_fd(fd, cep_id, msg->qos_cube)) {
+                        pthread_rwlock_unlock(&fmgr.np1_flows_lock);
+                        flow_alloc_msg__free_unpacked(msg, NULL);
+                        LOG_ERR("Failed to add np1 flow.");
+                        return -1;
+                }
 
-                list_add(&flow->next, &fmgr.n_flows);
                 break;
         case FLOW_ALLOC_CODE__FLOW_REPLY:
-                flow = get_n_flow_by_frct_i(frct_i);
+                flow = fmgr.np1_flows_cep[cep_id];
                 if (flow == NULL) {
-                        pthread_mutex_unlock(&fmgr.n_flows_lock);
+                        pthread_rwlock_unlock(&fmgr.np1_flows_lock);
                         flow_alloc_msg__free_unpacked(msg, NULL);
                         LOG_ERR("No such flow in flow manager.");
                         return -1;
@@ -515,20 +577,25 @@ int fmgr_flow_alloc_msg(struct frct_i * frct_i, buffer_t * buf)
 
                 ret = ipcp_flow_alloc_reply(flow->fd, msg->response);
                 if (msg->response < 0) {
-                        list_del(&flow->next);
+                        fmgr.np1_flows[flow->fd] = NULL;
+                        fmgr.np1_flows_cep[cep_id] = NULL;
                         free(flow);
+                } else {
+                        flow_set_add(fmgr.np1_set, flow->fd);
                 }
 
                 break;
         case FLOW_ALLOC_CODE__FLOW_DEALLOC:
-                flow = get_n_flow_by_frct_i(frct_i);
+                flow = fmgr.np1_flows_cep[cep_id];
                 if (flow == NULL) {
-                        pthread_mutex_unlock(&fmgr.n_flows_lock);
+                        pthread_rwlock_unlock(&fmgr.np1_flows_lock);
                         flow_alloc_msg__free_unpacked(msg, NULL);
                         LOG_ERR("No such flow in flow manager.");
                         return -1;
                 }
 
+                flow_set_del(fmgr.np1_set, flow->fd);
+
                 ret = flow_dealloc(flow->fd);
                 break;
         default:
@@ -537,9 +604,160 @@ int fmgr_flow_alloc_msg(struct frct_i * frct_i, buffer_t * buf)
                 break;
         }
 
-        pthread_mutex_unlock(&fmgr.n_flows_lock);
+        pthread_rwlock_unlock(&fmgr.np1_flows_lock);
 
         flow_alloc_msg__free_unpacked(msg, NULL);
 
         return ret;
 }
+
+int fmgr_np1_post_sdu(cep_id_t             cep_id,
+                      struct shm_du_buff * sdb)
+{
+        struct np1_flow * flow;
+
+        pthread_rwlock_rdlock(&fmgr.np1_flows_lock);
+
+        flow = fmgr.np1_flows_cep[cep_id];
+        if (flow == NULL) {
+                pthread_rwlock_unlock(&fmgr.np1_flows_lock);
+                LOG_ERR("Failed to find N flow.");
+                return -1;
+        }
+
+        if (ipcp_flow_write(flow->fd, sdb)) {
+                pthread_rwlock_unlock(&fmgr.np1_flows_lock);
+                LOG_ERR("Failed to hand SDU to N flow.");
+                return -1;
+        }
+
+        pthread_rwlock_unlock(&fmgr.np1_flows_lock);
+
+        return 0;
+}
+
+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;
+        }
+
+        result = flow_alloc_res(fd);
+        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;
+        }
+
+        return 0;
+}
+
+int fmgr_nm1_dt_flow(char * dst_name,
+                     enum qos_cube qos)
+{
+        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;
+        }
+
+        result = flow_alloc_res(fd);
+        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)) {
+                LOG_ERR("Failed to add file descriptor to list.");
+                flow_dealloc(fd);
+                free(ae_name);
+                return -1;
+        }
+
+        return 0;
+}
+
+int fmgr_nm1_write_sdu(struct pci *         pci,
+                       struct shm_du_buff * sdb)
+{
+        if (pci == NULL || sdb == NULL)
+                return -1;
+
+        if (shm_pci_ser(sdb, pci)) {
+                LOG_ERR("Failed to serialize PDU.");
+                ipcp_flow_del(sdb);
+                return -1;
+        }
+
+        if (ipcp_flow_write(fmgr.fd, sdb)) {
+                LOG_ERR("Failed to write SDU to fd %d.", fmgr.fd);
+                ipcp_flow_del(sdb);
+                return -1;
+        }
+
+        return 0;
+}
+
+int fmgr_nm1_write_buf(struct pci * pci,
+                       buffer_t *   buf)
+{
+        buffer_t * buffer;
+
+        if (pci == NULL || buf == NULL || buf->data == NULL)
+                return -1;
+
+        buffer = shm_pci_ser_buf(buf, pci);
+        if (buffer == NULL) {
+                LOG_ERR("Failed to serialize buffer.");
+                free(buf->data);
+                return -1;
+        }
+
+        if (flow_write(fmgr.fd, buffer->data, buffer->len) == -1) {
+                LOG_ERR("Failed to write buffer to fd.");
+                free(buffer);
+                return -1;
+        }
+
+        free(buffer);
+        return 0;
+}