X-Git-Url: http://git.meshlink.io/?a=blobdiff_plain;f=utcp.c;h=6b882c6f4cd4a0a888aff9dfafe18d080681dc5c;hb=ff52b8b4f37eeb4e0bfafb00d403f28395c2c957;hp=c21d6ab664b0146030c44d60e6feb730e924f0d8;hpb=7034cc75e0324210e96b3da5ac09c979b73173b7;p=utcp diff --git a/utcp.c b/utcp.c index c21d6ab..6b882c6 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 @@ -70,12 +70,12 @@ static void debug(const char *format, ...) { static void print_packet(struct utcp *utcp, const char *dir, const void *pkt, size_t len) { struct hdr hdr; if(len < sizeof hdr) { - debug("%p %s: short packet (%zu bytes)\n", utcp, dir, len); + debug("%p %s: short packet (%lu bytes)\n", utcp, dir, (unsigned long)len); return; } memcpy(&hdr, pkt, sizeof hdr); - 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); + debug("%p %s: len=%lu, src=%u dst=%u seq=%u ack=%u wnd=%u ctl=", utcp, dir, (unsigned long)len, hdr.src, hdr.dst, hdr.seq, hdr.ack, hdr.wnd); if(hdr.ctl & SYN) debug("SYN"); if(hdr.ctl & RST) @@ -86,11 +86,18 @@ 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; + const uint8_t *data = (uint8_t *)pkt + sizeof hdr; + char str[datalen * 2 + 1]; + char *p = str; + + for(uint32_t i = 0; i < datalen; i++) { + *p++ = "0123456789ABCDEF"[data[i] >> 4]; + *p++ = "0123456789ABCDEF"[data[i] & 15]; } + *p = 0; + + debug(" data=%s", str); } debug("\n"); @@ -135,16 +142,12 @@ static int32_t seqdiff(uint32_t a, uint32_t b) { // Store data into the buffer static ssize_t buffer_put_at(struct buffer *buf, size_t offset, const void *data, size_t len) { - if(buf->maxsize <= buf->used) - return 0; - - debug("buffer_put_at %zu %zu %zu\n", buf->used, offset, len); + debug("buffer_put_at %lu %lu %lu\n", (unsigned long)buf->used, (unsigned long)offset, (unsigned long)len); size_t required = offset + len; if(required > buf->maxsize) { if(offset >= buf->maxsize) return 0; - abort(); len = buf->maxsize - offset; required = buf->maxsize; } @@ -156,7 +159,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 +263,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 +309,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 +537,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 +556,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,11 +623,31 @@ 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) - abort(); + debug("sack_consume %lu\n", (unsigned long)len); + if(len > c->rcvbuf.used) { + debug("All SACK entries consumed"); + c->sacks[0].len = 0; + return; + } buffer_get(&c->rcvbuf, NULL, len); @@ -627,13 +656,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 +695,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 @@ -690,7 +721,7 @@ static void handle_out_of_order(struct utcp_connection *c, uint32_t offset, cons static void handle_in_order(struct utcp_connection *c, const void *data, size_t len) { // Check if we can process out-of-order data now. if(c->sacks[0].len && len >= c->sacks[0].offset) { // TODO: handle overlap with second SACK - debug("incoming packet len %zu connected with SACK at %u\n", len, c->sacks[0].offset); + debug("incoming packet len %lu connected with SACK at %u\n", (unsigned long)len, c->sacks[0].offset); buffer_put_at(&c->rcvbuf, 0, data, len); // TODO: handle return value len = max(len, c->sacks[0].offset + c->sacks[0].len); data = c->rcvbuf.data; @@ -815,8 +846,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,21 +883,25 @@ 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); + debug("Packet not acceptable, %u <= %u + %lu < %u\n", c->rcv.nxt, hdr.seq, (unsigned long)len, c->rcv.nxt + c->rcvbuf.maxsize); // Ignore unacceptable RST packets. if(hdr.ctl & RST) return 0; @@ -1017,6 +1054,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,7 +1202,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(c, prevrcvnxt != c->rcv.nxt); + ack(c, len || prevrcvnxt != c->rcv.nxt); return 0; reset: @@ -1241,11 +1279,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; @@ -1385,8 +1425,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; } @@ -1397,6 +1437,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]); }