ipcpd: Use capacity queue estimation for mb-ecn

The mb-ecn algorithm was using rbuff queue depths in packets to mark,
but sockets in the poa component report capacity in bytes. The tx
rings are now adaptive to block on queuing delay instead of when full
to prevent buffer bloat, controllable via fccntl (FLOWSTXQDLY and
FLOWGTXQDLY).

Signed-off-by: Dimitri Staessens <dimitri@ouroboros.rocks>
Signed-off-by: Sander Vrijders <sander@ouroboros.rocks>

1 files changed, 135 insertions, 10 deletions

diff --git a/src/lib/dev.c b/src/lib/dev.c
index eb706691..bce64c37 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -27,6 +27,7 @@
 #endif
 
 #include "config.h"
+#include "cap.h"
 #include "ssm.h"
 #include "poa/poa.h"
 
@@ -60,6 +61,7 @@
 #include <ouroboros/ssm_flow_set.h>
 #include <ouroboros/ssm_pool.h>
 #include <ouroboros/ssm_rbuff.h>
+#include <ouroboros/time.h>
 #include <ouroboros/tw.h>
 #include <ouroboros/utils.h>
 
@@ -85,6 +87,7 @@
 #define DONE_PART -2
 
 #define CRCLEN    (sizeof(uint32_t))
+#define FLOW_AVG_SHIFT 3
 #define SECMEMSZ  16384
 #define MSGBUFSZ  2048
 
@@ -123,7 +126,13 @@ struct flow {
 
         struct frcti *        frcti;
 
+        /* Mean written packet size (bytes), EWMA over the send path. */
+        size_t                mean_len;
+
         struct poa_flow *     poa;   /* NULL for shared memory flows */
+
+        /* Egress capacity estimator; armed by the IPCP, else NULL. */
+        struct cap_est *      cap;
 };
 
 struct flow_set {
@@ -730,6 +739,8 @@ static void do_flow_fini(int fd)
 
         crypt_destroy_ctx(proc.flows[fd].crypt);
 
+        free(proc.flows[fd].cap);
+
         flow_clear(fd);
 }
 
@@ -768,6 +779,7 @@ static int flow_init(struct flow_info * info,
                      struct poa_flow *  pf)
 {
         struct timespec now;
+        struct timespec txq;
         struct flow *   flow;
         int             fd;
         int             err = -ENOMEM;
@@ -795,6 +807,11 @@ static int flow_init(struct flow_info * info,
                 if (flow->tx_rb == NULL)
                         goto fail_tx_rb;
 
+                txq.tv_sec  = SSM_RBUFF_TXQ_DELAY / 1000;
+                txq.tv_nsec = (SSM_RBUFF_TXQ_DELAY % 1000) * MILLION;
+
+                ssm_rbuff_set_txq_target(flow->tx_rb, &txq);
+
                 flow->set = ssm_flow_set_open(info->n_1_pid);
                 if (flow->set == NULL)
                         goto fail_set;
@@ -1475,6 +1492,25 @@ int fccntl(int fd,
                         goto einval;
                 *maxp = flow_user_mtu(flow, flow->info.mtu);
                 break;
+        case FLOWSTXQDLY:
+                timeo = va_arg(l, struct timespec *);
+                if (timeo == NULL)
+                        goto einval;
+
+                if (flow->tx_rb == NULL)
+                        goto eperm;
+
+                ssm_rbuff_set_txq_target(flow->tx_rb, timeo);
+                break;
+        case FLOWGTXQDLY:
+                timeo = va_arg(l, struct timespec *);
+                if (timeo == NULL)
+                        goto einval;
+
+                if (flow->tx_rb == NULL)
+                        goto eperm;
+                ssm_rbuff_get_txq_target(flow->tx_rb, timeo);
+                break;
         case FLOWSFLAGS:
                 old_acc = flow->oflags & FLOWFACCMODE;
                 flow->oflags = va_arg(l, uint32_t);
@@ -1609,6 +1645,25 @@ int fccntl(int fd,
         return -EPERM;
 }
 
+/*
+ * The ring counts slots, so the queue is only bytes if we know what a
+ * packet weighs. Ordered so the unsigned arithmetic cannot wrap.
+ */
+static void flow_mean_len_update(struct flow * flow,
+                                 size_t        len)
+{
+        size_t avg = LOAD_RELAXED(&flow->mean_len);
+
+        if (avg == 0) {
+                STORE_RELAXED(&flow->mean_len, len);
+                return;
+        }
+
+        avg = avg + (len >> FLOW_AVG_SHIFT) - (avg >> FLOW_AVG_SHIFT);
+
+        STORE_RELAXED(&flow->mean_len, avg == 0 ? 1 : avg);
+}
+
 static int flow_tx_spb(struct flow *        flow,
                        struct ssm_pk_buff * spb,
                        uint16_t             flags,
@@ -1643,6 +1698,8 @@ static int flow_tx_spb(struct flow *        flow,
         if (flow->poa != NULL)
                 return poa_flow_tx(flow->poa, spb, block, abstime);
 
+        flow_mean_len_update(flow, ssm_pk_buff_len(spb));
+
         if (!block)
                 ret = ssm_rbuff_write(flow->tx_rb, idx);
         else
@@ -2941,25 +2998,92 @@ size_t ipcp_flow_queued(int fd)
         if (proc.flows[fd].poa != NULL)
                 return poa_flow_qlen(proc.flows[fd].poa);
 
-        return ssm_rbuff_queued(proc.flows[fd].tx_rb);
+        return ssm_rbuff_queued(proc.flows[fd].tx_rb)
+                * LOAD_RELAXED(&proc.flows[fd].mean_len);
 }
 
-int ipcp_flow_queue_id(int fd)
+size_t ipcp_flow_mean_len(int fd)
 {
-        int qid;
+        assert(fd >= 0 && fd < PROC_MAX_FLOWS);
+        assert(proc.flows[fd].info.id >= 0);
+
+        if (proc.flows[fd].poa != NULL)
+                return poa_flow_mean_len(proc.flows[fd].poa);
+
+        return LOAD_RELAXED(&proc.flows[fd].mean_len);
+}
+
+/* An update racing the arm seeds one bogus window; the filter absorbs. */
+int ipcp_flow_cap_arm(int fd)
+{
+        struct flow *    flow;
+        struct cap_est * e;
+
+        assert(fd >= 0 && fd < PROC_MAX_FLOWS);
+        assert(proc.flows[fd].info.id >= 0);
+
+        flow = &proc.flows[fd];
+        if (flow->poa != NULL) {
+                cap_clear(poa_flow_cap_est(flow->poa));
+                return 0;
+        }
+
+        e = flow->cap;
+        if (e != NULL) {
+                cap_clear(e);
+                return 0;
+        }
+
+        if (posix_memalign((void **) &e, CAP_ALIGN, sizeof(*e)) != 0)
+                return -ENOMEM;
+
+        cap_clear(e);
+
+        STORE_RELEASE(&flow->cap, e);
+
+        return 0;
+}
+
+void ipcp_flow_cap_update(int    fd,
+                          size_t qlen,
+                          size_t len)
+{
+        struct flow *    flow;
+        struct cap_est * e;
 
         assert(fd >= 0 && fd < PROC_MAX_FLOWS);
         assert(proc.flows[fd].info.id >= 0);
 
-        if (proc.flows[fd].poa == NULL)
-                return fd;
+        flow = &proc.flows[fd];
+        if (flow->poa != NULL) {
+                cap_update(poa_flow_cap_est(flow->poa), qlen, len);
+                return;
+        }
+
+        e = LOAD_ACQUIRE(&flow->cap);
+        if (e == NULL)
+                return;
 
-        /* An unidentified PoA answers for itself, never for an fd. */
-        qid = poa_flow_qid(proc.flows[fd].poa);
-        if (qid < 0 || qid >= POA_MAX_POAS)
-                return fd;
+        cap_update(e, qlen, len);
+}
+
+uint64_t ipcp_flow_cap(int fd)
+{
+        struct flow *    flow;
+        struct cap_est * e;
+
+        assert(fd >= 0 && fd < PROC_MAX_FLOWS);
+        assert(proc.flows[fd].info.id >= 0);
+
+        flow = &proc.flows[fd];
+        if (flow->poa != NULL)
+                return cap_rate(poa_flow_cap_est(flow->poa));
+
+        e = LOAD_ACQUIRE(&flow->cap);
+        if (e == NULL)
+                return 0;
 
-        return PROC_MAX_FLOWS + qid;
+        return cap_rate(e);
 }
 
 int local_flow_transfer(int               src_fd,
@@ -3028,4 +3152,5 @@ int local_flow_transfer(int               src_fd,
         return ret;
 }
 
+#include "cap.c"
 #include "poa/poa.c"