Merged in dstaesse/ouroboros/be-new-api (pull request #439)

lib: Revise flow allocation API

16 files changed, 184 insertions, 412 deletions

diff --git a/include/ouroboros/dev.h b/include/ouroboros/dev.h
index e92cdd1c..4984736c 100644
--- a/include/ouroboros/dev.h
+++ b/include/ouroboros/dev.h
@@ -24,6 +24,7 @@
 #include <ouroboros/qos.h>
 
 #include <unistd.h>
+#include <time.h>
 
 #ifndef OUROBOROS_DEV_H
 #define OUROBOROS_DEV_H
@@ -33,20 +34,14 @@ int     ap_init(const char * ap_name);
 
 void    ap_fini(void);
 
-/* Returns flow descriptor (> 0) and qos spec. */
-int     flow_accept(qosspec_t * spec);
+/* Returns flow descriptor, qs updates to supplied QoS. */
+int     flow_alloc(const char *      dst_name,
+                   qosspec_t *       qs,
+                   struct timespec * timeo);
 
-int     flow_alloc_resp(int fd,
-                        int response);
-
-/*
- * Returns flow descriptor (> 0).
- * On returning, spec will contain the actual supplied QoS.
- */
-int     flow_alloc(const char * dst_name,
-                   qosspec_t *  spec);
-
-int     flow_alloc_res(int fd);
+/* Returns flow descriptor, qs updates to supplied QoS. */
+int     flow_accept(qosspec_t *       qs,
+                    struct timespec * timeo);
 
 int     flow_dealloc(int fd);
 
@@ -58,4 +53,4 @@ ssize_t flow_read(int    fd,
                   void * buf,
                   size_t count);
 
-#endif
+#endif /* OUROBOROS_DEV_H */
diff --git a/src/ipcpd/normal/connmgr.c b/src/ipcpd/normal/connmgr.c
index b8314917..8068d173 100644
--- a/src/ipcpd/normal/connmgr.c
+++ b/src/ipcpd/normal/connmgr.c
@@ -126,18 +126,13 @@ static void * flow_acceptor(void * o)
 
                 pthread_rwlock_unlock(&ipcpi.state_lock);
 
-                fd = flow_accept(&qs);
+                fd = flow_accept(&qs, NULL);
                 if (fd < 0) {
                         if (fd != -EIRMD)
                                 log_warn("Flow accept failed: %d", fd);
                         continue;
                 }
 
-                if (flow_alloc_resp(fd, 0)) {
-                        log_err("Failed to respond to flow alloc request.");
-                        continue;
-                }
-
                 if (cacep_rcv(fd, &rcv_info)) {
                         log_err("Error establishing application connection.");
                         flow_dealloc(fd);
@@ -286,7 +281,7 @@ int connmgr_alloc(struct ae *   ae,
 
         memset(&conn->conn_info, 0, sizeof(conn->conn_info));
 
-        conn->flow_info.fd = flow_alloc(dst_name, qs);
+        conn->flow_info.fd = flow_alloc(dst_name, qs, NULL);
         if (conn->flow_info.fd < 0) {
                 log_err("Failed to allocate flow to %s.", dst_name);
                 return -1;
@@ -297,12 +292,6 @@ int connmgr_alloc(struct ae *   ae,
         else
                 memset(&conn->flow_info.qs, 0, sizeof(conn->flow_info.qs));
 
-        if (flow_alloc_res(conn->flow_info.fd)) {
-                log_err("Flow allocation to %s failed.", dst_name);
-                flow_dealloc(conn->flow_info.fd);
-                return -1;
-        }
-
         if (cacep_snd(conn->flow_info.fd, &ae->info)) {
                 log_err("Failed to create application connection.");
                 flow_dealloc(conn->flow_info.fd);
diff --git a/src/irmd/ipcp.c b/src/irmd/ipcp.c
index 06b66d3b..a8263580 100644
--- a/src/irmd/ipcp.c
+++ b/src/irmd/ipcp.c
@@ -176,18 +176,11 @@ pid_t ipcp_create(char *         name,
 
 int ipcp_destroy(pid_t api)
 {
-        int status;
-
         if (kill(api, SIGTERM)) {
                 log_err("Failed to destroy IPCP");
                 return -1;
         }
 
-        if (waitpid(api, &status, 0) < 0) {
-                log_err("Failed to destroy IPCP");
-                return -1;
-        }
-
         return 0;
 }
 
diff --git a/src/irmd/irm_flow.c b/src/irmd/irm_flow.c
index 99966561..4e7c22ef 100644
--- a/src/irmd/irm_flow.c
+++ b/src/irmd/irm_flow.c
@@ -45,6 +45,7 @@ struct irm_flow * irm_flow_create(pid_t n_api,
         }
 
         if (pthread_mutex_init(&f->state_lock, NULL)) {
+                pthread_cond_destroy(&f->state_cond);
                 free(f);
                 return NULL;
         }
@@ -63,6 +64,9 @@ struct irm_flow * irm_flow_create(pid_t n_api,
         f->n_1_rb = shm_rbuff_create(n_1_api, port_id);
         if (f->n_1_rb == NULL) {
                 log_err("Could not create ringbuffer for AP-I %d.", n_1_api);
+                shm_rbuff_destroy(f->n_rb);
+                pthread_mutex_destroy(&f->state_lock);
+                pthread_cond_destroy(&f->state_cond);
                 free(f);
                 return NULL;
         }
@@ -122,7 +126,8 @@ enum flow_state irm_flow_get_state(struct irm_flow * f)
         return state;
 }
 
-void irm_flow_set_state(struct irm_flow * f, enum flow_state state)
+void irm_flow_set_state(struct irm_flow * f,
+                        enum flow_state   state)
 {
         assert(f);
         assert(state != FLOW_DESTROY);
@@ -135,7 +140,8 @@ void irm_flow_set_state(struct irm_flow * f, enum flow_state state)
         pthread_mutex_unlock(&f->state_lock);
 }
 
-enum flow_state irm_flow_wait_state(struct irm_flow * f, enum flow_state state)
+enum flow_state irm_flow_wait_state(struct irm_flow * f,
+                                    enum flow_state   state)
 {
         assert(f);
         assert(state != FLOW_NULL);
@@ -143,6 +149,8 @@ enum flow_state irm_flow_wait_state(struct irm_flow * f, enum flow_state state)
 
         pthread_mutex_lock(&f->state_lock);
 
+        assert(f->state != FLOW_NULL);
+
         while (!(f->state == state || f->state == FLOW_DESTROY))
                 pthread_cond_wait(&f->state_cond, &f->state_lock);
 
diff --git a/src/irmd/main.c b/src/irmd/main.c
index 9901a608..c7adf386 100644
--- a/src/irmd/main.c
+++ b/src/irmd/main.c
@@ -133,7 +133,8 @@ static struct irm_flow * get_irm_flow_n(pid_t n_api)
 
         list_for_each(pos, &irmd->irm_flows) {
                 struct irm_flow * e = list_entry(pos, struct irm_flow, next);
-                if (e->n_api == n_api)
+                if (e->n_api == n_api &&
+                    irm_flow_get_state(e) == FLOW_ALLOC_PENDING)
                         return e;
         }
 
@@ -982,7 +983,12 @@ static struct irm_flow * flow_accept(pid_t       api,
         struct irm_flow *  f  = NULL;
         struct api_entry * e  = NULL;
         struct reg_entry * re = NULL;
-        struct list_head * p;
+        struct list_head * p  = NULL;
+
+        pid_t api_n1;
+        pid_t api_n;
+        int   port_id;
+        int   ret;
 
         pthread_rwlock_rdlock(&irmd->state_lock);
 
@@ -1016,7 +1022,7 @@ static struct irm_flow * flow_accept(pid_t       api,
         pthread_rwlock_unlock(&irmd->reg_lock);
         pthread_rwlock_unlock(&irmd->state_lock);
 
-        while (api_entry_sleep(e) == -ETIMEDOUT) {
+        while ((ret = api_entry_sleep(e)) == -ETIMEDOUT) {
                 pthread_rwlock_rdlock(&irmd->state_lock);
                 if (irmd->state != IRMD_RUNNING) {
                         pthread_rwlock_unlock(&irmd->state_lock);
@@ -1025,126 +1031,76 @@ static struct irm_flow * flow_accept(pid_t       api,
                 pthread_rwlock_unlock(&irmd->state_lock);
         }
 
-        pthread_rwlock_rdlock(&irmd->state_lock);
-
-        if (irmd->state != IRMD_RUNNING) {
-                reg_entry_set_state(re, REG_NAME_NULL);
-                pthread_rwlock_unlock(&irmd->state_lock);
+        if (ret == -1) {
+                /* The process died, we can exit here. */
                 return NULL;
         }
 
-        pthread_rwlock_rdlock(&irmd->reg_lock);
+        pthread_rwlock_rdlock(&irmd->state_lock);
 
-        e = api_table_get(&irmd->api_table, api);
-        if (e == NULL) {
-                pthread_rwlock_unlock(&irmd->reg_lock);
+        if (irmd->state != IRMD_RUNNING) {
+                reg_entry_set_state(re, REG_NAME_NULL);
                 pthread_rwlock_unlock(&irmd->state_lock);
-                log_dbg("Process gone while accepting flow.");
                 return NULL;
         }
 
-        pthread_mutex_lock(&e->state_lock);
-
-        re = e->re;
-
-        pthread_mutex_unlock(&e->state_lock);
-
-        if (reg_entry_get_state(re) != REG_NAME_FLOW_ARRIVED) {
-                pthread_rwlock_unlock(&irmd->reg_lock);
-                pthread_rwlock_unlock(&irmd->state_lock);
-                log_err("Entry in wrong state.");
-                return NULL;
-        }
-        pthread_rwlock_unlock(&irmd->reg_lock);
         pthread_rwlock_rdlock(&irmd->flows_lock);
 
         f = get_irm_flow_n(api);
         if (f == NULL) {
                 pthread_rwlock_unlock(&irmd->flows_lock);
                 pthread_rwlock_unlock(&irmd->state_lock);
-                log_err("Port_id was not created yet.");
+                log_warn("Port_id was not created yet.");
                 return NULL;
         }
 
         *cube = re->qos;
 
+        api_n   = f->n_api;
+        api_n1  = f->n_1_api;
+        port_id = f->port_id;
+
         log_info("Flow on port_id %d allocated.", f->port_id);
 
         pthread_rwlock_unlock(&irmd->flows_lock);
-        pthread_rwlock_unlock(&irmd->state_lock);
-
-        return f;
-}
-
-static int flow_alloc_resp(pid_t n_api,
-                           int   port_id,
-                           int   response)
-{
-        struct irm_flow *  f  = NULL;
-        struct reg_entry * re = NULL;
-        struct api_entry * e  = NULL;
-        int ret = -1;
-
-        pid_t api_n1;
-        pid_t api_n;
-
-        pthread_rwlock_rdlock(&irmd->state_lock);
-
-        if (irmd->state != IRMD_RUNNING) {
-                pthread_rwlock_unlock(&irmd->state_lock);
-                return -1;
-        }
-
-        pthread_rwlock_wrlock(&irmd->reg_lock);
+        pthread_rwlock_rdlock(&irmd->reg_lock);
 
-        e = api_table_get(&irmd->api_table, n_api);
+        e = api_table_get(&irmd->api_table, api);
         if (e == NULL) {
                 pthread_rwlock_unlock(&irmd->reg_lock);
                 pthread_rwlock_unlock(&irmd->state_lock);
-                log_err("Unknown AP-I %d responding for port_id %d.",
-                        n_api, port_id);
-                return -1;
+                ipcp_flow_alloc_resp(api_n1, port_id, api_n, -1);
+                log_dbg("Process gone while accepting flow.");
+                return NULL;
         }
 
+        pthread_mutex_lock(&e->state_lock);
+
         re = e->re;
-        if (re == NULL) {
-                pthread_rwlock_unlock(&irmd->reg_lock);
-                pthread_rwlock_unlock(&irmd->state_lock);
-                log_err("AP-I %d is not handling a flow request.", n_api);
-                return -1;
-        }
+
+        pthread_mutex_unlock(&e->state_lock);
 
         if (reg_entry_get_state(re) != REG_NAME_FLOW_ARRIVED) {
                 pthread_rwlock_unlock(&irmd->reg_lock);
                 pthread_rwlock_unlock(&irmd->state_lock);
-                log_err("Name %s has no pending flow request.", re->name);
-                return -1;
+                ipcp_flow_alloc_resp(api_n1, port_id, api_n, -1);
+                log_err("Entry in wrong state.");
+                return NULL;
         }
 
-        registry_del_api(&irmd->registry, n_api);
+        registry_del_api(&irmd->registry, api);
 
         pthread_rwlock_unlock(&irmd->reg_lock);
-        pthread_rwlock_wrlock(&irmd->flows_lock);
-
-        f = get_irm_flow(port_id);
-        if (f == NULL) {
-                pthread_rwlock_unlock(&irmd->flows_lock);
-                pthread_rwlock_unlock(&irmd->state_lock);
-                return -1;
-        }
-
-        api_n  = f->n_api;
-        api_n1 = f->n_1_api;
-
-        pthread_rwlock_unlock(&irmd->flows_lock);
         pthread_rwlock_unlock(&irmd->state_lock);
 
-        ret = ipcp_flow_alloc_resp(api_n1, port_id, api_n, response);
+        if (ipcp_flow_alloc_resp(api_n1, port_id, api_n, 0)) {
+                log_dbg("Failed to respond to alloc.");
+                return NULL;
+        }
 
-        if (!(response || ret))
-                irm_flow_set_state(f, FLOW_ALLOCATED);
+        irm_flow_set_state(f, FLOW_ALLOCATED);
 
-        return ret;
+        return f;
 }
 
 static struct irm_flow * flow_alloc(pid_t     api,
@@ -1196,6 +1152,8 @@ static struct irm_flow * flow_alloc(pid_t     api,
         pthread_rwlock_unlock(&irmd->flows_lock);
         pthread_rwlock_unlock(&irmd->state_lock);
 
+        assert(irm_flow_get_state(f) == FLOW_ALLOC_PENDING);
+
         if (ipcp_flow_alloc(ipcp, port_id, api,
                             dst_name, cube) < 0) {
                 pthread_rwlock_rdlock(&irmd->state_lock);
@@ -1210,54 +1168,16 @@ static struct irm_flow * flow_alloc(pid_t     api,
                 return NULL;
         }
 
-        return f;
-}
-
-static int flow_alloc_res(int port_id)
-{
-        struct irm_flow * f;
-
-        pthread_rwlock_rdlock(&irmd->state_lock);
-
-        if (irmd->state != IRMD_RUNNING) {
-                pthread_rwlock_unlock(&irmd->state_lock);
-                return -1;
-        }
-        pthread_rwlock_rdlock(&irmd->flows_lock);
-
-        f = get_irm_flow(port_id);
-        if (f == NULL) {
-                pthread_rwlock_unlock(&irmd->flows_lock);
-                pthread_rwlock_unlock(&irmd->state_lock);
-                log_err("Could not find port %d.", port_id);
-                return -1;
-        }
-
-        if (irm_flow_get_state(f) == FLOW_NULL) {
-                pthread_rwlock_unlock(&irmd->flows_lock);
-                pthread_rwlock_unlock(&irmd->state_lock);
-                log_info("Port %d is deprecated.", port_id);
-                return -1;
-        }
-
-        if (irm_flow_get_state(f) == FLOW_ALLOCATED) {
-                log_info("Flow on port_id %d allocated.", port_id);
-                pthread_rwlock_unlock(&irmd->flows_lock);
-                pthread_rwlock_unlock(&irmd->state_lock);
-                return 0;
+        if (irm_flow_wait_state(f, FLOW_ALLOCATED) != FLOW_ALLOCATED) {
+                log_info("Pending flow on port_id %d torn down.", port_id);
+                return NULL;
         }
 
-        pthread_rwlock_unlock(&irmd->flows_lock);
-        pthread_rwlock_unlock(&irmd->state_lock);
-
-        if (irm_flow_wait_state(f, FLOW_ALLOCATED) == FLOW_ALLOCATED) {
-                log_info("Flow on port_id %d allocated.", port_id);
-                return 0;
-        }
+        assert(irm_flow_get_state(f) == FLOW_ALLOCATED);
 
-        log_info("Pending flow on port_id %d torn down.", port_id);
+        log_info("Flow on port_id %d allocated.", port_id);
 
-        return -1;
+        return f;
 }
 
 static int flow_dealloc(pid_t api,
@@ -1293,6 +1213,9 @@ static int flow_dealloc(pid_t api,
 
         if (irm_flow_get_state(f) == FLOW_DEALLOC_PENDING) {
                 list_del(&f->next);
+                if ((kill(f->n_api, 0) < 0 && f->n_1_api == -1) ||
+                    (kill (f->n_1_api, 0) < 0 && f->n_api == -1))
+                        irm_flow_set_state(f, FLOW_NULL);
                 clear_irm_flow(f);
                 irm_flow_destroy(f);
                 bmp_release(irmd->port_ids, port_id);
@@ -1305,12 +1228,11 @@ static int flow_dealloc(pid_t api,
         }
 
         pthread_rwlock_unlock(&irmd->flows_lock);
+        pthread_rwlock_unlock(&irmd->state_lock);
 
         if (n_1_api != -1)
                 ret = ipcp_flow_dealloc(n_1_api, port_id);
 
-        pthread_rwlock_unlock(&irmd->state_lock);
-
         return ret;
 }
 
@@ -1501,7 +1423,7 @@ static int flow_alloc_reply(int port_id,
         struct irm_flow * f;
 
         pthread_rwlock_rdlock(&irmd->state_lock);
-        pthread_rwlock_wrlock(&irmd->flows_lock);
+        pthread_rwlock_rdlock(&irmd->flows_lock);
 
         f = get_irm_flow(port_id);
         if (f == NULL) {
@@ -1551,18 +1473,19 @@ static void irm_destroy(void)
         list_for_each_safe(p, h, &irmd->ipcps) {
                 struct ipcp_entry * e = list_entry(p, struct ipcp_entry, next);
                 list_del(&e->next);
-                ipcp_destroy(e->api);
-                clear_spawned_api(e->api);
-                registry_del_api(&irmd->registry, e->api);
                 ipcp_entry_destroy(e);
         }
 
-        list_for_each_safe(p, h, &irmd->spawned_apis) {
+        list_for_each(p, &irmd->spawned_apis) {
                 struct pid_el * e = list_entry(p, struct pid_el, next);
-                int status;
                 if (kill(e->pid, SIGTERM))
                         log_dbg("Could not send kill signal to %d.", e->pid);
-                else if (waitpid(e->pid, &status, 0) < 0)
+        }
+
+        list_for_each_safe(p, h, &irmd->spawned_apis) {
+                struct pid_el * e = list_entry(p, struct pid_el, next);
+                int status;
+                if (waitpid(e->pid, &status, 0) < 0)
                         log_dbg("Error waiting for %d to exit.", e->pid);
                 list_del(&e->next);
                 registry_del_api(&irmd->registry, e->pid);
@@ -1940,12 +1863,6 @@ void * mainloop(void * o)
                         ret_msg.has_api     = true;
                         ret_msg.api         = e->n_1_api;
                         break;
-                case IRM_MSG_CODE__IRM_FLOW_ALLOC_RESP:
-                        ret_msg.has_result = true;
-                        ret_msg.result = flow_alloc_resp(msg->api,
-                                                         msg->port_id,
-                                                         msg->response);
-                        break;
                 case IRM_MSG_CODE__IRM_FLOW_ALLOC:
                         e = flow_alloc(msg->api,
                                        msg->dst_name,
@@ -1960,10 +1877,6 @@ void * mainloop(void * o)
                         ret_msg.has_api     = true;
                         ret_msg.api         = e->n_1_api;
                         break;
-                case IRM_MSG_CODE__IRM_FLOW_ALLOC_RES:
-                        ret_msg.has_result = true;
-                        ret_msg.result = flow_alloc_res(msg->port_id);
-                        break;
                 case IRM_MSG_CODE__IRM_FLOW_DEALLOC:
                         ret_msg.has_result = true;
                         ret_msg.result = flow_dealloc(msg->api, msg->port_id);
diff --git a/src/lib/dev.c b/src/lib/dev.c
index 79797b92..e19083c3 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -161,21 +161,21 @@ struct {
 } ai;
 
 /* FIXME: translate real spec to cube */
-static qoscube_t spec_to_cube(qosspec_t * spec)
+static qoscube_t spec_to_cube(qosspec_t * qs)
 {
-        if (spec == NULL)
+        if (qs == NULL)
                 return QOS_CUBE_BE;
 
-        return spec->cube;
+        return qs->cube;
 }
 
 /* FIXME: fill real spec */
-static void fill_qosspec(qosspec_t * spec,
+static void fill_qosspec(qosspec_t * qs,
                          qoscube_t   cube)
 {
-        assert(spec);
+        assert(qs);
 
-        spec->cube = cube;
+        qs->cube = cube;
 }
 
 static int api_announce(char * ap_name)
@@ -209,6 +209,17 @@ static int api_announce(char * ap_name)
         return ret;
 }
 
+static void init_flow(int fd)
+{
+        assert(!(fd < 0));
+
+        memset(&ai.flows[fd], 0, sizeof(ai.flows[fd]));
+
+        ai.flows[fd].port_id  = -1;
+        ai.flows[fd].api      = -1;
+        ai.flows[fd].cube     = QOS_CUBE_BE;
+}
+
 static void reset_flow(int fd)
 {
         assert (!(fd < 0));
@@ -216,25 +227,17 @@ static void reset_flow(int fd)
         if (ai.flows[fd].port_id != -1)
                 port_destroy(&ai.ports[ai.flows[fd].port_id]);
 
-        ai.flows[fd].port_id = -1;
-        if (ai.flows[fd].rx_rb != NULL) {
+        if (ai.flows[fd].rx_rb != NULL)
                 shm_rbuff_close(ai.flows[fd].rx_rb);
-                ai.flows[fd].rx_rb = NULL;
-        }
-        if (ai.flows[fd].tx_rb != NULL) {
+
+        if (ai.flows[fd].tx_rb != NULL)
                 shm_rbuff_close(ai.flows[fd].tx_rb);
-                ai.flows[fd].tx_rb = NULL;
-        }
 
-        if (ai.flows[fd].set != NULL) {
+        if (ai.flows[fd].set != NULL)
                 shm_flow_set_close(ai.flows[fd].set);
-                ai.flows[fd].set = NULL;
-        }
 
-        ai.flows[fd].oflags = 0;
-        ai.flows[fd].api = -1;
-        ai.flows[fd].timesout = false;
-        ai.flows[fd].cube = QOS_CUBE_BE;
+        init_flow(fd);
+
 }
 
 int ap_init(const char * ap_name)
@@ -280,16 +283,8 @@ int ap_init(const char * ap_name)
                 return -1;
         }
 
-        for (i = 0; i < AP_MAX_FLOWS; ++i) {
-                ai.flows[i].rx_rb    = NULL;
-                ai.flows[i].tx_rb    = NULL;
-                ai.flows[i].set      = NULL;
-                ai.flows[i].port_id  = -1;
-                ai.flows[i].oflags   = 0;
-                ai.flows[i].api      = -1;
-                ai.flows[i].timesout = false;
-                ai.flows[i].cube     = QOS_CUBE_BE;
-        }
+        for (i = 0; i < AP_MAX_FLOWS; ++i)
+                init_flow(i);
 
         ai.ports = malloc(sizeof(*ai.ports) * IRMD_MAX_FLOWS);
         if (ai.ports == NULL) {
@@ -382,7 +377,8 @@ void ap_fini()
         pthread_rwlock_destroy(&ai.data_lock);
 }
 
-int flow_accept(qosspec_t * spec)
+int flow_accept(qosspec_t *       qs,
+                struct timespec * timeo)
 {
         irm_msg_t msg = IRM_MSG__INIT;
         irm_msg_t * recv_msg = NULL;
@@ -391,6 +387,13 @@ int flow_accept(qosspec_t * spec)
         msg.code    = IRM_MSG_CODE__IRM_FLOW_ACCEPT;
         msg.has_api = true;
 
+        if (timeo != NULL) {
+                msg.has_timeo_sec = true;
+                msg.has_timeo_usec = true;
+                msg.timeo_sec  = timeo->tv_sec;
+                msg.timeo_usec = timeo->tv_nsec / 1000;
+        }
+
         pthread_rwlock_rdlock(&ai.data_lock);
 
         msg.api     = ai.api;
@@ -424,7 +427,6 @@ int flow_accept(qosspec_t * spec)
 
         ai.flows[fd].rx_rb = shm_rbuff_open(ai.api, recv_msg->port_id);
         if (ai.flows[fd].rx_rb == NULL) {
-                reset_flow(fd);
                 bmp_release(ai.fds, fd);
                 pthread_rwlock_unlock(&ai.flows_lock);
                 pthread_rwlock_unlock(&ai.data_lock);
@@ -435,8 +437,10 @@ int flow_accept(qosspec_t * spec)
         ai.flows[fd].tx_rb = shm_rbuff_open(recv_msg->api, recv_msg->port_id);
         if (ai.flows[fd].tx_rb == NULL) {
                 reset_flow(fd);
+                bmp_release(ai.fds, fd);
                 pthread_rwlock_unlock(&ai.flows_lock);
                 pthread_rwlock_unlock(&ai.data_lock);
+                irm_msg__free_unpacked(recv_msg, NULL);
                 return -1;
         }
 
@@ -455,8 +459,8 @@ int flow_accept(qosspec_t * spec)
         ai.flows[fd].api     = recv_msg->api;
         ai.flows[fd].cube    = recv_msg->qoscube;
 
-        if (spec != NULL)
-                fill_qosspec(spec, ai.flows[fd].cube);
+        if (qs != NULL)
+                fill_qosspec(qs, ai.flows[fd].cube);
 
         ai.ports[recv_msg->port_id].fd    = fd;
         ai.ports[recv_msg->port_id].state = PORT_ID_ASSIGNED;
@@ -469,69 +473,27 @@ int flow_accept(qosspec_t * spec)
         return fd;
 }
 
-int flow_alloc_resp(int fd,
-                    int response)
+int flow_alloc(const char *      dst_name,
+               qosspec_t *       qs,
+               struct timespec * timeo)
 {
         irm_msg_t msg = IRM_MSG__INIT;
         irm_msg_t * recv_msg = NULL;
-        int ret = -1;
-
-        if (fd < 0 || fd >= AP_MAX_FLOWS)
-                return -EBADF;
-
-        msg.code         = IRM_MSG_CODE__IRM_FLOW_ALLOC_RESP;
-        msg.has_api      = true;
-        msg.api          = ai.api;
-        msg.has_port_id  = true;
-
-        pthread_rwlock_rdlock(&ai.data_lock);
-        pthread_rwlock_rdlock(&ai.flows_lock);
 
-        if (ai.flows[fd].port_id < 0) {
-                pthread_rwlock_unlock(&ai.flows_lock);
-                pthread_rwlock_unlock(&ai.data_lock);
-                return -ENOTALLOC;
-        }
-
-        msg.port_id      = ai.flows[fd].port_id;
-
-        pthread_rwlock_unlock(&ai.flows_lock);
-        pthread_rwlock_unlock(&ai.data_lock);
-
-        msg.has_response = true;
-        msg.response     = response;
-
-        recv_msg = send_recv_irm_msg(&msg);
-        if (recv_msg == NULL)
-                return -EIRMD;
-
-        if (!recv_msg->has_result) {
-                irm_msg__free_unpacked(recv_msg, NULL);
-                return -1;
-        }
-
-        ret = recv_msg->result;
-
-        irm_msg__free_unpacked(recv_msg, NULL);
-
-        return ret;
-}
-
-int flow_alloc(const char * dst_name,
-               qosspec_t *  spec)
-{
-        irm_msg_t msg = IRM_MSG__INIT;
-        irm_msg_t * recv_msg = NULL;
-        int fd = -1;
-
-        if (dst_name == NULL)
-                return -EINVAL;
+        int fd;
 
         msg.code        = IRM_MSG_CODE__IRM_FLOW_ALLOC;
         msg.dst_name    = (char *) dst_name;
         msg.has_api     = true;
         msg.has_qoscube = true;
-        msg.qoscube     = spec_to_cube(spec);
+        msg.qoscube     = spec_to_cube(qs);
+
+        if (timeo != NULL) {
+                msg.has_timeo_sec = true;
+                msg.has_timeo_usec = true;
+                msg.timeo_sec  = timeo->tv_sec;
+                msg.timeo_usec = timeo->tv_nsec / 1000;
+        }
 
         pthread_rwlock_rdlock(&ai.data_lock);
 
@@ -561,7 +523,6 @@ int flow_alloc(const char * dst_name,
 
         ai.flows[fd].rx_rb = shm_rbuff_open(ai.api, recv_msg->port_id);
         if (ai.flows[fd].rx_rb == NULL) {
-                reset_flow(fd);
                 bmp_release(ai.fds, fd);
                 pthread_rwlock_unlock(&ai.flows_lock);
                 pthread_rwlock_unlock(&ai.data_lock);
@@ -571,16 +532,21 @@ int flow_alloc(const char * dst_name,
 
         ai.flows[fd].tx_rb = shm_rbuff_open(recv_msg->api, recv_msg->port_id);
         if (ai.flows[fd].tx_rb == NULL) {
+                reset_flow(fd);
+                bmp_release(ai.fds, fd);
                 pthread_rwlock_unlock(&ai.flows_lock);
                 pthread_rwlock_unlock(&ai.data_lock);
+                irm_msg__free_unpacked(recv_msg, NULL);
                 return -1;
         }
 
         ai.flows[fd].set = shm_flow_set_open(recv_msg->api);
         if (ai.flows[fd].set == NULL) {
                 reset_flow(fd);
+                bmp_release(ai.fds, fd);
                 pthread_rwlock_unlock(&ai.flows_lock);
                 pthread_rwlock_unlock(&ai.data_lock);
+                irm_msg__free_unpacked(recv_msg, NULL);
                 return -1;
         }
 
@@ -589,7 +555,6 @@ int flow_alloc(const char * dst_name,
         ai.flows[fd].api     = recv_msg->api;
         ai.flows[fd].cube    = recv_msg->qoscube;
 
-        ai.ports[recv_msg->port_id].fd    = fd;
         ai.ports[recv_msg->port_id].state = PORT_ID_ASSIGNED;
 
         pthread_rwlock_unlock(&ai.flows_lock);
@@ -600,48 +565,6 @@ int flow_alloc(const char * dst_name,
         return fd;
 }
 
-int flow_alloc_res(int fd)
-{
-        irm_msg_t msg = IRM_MSG__INIT;
-        irm_msg_t * recv_msg = NULL;
-        int result = 0;
-
-        if (fd < 0 || fd >= AP_MAX_FLOWS)
-                return -EBADF;
-
-        msg.code         = IRM_MSG_CODE__IRM_FLOW_ALLOC_RES;
-        msg.has_port_id  = true;
-
-        pthread_rwlock_rdlock(&ai.data_lock);
-        pthread_rwlock_rdlock(&ai.flows_lock);
-
-        if (ai.flows[fd].port_id < 0) {
-                pthread_rwlock_unlock(&ai.flows_lock);
-                pthread_rwlock_unlock(&ai.data_lock);
-                return -ENOTALLOC;
-        }
-
-        msg.port_id = ai.flows[fd].port_id;
-
-        pthread_rwlock_unlock(&ai.flows_lock);
-        pthread_rwlock_unlock(&ai.data_lock);
-
-        recv_msg = send_recv_irm_msg_b(&msg);
-        if (recv_msg == NULL)
-                return -EIRMD;
-
-        if (!recv_msg->has_result) {
-                irm_msg__free_unpacked(recv_msg, NULL);
-                return -1;
-        }
-
-        result = recv_msg->result;
-
-        irm_msg__free_unpacked(recv_msg, NULL);
-
-        return result;
-}
-
 int flow_dealloc(int fd)
 {
         irm_msg_t msg = IRM_MSG__INIT;
@@ -804,9 +727,9 @@ int flow_set_timeout(int                     fd,
 }
 
 int flow_get_qosspec(int         fd,
-                     qosspec_t * spec)
+                     qosspec_t * qs)
 {
-        if (fd < 0 || fd >= AP_MAX_FLOWS || spec == NULL)
+        if (fd < 0 || fd >= AP_MAX_FLOWS || qs == NULL)
                 return -EINVAL;
 
         pthread_rwlock_rdlock(&ai.data_lock);
@@ -818,7 +741,7 @@ int flow_get_qosspec(int         fd,
                 return -ENOTALLOC;
         }
 
-        fill_qosspec(spec, ai.flows[fd].cube);
+        fill_qosspec(qs, ai.flows[fd].cube);
 
         pthread_rwlock_unlock(&ai.flows_lock);
         pthread_rwlock_unlock(&ai.data_lock);
diff --git a/src/lib/irmd_messages.proto b/src/lib/irmd_messages.proto
index c25d2c18..4fbd676e 100644
--- a/src/lib/irmd_messages.proto
+++ b/src/lib/irmd_messages.proto
@@ -39,14 +39,12 @@ enum irm_msg_code {
         IRM_UNBIND_API        = 11;
         IRM_REG               = 12;
         IRM_UNREG             = 13;
-        IRM_FLOW_ACCEPT       = 14;
-        IRM_FLOW_ALLOC_RESP   = 15;
-        IRM_FLOW_ALLOC        = 16;
-        IRM_FLOW_ALLOC_RES    = 17;
-        IRM_FLOW_DEALLOC      = 18;
-        IPCP_FLOW_REQ_ARR     = 19;
-        IPCP_FLOW_ALLOC_REPLY = 20;
-        IRM_REPLY             = 21;
+        IRM_FLOW_ALLOC        = 14;
+        IRM_FLOW_ACCEPT       = 15;
+        IRM_FLOW_DEALLOC      = 16;
+        IPCP_FLOW_REQ_ARR     = 17;
+        IPCP_FLOW_ALLOC_REPLY = 18;
+        IRM_REPLY             = 19;
 };
 
 message irm_msg {
@@ -63,5 +61,7 @@ message irm_msg {
         optional dif_config_msg conf = 11;
         optional uint32 opts         = 12;
         repeated sint32 apis         = 13;
-        optional sint32 result       = 14;
+        optional uint32 timeo_sec    = 14;
+        optional uint32 timeo_usec   = 15;
+        optional sint32 result       = 16;
 };
diff --git a/src/lib/shm_flow_set.c b/src/lib/shm_flow_set.c
index 615fbd2b..67abbb5b 100644
--- a/src/lib/shm_flow_set.c
+++ b/src/lib/shm_flow_set.c
@@ -302,7 +302,8 @@ int shm_flow_set_has(struct shm_flow_set * set,
         return ret;
 }
 
-void shm_flow_set_notify(struct shm_flow_set * set, int port_id)
+void shm_flow_set_notify(struct shm_flow_set * set,
+                         int                   port_id)
 {
         assert(set);
         assert(!(port_id < 0) && port_id < IRMD_MAX_FLOWS);
diff --git a/src/tools/cbr/cbr_client.c b/src/tools/cbr/cbr_client.c
index 16ade13d..5ec1d560 100644
--- a/src/tools/cbr/cbr_client.c
+++ b/src/tools/cbr/cbr_client.c
@@ -63,7 +63,6 @@ int client_main(char * server,
         struct sigaction sig_act;
 
         int fd = 0;
-        int result = 0;
         char buf[size];
         long seqnr = 0;
         long gap = size * 8.0 * (BILLION / (double) rate);
@@ -90,19 +89,12 @@ int client_main(char * server,
         printf("Client started, duration %d, rate %lu b/s, size %d B.\n",
                duration, rate, size);
 
-        fd = flow_alloc(server, NULL);
+        fd = flow_alloc(server, NULL, NULL);
         if (fd < 0) {
                 printf("Failed to allocate flow.\n");
                 return -1;
         }
 
-        result = flow_alloc_res(fd);
-        if (result < 0) {
-                printf("Flow allocation refused.\n");
-                flow_dealloc(fd);
-                return -1;
-        }
-
         clock_gettime(CLOCK_REALTIME, &start);
         if (!flood) {
                 while (!stop) {
diff --git a/src/tools/cbr/cbr_server.c b/src/tools/cbr/cbr_server.c
index 9198858c..1a963a64 100644
--- a/src/tools/cbr/cbr_server.c
+++ b/src/tools/cbr/cbr_server.c
@@ -146,6 +146,8 @@ static void * worker(void * o)
 
                 pthread_mutex_lock(&fds_lock);
                 fds_count--;
+
+                pthread_cond_signal(&fds_signal);
                 pthread_mutex_unlock(&fds_lock);
         }
 
@@ -154,8 +156,7 @@ static void * worker(void * o)
 
 static void * listener(void * o)
 {
-        int client_fd = 0;
-        int response = 0;
+        int fd = 0;
         qosspec_t qs;
 
         (void) o;
@@ -164,8 +165,19 @@ static void * listener(void * o)
                server_settings.interval, server_settings.timeout);
 
         while (true) {
-                client_fd = flow_accept(&qs);
-                if (client_fd < 0) {
+                pthread_mutex_lock(&fds_lock);
+                pthread_cleanup_push((void(*)(void *)) pthread_mutex_unlock,
+                                     (void *) &fds_lock);
+
+                while (fds_count == THREADS_SIZE) {
+                        printf("Can't accept any more flows, waiting.\n");
+                        pthread_cond_wait(&fds_signal, &fds_lock);
+                }
+
+                pthread_cleanup_pop(true);
+
+                fd = flow_accept(&qs, NULL);
+                if (fd < 0) {
                         printf("Failed to accept flow.\n");
                         break;
                 }
@@ -174,26 +186,12 @@ static void * listener(void * o)
 
                 pthread_mutex_lock(&fds_lock);
 
-                response = (fds_count < THREADS_SIZE) ? 0 : -1;
-
-                if (flow_alloc_resp(client_fd, response)) {
-                        printf("Failed to give an allocate response.\n");
-                        flow_dealloc(client_fd);
-                        pthread_mutex_unlock(&fds_lock);
-                        continue;
-                }
-
-                if (response) {
-                        printf("Can't accept any more flows, denying.\n");
-                        continue;
-                }
-
                 fds_count++;
                 fds_index = (fds_index + 1) % THREADS_SIZE;
-                fds[fds_index] = client_fd;
+                fds[fds_index] = fd;
 
-                pthread_mutex_unlock(&fds_lock);
                 pthread_cond_signal(&fds_signal);
+                pthread_mutex_unlock(&fds_lock);
         }
 
         return 0;
diff --git a/src/tools/echo/echo_client.c b/src/tools/echo/echo_client.c
index f84de73a..5ec2051f 100644
--- a/src/tools/echo/echo_client.c
+++ b/src/tools/echo/echo_client.c
@@ -26,25 +26,17 @@
 int client_main(void)
 {
         int fd = 0;
-        int result = 0;
         char buf[BUF_SIZE];
         char * message  = "Client says hi!";
         ssize_t count = 0;
 
-        fd = flow_alloc("echo", NULL);
+        fd = flow_alloc("echo", NULL, NULL);
         if (fd < 0) {
                 printf("Failed to allocate flow.\n");
                 return -1;
         }
 
-        result = flow_alloc_res(fd);
-        if (result < 0) {
-                printf("Flow allocation refused.\n");
-                flow_dealloc(fd);
-                return -1;
-        }
-
-        if (flow_write(fd, message, strlen(message) + 1) == -1) {
+        if (flow_write(fd, message, strlen(message) + 1) < 0) {
                 printf("Failed to write SDU.\n");
                 flow_dealloc(fd);
                 return -1;
diff --git a/src/tools/echo/echo_server.c b/src/tools/echo/echo_server.c
index aa136485..771155f4 100644
--- a/src/tools/echo/echo_server.c
+++ b/src/tools/echo/echo_server.c
@@ -37,7 +37,7 @@ void shutdown_server(int signo)
 
 int server_main(void)
 {
-        int    client_fd = 0;
+        int    fd = 0;
         char   buf[BUF_SIZE];
         ssize_t count = 0;
         qosspec_t qs;
@@ -51,36 +51,30 @@ int server_main(void)
         }
 
         while (true) {
-                client_fd = flow_accept(&qs);
-                if (client_fd < 0) {
+                fd = flow_accept(&qs, NULL);
+                if (fd < 0) {
                         printf("Failed to accept flow.\n");
                         break;
                 }
 
                 printf("New flow.\n");
 
-                if (flow_alloc_resp(client_fd, 0)) {
-                        printf("Failed to give an allocate response.\n");
-                        flow_dealloc(client_fd);
-                        continue;
-                }
-
-                count = flow_read(client_fd, &buf, BUF_SIZE);
+                count = flow_read(fd, &buf, BUF_SIZE);
                 if (count < 0) {
                         printf("Failed to read SDU.\n");
-                        flow_dealloc(client_fd);
+                        flow_dealloc(fd);
                         continue;
                 }
 
                 printf("Message from client is %.*s.\n", (int) count, buf);
 
-                if (flow_write(client_fd, buf, count) == -1) {
+                if (flow_write(fd, buf, count) == -1) {
                         printf("Failed to write SDU.\n");
-                        flow_dealloc(client_fd);
+                        flow_dealloc(fd);
                         continue;
                 }
 
-                flow_dealloc(client_fd);
+                flow_dealloc(fd);
         }
 
         return 0;
diff --git a/src/tools/operf/operf_client.c b/src/tools/operf/operf_client.c
index d2f08ef4..7827b62b 100644
--- a/src/tools/operf/operf_client.c
+++ b/src/tools/operf/operf_client.c
@@ -182,18 +182,12 @@ int client_main(void)
         client.sent = 0;
         client.rcvd = 0;
 
-        fd = flow_alloc(client.s_apn, NULL);
+        fd = flow_alloc(client.s_apn, NULL, NULL);
         if (fd < 0) {
                 printf("Failed to allocate flow.\n");
                 return -1;
         }
 
-        if (flow_alloc_res(fd)) {
-                printf("Flow allocation refused.\n");
-                flow_dealloc(fd);
-                return -1;
-        }
-
         clock_gettime(CLOCK_REALTIME, &tic);
 
         pthread_create(&client.reader_pt, NULL, reader, &fd);
diff --git a/src/tools/operf/operf_server.c b/src/tools/operf/operf_server.c
index 3665d4cc..b17a4f7b 100644
--- a/src/tools/operf/operf_server.c
+++ b/src/tools/operf/operf_server.c
@@ -108,7 +108,7 @@ void * accept_thread(void * o)
         printf("Ouroboros perf server started.\n");
 
         while (true) {
-                fd = flow_accept(&qs);
+                fd = flow_accept(&qs, NULL);
                 if (fd < 0) {
                         printf("Failed to accept flow.\n");
                         break;
@@ -116,12 +116,6 @@ void * accept_thread(void * o)
 
                 printf("New flow %d.\n", fd);
 
-                if (flow_alloc_resp(fd, 0)) {
-                        printf("Failed to give an allocate response.\n");
-                        flow_dealloc(fd);
-                        continue;
-                }
-
                 clock_gettime(CLOCK_REALTIME, &now);
 
                 pthread_mutex_lock(&server.lock);
diff --git a/src/tools/oping/oping_client.c b/src/tools/oping/oping_client.c
index a91a126c..77a08db7 100644
--- a/src/tools/oping/oping_client.c
+++ b/src/tools/oping/oping_client.c
@@ -176,7 +176,6 @@ static int client_init(void)
         client.rtt_m2 = 0;
 
         pthread_mutex_init(&client.lock, NULL);
-        pthread_mutex_lock(&client.lock);
 
         return 0;
 }
@@ -213,21 +212,13 @@ int client_main(void)
                 return -1;
         }
 
-        fd = flow_alloc(client.s_apn, NULL);
+        fd = flow_alloc(client.s_apn, NULL, NULL);
         if (fd < 0) {
                 printf("Failed to allocate flow.\n");
-                return -1;
-        }
-
-        if (flow_alloc_res(fd)) {
-                printf("Flow allocation refused.\n");
-                flow_dealloc(fd);
                 client_fini();
                 return -1;
         }
 
-        pthread_mutex_unlock(&client.lock);
-
         clock_gettime(CLOCK_REALTIME, &tic);
 
         pthread_create(&client.reader_pt, NULL, reader, &fd);
diff --git a/src/tools/oping/oping_server.c b/src/tools/oping/oping_server.c
index e20e236d..44a301ba 100644
--- a/src/tools/oping/oping_server.c
+++ b/src/tools/oping/oping_server.c
@@ -57,6 +57,7 @@ void * cleaner_thread(void * o)
                 for (i = 0; i < OPING_MAX_FLOWS; ++i)
                         if (flow_set_has(server.flows, i) &&
                             ts_diff_ms(&server.times[i], &now) > deadline_ms) {
+                                printf("Flow %d timed out.\n", i);
                                 flow_set_del(server.flows, i);
                                 flow_dealloc(i);
                         }
@@ -110,8 +111,8 @@ void * server_thread(void *o)
 
 void * accept_thread(void * o)
 {
-        int fd = 0;
-        struct timespec now = {0, 0};
+        int fd;
+        struct timespec now;
         qosspec_t qs;
 
         (void) o;
@@ -119,7 +120,7 @@ void * accept_thread(void * o)
         printf("Ouroboros ping server started.\n");
 
         while (true) {
-                fd = flow_accept(&qs);
+                fd = flow_accept(&qs, NULL);
                 if (fd < 0) {
                         printf("Failed to accept flow.\n");
                         break;
@@ -127,12 +128,6 @@ void * accept_thread(void * o)
 
                 printf("New flow %d.\n", fd);
 
-                if (flow_alloc_resp(fd, 0)) {
-                        printf("Failed to give an allocate response.\n");
-                        flow_dealloc(fd);
-                        continue;
-                }
-
                 clock_gettime(CLOCK_REALTIME, &now);
 
                 pthread_mutex_lock(&server.lock);