X-Git-Url: http://git.meshlink.io/?a=blobdiff_plain;f=utcp.c;h=75553d72567be387667b7ced7871cac32d0d6a46;hb=5d25d40b5baf4bd60ccc1dc6db65b3e5b094e2cd;hp=1fae32cd2eaf44430f93311a0787034c58e2861d;hpb=d5888c10a181d6247e407f480524d36be6319b3e;p=utcp diff --git a/utcp.c b/utcp.c index 1fae32c..75553d7 100644 --- a/utcp.c +++ b/utcp.c @@ -19,6 +19,7 @@ #define _GNU_SOURCE +#include #include #include #include @@ -31,6 +32,27 @@ #include "utcp_priv.h" +#ifndef EBADMSG +#define EBADMSG 104 +#endif + +#ifndef SHUT_RDWR +#define SHUT_RDWR 2 +#endif + +#ifdef poll +#undef poll +#endif + +#ifndef timersub +#define timersub(a, b, r) do {\ + (r)->tv_sec = (a)->tv_sec - (b)->tv_sec;\ + (r)->tv_usec = (a)->tv_usec - (b)->tv_usec;\ + if((r)->tv_usec < 0)\ + (r)->tv_sec--, (r)->tv_usec += 1000000;\ +} while (0) +#endif + #ifdef UTCP_DEBUG #include @@ -49,7 +71,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 +119,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,8 +143,8 @@ 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); @@ -200,6 +223,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; @@ -239,9 +263,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; @@ -377,7 +399,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; @@ -596,20 +618,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) { @@ -632,7 +672,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(); } } } @@ -640,7 +681,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); } @@ -714,10 +755,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) @@ -769,14 +815,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: @@ -797,7 +843,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; @@ -997,6 +1043,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; @@ -1072,6 +1121,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) @@ -1097,3 +1150,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; +}