]> git.meshlink.io Git - utcp/blobdiff - utcp.c
Add a poll callback to UTCP connections.
[utcp] / utcp.c
diff --git a/utcp.c b/utcp.c
index 730f12b98e8102a42d85c2d6f1fb55971728beec..ca3c132f60c9132da69b019e3583d83db11974df 100644 (file)
--- a/utcp.c
+++ b/utcp.c
@@ -19,6 +19,7 @@
 
 #define _GNU_SOURCE
 
+#include <assert.h>
 #include <errno.h>
 #include <stdio.h>
 #include <stdlib.h>
@@ -49,7 +50,7 @@ static void print_packet(struct utcp *utcp, const char *dir, const void *pkt, si
        }
 
        memcpy(&hdr, pkt, sizeof hdr);
-       fprintf (stderr, "%p %s: src=%u dst=%u seq=%u ack=%u wnd=%u ctl=", utcp, dir, hdr.src, hdr.dst, hdr.seq, hdr.ack, hdr.wnd);
+       fprintf (stderr, "%p %s: len=%zu, src=%u dst=%u seq=%u ack=%u wnd=%u ctl=", utcp, dir, len, hdr.src, hdr.dst, hdr.seq, hdr.ack, hdr.wnd);
        if(hdr.ctl & SYN)
                debug("SYN");
        if(hdr.ctl & RST)
@@ -97,8 +98,9 @@ static int32_t seqdiff(uint32_t a, uint32_t b) {
 static int compare(const void *va, const void *vb) {
        const struct utcp_connection *a = *(struct utcp_connection **)va;
        const struct utcp_connection *b = *(struct utcp_connection **)vb;
-       if(!a->src || !b->src)
-               abort();
+
+       assert(a->src && b->src);
+
        int c = (int)a->src - (int)b->src;
        if(c)
                return c;
@@ -120,13 +122,14 @@ static struct utcp_connection *find_connection(const struct utcp *utcp, uint16_t
 static void free_connection(struct utcp_connection *c) {
        struct utcp *utcp = c->utcp;
        struct utcp_connection **cp = bsearch(&c, utcp->connections, utcp->nconnections, sizeof *utcp->connections, compare);
-       if(!cp)
-               abort();
+
+       assert(cp);
 
        int i = cp - utcp->connections;
        memmove(cp + i, cp + i + 1, (utcp->nconnections - i - 1) * sizeof *cp);
        utcp->nconnections--;
 
+       free(c->sndbuf);
        free(c);
 }
 
@@ -199,6 +202,7 @@ struct utcp_connection *utcp_connect(struct utcp *utcp, uint16_t dst, utcp_recv_
                return NULL;
 
        c->recv = recv;
+       c->priv = priv;
 
        struct hdr hdr;
 
@@ -206,8 +210,9 @@ struct utcp_connection *utcp_connect(struct utcp *utcp, uint16_t dst, utcp_recv_
        hdr.dst = c->dst;
        hdr.seq = c->snd.iss;
        hdr.ack = 0;
-       hdr.ctl = SYN;
        hdr.wnd = c->rcv.wnd;
+       hdr.ctl = SYN;
+       hdr.aux = 0;
 
        set_state(c, SYN_SENT);
 
@@ -237,9 +242,7 @@ static void ack(struct utcp_connection *c, bool sendatleastone) {
        int32_t cwndleft = c->snd.cwnd - seqdiff(c->snd.nxt, c->snd.una);
        char *data = c->sndbuf + seqdiff(c->snd.nxt, c->snd.una);
 
-       fprintf(stderr, "ack, left=%d, cwndleft=%d, sendatleastone=%d\n", left, cwndleft, sendatleastone);
-       if(left < 0)
-               abort();
+       assert(left >= 0);
 
        if(cwndleft <= 0)
                cwndleft = 0;
@@ -260,6 +263,7 @@ static void ack(struct utcp_connection *c, bool sendatleastone) {
        pkt.hdr.ack = c->rcv.nxt;
        pkt.hdr.wnd = c->snd.wnd;
        pkt.hdr.ctl = ACK;
+       pkt.hdr.aux = 0;
 
        do {
                uint32_t seglen = left > c->utcp->mtu ? c->utcp->mtu : left;
@@ -374,7 +378,7 @@ static void swap_ports(struct hdr *hdr) {
        hdr->dst = tmp;
 }
 
-int utcp_recv(struct utcp *utcp, const void *data, size_t len) {
+ssize_t utcp_recv(struct utcp *utcp, const void *data, size_t len) {
        if(!utcp) {
                errno = EFAULT;
                return -1;
@@ -593,20 +597,38 @@ int utcp_recv(struct utcp *utcp, const void *data, size_t len) {
        // 3. Advance snd.una
 
        uint32_t advanced = seqdiff(hdr.ack, c->snd.una);
-       c->snd.una = hdr.ack;
+       uint32_t prevrcvnxt = c->rcv.nxt;
 
        if(advanced) {
-               debug("%p advanced %u\n", utcp, advanced);
+               int32_t data_acked = advanced;
+
+               switch(c->state) {
+                       case SYN_SENT:
+                       case SYN_RECEIVED:
+                               data_acked--;
+                               break;
+                       // TODO: handle FIN as well.
+                       default:
+                               break;
+               }
+
+               assert(data_acked >= 0);
+
+               int32_t bufused = seqdiff(c->snd.last, c->snd.una);
+               assert(data_acked <= bufused);
+
                // Make room in the send buffer.
                // TODO: try to avoid memmoving too much. Circular buffer?
-               uint32_t left = seqdiff(c->snd.nxt, hdr.ack);
-               if(left)
-                       memmove(c->sndbuf, c->sndbuf + advanced, left);
+               uint32_t left = bufused - data_acked;
+               if(data_acked && left)
+                       memmove(c->sndbuf, c->sndbuf + data_acked, left);
+
+               c->snd.una = hdr.ack;
+
                c->dupack = 0;
                c->snd.cwnd += utcp->mtu;
                if(c->snd.cwnd > c->maxsndbufsize)
                        c->snd.cwnd = c->maxsndbufsize;
-               debug("%p increasing cwnd to %u\n", utcp, c->snd.cwnd);
 
                // Check if we have sent a FIN that is now ACKed.
                switch(c->state) {
@@ -629,7 +651,8 @@ int utcp_recv(struct utcp *utcp, const void *data, size_t len) {
                        c->dupack++;
                        if(c->dupack >= 3) {
                                debug("Triplicate ACK\n");
-                               abort();
+                               //TODO: Resend one packet and go to fast recovery mode. See RFC 6582.
+                               //abort();
                        }
                }
        }
@@ -637,7 +660,7 @@ int utcp_recv(struct utcp *utcp, const void *data, size_t len) {
        // 4. Update timers
 
        if(advanced) {
-               timerclear(&c->conn_timeout); // It should be set anew in utcp_timeout() if c->snd.una != c->snd.nxt.
+               timerclear(&c->conn_timeout); // It will be set anew in utcp_timeout() if c->snd.una != c->snd.nxt.
                if(c->snd.una == c->snd.nxt)
                        timerclear(&c->rtrx_timeout);
        }
@@ -711,10 +734,15 @@ int utcp_recv(struct utcp *utcp, const void *data, size_t len) {
                        abort();
                }
 
-               int rxd;
+               ssize_t rxd;
 
                if(c->recv) {
                        rxd = c->recv(c, data, len);
+                       if(rxd != len) {
+                               // TODO: once we have a receive buffer, handle the application not accepting all data.
+                               fprintf(stderr, "c->recv(%p, %p, %zu) returned %zd\n", c, data, len, rxd);
+                               abort();
+                       }
                        if(rxd < 0)
                                rxd = 0;
                        else if(rxd > len)
@@ -757,6 +785,7 @@ int utcp_recv(struct utcp *utcp, const void *data, size_t len) {
 
                // FIN counts as one sequence number
                c->rcv.nxt++;
+               len++;
 
                // Inform the application that the peer closed the connection.
                if(c->recv) {
@@ -765,14 +794,14 @@ int utcp_recv(struct utcp *utcp, const void *data, size_t len) {
                }
        }
 
-       if(!len && !advanced)
-               return 0;
-
-       if(!len && !(hdr.ctl & SYN) && !(hdr.ctl & FIN))
-               return 0;
+       // Now we send something back if:
+       // - we advanced rcv.nxt (ie, we got some data that needs to be ACKed)
+       //   -> sendatleastone = true
+       // - or we got an ack, so we should maybe send a bit more data
+       //   -> sendatleastone = false
 
 ack:
-       ack(c, true);
+       ack(c, prevrcvnxt != c->rcv.nxt);
        return 0;
 
 reset:
@@ -793,7 +822,7 @@ reset:
 }
 
 int utcp_shutdown(struct utcp_connection *c, int dir) {
-       debug("%p shutdown %d\n", c->utcp, dir);
+       debug("%p shutdown %d\n", c ? c->utcp : NULL, dir);
        if(!c) {
                errno = EFAULT;
                return -1;
@@ -993,6 +1022,9 @@ int utcp_timeout(struct utcp *utcp) {
                        retransmit(c);
                }
 
+               if(c->poll && c->sndbufsize < c->maxsndbufsize / 2)
+                       c->poll(c, c->maxsndbufsize - c->sndbufsize);
+
                if(timerisset(&c->conn_timeout) && timercmp(&c->conn_timeout, &next, <))
                        next = c->conn_timeout;
 
@@ -1037,8 +1069,13 @@ struct utcp *utcp_init(utcp_accept_t accept, utcp_pre_accept_t pre_accept, utcp_
 void utcp_exit(struct utcp *utcp) {
        if(!utcp)
                return;
-       for(int i = 0; i < utcp->nconnections; i++)
-               free_connection(utcp->connections[i]);
+       for(int i = 0; i < utcp->nconnections; i++) {
+               if(!utcp->connections[i]->reapable)
+                       debug("Warning, freeing unclosed connection %p\n", utcp->connections[i]);
+               free(utcp->connections[i]->sndbuf);
+               free(utcp->connections[i]);
+       }
+       free(utcp->connections);
        free(utcp);
 }
 
@@ -1063,6 +1100,10 @@ size_t utcp_get_sndbuf(struct utcp_connection *c) {
        return c->maxsndbufsize;
 }
 
+size_t utcp_get_sndbuf_free(struct utcp_connection *c) {
+       return c->maxsndbufsize - c->sndbufsize;
+}
+
 void utcp_set_sndbuf(struct utcp_connection *c, size_t size) {
        c->maxsndbufsize = size;
        if(c->maxsndbufsize != size)
@@ -1088,3 +1129,11 @@ void utcp_set_keepalive(struct utcp_connection *c, bool keepalive) {
 size_t utcp_get_outq(struct utcp_connection *c) {
        return seqdiff(c->snd.nxt, c->snd.una);
 }
+
+void utcp_set_recv_cb(struct utcp_connection *c, utcp_recv_t recv) {
+       c->recv = recv;
+}
+
+void utcp_set_poll_cb(struct utcp_connection *c, utcp_poll_t poll) {
+       c->poll = poll;
+}