Merged in sandervrijders/ouroboros/be-qos (pull request #517)

lib, ipcpd, irmd: Add full-fledged QoS

5 files changed, 253 insertions, 33 deletions

diff --git a/src/lib/CMakeLists.txt b/src/lib/CMakeLists.txt
index 263065dd..99dd96fc 100644
--- a/src/lib/CMakeLists.txt
+++ b/src/lib/CMakeLists.txt
@@ -11,6 +11,7 @@ protobuf_generate_c(DIF_CONFIG_PROTO_SRCS DIF_CONFIG_PROTO_HDRS
 protobuf_generate_c(CDAP_PROTO_SRCS CDAP_PROTO_HDRS cdap.proto)
 protobuf_generate_c(RO_PROTO_SRCS RO_PROTO_HDRS ro.proto)
 protobuf_generate_c(CACEP_PROTO_SRCS CACEP_PROTO_HDRS cacep.proto)
+protobuf_generate_c(FRCT_ENROLL_SRCS FRCT_ENROLL_HDRS frct_enroll.proto)
 
 if (NOT APPLE)
   find_library(LIBRT_LIBRARIES rt)
@@ -43,6 +44,8 @@ set(SOURCE_FILES
   logs.c
   md5.c
   nsm.c
+  qos.c
+  qoscube.c
   rib.c
   sha3.c
   shm_flow_set.c
@@ -55,7 +58,7 @@ set(SOURCE_FILES
 
 add_library(ouroboros SHARED ${SOURCE_FILES} ${IRM_PROTO_SRCS}
   ${IPCP_PROTO_SRCS} ${DIF_CONFIG_PROTO_SRCS} ${CDAP_PROTO_SRCS}
-  ${CACEP_PROTO_SRCS} ${RO_PROTO_SRCS})
+  ${CACEP_PROTO_SRCS} ${RO_PROTO_SRCS} ${FRCT_ENROLL_SRCS})
 
 target_link_libraries(ouroboros ${LIBRT_LIBRARIES}
   ${LIBPTHREAD_LIBRARIES} ${PROTOBUF_C_LIBRARY})
diff --git a/src/lib/dev.c b/src/lib/dev.c
index 3cdcc4d3..e4fd32cc 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -34,11 +34,14 @@
 #include <ouroboros/shm_rbuff.h>
 #include <ouroboros/utils.h>
 #include <ouroboros/fqueue.h>
+#include <ouroboros/qoscube.h>
 
 #include <stdlib.h>
 #include <string.h>
 #include <stdio.h>
 
+#define BUF_SIZE 1500
+
 struct flow_set {
         size_t idx;
 };
@@ -72,6 +75,7 @@ struct flow {
         int                   port_id;
         int                   oflags;
         qoscube_t             cube;
+        qosspec_t             spec;
 
         pid_t                 api;
 
@@ -174,24 +178,6 @@ static enum port_state port_wait_assign(int port_id)
         return state;
 }
 
-/* FIXME: translate real spec to cube */
-static qoscube_t spec_to_cube(qosspec_t * qs)
-{
-        if (qs == NULL)
-                return QOS_CUBE_BE;
-
-        return qs->cube;
-}
-
-/* FIXME: fill real spec */
-static void fill_qosspec(qosspec_t * qs,
-                         qoscube_t   cube)
-{
-        assert(qs);
-
-        qs->cube = cube;
-}
-
 static int api_announce(char * ap_name)
 {
         irm_msg_t msg = IRM_MSG__INIT;
@@ -382,9 +368,13 @@ void ouroboros_fini()
 int flow_accept(qosspec_t *             qs,
                 const struct timespec * timeo)
 {
-        irm_msg_t msg = IRM_MSG__INIT;
-        irm_msg_t * recv_msg = NULL;
-        int fd = -1;
+        irm_msg_t           msg      = IRM_MSG__INIT;
+        irm_msg_t *         recv_msg = NULL;
+        int                 fd       = -1;
+        frct_enroll_msg_t * frct_enroll;
+        qosspec_t           spec;
+        uint8_t             data[BUF_SIZE];
+        ssize_t             n;
 
         msg.code    = IRM_MSG_CODE__IRM_FLOW_ACCEPT;
         msg.has_api = true;
@@ -412,7 +402,8 @@ int flow_accept(qosspec_t *             qs,
                 return res;
         }
 
-        if (!recv_msg->has_api || !recv_msg->has_port_id) {
+        if (!recv_msg->has_api || !recv_msg->has_port_id ||
+            !recv_msg->has_qoscube) {
                 irm_msg__free_unpacked(recv_msg, NULL);
                 return -EIRMD;
         }
@@ -459,8 +450,7 @@ int flow_accept(qosspec_t *             qs,
 
         assert(ai.ports[ai.flows[fd].port_id].state == PORT_INIT);
 
-        if (qs != NULL)
-                fill_qosspec(qs, ai.flows[fd].cube);
+        spec = qos_cube_to_spec(recv_msg->qoscube);
 
         ai.ports[recv_msg->port_id].fd    = fd;
         ai.ports[recv_msg->port_id].state = PORT_ID_ASSIGNED;
@@ -469,6 +459,33 @@ int flow_accept(qosspec_t *             qs,
 
         irm_msg__free_unpacked(recv_msg, NULL);
 
+        n = flow_read(fd, data, BUF_SIZE);
+        if (n < 0) {
+                flow_dealloc(fd);
+                return n;
+        }
+
+        frct_enroll = frct_enroll_msg__unpack(NULL, n, data);
+        if (frct_enroll == NULL) {
+                flow_dealloc(fd);
+                return -1;
+        }
+
+        spec.resource_control = frct_enroll->resource_control;
+        spec.reliable = frct_enroll->reliable;
+        spec.error_check = frct_enroll->error_check;
+        spec.ordered = frct_enroll->ordered;
+        spec.partial = frct_enroll->partial;
+
+        frct_enroll_msg__free_unpacked(frct_enroll, NULL);
+
+        pthread_rwlock_wrlock(&ai.flows_lock);
+        ai.flows[fd].spec = spec;
+        pthread_rwlock_unlock(&ai.flows_lock);
+
+        if (qs != NULL)
+                *qs = spec;
+
         return fd;
 }
 
@@ -476,18 +493,33 @@ int flow_alloc(const char *            dst_name,
                qosspec_t *             qs,
                const struct timespec * timeo)
 {
-        irm_msg_t msg = IRM_MSG__INIT;
-        irm_msg_t * recv_msg = NULL;
-
-        int fd;
+        irm_msg_t         msg         = IRM_MSG__INIT;
+        frct_enroll_msg_t frct_enroll = FRCT_ENROLL_MSG__INIT;
+        irm_msg_t *       recv_msg    = NULL;
+        qoscube_t         qc          = QOS_CUBE_BE;
+        int               fd;
+        ssize_t           len;
+        uint8_t *         data;
+        int               ret;
 
         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(qs);
         msg.api         = ai.api;
 
+        if (qs != NULL) {
+                frct_enroll.resource_control = qs->resource_control;
+                frct_enroll.reliable = qs->reliable;
+                frct_enroll.error_check = qs->error_check;
+                frct_enroll.ordered = qs->ordered;
+                frct_enroll.partial = qs->partial;
+
+                qc = qos_spec_to_cube(*qs);
+        }
+
+        msg.qoscube = qc;
+
         if (timeo != NULL) {
                 msg.has_timeo_sec = true;
                 msg.has_timeo_nsec = true;
@@ -560,9 +592,29 @@ int flow_alloc(const char *            dst_name,
         ai.ports[recv_msg->port_id].fd    = fd;
         ai.ports[recv_msg->port_id].state = PORT_ID_ASSIGNED;
 
+        irm_msg__free_unpacked(recv_msg, NULL);
+
         pthread_rwlock_unlock(&ai.flows_lock);
 
-        irm_msg__free_unpacked(recv_msg, NULL);
+        len = frct_enroll_msg__get_packed_size(&frct_enroll);
+        if (len < 0) {
+                flow_dealloc(fd);
+                return -1;
+        }
+
+        data = malloc(len);
+        if (data == NULL) {
+                flow_dealloc(fd);
+                return -ENOMEM;
+        }
+
+        frct_enroll_msg__pack(&frct_enroll, data);
+
+        ret = flow_write(fd, data, len);
+        if (ret < 0) {
+                flow_dealloc(fd);
+                return ret;
+        }
 
         return fd;
 }
@@ -582,7 +634,7 @@ int flow_dealloc(int fd)
 
         pthread_rwlock_rdlock(&ai.flows_lock);
 
-        assert (!(ai.flows[fd].port_id < 0));
+        assert(!(ai.flows[fd].port_id < 0));
 
         msg.port_id = ai.flows[fd].port_id;
 
@@ -722,7 +774,7 @@ int flow_get_qosspec(int         fd,
                 return -ENOTALLOC;
         }
 
-        fill_qosspec(qs, ai.flows[fd].cube);
+        *qs = ai.flows[fd].spec;
 
         pthread_rwlock_unlock(&ai.flows_lock);
 
diff --git a/src/lib/frct_enroll.proto b/src/lib/frct_enroll.proto
new file mode 100644
index 00000000..497d6acc
--- /dev/null
+++ b/src/lib/frct_enroll.proto
@@ -0,0 +1,32 @@
+/*
+ * Ouroboros - Copyright (C) 2016 - 2017
+ *
+ * QoS messages
+ *
+ *    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., 51 Franklin Street, Fifth Floor, Boston, MA
+ * 02110-1301 USA
+ */
+
+syntax = "proto2";
+
+message frct_enroll_msg {
+        required bool resource_control = 1;
+        required bool reliable         = 2;
+        required bool error_check      = 3;
+        required bool ordered          = 4;
+        required bool partial          = 5;
+};
diff --git a/src/lib/qos.c b/src/lib/qos.c
new file mode 100644
index 00000000..035b9049
--- /dev/null
+++ b/src/lib/qos.c
@@ -0,0 +1,57 @@
+/*
+ * Ouroboros - Copyright (C) 2016 - 2017
+ *
+ * Quality of Service cube specifications
+ *
+ *    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., 51 Franklin Street, Fifth Floor, Boston, MA
+ * 02110-1301 USA
+ */
+
+#include <ouroboros/qos.h>
+#include <ouroboros/errno.h>
+
+#include <stdint.h>
+#include <stddef.h>
+
+int qosspec_init(qosspec_t * qs)
+{
+        if (qs == NULL)
+                return -EINVAL;
+
+        qs->delay = UINT32_MAX;
+        qs->bandwidth = UINT64_MAX;
+        qs->availability = 0;
+        qs->maximum_interruption = UINT32_MAX;
+
+        qs->resource_control = true;
+        qs->reliable = false;
+        qs->error_check = true;
+        qs->ordered = true;
+        qs->partial = false;
+
+        return 0;
+}
+
+int qosspec_fini(qosspec_t * qs)
+{
+        if (qs == NULL)
+                return -EINVAL;
+
+        qs = NULL;
+
+        return 0;
+}
diff --git a/src/lib/qoscube.c b/src/lib/qoscube.c
new file mode 100644
index 00000000..4ab827a6
--- /dev/null
+++ b/src/lib/qoscube.c
@@ -0,0 +1,76 @@
+/*
+ * Ouroboros - Copyright (C) 2016 - 2017
+ *
+ * Quality of Service cube
+ *
+ *    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., 51 Franklin Street, Fifth Floor, Boston, MA
+ * 02110-1301 USA
+ */
+
+#include <ouroboros/qoscube.h>
+
+#include <string.h>
+
+static struct qos_spec qos_best_effort = {
+        .delay                = UINT32_MAX,
+        .bandwidth            = UINT64_MAX,
+        .availability         = 0,
+        .maximum_interruption = UINT32_MAX
+};
+
+static struct qos_spec qos_video = {
+        .delay                = 100,
+        .bandwidth            = UINT64_MAX,
+        .availability         = 3,
+        .maximum_interruption = 100
+};
+
+static struct qos_spec qos_voice = {
+        .delay                = 10,
+        .bandwidth            = 100000,
+        .availability         = 5,
+        .maximum_interruption = 50
+};
+
+qoscube_t qos_spec_to_cube(qosspec_t qs)
+{
+        if (qs.delay <= qos_voice.delay &&
+            qs.bandwidth <= qos_voice.bandwidth &&
+            qs.availability >= qos_voice.availability &&
+            qs.maximum_interruption <= qos_voice.maximum_interruption)
+                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)
+                return QOS_CUBE_VIDEO;
+        else
+                return QOS_CUBE_BE;
+}
+
+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:
+        default:
+                return qos_best_effort;
+        }
+}