1 files changed, 42 insertions, 33 deletions

diff --git a/src/lib/frct.c b/src/lib/frct.c
index 200a9fe7..04b8a758 100644
--- a/src/lib/frct.c
+++ b/src/lib/frct.c
@@ -39,8 +39,8 @@ struct frct_cr {
         uint8_t  cflags;
         uint32_t seqno;
 
-        time_t   act;
-        time_t   inact;
+        time_t   act;     /* s */
+        time_t   inact;   /* s */
 };
 
 struct frcti {
@@ -50,24 +50,19 @@ struct frcti {
         time_t           a;
         time_t           r;
 
+        time_t           rto; /* ms */
+
         struct frct_cr   snd_cr;
         struct frct_cr   rcv_cr;
 
         ssize_t          rq[RQ_SIZE];
-
-        struct timespec  rtt;
-
         pthread_rwlock_t lock;
 };
 
-struct {
-        struct timerwheel * tw;
-} frct;
-
 enum frct_flags {
         FRCT_DATA = 0x01, /* PDU carries data */
         FRCT_DRF  = 0x02, /* Data run flag    */
-        FRCT_ACK  = 0x03, /* ACK field valid  */
+        FRCT_ACK  = 0x04, /* ACK field valid  */
         FRCT_FC   = 0x08, /* FC window valid  */
         FRCT_RDVZ = 0x10, /* Rendez-vous      */
         FRCT_MFGM = 0x20, /* More fragments   */
@@ -83,21 +78,7 @@ struct frct_pci {
         uint32_t ackno;
 } __attribute__((packed));
 
-static int frct_init(void)
-{
-        frct.tw = timerwheel_create(TW_RESOLUTION, TW_RESOLUTION * TW_ELEMENTS);
-        if (frct.tw == NULL)
-                return -1;
-
-        return 0;
-}
-
-static void frct_fini(void)
-{
-        assert(frct.tw);
-
-        timerwheel_destroy(frct.tw);
-}
+#include <rxmwheel.c>
 
 static struct frcti * frcti_create(int fd)
 {
@@ -129,9 +110,13 @@ 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);
+        /* Initial rto. FIXME: recalc using Karn algorithm. */
+        frcti->rto           = 120;
 
-        if (ai.flows[fd].spec.loss == 0)
+        if (ai.flows[fd].spec.loss == 0) {
                 frcti->snd_cr.cflags |= FRCTFRTX;
+                frcti->rcv_cr.cflags |= FRCTFRTX;
+        }
 
         frcti->rcv_cr.inact  = 2 * delta_t;
         frcti->rcv_cr.act    = now.tv_sec - (frcti->rcv_cr.inact + 1);
@@ -151,6 +136,8 @@ static void frcti_destroy(struct frcti * frcti)
          * make sure everything we sent is acked.
          */
 
+        rxmwheel_clear(frcti->fd);
+
         pthread_rwlock_destroy(&frcti->lock);
 
         free(frcti);
@@ -204,7 +191,7 @@ static ssize_t __frcti_queued_pdu(struct frcti * frcti)
 
 static struct frct_pci * frcti_alloc_head(struct shm_du_buff * sdb)
 {
-        struct frct_pci * pci = NULL;
+        struct frct_pci * pci;
 
         pci = (struct frct_pci *) shm_du_buff_head_alloc(sdb, FRCT_PCILEN);
         if (pci != NULL)
@@ -219,10 +206,14 @@ static int __frcti_snd(struct frcti *       frcti,
         struct frct_pci * pci;
         struct timespec   now;
         struct frct_cr *  snd_cr;
+        struct frct_cr *  rcv_cr;
 
         assert(frcti);
 
         snd_cr = &frcti->snd_cr;
+        rcv_cr = &frcti->rcv_cr;
+
+        rxmwheel_move();
 
         pci = frcti_alloc_head(sdb);
         if (pci == NULL)
@@ -250,13 +241,19 @@ static int __frcti_snd(struct frcti *       frcti,
                 frcti->snd_cr.lwe = snd_cr->seqno;
         }
 
-        pci->seqno = hton32(snd_cr->seqno++);
-        if (!(snd_cr->cflags & FRCTFRTX))
-                snd_cr->lwe++;
-        else
-                /* TODO: update on ACK */
+        pci->seqno = hton32(snd_cr->seqno);
+        if (!(snd_cr->cflags & FRCTFRTX)) {
                 snd_cr->lwe++;
+        } else if (now.tv_sec - rcv_cr->act <= rcv_cr->inact) {
+                rxmwheel_add(frcti, snd_cr->seqno, sdb);
+                if (rcv_cr->lwe != rcv_cr->seqno) {
+                        pci->flags |= FRCT_ACK;
+                        pci->ackno = hton32(rcv_cr->seqno);
+                        rcv_cr->lwe = rcv_cr->seqno;
+                }
+        }
 
+        snd_cr->seqno++;
         snd_cr->act  = now.tv_sec;
 
         pthread_rwlock_unlock(&frcti->lock);
@@ -271,6 +268,7 @@ static int __frcti_rcv(struct frcti *       frcti,
         ssize_t           idx;
         struct frct_pci * pci;
         struct timespec   now;
+        struct frct_cr *  snd_cr;
         struct frct_cr *  rcv_cr;
         uint32_t          seqno;
         int               ret = 0;
@@ -278,6 +276,7 @@ static int __frcti_rcv(struct frcti *       frcti,
         assert(frcti);
 
         rcv_cr = &frcti->rcv_cr;
+        snd_cr = &frcti->snd_cr;
 
         pci = (struct frct_pci *) shm_du_buff_head_release(sdb, FRCT_PCILEN);
 
@@ -306,7 +305,7 @@ static int __frcti_rcv(struct frcti *       frcti,
 
                 if (rcv_cr->cflags & FRCTFRTX) {
                         size_t pos = seqno & (RQ_SIZE - 1);
-                        if ((seqno - rcv_cr->seqno) > RQ_SIZE /* Out of rq. */
+                        if ((seqno - rcv_cr->lwe) > RQ_SIZE /* Out of rq. */
                             || frcti->rq[pos] != -1) /* Duplicate in rq. */
                                 goto drop_packet;
                         /* Queue. */
@@ -317,6 +316,13 @@ static int __frcti_rcv(struct frcti *       frcti,
                 }
         }
 
+        if (rcv_cr->cflags & FRCTFRTX && pci->flags & FRCT_ACK) {
+                uint32_t ackno = ntoh32(pci->ackno);
+                /* Check for duplicate (old) acks. */
+                if ((int32_t)(ackno - snd_cr->lwe) >= 0)
+                        snd_cr->lwe = ackno;
+        }
+
         rcv_cr->act = now.tv_sec;
 
         pthread_rwlock_unlock(&frcti->lock);
@@ -324,10 +330,13 @@ static int __frcti_rcv(struct frcti *       frcti,
         if (!(pci->flags & FRCT_DATA))
                 shm_rdrbuff_remove(ai.rdrb, idx);
 
+        rxmwheel_move();
+
         return ret;
 
  drop_packet:
         pthread_rwlock_unlock(&frcti->lock);
         shm_rdrbuff_remove(ai.rdrb, idx);
+        rxmwheel_move();
         return -EAGAIN;
 }