lib: Stabilize FRCP under packet loss conditions0.17.3

There were a bunch of bugs in FRCP that urgently needed fixing. Now
data QoS is usable even with heavy packet loss (within some
parameters). The current RTT estimator is the IETF one. It should be
updated to the improved one used in the Linux kernel once the A-timer
(ACKs without data) and graceful shutdown are implemented.

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

1 files changed, 21 insertions, 15 deletions

diff --git a/src/lib/rxmwheel.c b/src/lib/rxmwheel.c
index ce7ef8e4..3f01a0d3 100644
--- a/src/lib/rxmwheel.c
+++ b/src/lib/rxmwheel.c
@@ -22,11 +22,11 @@
 
 #include <ouroboros/list.h>
 
-#define RXMQ_S     16                 /* defines #slots     */
-#define RXMQ_M     24                 /* defines max delay  */
-#define RXMQ_R     (RXMQ_M - RXMQ_S)  /* defines resolution */
+#define RXMQ_S     16                 /* defines #slots           */
+#define RXMQ_M     24                 /* defines max delay  (us)  */
+#define RXMQ_R     (RXMQ_M - RXMQ_S)  /* defines resolution (us)  */
 #define RXMQ_SLOTS (1 << RXMQ_S)
-#define RXMQ_MAX   (1 << RXMQ_M)      /* ms                 */
+#define RXMQ_MAX   (1 << RXMQ_M)      /* us                       */
 
 /* Small inacurracy to avoid slow division by MILLION. */
 #define ts_to_ms(ts) (ts.tv_sec * 1000 + (ts.tv_nsec >> 20))
@@ -119,12 +119,11 @@ static void check_probe(struct frcti * frcti,
 
         if (frcti->probe && ((frcti->rttseq + 1) == seqno)) {
                 /* Backoff to avoid never updating rtt */
-                frcti->srtt_us <<= 1;
+                frcti->srtt_us += frcti->mdev_us;
                 frcti->probe = false;
         }
 }
 
-#define rto(frcti) (frcti->srtt_us + (frcti->mdev_us << 1))
 /* Return fd on r-timer expiry. */
 static int rxmwheel_move(void)
 {
@@ -136,12 +135,17 @@ static int rxmwheel_move(void)
 
         clock_gettime(PTHREAD_COND_CLOCK, &now);
 
+        pthread_mutex_lock(&rw.lock);
+
         slot = ts_to_slot(now);
 
-        pthread_mutex_lock(&rw.lock);
+        i = rw.prv;
 
-        for (i = rw.prv; (ssize_t) (i - slot) <= 0; ++i) {
-                list_for_each_safe(p, h, &rw.wheel[i]) {
+        if (slot < i)
+                slot += RXMQ_SLOTS;
+
+        while (i++ < slot) {
+                list_for_each_safe(p, h, &rw.wheel[i & (RXMQ_SLOTS - 1)]) {
                         struct rxm *         r;
                         struct frct_cr *     snd_cr;
                         struct frct_cr *     rcv_cr;
@@ -158,7 +162,7 @@ static int rxmwheel_move(void)
                         snd_cr = &r->frcti->snd_cr;
                         rcv_cr = &r->frcti->rcv_cr;
                         /* Has been ack'd, remove. */
-                        if ((int) (r->seqno - snd_cr->lwe) <= 0) {
+                        if ((int) (r->seqno - snd_cr->lwe) < 0) {
                                 shm_du_buff_ack(r->sdb);
                                 ipcp_sdb_release(r->sdb);
                                 free(r);
@@ -180,7 +184,7 @@ static int rxmwheel_move(void)
 
                         /* Copy the payload, safe rtx in other layers. */
                         if (ipcp_sdb_reserve(&sdb, r->tail - r->head)) {
-                                /* FIXME: reschedule send? */
+                                /* FIXME: reschedule send instead of failing? */
                                 int fd = r->frcti->fd;
                                 pthread_mutex_unlock(&rw.lock);
                                 shm_du_buff_ack(r->sdb);
@@ -206,10 +210,12 @@ static int rxmwheel_move(void)
 
                         /* Retransmit the copy. */
                         if (shm_rbuff_write(f->tx_rb, idx)) {
+                                int fd = r->frcti->fd;
+                                pthread_mutex_unlock(&rw.lock);
                                 ipcp_sdb_release(sdb);
                                 free(r);
                                 /* FIXME: reschedule send? */
-                                continue;
+                                return fd;
                         }
 
                         shm_flow_set_notify(f->set, f->flow_id, FLOW_PKT);
@@ -221,14 +227,14 @@ static int rxmwheel_move(void)
                         r->tail = shm_du_buff_tail(sdb);
                         r->sdb  = sdb;
 
-                        newtime = ts_to_us(now) + rto(f->frcti);
+                        newtime = ts_to_us(now) + f->frcti->rto;
                         rslot   = (newtime >> RXMQ_R) & (RXMQ_SLOTS - 1);
 
                         list_add_tail(&r->next, &rw.wheel[rslot]);
                 }
         }
 
-        rw.prv = slot;
+        rw.prv = slot & (RXMQ_SLOTS - 1);
 
         pthread_mutex_unlock(&rw.lock);
 
@@ -259,7 +265,7 @@ static int rxmwheel_add(struct frcti *       frcti,
         r->tail  = shm_du_buff_tail(sdb);
         r->frcti = frcti;
 
-        slot = ((r->t0 + rto(frcti)) >> RXMQ_R) & (RXMQ_SLOTS - 1);
+        slot = ((r->t0 + frcti->rto) >> RXMQ_R) & (RXMQ_SLOTS - 1);
 
         list_add_tail(&r->next, &rw.wheel[slot]);