X-Git-Url: http://git.meshlink.io/?p=utcp;a=blobdiff_plain;f=utcp.c;h=c650ab700ad150fa876b0944a6872ea39298f352;hp=4f3510e8c0da3f8ee130e5089d23aa244f22cb0c;hb=5ded633782709710c42648c71154d6062e83b7a6;hpb=e7e607d6d1b4e53f6f2a6094c02240e4fd8f59d4 diff --git a/utcp.c b/utcp.c index 4f3510e..c650ab7 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"); @@ -120,6 +127,10 @@ static bool fin_wanted(struct utcp_connection *c, uint32_t seq) { } } +static bool is_reliable(struct utcp_connection *c) { + return c->flags & UTCP_RELIABLE; +} + static inline void list_connections(struct utcp *utcp) { debug("%p has %d connections:\n", utcp, utcp->nconnections); for(int i = 0; i < utcp->nconnections; i++) @@ -135,16 +146,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 +163,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 +267,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 +313,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; } @@ -358,28 +367,53 @@ static void update_rtt(struct utcp_connection *c, uint32_t rtt) { debug("rtt %u srtt %u rttvar %u rto %u\n", rtt, utcp->srtt, utcp->rttvar, utcp->rto); } -struct utcp_connection *utcp_connect(struct utcp *utcp, uint16_t dst, utcp_recv_t recv, void *priv) { +static void start_retransmit_timer(struct utcp_connection *c) { + gettimeofday(&c->rtrx_timeout, NULL); + c->rtrx_timeout.tv_usec += c->utcp->rto; + while(c->rtrx_timeout.tv_usec >= 1000000) { + c->rtrx_timeout.tv_usec -= 1000000; + c->rtrx_timeout.tv_sec++; + } + debug("timeout set to %lu.%06lu (%u)\n", c->rtrx_timeout.tv_sec, c->rtrx_timeout.tv_usec, c->utcp->rto); +} + +static void stop_retransmit_timer(struct utcp_connection *c) { + timerclear(&c->rtrx_timeout); + debug("timeout cleared\n"); +} + +struct utcp_connection *utcp_connect_ex(struct utcp *utcp, uint16_t dst, utcp_recv_t recv, void *priv, uint32_t flags) { struct utcp_connection *c = allocate_connection(utcp, 0, dst); if(!c) return NULL; + assert((flags & ~0xf) == 0); + + c->flags = flags; c->recv = recv; c->priv = priv; - struct hdr hdr; - - hdr.src = c->src; - hdr.dst = c->dst; - hdr.seq = c->snd.iss; - hdr.ack = 0; - hdr.wnd = c->rcv.wnd; - hdr.ctl = SYN; - hdr.aux = 0; + struct { + struct hdr hdr; + uint8_t init[4]; + } pkt; + + pkt.hdr.src = c->src; + pkt.hdr.dst = c->dst; + pkt.hdr.seq = c->snd.iss; + pkt.hdr.ack = 0; + pkt.hdr.wnd = c->rcv.wnd; + pkt.hdr.ctl = SYN; + pkt.hdr.aux = 0x0101; + pkt.init[0] = 1; + pkt.init[1] = 0; + pkt.init[2] = 0; + pkt.init[3] = flags & 0x7; set_state(c, SYN_SENT); - print_packet(utcp, "send", &hdr, sizeof hdr); - utcp->send(utcp, &hdr, sizeof hdr); + print_packet(utcp, "send", &pkt, sizeof pkt); + utcp->send(utcp, &pkt, sizeof pkt); gettimeofday(&c->conn_timeout, NULL); c->conn_timeout.tv_sec += utcp->timeout; @@ -387,6 +421,10 @@ struct utcp_connection *utcp_connect(struct utcp *utcp, uint16_t dst, utcp_recv_ return c; } +struct utcp_connection *utcp_connect(struct utcp *utcp, uint16_t dst, utcp_recv_t recv, void *priv) { + return utcp_connect_ex(utcp, dst, recv, priv, UTCP_TCP); +} + void utcp_accept(struct utcp_connection *c, utcp_recv_t recv, void *priv) { if(c->reapable || c->state != SYN_RECEIVED) { debug("Error: accept() called on invalid connection %p in state %s\n", c, strstate[c->state]); @@ -487,7 +525,7 @@ ssize_t utcp_send(struct utcp_connection *c, const void *data, size_t len) { return -1; } - // Add data to send buffer + // Exit early if we have nothing to send. if(!len) return 0; @@ -497,6 +535,8 @@ ssize_t utcp_send(struct utcp_connection *c, const void *data, size_t len) { return -1; } + // Add data to send buffer. + len = buffer_put(&c->sndbuf, data, len); if(len <= 0) { errno = EWOULDBLOCK; @@ -505,6 +545,12 @@ ssize_t utcp_send(struct utcp_connection *c, const void *data, size_t len) { c->snd.last += len; ack(c, false); + if(!is_reliable(c)) { + c->snd.una = c->snd.nxt = c->snd.last; + buffer_get(&c->sndbuf, NULL, c->sndbuf.used); + } + if(is_reliable(c) && !timerisset(&c->rtrx_timeout)) + start_retransmit_timer(c); return len; } @@ -515,8 +561,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; @@ -531,13 +580,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); @@ -583,22 +633,45 @@ static void retransmit(struct utcp_connection *c) { #ifdef UTCP_DEBUG abort(); #endif - timerclear(&c->rtrx_timeout); - break; + stop_retransmit_timer(c); + goto cleanup; } + start_retransmit_timer(c); utcp->rto *= 2; if(utcp->rto > MAX_RTO) utcp->rto = MAX_RTO; c->rtt_start.tv_sec = 0; // invalidate RTT timer + +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); @@ -607,13 +680,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; @@ -646,6 +719,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 @@ -670,7 +745,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; @@ -692,6 +767,12 @@ static void handle_in_order(struct utcp_connection *c, const void *data, size_t static void handle_incoming_data(struct utcp_connection *c, uint32_t seq, const void *data, size_t len) { + if(!is_reliable(c)) { + c->recv(c, data, len); + c->rcv.nxt = seq + len; + return; + } + uint32_t offset = seqdiff(seq, c->rcv.nxt); if(offset + len > c->rcvbuf.maxsize) abort(); @@ -768,6 +849,19 @@ ssize_t utcp_recv(struct utcp *utcp, const void *data, size_t len) { goto reset; } + // Parse auxilliary information + if(hdr.aux) { + if(hdr.aux != 0x0101 || len < 4 || ((uint8_t *)data)[0] != 1) { + len = 1; + goto reset; + } + c->flags = ((uint8_t *)data)[3] & 0x7; + data += 4; + len -= 4; + } else { + c->flags = UTCP_TCP; + } + // Return SYN+ACK, go to SYN_RECEIVED state c->snd.wnd = hdr.wnd; c->rcv.irs = hdr.seq; @@ -795,8 +889,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. @@ -830,27 +926,31 @@ 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 - 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 + else { + int32_t rcv_offset = seqdiff(hdr.seq, c->rcv.nxt); + + // cut already accepted front overlapping + if(rcv_offset < 0) { + acceptable = len > -rcv_offset; + if(acceptable) { + data -= rcv_offset; + len += rcv_offset; + hdr.seq -= rcv_offset; + } + } else { + acceptable = seqdiff(hdr.seq, c->rcv.nxt) >= 0 && seqdiff(hdr.seq, c->rcv.nxt) + len <= c->rcvbuf.maxsize; + } + } 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; - // Otherwise, send an ACK back in the hope things improve. - ack(c, true); - return 0; + // Otherwise, continue processing. + len = 0; } c->snd.wnd = hdr.wnd; // TODO: move below @@ -859,10 +959,7 @@ ssize_t utcp_recv(struct utcp *utcp, const void *data, size_t len) { // ackno should not roll back, and it should also not be bigger than what we ever could have sent // (= snd.una + c->sndbuf.used). - if(hdr.ctl & ACK && - ((seqdiff(hdr.ack, c->snd.una + c->sndbuf.used) > 0 && - seqdiff(hdr.ack, c->snd.nxt) > 0) // TODO: simplify this if - || seqdiff(hdr.ack, c->snd.una) < 0)) { + if(hdr.ctl & ACK && (seqdiff(hdr.ack, c->snd.last) > 0 || seqdiff(hdr.ack, c->snd.una) < 0)) { debug("Packet ack seqno out of range, %u <= %u < %u\n", c->snd.una, hdr.ack, c->snd.una + c->sndbuf.used); // Ignore unacceptable RST packets. if(hdr.ctl & RST) @@ -991,7 +1088,7 @@ ssize_t utcp_recv(struct utcp *utcp, const void *data, size_t len) { break; } } else { - if(!len) { + if(!len && is_reliable(c)) { c->dupack++; if(c->dupack == 3) { debug("Triplicate ACK\n"); @@ -1000,6 +1097,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); } } } @@ -1008,8 +1106,10 @@ ssize_t utcp_recv(struct utcp *utcp, const void *data, size_t len) { if(advanced) { 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); + if(c->snd.una == c->snd.last) + stop_retransmit_timer(c); + else if(is_reliable(c)) + start_retransmit_timer(c); } // 5. Process SYN stuff @@ -1145,8 +1245,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: @@ -1223,11 +1322,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; @@ -1303,6 +1404,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); @@ -1321,6 +1423,7 @@ struct timeval utcp_timeout(struct utcp *utcp) { } if(timerisset(&c->rtrx_timeout) && timercmp(&c->rtrx_timeout, &now, <)) { + debug("retransmit()\n"); retransmit(c); } @@ -1330,13 +1433,6 @@ struct timeval utcp_timeout(struct utcp *utcp) { if(timerisset(&c->conn_timeout) && timercmp(&c->conn_timeout, &next, <)) next = c->conn_timeout; - if(c->snd.nxt != c->snd.una) { - c->rtrx_timeout = now; - c->rtrx_timeout.tv_sec++; - } else { - timerclear(&c->rtrx_timeout); - } - if(timerisset(&c->rtrx_timeout) && timercmp(&c->rtrx_timeout, &next, <)) next = c->rtrx_timeout; } @@ -1372,8 +1468,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; } @@ -1384,6 +1480,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]); }