lib: Pass qosspec at flow allocation

The flow allocator now passes the full qos specification to the
endpoint, instead of just a cube. This is a more flexible
architecture, as it makes QoS cubes internal to the layers.

Adds endianness transforms for the flow allocator protocol in the
normal IPCP.

Signed-off-by: Dimitri Staessens <[email protected]>
Signed-off-by: Sander Vrijders <[email protected]>

9 files changed, 167 insertions, 130 deletions

diff --git a/src/lib/CMakeLists.txt b/src/lib/CMakeLists.txt
index e7e07802..aa4e5bf3 100644
--- a/src/lib/CMakeLists.txt
+++ b/src/lib/CMakeLists.txt
@@ -6,6 +6,8 @@ include_directories(${CMAKE_BINARY_DIR}/include)
 
 protobuf_generate_c(IRM_PROTO_SRCS IRM_PROTO_HDRS irmd_messages.proto)
 protobuf_generate_c(IPCP_PROTO_SRCS IPCP_PROTO_HDRS ipcpd_messages.proto)
+protobuf_generate_c(QOSSPEC_PROTO_SRCS QOSSPEC_PROTO_HDRS
+  qosspec.proto)
 protobuf_generate_c(LAYER_CONFIG_PROTO_SRCS LAYER_CONFIG_PROTO_HDRS
   ipcp_config.proto)
 protobuf_generate_c(CACEP_PROTO_SRCS CACEP_PROTO_HDRS cacep.proto)
@@ -214,7 +216,7 @@ configure_file("${CMAKE_CURRENT_SOURCE_DIR}/config.h.in"
   "${CMAKE_CURRENT_BINARY_DIR}/config.h" @ONLY)
 
 add_library(ouroboros-common SHARED ${SOURCE_FILES_COMMON} ${IRM_PROTO_SRCS}
-  ${IPCP_PROTO_SRCS} ${LAYER_CONFIG_PROTO_SRCS})
+  ${IPCP_PROTO_SRCS} ${LAYER_CONFIG_PROTO_SRCS} ${QOSSPEC_PROTO_SRCS})
 
 add_library(ouroboros-dev SHARED ${SOURCE_FILES_DEV} ${CACEP_PROTO_SRCS})
 
diff --git a/src/lib/dev.c b/src/lib/dev.c
index 3d9e1d49..a92c1e42 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -44,7 +44,6 @@
 #include <ouroboros/shm_rbuff.h>
 #include <ouroboros/utils.h>
 #include <ouroboros/fqueue.h>
-#include <ouroboros/qoscube.h>
 #include <ouroboros/timerwheel.h>
 
 #include <stdlib.h>
@@ -94,7 +93,6 @@ struct flow {
         struct shm_flow_set * set;
         int                   port_id;
         int                   oflags;
-        qoscube_t             cube;
         qosspec_t             spec;
         ssize_t               part_idx;
 
@@ -235,7 +233,6 @@ static void flow_clear(int fd)
 
         ai.flows[fd].port_id  = -1;
         ai.flows[fd].pid      = -1;
-        ai.flows[fd].cube     = QOS_CUBE_BE;
 }
 
 static void flow_fini(int fd)
@@ -272,7 +269,7 @@ static void flow_fini(int fd)
 
 static int flow_init(int       port_id,
                      pid_t     pid,
-                     qoscube_t qc)
+                     qosspec_t qs)
 {
         int fd;
         int err = -ENOMEM;
@@ -300,9 +297,8 @@ static int flow_init(int       port_id,
         ai.flows[fd].port_id  = port_id;
         ai.flows[fd].oflags   = FLOWFDEFAULT;
         ai.flows[fd].pid      = pid;
-        ai.flows[fd].cube     = qc;
-        ai.flows[fd].spec     = qos_cube_to_spec(qc);
         ai.flows[fd].part_idx = NO_PART;
+        ai.flows[fd].spec     = qs;
 
         ai.ports[port_id].fd = fd;
 
@@ -499,7 +495,6 @@ int flow_accept(qosspec_t *             qs,
         irm_msg_t   msg = IRM_MSG__INIT;
         irm_msg_t * recv_msg;
         int         fd;
-        qoscube_t   qc;
 
         msg.code    = IRM_MSG_CODE__IRM_FLOW_ACCEPT;
         msg.has_pid = true;
@@ -528,14 +523,13 @@ int flow_accept(qosspec_t *             qs,
         }
 
         if (!recv_msg->has_pid || !recv_msg->has_port_id ||
-            !recv_msg->has_qoscube) {
+            recv_msg->qosspec == NULL) {
                 irm_msg__free_unpacked(recv_msg, NULL);
                 return -EIRMD;
         }
 
-        qc = recv_msg->qoscube;
-
-        fd = flow_init(recv_msg->port_id, recv_msg->pid, recv_msg->qoscube);
+        fd = flow_init(recv_msg->port_id, recv_msg->pid,
+                       msg_to_spec(recv_msg->qosspec));
 
         irm_msg__free_unpacked(recv_msg, NULL);
 
@@ -544,12 +538,10 @@ int flow_accept(qosspec_t *             qs,
 
         pthread_rwlock_wrlock(&ai.lock);
 
-        /* FIXME: check if FRCT is needed based on qc? */
-
         assert(ai.flows[fd].frcti == NULL);
 
-        if (qc != QOS_CUBE_RAW) {
-                ai.flows[fd].frcti = frcti_create(fd, qc);
+        if (ai.flows[fd].spec.in_order != 0) {
+                ai.flows[fd].frcti = frcti_create(fd);
                 if (ai.flows[fd].frcti == NULL) {
                         flow_fini(fd);
                         pthread_rwlock_unlock(&ai.lock);
@@ -569,21 +561,17 @@ int flow_alloc(const char *            dst,
                qosspec_t *             qs,
                const struct timespec * timeo)
 {
-        irm_msg_t   msg      = IRM_MSG__INIT;
-        irm_msg_t * recv_msg;
-        qoscube_t   qc       = QOS_CUBE_RAW;
-        int         fd;
-
-        msg.code        = IRM_MSG_CODE__IRM_FLOW_ALLOC;
-        msg.dst         = (char *) dst;
-        msg.has_pid     = true;
-        msg.has_qoscube = true;
-        msg.pid         = ai.pid;
-
-        if (qs != NULL)
-                qc = qos_spec_to_cube(*qs);
+        irm_msg_t     msg    = IRM_MSG__INIT;
+        qosspec_msg_t qs_msg = QOSSPEC_MSG__INIT;
+        irm_msg_t *   recv_msg;
+        int           fd;
 
-        msg.qoscube = qc;
+        msg.code    = IRM_MSG_CODE__IRM_FLOW_ALLOC;
+        msg.dst     = (char *) dst;
+        msg.has_pid = true;
+        msg.pid     = ai.pid;
+        qs_msg      = spec_to_msg(qs);
+        msg.qosspec = &qs_msg;
 
         if (timeo != NULL) {
                 msg.has_timeo_sec = true;
@@ -612,7 +600,8 @@ int flow_alloc(const char *            dst,
                 return -EIRMD;
         }
 
-        fd = flow_init(recv_msg->port_id, recv_msg->pid, qc);
+        fd = flow_init(recv_msg->port_id, recv_msg->pid,
+                       qs == NULL ? qos_raw : *qs);
 
         irm_msg__free_unpacked(recv_msg, NULL);
 
@@ -623,8 +612,8 @@ int flow_alloc(const char *            dst,
 
         assert(ai.flows[fd].frcti == NULL);
 
-        if (qc != QOS_CUBE_RAW) {
-                ai.flows[fd].frcti = frcti_create(fd, qc);
+        if (ai.flows[fd].spec.in_order != 0) {
+                ai.flows[fd].frcti = frcti_create(fd);
                 if (ai.flows[fd].frcti == NULL) {
                         flow_fini(fd);
                         pthread_rwlock_unlock(&ai.lock);
@@ -1178,9 +1167,9 @@ int fevent(struct flow_set *       set,
 
 int np1_flow_alloc(pid_t     n_pid,
                    int       port_id,
-                   qoscube_t qc)
+                   qosspec_t qs)
 {
-        return flow_init(port_id, n_pid, qc);
+        return flow_init(port_id, n_pid, qs);
 }
 
 int np1_flow_dealloc(int port_id)
@@ -1243,25 +1232,25 @@ int ipcp_create_r(pid_t pid,
 int ipcp_flow_req_arr(pid_t           pid,
                       const uint8_t * dst,
                       size_t          len,
-                      qoscube_t       qc)
+                      qosspec_t       qs)
 {
-        irm_msg_t   msg = IRM_MSG__INIT;
-        irm_msg_t * recv_msg;
-        int         fd;
+        irm_msg_t     msg = IRM_MSG__INIT;
+        irm_msg_t *   recv_msg;
+        qosspec_msg_t qs_msg;
+        int           fd;
 
         assert(dst != NULL);
 
-        msg.code        = IRM_MSG_CODE__IPCP_FLOW_REQ_ARR;
-        msg.has_pid     = true;
-        msg.pid         = pid;
-        msg.has_hash    = true;
-        msg.hash.len    = len;
-        msg.hash.data   = (uint8_t *) dst;
-        msg.has_qoscube = true;
-        msg.qoscube     = qc;
+        msg.code      = IRM_MSG_CODE__IPCP_FLOW_REQ_ARR;
+        msg.has_pid   = true;
+        msg.pid       = pid;
+        msg.has_hash  = true;
+        msg.hash.len  = len;
+        msg.hash.data = (uint8_t *) dst;
+        qs_msg        = spec_to_msg(&qs);
+        msg.qosspec   = &qs_msg;
 
         recv_msg = send_recv_irm_msg(&msg);
-
         if (recv_msg == NULL)
                 return -EIRMD;
 
@@ -1275,7 +1264,7 @@ int ipcp_flow_req_arr(pid_t           pid,
                 return -1;
         }
 
-        fd = flow_init(recv_msg->port_id, recv_msg->pid, qc);
+        fd = flow_init(recv_msg->port_id, recv_msg->pid, qs);
 
         irm_msg__free_unpacked(recv_msg, NULL);
 
@@ -1457,7 +1446,7 @@ int ipcp_flow_get_qoscube(int         fd,
 
         assert(ai.flows[fd].port_id >= 0);
 
-        *cube = ai.flows[fd].cube;
+        *cube = qos_spec_to_cube(ai.flows[fd].spec);
 
         pthread_rwlock_unlock(&ai.lock);
 
diff --git a/src/lib/frct.c b/src/lib/frct.c
index 516c958b..e9acb1dc 100644
--- a/src/lib/frct.c
+++ b/src/lib/frct.c
@@ -101,8 +101,7 @@ static void frct_fini(void)
         timerwheel_destroy(frct.tw);
 }
 
-static struct frcti * frcti_create(int       fd,
-                                   qoscube_t qc)
+static struct frcti * frcti_create(int fd)
 {
         struct frcti *  frcti;
         time_t          delta_t;
@@ -133,7 +132,7 @@ static struct frcti * frcti_create(int       fd,
         frcti->snd_cr.inact  = 3 * delta_t;
         frcti->snd_cr.act    = now.tv_sec - (frcti->snd_cr.inact + 1);
 
-        if (qc == QOS_CUBE_DATA)
+        if (ai.flows[fd].spec.loss == 0)
                 frcti->snd_cr.cflags |= FRCTFRTX;
 
         frcti->rcv_cr.inact  = 2 * delta_t;
diff --git a/src/lib/ipcpd_messages.proto b/src/lib/ipcpd_messages.proto
index 454af0dc..48198a5b 100644
--- a/src/lib/ipcpd_messages.proto
+++ b/src/lib/ipcpd_messages.proto
@@ -23,6 +23,7 @@
 syntax = "proto2";
 
 import "ipcp_config.proto";
+import "qosspec.proto";
 
 enum ipcp_msg_code {
         IPCP_BOOTSTRAP       =  1;
@@ -43,7 +44,7 @@ message ipcp_msg {
         optional bytes hash                =  2;
         optional int32 port_id             =  3;
         optional string dst                =  4;
-        optional uint32 qoscube            =  5;
+        optional qosspec_msg qosspec       =  5;
         optional ipcp_config_msg conf      =  6;
         optional int32 pid                 =  7;
         optional layer_info_msg layer_info =  8;
diff --git a/src/lib/irmd_messages.proto b/src/lib/irmd_messages.proto
index 16dfe828..2ed2ec37 100644
--- a/src/lib/irmd_messages.proto
+++ b/src/lib/irmd_messages.proto
@@ -23,6 +23,7 @@
 syntax = "proto2";
 
 import "ipcp_config.proto";
+import "qosspec.proto";
 
 enum irm_msg_code {
         IRM_CREATE_IPCP       =  1;
@@ -67,7 +68,7 @@ message irm_msg {
         optional string dst           =  9;
         optional bytes  hash          = 10;
         optional sint32 port_id       = 11;
-        optional sint32 qoscube       = 12;
+        optional qosspec_msg qosspec  = 12;
         optional ipcp_config_msg conf = 13;
         optional uint32 opts          = 14;
         repeated ipcp_info_msg ipcps  = 15;
diff --git a/src/lib/qos.c b/src/lib/qos.c
index bee6ed71..8607031e 100644
--- a/src/lib/qos.c
+++ b/src/lib/qos.c
@@ -28,66 +28,61 @@
 #include <string.h>
 
 qosspec_t qos_raw = {
-        .delay                = UINT32_MAX,
-        .bandwidth            = 0,
-        .availability         = 0,
-        .loss                 = 1,
-        .in_order             = 0,
-        .maximum_interruption = UINT32_MAX
+        .delay        = UINT32_MAX,
+        .bandwidth    = 0,
+        .availability = 0,
+        .loss         = 1,
+        .ber          = 1,
+        .in_order     = 0,
+        .max_gap      = UINT32_MAX
+};
+
+qosspec_t qos_raw_no_errors = {
+        .delay        = UINT32_MAX,
+        .bandwidth    = 0,
+        .availability = 0,
+        .loss         = 1,
+        .ber          = 0,
+        .in_order     = 0,
+        .max_gap      = UINT32_MAX
 };
 
 qosspec_t qos_best_effort = {
-        .delay                = UINT32_MAX,
-        .bandwidth            = 0,
-        .availability         = 0,
-        .loss                 = 1,
-        .in_order             = 1,
-        .maximum_interruption = UINT32_MAX
+        .delay        = UINT32_MAX,
+        .bandwidth    = 0,
+        .availability = 0,
+        .loss         = 1,
+        .ber          = 0,
+        .in_order     = 1,
+        .max_gap      = UINT32_MAX
 };
 
-qosspec_t qos_video = {
-        .delay                = 100,
-        .bandwidth            = UINT64_MAX,
-        .availability         = 3,
-        .loss                 = 1,
-        .in_order             = 1,
-        .maximum_interruption = 100
+qosspec_t qos_video   = {
+        .delay        = 100,
+        .bandwidth    = UINT64_MAX,
+        .availability = 3,
+        .loss         = 1,
+        .ber          = 0,
+        .in_order     = 1,
+        .max_gap      = 100
 };
 
 qosspec_t qos_voice = {
-        .delay                = 50,
-        .bandwidth            = 100000,
-        .availability         = 5,
-        .loss                 = 1,
-        .in_order             = 1,
-        .maximum_interruption = 50
+        .delay        = 50,
+        .bandwidth    = 100000,
+        .availability = 5,
+        .loss         = 1,
+        .ber          = 0,
+        .in_order     = 1,
+        .max_gap      = 50
 };
 
 qosspec_t qos_data = {
-        .delay                = 1000,
-        .bandwidth            = 0,
-        .availability         = 0,
-        .in_order             = 1,
-        .loss                 = 0,
-        .maximum_interruption = 2000
+        .delay        = 1000,
+        .bandwidth    = 0,
+        .availability = 0,
+        .loss         = 0,
+        .ber          = 0,
+        .in_order     = 1,
+        .max_gap      = 2000
 };
-
-int qosspec_init(qosspec_t * qs)
-{
-        if (qs == NULL)
-                return -EINVAL;
-
-        *qs = qos_best_effort;
-
-        return 0;
-}
-
-int qosspec_fini(qosspec_t * qs)
-{
-        if (qs == NULL)
-                return -EINVAL;
-
-        memset(qs, 0, sizeof(*qs));
-
-        return 0;
-}
diff --git a/src/lib/qoscube.c b/src/lib/qoscube.c
index 5dfa35ad..efca0e42 100644
--- a/src/lib/qoscube.c
+++ b/src/lib/qoscube.c
@@ -25,38 +25,20 @@
 
 #include <string.h>
 
+
+
 qoscube_t qos_spec_to_cube(qosspec_t qs)
 {
-        if (qs.loss == 0)
-                return QOS_CUBE_DATA;
-        else if (qs.delay <= qos_voice.delay &&
+        if (qs.delay <= qos_voice.delay &&
             qs.bandwidth <= qos_voice.bandwidth &&
             qs.availability >= qos_voice.availability &&
-            qs.maximum_interruption <= qos_voice.maximum_interruption)
+            qs.max_gap <= qos_voice.max_gap)
                 return QOS_CUBE_VOICE;
         else if (qs.delay <= qos_video.delay &&
                  qs.bandwidth <= qos_video.bandwidth &&
                  qs.availability >= qos_video.availability &&
-                 qs.maximum_interruption <= qos_video.maximum_interruption)
+                 qs.max_gap <= qos_video.max_gap)
                 return QOS_CUBE_VIDEO;
-        else if (qs.in_order == 1)
-                return QOS_CUBE_BE;
         else
-                return QOS_CUBE_RAW;
-}
-
-qosspec_t qos_cube_to_spec(qoscube_t qc)
-{
-        switch (qc) {
-        case QOS_CUBE_VOICE:
-                return qos_voice;
-        case QOS_CUBE_VIDEO:
-                return qos_video;
-        case QOS_CUBE_BE:
-                return qos_best_effort;
-        case QOS_CUBE_DATA:
-                return qos_data;
-        default:
-                return qos_raw;
-        }
+                return QOS_CUBE_BE;
 }
diff --git a/src/lib/qosspec.proto b/src/lib/qosspec.proto
new file mode 100644
index 00000000..f355e345
--- /dev/null
+++ b/src/lib/qosspec.proto
@@ -0,0 +1,33 @@
+/*
+ * Ouroboros - Copyright (C) 2016 - 2018
+ *
+ * QoS specification message
+ *
+ *    Dimitri Staessens <[email protected]>
+ *    Sander Vrijders   <[email protected]>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public License
+ * version 2.1 as published by the Free Software Foundation.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., http://www.fsf.org/about/contact/.
+ */
+
+syntax = "proto2";
+
+message qosspec_msg {
+        required uint32 delay        = 1; /* In ms */
+        required uint64 bandwidth    = 2; /* In bits/s */
+        required uint32 availability = 3; /* Class of 9s */
+        required uint32 loss         = 4; /* Packet loss */
+        required uint32 ber          = 5; /* Bit error rate, ppb */
+        required uint32 in_order     = 6; /* In-order delivery */
+        required uint32 max_gap      = 7; /* In ms */
+};
diff --git a/src/lib/sockets.c b/src/lib/sockets.c
index b148b7ca..85726783 100644
--- a/src/lib/sockets.c
+++ b/src/lib/sockets.c
@@ -165,3 +165,38 @@ char * ipcp_sock_path(pid_t pid)
 
         return full_name;
 }
+
+qosspec_msg_t spec_to_msg(qosspec_t * qs)
+{
+        qosspec_t     spec;
+        qosspec_msg_t msg = QOSSPEC_MSG__INIT;
+
+        spec = (qs == NULL ? qos_raw : *qs);
+
+        msg.delay        = spec.delay;
+        msg.bandwidth    = spec.bandwidth;
+        msg.availability = spec.availability;
+        msg.loss         = spec.loss;
+        msg.ber          = spec.ber;
+        msg.in_order     = spec.in_order;
+        msg.max_gap      = spec.max_gap;
+
+        return msg;
+}
+
+qosspec_t msg_to_spec(qosspec_msg_t * msg)
+{
+        qosspec_t     spec;
+
+        assert(msg);
+
+        spec.delay        = msg->delay;
+        spec.bandwidth    = msg->bandwidth;
+        spec.availability = msg->availability;
+        spec.loss         = msg->loss;
+        spec.ber          = msg->ber;
+        spec.in_order     = msg->in_order;
+        spec.max_gap      = msg->max_gap;
+
+        return spec;
+}