X-Git-Url: http://git.meshlink.io/?a=blobdiff_plain;f=utcp.c;h=13f46584484bef5fd71ae74a77792671c1fb7637;hb=16ecd6d75bfabb97193581bcc8095652759cdb8e;hp=1b498a2897d0c4063b568040c24804c6235f636d;hpb=0c29e6bb0a943f9d06067628f894495dcc01c5ba;p=utcp diff --git a/utcp.c b/utcp.c index 1b498a2..13f4658 100644 --- a/utcp.c +++ b/utcp.c @@ -49,13 +49,13 @@ (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;\ + (r)->tv_sec--, (r)->tv_usec += USEC_PER_SEC;\ } while (0) #endif -#ifndef max -#define max(a, b) ((a) > (b) ? (a) : (b)) -#endif +static inline size_t max(size_t a, size_t b) { + return a > b ? a : b; +} #ifdef UTCP_DEBUG #include @@ -86,11 +86,28 @@ static void print_packet(struct utcp *utcp, const char *dir, const void *pkt, si debug("ACK"); if(len > sizeof hdr) { - debug(" data="); - for(int i = sizeof hdr; i < len; i++) { - const char *data = pkt; - debug("%c", data[i] >= 32 ? data[i] : '.'); + uint32_t datalen = len - sizeof hdr; + uint8_t *str = malloc((datalen << 1) + 7); + if(!str) { + debug("out of memory"); + return; + } + memcpy(str, " data=", 6); + uint8_t *strptr = str + 6; + const uint8_t *data = pkt; + const uint8_t *dataend = data + datalen; + + while(data != dataend) { + *strptr = (*data >> 4) > 9? (*data >> 4) + 55 : (*data >> 4) + 48; + ++strptr; + *strptr = (*data & 0xf) > 9? (*data & 0xf) + 55 : (*data & 0xf) + 48; + ++strptr; + ++data; } + *strptr = 0; + + debug(str); + free(str); } debug("\n"); @@ -156,7 +173,7 @@ static ssize_t buffer_put_at(struct buffer *buf, size_t offset, const void *data } else { do { newsize *= 2; - } while(newsize < buf->used + len); + } while(newsize < required); } if(newsize > buf->maxsize) newsize = buf->maxsize; @@ -260,6 +277,7 @@ static void free_connection(struct utcp_connection *c) { memmove(cp, cp + 1, (utcp->nconnections - i - 1) * sizeof *cp); utcp->nconnections--; + buffer_exit(&c->rcvbuf); buffer_exit(&c->sndbuf); free(c); } @@ -305,6 +323,7 @@ static struct utcp_connection *allocate_connection(struct utcp *utcp, uint16_t s } if(!buffer_init(&c->rcvbuf, DEFAULT_RCVBUFSIZE, DEFAULT_MAXRCVBUFSIZE)) { + buffer_exit(&c->sndbuf); free(c); return NULL; } @@ -532,8 +551,11 @@ static void swap_ports(struct hdr *hdr) { } static void retransmit(struct utcp_connection *c) { - if(c->state == CLOSED || c->snd.nxt == c->snd.una) + if(c->state == CLOSED || c->snd.last == c->snd.una) { + debug("Retransmit() called but nothing to retransmit!\n"); + stop_retransmit_timer(c); return; + } struct utcp *utcp = c->utcp; @@ -548,13 +570,14 @@ static void retransmit(struct utcp_connection *c) { pkt->hdr.src = c->src; pkt->hdr.dst = c->dst; + pkt->hdr.wnd = c->rcv.wnd; + pkt->hdr.aux = 0; switch(c->state) { case SYN_SENT: // Send our SYN again pkt->hdr.seq = c->snd.iss; pkt->hdr.ack = 0; - pkt->hdr.wnd = c->rcv.wnd; pkt->hdr.ctl = SYN; print_packet(c->utcp, "rtrx", pkt, sizeof pkt->hdr); utcp->send(utcp, pkt, sizeof pkt->hdr); @@ -614,7 +637,24 @@ cleanup: free(pkt); } -// Update receive buffer and SACK entries after consuming data. +/* Update receive buffer and SACK entries after consuming data. + * + * Situation: + * + * |.....0000..1111111111.....22222......3333| + * |---------------^ + * + * 0..3 represent the SACK entries. The ^ indicates up to which point we want + * to remove data from the receive buffer. The idea is to substract "len" + * from the offset of all the SACK entries, and then remove/cut down entries + * that are shifted to before the start of the receive buffer. + * + * There are three cases: + * - the SACK entry is after ^, in that case just change the offset. + * - the SACK entry starts before and ends after ^, so we have to + * change both its offset and size. + * - the SACK entry is completely before ^, in that case delete it. + */ static void sack_consume(struct utcp_connection *c, size_t len) { debug("sack_consume %zu\n", len); if(len > c->rcvbuf.used) @@ -627,13 +667,13 @@ static void sack_consume(struct utcp_connection *c, size_t len) { c->sacks[i].offset -= len; i++; } else if(len < c->sacks[i].offset + c->sacks[i].len) { - c->sacks[i].offset = 0; c->sacks[i].len -= len - c->sacks[i].offset; + c->sacks[i].offset = 0; i++; } else { if(i < NSACKS - 1) { memmove(&c->sacks[i], &c->sacks[i + 1], (NSACKS - 1 - i) * sizeof c->sacks[i]); - c->sacks[i + 1].len = 0; + c->sacks[NSACKS - 1].len = 0; } else { c->sacks[i].len = 0; break; @@ -666,6 +706,8 @@ static void handle_out_of_order(struct utcp_connection *c, uint32_t offset, cons memmove(&c->sacks[i + 1], &c->sacks[i], (NSACKS - i - 1) * sizeof c->sacks[i]); c->sacks[i].offset = offset; c->sacks[i].len = rxd; + } else { + debug("SACK entries full, dropping packet\n"); } break; } else { // merge @@ -815,8 +857,10 @@ ssize_t utcp_recv(struct utcp *utcp, const void *data, size_t len) { // In case this is for a CLOSED connection, ignore the packet. // TODO: make it so incoming packets can never match a CLOSED connection. - if(c->state == CLOSED) + if(c->state == CLOSED) { + debug("Got packet for closed connection\n"); return 0; + } // It is for an existing connection. @@ -850,18 +894,22 @@ ssize_t utcp_recv(struct utcp *utcp, const void *data, size_t len) { if(c->state == SYN_SENT) acceptable = true; - - // TODO: handle packets overlapping c->rcv.nxt. -#if 1 - // Only use this when accepting out-of-order packets. else if(len == 0) acceptable = seqdiff(hdr.seq, c->rcv.nxt) >= 0; - else + else { + int32_t rcv_offset = seqdiff(hdr.seq, c->rcv.nxt); + + // cut already accepted front overlapping + if(rcv_offset < 0) { + acceptable = rcv_offset + len >= 0; + if(acceptable) { + data -= rcv_offset; + len += rcv_offset; + } + } + acceptable = seqdiff(hdr.seq, c->rcv.nxt) >= 0 && seqdiff(hdr.seq, c->rcv.nxt) + len <= c->rcvbuf.maxsize; -#else - if(c->state != SYN_SENT) - acceptable = hdr.seq == c->rcv.nxt; -#endif + } if(!acceptable) { debug("Packet not acceptable, %u <= %u + %zu < %u\n", c->rcv.nxt, hdr.seq, len, c->rcv.nxt + c->rcvbuf.maxsize); @@ -1017,6 +1065,7 @@ ssize_t utcp_recv(struct utcp *utcp, const void *data, size_t len) { //Reset the congestion window so we wait for ACKs. c->snd.nxt = c->snd.una; c->snd.cwnd = utcp->mtu; + start_retransmit_timer(c); } } } @@ -1164,8 +1213,7 @@ ssize_t utcp_recv(struct utcp *utcp, const void *data, size_t len) { // - or we got an ack, so we should maybe send a bit more data // -> sendatleastone = false -ack: - ack(c, prevrcvnxt != c->rcv.nxt); + ack(c, len || prevrcvnxt != c->rcv.nxt); return 0; reset: @@ -1242,11 +1290,13 @@ int utcp_shutdown(struct utcp_connection *c, int dir) { c->snd.last++; ack(c, false); + if(!timerisset(&c->rtrx_timeout)) + start_retransmit_timer(c); return 0; } int utcp_close(struct utcp_connection *c) { - if(utcp_shutdown(c, SHUT_RDWR)) + if(utcp_shutdown(c, SHUT_RDWR) && errno != ENOTCONN) return -1; c->recv = NULL; c->poll = NULL; @@ -1322,6 +1372,7 @@ struct timeval utcp_timeout(struct utcp *utcp) { if(!c) continue; + // delete connections that have been utcp_close()d. if(c->state == CLOSED) { if(c->reapable) { debug("Reaping %p\n", c); @@ -1340,6 +1391,7 @@ struct timeval utcp_timeout(struct utcp *utcp) { } if(timerisset(&c->rtrx_timeout) && timercmp(&c->rtrx_timeout, &now, <)) { + debug("retransmit()\n"); retransmit(c); } @@ -1384,8 +1436,8 @@ struct utcp *utcp_init(utcp_accept_t accept, utcp_pre_accept_t pre_accept, utcp_ utcp->send = send; utcp->priv = priv; utcp->mtu = DEFAULT_MTU; - utcp->timeout = DEFAULT_USER_TIMEOUT; // s - utcp->rto = START_RTO; // us + utcp->timeout = DEFAULT_USER_TIMEOUT; // sec + utcp->rto = START_RTO; // usec return utcp; } @@ -1396,6 +1448,7 @@ void utcp_exit(struct utcp *utcp) { for(int i = 0; i < utcp->nconnections; i++) { if(!utcp->connections[i]->reapable) debug("Warning, freeing unclosed connection %p\n", utcp->connections[i]); + buffer_exit(&utcp->connections[i]->rcvbuf); buffer_exit(&utcp->connections[i]->sndbuf); free(utcp->connections[i]); }