lib: Add fccntl configuration command

This replaces the flow_set_* commands with a single fccntl command
that can configure flows and the FRCT instance.

For more details, see "man 3 fccntl".

8 files changed, 140 insertions, 163 deletions

diff --git a/src/lib/cdap.c b/src/lib/cdap.c
index 7d2feae3..d9cb2036 100644
--- a/src/lib/cdap.c
+++ b/src/lib/cdap.c
@@ -26,7 +26,6 @@
 #include <ouroboros/bitmap.h>
 #include <ouroboros/dev.h>
 #include <ouroboros/fqueue.h>
-#include <ouroboros/fcntl.h>
 #include <ouroboros/errno.h>
 
 #include "cdap_req.h"
diff --git a/src/lib/dev.c b/src/lib/dev.c
index d5044fb3..e02d66d0 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -29,7 +29,7 @@
 #include <ouroboros/ipcp-dev.h>
 #include <ouroboros/local-dev.h>
 #include <ouroboros/sockets.h>
-#include <ouroboros/fcntl.h>
+#include <ouroboros/fccntl.h>
 #include <ouroboros/bitmap.h>
 #include <ouroboros/shm_flow_set.h>
 #include <ouroboros/shm_rdrbuff.h>
@@ -43,6 +43,7 @@
 #include <stdlib.h>
 #include <string.h>
 #include <stdio.h>
+#include <stdarg.h>
 
 #define BUF_SIZE 1500
 
@@ -86,7 +87,7 @@ struct frcti {
         uint64_t        rcv_lwe;
         uint64_t        rcv_rwe;
 
-        uint8_t         conf_flags;
+        uint16_t        conf_flags;
 };
 
 struct port {
@@ -108,7 +109,9 @@ struct flow {
 
         pid_t                 api;
 
-        bool                  timesout;
+        bool                  snd_timesout;
+        bool                  rcv_timesout;
+        struct timespec       snd_timeo;
         struct timespec       rcv_timeo;
 };
 
@@ -309,7 +312,7 @@ static int frcti_send(int                  fd,
 
         pci->seqno = frcti->snd_lwe++;
 
-        if (frct_pci_ser(sdb, pci, frcti->conf_flags & CONF_ERROR_CHECK)) {
+        if (frct_pci_ser(sdb, pci, frcti->conf_flags & FRCTFERRCHCK)) {
                 pthread_rwlock_unlock(&ai.lock);
                 return -1;
         }
@@ -325,8 +328,8 @@ static int frcti_send(int                  fd,
 }
 
 
-static int frcti_configure(int         fd,
-                           qosspec_t * qos)
+static int frcti_configure(int      fd,
+                           uint16_t flags)
 {
         struct frcti *       frcti;
         struct frct_pci      pci;
@@ -336,22 +339,10 @@ static int frcti_configure(int         fd,
 
         memset(&pci, 0, sizeof(pci));
 
-        if (qos == NULL)
-                return 0;
-
         if (ipcp_sdb_reserve(&sdb, 0))
                 return -1;
 
-        if (qos->resource_control)
-                pci.conf_flags |= CONF_RESOURCE_CONTROL;
-        if (qos->reliable)
-                pci.conf_flags |= CONF_RELIABLE;
-        if (qos->error_check)
-                pci.conf_flags |= CONF_ERROR_CHECK;
-        if (qos->ordered)
-                pci.conf_flags |= CONF_ORDERED;
-        if (qos->partial)
-                pci.conf_flags |= CONF_PARTIAL;
+        pci.conf_flags = flags;
 
         /* Always set the DRF on a configure message. */
         pci.flags |= FLAG_DATA_RUN;
@@ -385,49 +376,50 @@ static int frcti_write(int                  fd,
 
 static ssize_t frcti_read(int fd)
 {
-        ssize_t              idx = -1;
-        struct timespec      abstime;
+        ssize_t              idx;
         struct frcti *       frcti;
         struct frct_pci      pci;
         struct shm_du_buff * sdb;
-        struct timespec      now = {0, 0};
 
         do {
-                pthread_rwlock_rdlock(&ai.lock);
+                struct timespec    now;
+                struct timespec    abs;
+                struct timespec *  abstime = NULL;
+                struct shm_rbuff * rb;
+                bool               noblock;
 
-                if (ai.flows[fd].oflags & FLOW_O_NONBLOCK) {
-                        idx = shm_rbuff_read(ai.flows[fd].rx_rb);
-                        pthread_rwlock_unlock(&ai.lock);
-                } else {
-                        struct shm_rbuff * rb   = ai.flows[fd].rx_rb;
-                        bool timeo = ai.flows[fd].timesout;
-                        struct timespec timeout = ai.flows[fd].rcv_timeo;
+                clock_gettime(PTHREAD_COND_CLOCK, &now);
 
-                        pthread_rwlock_unlock(&ai.lock);
+                pthread_rwlock_rdlock(&ai.lock);
+
+                noblock = ai.flows[fd].oflags & FLOWFNONBLOCK;
+                rb      = ai.flows[fd].rx_rb;
 
-                        if (timeo) {
-                                clock_gettime(PTHREAD_COND_CLOCK, &abstime);
-                                ts_add(&abstime, &timeout, &abstime);
-                                idx = shm_rbuff_read_b(rb, &abstime);
-                        } else {
-                                idx = shm_rbuff_read_b(rb, NULL);
-                        }
+                if (ai.flows[fd].rcv_timesout) {
+                        ts_add(&now, &ai.flows[fd].rcv_timeo, &abs);
+                        abstime = &abs;
                 }
 
+                pthread_rwlock_unlock(&ai.lock);
+
+                if (noblock)
+                        idx = shm_rbuff_read(rb);
+                else
+                        idx = shm_rbuff_read_b(rb, abstime);
+
                 if (idx < 0)
                         return idx;
 
                 clock_gettime(CLOCK_REALTIME_COARSE, &now);
 
-                pthread_rwlock_wrlock(&ai.lock);
-
                 frcti = &(ai.frcti[fd]);
 
                 sdb = shm_rdrbuff_get(ai.rdrb, idx);
 
+                pthread_rwlock_wrlock(&ai.lock);
+
                 /* SDU may be corrupted. */
-                if (frct_pci_des(sdb, &pci,
-                                 frcti->conf_flags & CONF_ERROR_CHECK)) {
+                if (frct_pci_des(sdb, &pci, frcti->conf_flags & FRCTFERRCHCK)) {
                         pthread_rwlock_unlock(&ai.lock);
                         shm_rdrbuff_remove(ai.rdrb, idx);
                         return -EAGAIN;
@@ -538,7 +530,7 @@ static int flow_init(int       port_id,
         }
 
         ai.flows[fd].port_id = port_id;
-        ai.flows[fd].oflags  = FLOW_O_DEFAULT;
+        ai.flows[fd].oflags  = FLOWFDEFAULT;
         ai.flows[fd].api     = api;
         ai.flows[fd].cube    = qc;
         ai.flows[fd].spec    = qos_cube_to_spec(qc);
@@ -821,7 +813,7 @@ int flow_alloc(const char *            dst_name,
 
         pthread_rwlock_unlock(&ai.lock);
 
-        if (frcti_configure(fd, qs)) {
+        if (frcti_configure(fd, FRCTFORDERING | FRCTFERRCHCK)) {
                 flow_fini(fd);
                 bmp_release(ai.fds, fd);
                 return -1;
@@ -872,123 +864,113 @@ int flow_dealloc(int fd)
         return 0;
 }
 
-int flow_set_flags(int fd,
-                   int flags)
+int fccntl(int fd,
+           int cmd,
+           ...)
 {
-        int old;
+        uint32_t *        fflags;
+        uint16_t *        cflags;
+        va_list           l;
+        struct timespec * timeo;
+        qosspec_t *       qs;
 
         if (fd < 0 || fd >= AP_MAX_FLOWS)
                 return -EBADF;
 
-        pthread_rwlock_wrlock(&ai.lock);
-
-        if (ai.flows[fd].port_id < 0) {
-                pthread_rwlock_unlock(&ai.lock);
-                return -ENOTALLOC;
-        }
-
-        old = ai.flows[fd].oflags;
-
-        ai.flows[fd].oflags = flags;
-        if (flags & FLOW_O_WRONLY)
-                shm_rbuff_block(ai.flows[fd].rx_rb);
-        if (flags & FLOW_O_RDWR)
-                shm_rbuff_unblock(ai.flows[fd].rx_rb);
-
-        pthread_rwlock_unlock(&ai.lock);
-
-        return old;
-}
-
-int flow_get_flags(int fd)
-{
-        int old;
-
-        if (fd < 0 || fd >= AP_MAX_FLOWS)
-                return -EBADF;
+        va_start(l, cmd);
 
         pthread_rwlock_wrlock(&ai.lock);
 
         if (ai.flows[fd].port_id < 0) {
                 pthread_rwlock_unlock(&ai.lock);
+                va_end(l);
                 return -ENOTALLOC;
         }
 
-        old = ai.flows[fd].oflags;
-
-        pthread_rwlock_unlock(&ai.lock);
-
-        return old;
-}
-
-int flow_get_timeout(int               fd,
-                     struct timespec * timeo)
-{
-        int ret = 0;
-
-        if (fd < 0 || fd > AP_MAX_FLOWS || timeo == NULL)
-                return -EINVAL;
-
-        pthread_rwlock_wrlock(&ai.lock);
-
-        if (ai.flows[fd].port_id < 0) {
+        switch(cmd) {
+        case FLOWSSNDTIMEO:
+                timeo = va_arg(l, struct timespec *);
+                if (timeo == NULL) {
+                        ai.flows[fd].snd_timesout = false;
+                } else {
+                        ai.flows[fd].snd_timesout = true;
+                        ai.flows[fd].snd_timeo    = *timeo;
+                }
+                break;
+        case FLOWGSNDTIMEO:
+                timeo = va_arg(l, struct timespec *);
+                if (timeo == NULL)
+                        goto einval;
+                if (!ai.flows[fd].snd_timesout)
+                        goto eperm;
+                *timeo = ai.flows[fd].snd_timeo;
+                break;
+        case FLOWSRCVTIMEO:
+                timeo = va_arg(l, struct timespec *);
+                if (timeo == NULL) {
+                        ai.flows[fd].rcv_timesout = false;
+                } else {
+                        ai.flows[fd].rcv_timesout = true;
+                        ai.flows[fd].rcv_timeo    = *timeo;
+                }
+                break;
+        case FLOWGRCVTIMEO:
+                timeo = va_arg(l, struct timespec *);
+                if (timeo == NULL)
+                        goto einval;
+                if (!ai.flows[fd].rcv_timesout)
+                        goto eperm;
+                *timeo = ai.flows[fd].snd_timeo;
+                break;
+        case FLOWGQOSSPEC:
+                qs = va_arg(l, qosspec_t *);
+                if (qs == NULL)
+                        goto einval;
+                *qs = ai.flows[fd].spec;
+                break;
+        case FLOWSFLAGS:
+                ai.flows[fd].oflags = va_arg(l, uint32_t);
+                if (ai.flows[fd].oflags & FLOWFWRONLY)
+                        shm_rbuff_block(ai.flows[fd].rx_rb);
+                if (ai.flows[fd].oflags & FLOWFRDWR)
+                        shm_rbuff_unblock(ai.flows[fd].rx_rb);
+                break;
+        case FLOWGFLAGS:
+                fflags = va_arg(l, uint32_t *);
+                if (fflags == NULL)
+                        goto einval;
+                *fflags = ai.flows[fd].oflags;
+                break;
+        case FRCTSFLAGS:
+                ai.frcti[fd].conf_flags = (uint16_t) va_arg(l, int);
+                break;
+        case FRCTGFLAGS:
+                cflags = (uint16_t *) va_arg(l, int *);
+                if (cflags == NULL)
+                        goto einval;
+                *cflags = ai.frcti[fd].conf_flags;
+                break;
+        default:
                 pthread_rwlock_unlock(&ai.lock);
-                return -ENOTALLOC;
-        }
+                va_end(l);
+                return -ENOTSUP;
 
-        if (ai.flows[fd].timesout)
-                *timeo = ai.flows[fd].rcv_timeo;
-        else
-                ret = -EPERM;
+        };
 
         pthread_rwlock_unlock(&ai.lock);
 
-        return ret;
-}
-
-int flow_set_timeout(int                     fd,
-                     const struct timespec * timeo)
-{
-        if (fd < 0 || fd > AP_MAX_FLOWS)
-                return -EINVAL;
-
-        pthread_rwlock_wrlock(&ai.lock);
-
-        if (ai.flows[fd].port_id < 0) {
-                pthread_rwlock_unlock(&ai.lock);
-                return -ENOTALLOC;
-        }
-
-        if (timeo == NULL) {
-                ai.flows[fd].timesout = false;
-        } else {
-                ai.flows[fd].timesout = true;
-                ai.flows[fd].rcv_timeo = *timeo;
-        }
-
-        pthread_rwlock_unlock(&ai.lock);
+        va_end(l);
 
         return 0;
-}
-
-int flow_get_qosspec(int         fd,
-                     qosspec_t * qs)
-{
-        if (fd < 0 || fd > AP_MAX_FLOWS || qs == NULL)
-                return -EINVAL;
-
-        pthread_rwlock_wrlock(&ai.lock);
-
-        if (ai.flows[fd].port_id < 0) {
-                pthread_rwlock_unlock(&ai.lock);
-                return -ENOTALLOC;
-        }
-
-        *qs = ai.flows[fd].spec;
 
+ einval:
         pthread_rwlock_unlock(&ai.lock);
-
-        return 0;
+        va_end(l);
+        return -EINVAL;
+ eperm:
+        pthread_rwlock_unlock(&ai.lock);
+        va_end(l);
+        return -EPERM;
 }
 
 ssize_t flow_write(int          fd,
@@ -1010,12 +992,12 @@ ssize_t flow_write(int          fd,
                 return -ENOTALLOC;
         }
 
-        if ((ai.flows[fd].oflags & FLOW_O_ACCMODE) == FLOW_O_RDONLY) {
+        if ((ai.flows[fd].oflags & FLOWFACCMODE) == FLOWFRDONLY) {
                 pthread_rwlock_unlock(&ai.lock);
                 return -EPERM;
         }
 
-        if (ai.flows[fd].oflags & FLOW_O_NONBLOCK) {
+        if (ai.flows[fd].oflags & FLOWFNONBLOCK) {
                 idx = shm_rdrbuff_write(ai.rdrb,
                                        DU_BUFF_HEADSPACE,
                                        DU_BUFF_TAILSPACE,
@@ -1062,10 +1044,11 @@ ssize_t flow_read(int    fd,
                   void * buf,
                   size_t count)
 {
-        ssize_t   idx = -1;
-        ssize_t   n;
-        uint8_t * sdu;
-        bool      used;
+        ssize_t            idx;
+        ssize_t            n;
+        uint8_t *          sdu;
+        bool               used;
+        struct shm_rbuff * rb;
 
         if (fd < 0 || fd > AP_MAX_FLOWS)
                 return -EBADF;
@@ -1078,11 +1061,12 @@ ssize_t flow_read(int    fd,
         }
 
         used = ai.frcti[fd].used;
+        rb   = ai.flows[fd].rx_rb;
 
         pthread_rwlock_unlock(&ai.lock);
 
         if (!used)
-                idx = shm_rbuff_read(ai.flows[fd].rx_rb);
+                idx = shm_rbuff_read(rb);
         else
                 idx = frcti_read(fd);
 
@@ -1477,7 +1461,7 @@ int ipcp_flow_write(int                  fd,
                 return -ENOTALLOC;
         }
 
-        if ((ai.flows[fd].oflags & FLOW_O_ACCMODE) == FLOW_O_RDONLY) {
+        if ((ai.flows[fd].oflags & FLOWFACCMODE) == FLOWFRDONLY) {
                 pthread_rwlock_unlock(&ai.lock);
                 return -EPERM;
         }
@@ -1527,7 +1511,7 @@ void ipcp_flow_fini(int fd)
 {
         struct shm_rbuff * rx_rb;
 
-        flow_set_flags(fd, FLOW_O_WRONLY);
+        fccntl(fd, FLOWSFLAGS, FLOWFWRONLY);
 
         pthread_rwlock_rdlock(&ai.lock);
 
diff --git a/src/lib/frct_pci.c b/src/lib/frct_pci.c
index 4fa9ddc2..e44554f2 100644
--- a/src/lib/frct_pci.c
+++ b/src/lib/frct_pci.c
@@ -30,7 +30,7 @@
 #define TYPE_SIZE       1
 #define SEQNO_SIZE      8
 #define FLAGS_SIZE      1
-#define CONF_FLAGS_SIZE 1
+#define CONF_FLAGS_SIZE sizeof(((struct frct_pci *) NULL)->conf_flags)
 #define BASE_SIZE       TYPE_SIZE + FLAGS_SIZE + SEQNO_SIZE
 #define CONFIG_SIZE     CONF_FLAGS_SIZE
 
diff --git a/src/lib/qos.c b/src/lib/qos.c
index 35e16f97..09866c4e 100644
--- a/src/lib/qos.c
+++ b/src/lib/qos.c
@@ -36,12 +36,6 @@ int qosspec_init(qosspec_t * qs)
         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;
 }
 
diff --git a/src/tools/cbr/cbr_server.c b/src/tools/cbr/cbr_server.c
index a2e930e5..b25b09e8 100644
--- a/src/tools/cbr/cbr_server.c
+++ b/src/tools/cbr/cbr_server.c
@@ -22,7 +22,7 @@
 
 #include <ouroboros/dev.h>
 #include <ouroboros/time_utils.h>
-#include <ouroboros/fcntl.h>
+#include <ouroboros/fccntl.h>
 
 #include <stdbool.h>
 
@@ -84,7 +84,7 @@ static void handle_flow(int fd)
         alive = iv_start;
         ts_add(&iv_start, &intv, &iv_end);
 
-        flow_set_flags(fd, FLOW_O_NONBLOCK);
+        fccntl(fd, FLOWSFLAGS, FLOWFNONBLOCK);
 
         while (!stop) {
                 clock_gettime(CLOCK_REALTIME, &now);
diff --git a/src/tools/operf/operf_client.c b/src/tools/operf/operf_client.c
index 4ad26d13..df486380 100644
--- a/src/tools/operf/operf_client.c
+++ b/src/tools/operf/operf_client.c
@@ -21,7 +21,7 @@
  */
 
 #include <ouroboros/dev.h>
-#include <ouroboros/fcntl.h>
+#include <ouroboros/fccntl.h>
 #include <ouroboros/time_utils.h>
 
 #include <signal.h>
@@ -68,7 +68,7 @@ void * reader(void * o)
         int fd = *((int *) o);
         int msg_len = 0;
 
-        flow_set_timeout(fd, &timeout);
+        fccntl(fd, FLOWSRCVTIMEO, &timeout);
 
         while ((msg_len = flow_read(fd, buf, OPERF_BUF_SIZE)) != -ETIMEDOUT) {
                 if (msg_len != client.size) {
diff --git a/src/tools/oping/oping_client.c b/src/tools/oping/oping_client.c
index db0ef199..ca8c0e72 100644
--- a/src/tools/oping/oping_client.c
+++ b/src/tools/oping/oping_client.c
@@ -21,7 +21,7 @@
  */
 
 #include <ouroboros/dev.h>
-#include <ouroboros/fcntl.h>
+#include <ouroboros/fccntl.h>
 #include <ouroboros/time_utils.h>
 
 #include <signal.h>
@@ -60,7 +60,7 @@ void * reader(void * o)
         double ms = 0;
         double d = 0;
 
-        flow_set_timeout(fd, &timeout);
+        fccntl(fd, FLOWSRCVTIMEO, &timeout);
 
         while (client.rcvd != client.count) {
                 msg_len = flow_read(fd, buf, OPING_BUF_SIZE);
diff --git a/src/tools/oping/oping_server.c b/src/tools/oping/oping_server.c
index 6a535caf..0371f90f 100644
--- a/src/tools/oping/oping_server.c
+++ b/src/tools/oping/oping_server.c
@@ -130,7 +130,7 @@ void * accept_thread(void * o)
                 server.times[fd] = now;
                 pthread_mutex_unlock(&server.lock);
 
-                flow_set_flags(fd, FLOW_O_NONBLOCK | FLOW_O_RDWR);
+                fccntl(fd, FLOWSFLAGS, FLOWFNONBLOCK | FLOWFRDWR);
         }
 
         return (void *) 0;