X-Git-Url: http://git.meshlink.io/?a=blobdiff_plain;f=utcp.c;h=f5484ab82b92e53e5365f7de75a822881218a7f4;hb=aef3cad25ea81f727608689ef97fce9e0dbe5285;hp=e6f70f7b5c18f4ad83dbe427418186837ab24e8d;hpb=c93ef1391476090d5afacb899632df5476f964ed;p=utcp diff --git a/utcp.c b/utcp.c index e6f70f7..f5484ab 100644 --- a/utcp.c +++ b/utcp.c @@ -19,6 +19,7 @@ #define _GNU_SOURCE +#include #include #include #include @@ -29,154 +30,169 @@ #include #include -#define UTCP_INTERNAL -#include "utcp.h" - -#define PREP(l) char pkt[(l) + sizeof struct hdr]; struct hdr *hdr = &pkt; - -#define SYN 1 -#define ACK 2 -#define FIN 4 -#define RST 8 - -struct hdr { - uint16_t src; // Source port - uint16_t dst; // Destination port - uint32_t seq; // Sequence number - uint32_t ack; // Acknowledgement number - uint32_t wnd; // Window size - uint16_t ctl; // Flags (SYN, ACK, FIN, RST) - uint16_t aux; // other stuff -}; - -enum state { - CLOSED, - LISTEN, - SYN_SENT, - SYN_RECEIVED, - ESTABLISHED, - FIN_WAIT_1, - FIN_WAIT_2, - CLOSE_WAIT, - CLOSING, - LAST_ACK, - TIME_WAIT -}; - -const char *strstate[] = { - "CLOSED", - "LISTEN", - "SYN_SENT", - "SYN_RECEIVED", - "ESTABLISHED", - "FIN_WAIT_1", - "FIN_WAIT_2", - "CLOSE_WAIT", - "CLOSING", - "LAST_ACK", - "TIME_WAIT" -}; - -struct utcp_connection { - void *priv; - struct utcp *utcp; - bool reapable; - - uint16_t src; - uint16_t dst; - enum state state; - - // The following two structures form the TCB - - struct { - uint32_t una; - uint32_t nxt; - uint32_t wnd; - uint32_t up; - uint32_t wl1; - uint32_t wl2; - uint32_t iss; - } snd; - - struct { - uint32_t nxt; - uint32_t wnd; - uint32_t up; - uint32_t irs; - } rcv; - - utcp_recv_t recv; - - struct timeval conn_timeout; - struct timeval rtrx_timeout; - - char *sndbuf; - uint32_t sndbufsize; -}; - -struct utcp { - void *priv; - - utcp_accept_t accept; - utcp_pre_accept_t pre_accept; - utcp_send_t send; - - uint16_t mtu; - - struct utcp_connection **connections; - int nconnections; - int nallocated; - int gap; -}; - -static void set_state(struct utcp_connection *c, enum state state) { - c->state = state; - fprintf(stderr, "%p new state: %s\n", c->utcp, strstate[state]); +#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 + +static void debug(const char *format, ...) { + va_list ap; + va_start(ap, format); + vfprintf(stderr, format, ap); + va_end(ap); } -static void print_packet(void *pkt, size_t len) { +static void print_packet(struct utcp *utcp, const char *dir, const void *pkt, size_t len) { struct hdr hdr; if(len < sizeof hdr) { - fprintf(stderr, "short packet (%zu bytes)\n", len); + debug("%p %s: short packet (%zu bytes)\n", utcp, dir, len); return; } memcpy(&hdr, pkt, sizeof hdr); - fprintf (stderr, "src=%u dst=%u seq=%u ack=%u wnd=%u ctl=", 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) - fprintf(stderr, "SYN"); + debug("SYN"); if(hdr.ctl & RST) - fprintf(stderr, "RST"); + debug("RST"); if(hdr.ctl & FIN) - fprintf(stderr, "FIN"); + debug("FIN"); if(hdr.ctl & ACK) - fprintf(stderr, "ACK"); + debug("ACK"); if(len > sizeof hdr) { - fprintf(stderr, " data="); + debug(" data="); for(int i = sizeof hdr; i < len; i++) { - char *data = pkt; - fprintf(stderr, "%c", data[i] >= 32 ? data[i] : '.'); + const char *data = pkt; + debug("%c", data[i] >= 32 ? data[i] : '.'); } } - fprintf(stderr, "\n"); + debug("\n"); +} +#else +#define debug(...) +#define print_packet(...) +#endif + +static void set_state(struct utcp_connection *c, enum state state) { + c->state = state; + if(state == ESTABLISHED) + timerclear(&c->conn_timeout); + debug("%p new state: %s\n", c->utcp, strstate[state]); } -static void list_connections(struct utcp *utcp) { - fprintf(stderr, "%p has %d connections:\n", utcp, utcp->nconnections); +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++) - fprintf(stderr, " %u -> %u state %s\n", utcp->connections[i]->src, utcp->connections[i]->dst, strstate[utcp->connections[i]->state]); + debug(" %u -> %u state %s\n", utcp->connections[i]->src, utcp->connections[i]->dst, strstate[utcp->connections[i]->state]); +} + +static int32_t seqdiff(uint32_t a, uint32_t b) { + return a - b; +} + +// Buffer functions +// TODO: convert to ringbuffers to avoid memmove() operations. + +// Store data into the buffer +static ssize_t buffer_put(struct buffer *buf, const void *data, size_t len) { + if(buf->maxsize <= buf->used) + return 0; + if(len > buf->maxsize - buf->used) + len = buf->maxsize - buf->used; + if(len > buf->size - buf->used) { + size_t newsize = buf->size; + do { + newsize *= 2; + } while(newsize < buf->used + len); + if(newsize > buf->maxsize) + newsize = buf->maxsize; + char *newdata = realloc(buf->data, newsize); + if(!newdata) + return -1; + buf->data = newdata; + buf->size = newsize; + } + memcpy(buf->data + buf->used, data, len); + buf->used += len; + return len; +} + +// Get data from the buffer. data can be NULL. +static ssize_t buffer_get(struct buffer *buf, void *data, size_t len) { + if(len > buf->used) + len = buf->used; + if(data) + memcpy(data, buf->data, len); + if(len < buf->used) + memmove(buf->data, buf->data + len, buf->used - len); + buf->used -= len; + return len; +} + +// Copy data from the buffer without removing it. +static ssize_t buffer_copy(struct buffer *buf, void *data, size_t offset, size_t len) { + if(offset >= buf->used) + return 0; + if(offset + len > buf->used) + len = buf->used - offset; + memcpy(data, buf->data + offset, len); + return len; +} + +static bool buffer_init(struct buffer *buf, uint32_t len, uint32_t maxlen) { + memset(buf, 0, sizeof *buf); + buf->data = malloc(len); + if(!len) + return false; + buf->size = len; + buf->maxsize = maxlen; + return true; +} + +static void buffer_exit(struct buffer *buf) { + free(buf->data); + memset(buf, 0, sizeof *buf); +} + +static uint32_t buffer_free(const struct buffer *buf) { + return buf->maxsize - buf->used; } // Connections are stored in a sorted list. // This gives O(log(N)) lookup time, O(N log(N)) insertion time and O(N) deletion time. static int compare(const void *va, const void *vb) { + assert(va && 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 && b); + assert(a->src && b->src); + int c = (int)a->src - (int)b->src; if(c) return c; @@ -198,19 +214,20 @@ 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); + memmove(cp, cp + 1, (utcp->nconnections - i - 1) * sizeof *cp); utcp->nconnections--; + buffer_exit(&c->sndbuf); free(c); } static struct utcp_connection *allocate_connection(struct utcp *utcp, uint16_t src, uint16_t dst) { // Check whether this combination of src and dst is free - + if(src) { if(find_connection(utcp, src, dst)) { errno = EADDRINUSE; @@ -234,16 +251,17 @@ static struct utcp_connection *allocate_connection(struct utcp *utcp, uint16_t s else utcp->nallocated *= 2; struct utcp_connection **new_array = realloc(utcp->connections, utcp->nallocated * sizeof *utcp->connections); - if(!new_array) { - errno = ENOMEM; + if(!new_array) return NULL; - } utcp->connections = new_array; } struct utcp_connection *c = calloc(1, sizeof *c); - if(!c) { - errno = ENOMEM; + if(!c) + return NULL; + + if(!buffer_init(&c->sndbuf, DEFAULT_SNDBUFSIZE, DEFAULT_MAXSNDBUFSIZE)) { + free(c); return NULL; } @@ -255,6 +273,8 @@ static struct utcp_connection *allocate_connection(struct utcp *utcp, uint16_t s c->snd.una = c->snd.iss; c->snd.nxt = c->snd.iss + 1; c->rcv.wnd = utcp->mtu; + c->snd.last = c->snd.nxt; + c->snd.cwnd = utcp->mtu; c->utcp = utcp; // Add it to the sorted list of connections @@ -271,6 +291,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; @@ -278,33 +299,95 @@ 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); + print_packet(utcp, "send", &hdr, sizeof hdr); utcp->send(utcp, &hdr, sizeof hdr); - // Set timeout? + gettimeofday(&c->conn_timeout, NULL); + c->conn_timeout.tv_sec += utcp->timeout; return c; } void utcp_accept(struct utcp_connection *c, utcp_recv_t recv, void *priv) { if(c->reapable || c->state != SYN_RECEIVED) { - fprintf(stderr, "Error: accept() called on invalid connection %p in state %s\n", c, strstate[c->state]); + debug("Error: accept() called on invalid connection %p in state %s\n", c, strstate[c->state]); return; } - fprintf(stderr, "%p accepted, %p %p\n", c, recv, priv); + debug("%p accepted, %p %p\n", c, recv, priv); c->recv = recv; c->priv = priv; set_state(c, ESTABLISHED); } -int utcp_send(struct utcp_connection *c, void *data, size_t len) { +static void ack(struct utcp_connection *c, bool sendatleastone) { + int32_t left = seqdiff(c->snd.last, c->snd.nxt); + int32_t cwndleft = c->snd.cwnd - seqdiff(c->snd.nxt, c->snd.una); + + assert(left >= 0); + + if(cwndleft <= 0) + cwndleft = 0; + + if(cwndleft < left) + left = cwndleft; + + if(!left && !sendatleastone) + return; + + struct { + struct hdr hdr; + char data[]; + } *pkt; + + pkt = malloc(sizeof pkt->hdr + c->utcp->mtu); + if(!pkt->data) + return; + + pkt->hdr.src = c->src; + pkt->hdr.dst = c->dst; + 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; + pkt->hdr.seq = c->snd.nxt; + + buffer_copy(&c->sndbuf, pkt->data, seqdiff(c->snd.nxt, c->snd.una), seglen); + + c->snd.nxt += seglen; + left -= seglen; + + if(c->state != ESTABLISHED && !left && seglen) { + switch(c->state) { + case FIN_WAIT_1: + case CLOSING: + seglen--; + pkt->hdr.ctl |= FIN; + break; + default: + break; + } + } + + print_packet(c->utcp, "send", pkt, sizeof pkt->hdr + seglen); + c->utcp->send(c->utcp, pkt, sizeof pkt->hdr + seglen); + } while(left); + + free(pkt); +} + +ssize_t utcp_send(struct utcp_connection *c, const void *data, size_t len) { if(c->reapable) { - fprintf(stderr, "Error: send() called on closed connection %p\n", c); + debug("Error: send() called on closed connection %p\n", c); errno = EBADF; return -1; } @@ -314,7 +397,7 @@ int utcp_send(struct utcp_connection *c, void *data, size_t len) { case LISTEN: case SYN_SENT: case SYN_RECEIVED: - fprintf(stderr, "Error: send() called on unconnected connection %p\n", c); + debug("Error: send() called on unconnected connection %p\n", c); errno = ENOTCONN; return -1; case ESTABLISHED: @@ -325,11 +408,13 @@ int utcp_send(struct utcp_connection *c, void *data, size_t len) { case CLOSING: case LAST_ACK: case TIME_WAIT: - fprintf(stderr, "Error: send() called on closing connection %p\n", c); + debug("Error: send() called on closing connection %p\n", c); errno = EPIPE; return -1; } + // Add data to send buffer + if(!len) return 0; @@ -338,30 +423,15 @@ int utcp_send(struct utcp_connection *c, void *data, size_t len) { return -1; } - - struct { - struct hdr hdr; - char data[len]; - } pkt; - - pkt.hdr.src = c->src; - pkt.hdr.dst = c->dst; - pkt.hdr.seq = c->snd.nxt; - pkt.hdr.ack = c->rcv.nxt; - pkt.hdr.wnd = c->snd.wnd; - pkt.hdr.ctl = ACK; - - memcpy(pkt.data, data, len); - - c->snd.nxt += len; - - c->utcp->send(c->utcp, &pkt, sizeof pkt.hdr + len); - // - // Can we add it to the send window? - - // Do we need to kick some timers? - - return 0; + len = buffer_put(&c->sndbuf, data, len); + if(len <= 0) { + errno = EWOULDBLOCK; + return 0; + } + + c->snd.last += len; + ack(c, false); + return len; } static void swap_ports(struct hdr *hdr) { @@ -370,7 +440,75 @@ static void swap_ports(struct hdr *hdr) { hdr->dst = tmp; } -int utcp_recv(struct utcp *utcp, void *data, size_t len) { +static void retransmit(struct utcp_connection *c) { + if(c->state == CLOSED || c->snd.nxt == c->snd.una) + return; + + struct utcp *utcp = c->utcp; + + struct { + struct hdr hdr; + char data[]; + } *pkt; + + pkt = malloc(sizeof pkt->hdr + c->utcp->mtu); + if(!pkt) + return; + + pkt->hdr.src = c->src; + pkt->hdr.dst = c->dst; + + switch(c->state) { + case LISTEN: + // TODO: this should not happen + break; + + case SYN_SENT: + 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); + break; + + case SYN_RECEIVED: + pkt->hdr.seq = c->snd.nxt; + pkt->hdr.ack = c->rcv.nxt; + pkt->hdr.ctl = SYN | ACK; + print_packet(c->utcp, "rtrx", pkt, sizeof pkt->hdr); + utcp->send(utcp, pkt, sizeof pkt->hdr); + break; + + case ESTABLISHED: + case FIN_WAIT_1: + pkt->hdr.seq = c->snd.una; + pkt->hdr.ack = c->rcv.nxt; + pkt->hdr.ctl = ACK; + uint32_t len = seqdiff(c->snd.nxt, c->snd.una); + if(c->state == FIN_WAIT_1) + len--; + if(len > utcp->mtu) + len = utcp->mtu; + else { + if(c->state == FIN_WAIT_1) + pkt->hdr.ctl |= FIN; + } + buffer_copy(&c->sndbuf, pkt->data, 0, len); + print_packet(c->utcp, "rtrx", pkt, sizeof pkt->hdr + len); + utcp->send(utcp, pkt, sizeof pkt->hdr + len); + break; + + default: + // TODO: implement + abort(); + } + + free(pkt); +} + + +ssize_t utcp_recv(struct utcp *utcp, const void *data, size_t len) { if(!utcp) { errno = EFAULT; return -1; @@ -384,8 +522,9 @@ int utcp_recv(struct utcp *utcp, void *data, size_t len) { return -1; } - fprintf(stderr, "%p got: ", utcp); - print_packet(data, len); + print_packet(utcp, "recv", data, len); + + // Drop packets smaller than the header struct hdr hdr; if(len < sizeof hdr) { @@ -393,27 +532,48 @@ int utcp_recv(struct utcp *utcp, void *data, size_t len) { return -1; } + // Make a copy from the potentially unaligned data to a struct hdr + memcpy(&hdr, data, sizeof hdr); data += sizeof hdr; len -= sizeof hdr; + // Drop packets with an unknown CTL flag + if(hdr.ctl & ~(SYN | ACK | RST | FIN)) { errno = EBADMSG; return -1; } - //list_connections(utcp); + // Try to match the packet to an existing connection struct utcp_connection *c = find_connection(utcp, hdr.dst, hdr.src); // Is it for a new connection? if(!c) { + // Ignore RST packets + if(hdr.ctl & RST) return 0; - if(hdr.ctl & SYN && !(hdr.ctl & ACK) && utcp->accept && (!utcp->pre_accept || utcp->pre_accept(utcp, hdr.dst)) && (c = allocate_connection(utcp, hdr.dst, hdr.src))) { // LISTEN - // Return SYN+ACK + // Is it a SYN packet and are we LISTENing? + + if(hdr.ctl & SYN && !(hdr.ctl & ACK) && utcp->accept) { + // If we don't want to accept it, send a RST back + if((utcp->pre_accept && !utcp->pre_accept(utcp, hdr.dst))) { + len = 1; + goto reset; + } + + // Try to allocate memory, otherwise send a RST back + c = allocate_connection(utcp, hdr.dst, hdr.src); + if(!c) { + len = 1; + goto reset; + } + + // Return SYN+ACK, go to SYN_RECEIVED state c->snd.wnd = hdr.wnd; c->rcv.irs = hdr.seq; c->rcv.nxt = c->rcv.irs + 1; @@ -424,121 +584,233 @@ int utcp_recv(struct utcp *utcp, void *data, size_t len) { hdr.ack = c->rcv.irs + 1; hdr.seq = c->snd.iss; hdr.ctl = SYN | ACK; + print_packet(c->utcp, "send", &hdr, sizeof hdr); utcp->send(utcp, &hdr, sizeof hdr); - return 0; - } else { // CLOSED + } else { + // No, we don't want your packets, send a RST back len = 1; goto reset; } + + return 0; } - fprintf(stderr, "%p state %s\n", c->utcp, strstate[c->state]); + debug("%p state %s\n", c->utcp, strstate[c->state]); - if(c->state == CLOSED) { - fprintf(stderr, "Error: packet recv()d on closed connection %p\n", c); - errno = EBADF; - return -1; - } + // 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) + return 0; // It is for an existing connection. - - if(c->state == SYN_SENT) { - if(hdr.ctl & ACK) { - if(hdr.ack <= c->snd.iss || hdr.ack > c->snd.nxt) { - fprintf(stderr, "Invalid ACK, %u %u %u\n", hdr.ack, c->snd.iss, c->snd.nxt); - goto reset; - } - } - if(hdr.ctl & RST) { - if(!(hdr.ctl & ACK)) - return 0; - set_state(c, CLOSED); - errno = ECONNREFUSED; - c->recv(c, NULL, 0); - return 0; - } - if(hdr.ctl & SYN) { - c->dst = hdr.src; - c->rcv.nxt = hdr.seq + 1; - c->rcv.irs = hdr.seq; - c->snd.wnd = hdr.wnd; - if(hdr.ctl & ACK) - c->snd.una = hdr.ack; - if(c->snd.una > c->snd.iss) { - set_state(c, ESTABLISHED); - // TODO: signal app? - swap_ports(&hdr); - hdr.seq = c->snd.nxt; - hdr.ack = c->rcv.nxt; - hdr.ctl = ACK; - } else { - set_state(c, SYN_RECEIVED); - swap_ports(&hdr); - hdr.seq = c->snd.iss; - hdr.ack = c->rcv.nxt; - hdr.ctl = SYN | ACK; - } - utcp->send(utcp, &hdr, sizeof hdr); - // TODO: queue any data? - } + uint32_t prevrcvnxt = c->rcv.nxt; - return 0; + // 1. Drop invalid packets. + + // 1a. Drop packets that should not happen in our current state. + + switch(c->state) { + case SYN_SENT: + case SYN_RECEIVED: + case ESTABLISHED: + case FIN_WAIT_1: + case FIN_WAIT_2: + case CLOSE_WAIT: + case CLOSING: + case LAST_ACK: + case TIME_WAIT: + break; + default: + abort(); } + // 1b. Drop packets with a sequence number not in our receive window. + bool acceptable; - if(len == 0) + if(c->state == SYN_SENT) + acceptable = true; + + // TODO: handle packets overlapping c->rcv.nxt. +#if 0 + // Only use this when accepting out-of-order packets. + else if(len == 0) if(c->rcv.wnd == 0) acceptable = hdr.seq == c->rcv.nxt; else - acceptable = (hdr.seq >= c->rcv.nxt && hdr.seq < c->rcv.nxt + c->rcv.wnd); + acceptable = (seqdiff(hdr.seq, c->rcv.nxt) >= 0 && seqdiff(hdr.seq, c->rcv.nxt + c->rcv.wnd) < 0); else if(c->rcv.wnd == 0) + // We don't accept data when the receive window is zero. acceptable = false; else - acceptable = (hdr.seq >= c->rcv.nxt && hdr.seq < c->rcv.nxt + c->rcv.wnd) - || (hdr.seq + len - 1 >= c->rcv.nxt && hdr.seq + len - 1 < c->rcv.nxt + c->rcv.wnd); + // Both start and end of packet must be within the receive window + acceptable = (seqdiff(hdr.seq, c->rcv.nxt) >= 0 && seqdiff(hdr.seq, c->rcv.nxt + c->rcv.wnd) < 0) + || (seqdiff(hdr.seq + len + 1, c->rcv.nxt) >= 0 && seqdiff(hdr.seq + len - 1, c->rcv.nxt + c->rcv.wnd) < 0); +#else + if(c->state != SYN_SENT) + acceptable = hdr.seq == c->rcv.nxt; +#endif if(!acceptable) { - fprintf(stderr, "Packet not acceptable, %u %u %u %zu\n", hdr.seq, c->rcv.nxt, c->rcv.wnd, len); + debug("Packet not acceptable, %u <= %u + %zu < %u\n", c->rcv.nxt, hdr.seq, len, c->rcv.nxt + c->rcv.wnd); + // 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; + } + + c->snd.wnd = hdr.wnd; // TODO: move below + + // 1c. Drop packets with an invalid ACK. + // ackno should not roll back, and it should also not be bigger than snd.nxt. + + if(hdr.ctl & ACK && (seqdiff(hdr.ack, c->snd.nxt) > 0 || seqdiff(hdr.ack, c->snd.una) < 0)) { + debug("Packet ack seqno out of range, %u %u %u\n", hdr.ack, c->snd.una, c->snd.nxt); + // Ignore unacceptable RST packets. if(hdr.ctl & RST) return 0; - goto ack_and_drop; + goto reset; } - c->snd.wnd = hdr.wnd; + // 2. Handle RST packets - // TODO: check whether segment really starts at rcv.nxt, otherwise trim it. - if(hdr.ctl & RST) { switch(c->state) { + case SYN_SENT: + if(!(hdr.ctl & ACK)) + return 0; + // The peer has refused our connection. + set_state(c, CLOSED); + errno = ECONNREFUSED; + if(c->recv) + c->recv(c, NULL, 0); + return 0; case SYN_RECEIVED: - // TODO: delete connection? - break; + if(hdr.ctl & ACK) + return 0; + // We haven't told the application about this connection yet. Silently delete. + free_connection(c); + return 0; case ESTABLISHED: case FIN_WAIT_1: case FIN_WAIT_2: case CLOSE_WAIT: + if(hdr.ctl & ACK) + return 0; + // The peer has aborted our connection. set_state(c, CLOSED); errno = ECONNRESET; - c->recv(c, NULL, 0); - break; + if(c->recv) + c->recv(c, NULL, 0); + return 0; case CLOSING: case LAST_ACK: case TIME_WAIT: - // TODO: delete connection? + if(hdr.ctl & ACK) + return 0; + // As far as the application is concerned, the connection has already been closed. + // If it has called utcp_close() already, we can immediately free this connection. + if(c->reapable) { + free_connection(c); + return 0; + } + // Otherwise, immediately move to the CLOSED state. + set_state(c, CLOSED); + return 0; + default: + abort(); + } + } + + // 3. Advance snd.una + + uint32_t advanced = seqdiff(hdr.ack, c->snd.una); + prevrcvnxt = c->rcv.nxt; + + if(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); + + if(data_acked) + buffer_get(&c->sndbuf, NULL, data_acked); + + c->snd.una = hdr.ack; + + c->dupack = 0; + c->snd.cwnd += utcp->mtu; + if(c->snd.cwnd > c->sndbuf.maxsize) + c->snd.cwnd = c->sndbuf.maxsize; + + // Check if we have sent a FIN that is now ACKed. + switch(c->state) { + case FIN_WAIT_1: + if(c->snd.una == c->snd.last) + set_state(c, FIN_WAIT_2); + break; + case CLOSING: + if(c->snd.una == c->snd.last) { + gettimeofday(&c->conn_timeout, NULL); + c->conn_timeout.tv_sec += 60; + set_state(c, TIME_WAIT); + } break; default: - // TODO: wtf? - return 0; + break; + } + } else { + if(!len) { + c->dupack++; + if(c->dupack == 3) { + debug("Triplicate ACK\n"); + //TODO: Resend one packet and go to fast recovery mode. See RFC 6582. + //We do a very simple variant here; reset the nxt pointer to the last acknowledged packet from the peer. + //This will cause us to start retransmitting, but at the same speed as the incoming ACKs arrive, + //thus preventing a drop in speed. + c->snd.nxt = c->snd.una; + } } - set_state(c, CLOSED); - return 0; } + // 4. Update timers + + 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); + } + + // 5. Process SYN stuff + if(hdr.ctl & SYN) { switch(c->state) { + case SYN_SENT: + // This is a SYNACK. It should always have ACKed the SYN. + if(!advanced) + goto reset; + c->rcv.irs = hdr.seq; + c->rcv.nxt = hdr.seq; + set_state(c, ESTABLISHED); + // TODO: notify application of this somehow. + break; case SYN_RECEIVED: case ESTABLISHED: case FIN_WAIT_1: @@ -547,121 +819,121 @@ int utcp_recv(struct utcp *utcp, void *data, size_t len) { case CLOSING: case LAST_ACK: case TIME_WAIT: - set_state(c, CLOSED); - errno = ECONNRESET; - c->recv(c, NULL, 0); + // Ehm, no. We should never receive a second SYN. goto reset; - break; default: - // TODO: wtf? - return 0; + abort(); } + + // SYN counts as one sequence number + c->rcv.nxt++; } - if(!(hdr.ctl & ACK)) - return 0; + // 6. Process new data - switch(c->state) { - case SYN_RECEIVED: - if(hdr.ack >= c->snd.una && hdr.ack <= c->snd.nxt) - c->utcp->accept(c, hdr.dst); - - if(c->state != ESTABLISHED) + if(c->state == SYN_RECEIVED) { + // This is the ACK after the SYNACK. It should always have ACKed the SYNACK. + if(!advanced) goto reset; - break; - case ESTABLISHED: - case CLOSE_WAIT: - if(hdr.ack < c->snd.una) - return 0; - if(hdr.ack > c->snd.nxt) - goto ack_and_drop; - if(hdr.ack > c->snd.una && hdr.ack <= c->snd.nxt) { - c->snd.una = hdr.ack; - if(c->snd.wl1 < hdr.seq || (c->snd.wl1 == hdr.seq && c->snd.wl2 <= hdr.ack)) { - c->snd.wnd = hdr.wnd; - c->snd.wl1 = hdr.seq; - c->snd.wl2 = hdr.ack; - } - } - break; - case FIN_WAIT_1: - if(hdr.ack == c->snd.nxt) - set_state(c, FIN_WAIT_2); - break; - case FIN_WAIT_2: - // TODO: If nothing left to send, close. - break; - case CLOSING: - if(hdr.ack == c->snd.nxt) { - set_state(c, TIME_WAIT); - } - break; - case LAST_ACK: - if(hdr.ack == c->snd.nxt) { + + // Are we still LISTENing? + if(utcp->accept) + utcp->accept(c, c->src); + + if(c->state != ESTABLISHED) { set_state(c, CLOSED); + c->reapable = true; + goto reset; } - return 0; - case TIME_WAIT: - // TODO: retransmission of remote FIN, ACK and restart 2 MSL timeout - break; - default: - goto reset; } - // Process data + if(len) { + switch(c->state) { + case SYN_SENT: + case SYN_RECEIVED: + // This should never happen. + abort(); + case ESTABLISHED: + case FIN_WAIT_1: + case FIN_WAIT_2: + break; + case CLOSE_WAIT: + case CLOSING: + case LAST_ACK: + case TIME_WAIT: + // Ehm no, We should never receive more data after a FIN. + goto reset; + default: + abort(); + } - switch(c->state) { - case ESTABLISHED: - case FIN_WAIT_1: - case FIN_WAIT_2: - // TODO: process the data, see page 74 - break; - case CLOSE_WAIT: - case CLOSING: - case LAST_ACK: - case TIME_WAIT: - break; - default: - abort(); + 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. + abort(); + } + if(rxd < 0) + rxd = 0; + else if(rxd > len) + rxd = len; // Bad application, bad! + } else { + rxd = len; + } + + c->rcv.nxt += len; } + // 7. Process FIN stuff + if(hdr.ctl & FIN) { switch(c->state) { - case CLOSED: - case LISTEN: case SYN_SENT: - return 0; case SYN_RECEIVED: + // This should never happen. + abort(); case ESTABLISHED: set_state(c, CLOSE_WAIT); - c->rcv.nxt++; - goto ack_and_drop; + break; case FIN_WAIT_1: set_state(c, CLOSING); - c->rcv.nxt++; - goto ack_and_drop; + break; case FIN_WAIT_2: + gettimeofday(&c->conn_timeout, NULL); + c->conn_timeout.tv_sec += 60; set_state(c, TIME_WAIT); - c->rcv.nxt++; - goto ack_and_drop; + break; case CLOSE_WAIT: case CLOSING: case LAST_ACK: case TIME_WAIT: - break; + // Ehm, no. We should never receive a second FIN. + goto reset; default: abort(); } - } - // Process the data - - if(len && c->recv) { - c->recv(c, data, len); - c->rcv.nxt += len; - goto ack_and_drop; + // FIN counts as one sequence number + c->rcv.nxt++; + len++; + + // Inform the application that the peer closed the connection. + if(c->recv) { + errno = 0; + c->recv(c, NULL, 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, prevrcvnxt != c->rcv.nxt); return 0; reset: @@ -675,30 +947,21 @@ reset: hdr.seq = 0; hdr.ctl = RST | ACK; } + print_packet(utcp, "send", &hdr, sizeof hdr); utcp->send(utcp, &hdr, sizeof hdr); return 0; -ack_and_drop: - swap_ports(&hdr); - hdr.seq = c->snd.nxt; - hdr.ack = c->rcv.nxt; - hdr.ctl = ACK; - utcp->send(utcp, &hdr, sizeof hdr); - if(c->state == CLOSE_WAIT || c->state == TIME_WAIT) { - errno = 0; - c->recv(c, NULL, 0); - } - return 0; } int utcp_shutdown(struct utcp_connection *c, int dir) { + debug("%p shutdown %d\n", c ? c->utcp : NULL, dir); if(!c) { errno = EFAULT; return -1; } if(c->reapable) { - fprintf(stderr, "Error: shutdown() called on closed connection %p\n", c); + debug("Error: shutdown() called on closed connection %p\n", c); errno = EBADF; return -1; } @@ -721,7 +984,7 @@ int utcp_shutdown(struct utcp_connection *c, int dir) { case FIN_WAIT_2: return 0; case CLOSE_WAIT: - set_state(c, LAST_ACK); + set_state(c, CLOSING); break; case CLOSING: @@ -730,26 +993,17 @@ int utcp_shutdown(struct utcp_connection *c, int dir) { return 0; } - // Send FIN + c->snd.last++; - struct hdr hdr; - - hdr.src = c->src; - hdr.dst = c->dst; - hdr.seq = c->snd.nxt; - hdr.ack = c->rcv.nxt; - hdr.wnd = c->snd.wnd; - hdr.ctl = FIN | ACK; - - c->snd.nxt += 1; - - c->utcp->send(c->utcp, &hdr, sizeof hdr); + ack(c, false); return 0; } int utcp_close(struct utcp_connection *c) { if(utcp_shutdown(c, SHUT_RDWR)) return -1; + c->recv = NULL; + c->poll = NULL; c->reapable = true; return 0; } @@ -761,11 +1015,13 @@ int utcp_abort(struct utcp_connection *c) { } if(c->reapable) { - fprintf(stderr, "Error: abort() called on closed connection %p\n", c); + debug("Error: abort() called on closed connection %p\n", c); errno = EBADF; return -1; } + c->recv = NULL; + c->poll = NULL; c->reapable = true; switch(c->state) { @@ -799,41 +1055,68 @@ int utcp_abort(struct utcp_connection *c) { hdr.wnd = 0; hdr.ctl = RST; + print_packet(c->utcp, "send", &hdr, sizeof hdr); c->utcp->send(c->utcp, &hdr, sizeof hdr); return 0; } -void utcp_timeout(struct utcp *utcp) { +/* Handle timeouts. + * One call to this function will loop through all connections, + * checking if something needs to be resent or not. + * The return value is the time to the next timeout in milliseconds, + * or maybe a negative value if the timeout is infinite. + */ +struct timeval utcp_timeout(struct utcp *utcp) { struct timeval now; gettimeofday(&now, NULL); + struct timeval next = {now.tv_sec + 3600, now.tv_usec}; for(int i = 0; i < utcp->nconnections; i++) { struct utcp_connection *c = utcp->connections[i]; if(!c) continue; - if(c->reapable) { - fprintf(stderr, "Reaping %p\n", c); - free_connection(c); + if(c->state == CLOSED) { + if(c->reapable) { + debug("Reaping %p\n", c); + free_connection(c); + i--; + } continue; } - if(c->state == CLOSED) - return; - - if(c->conn_timeout.tv_sec && timercmp(&c->conn_timeout, &now, <)) { - if(!c->reapable) { - errno = ETIMEDOUT; - c->recv(c, NULL, 0); - } + if(timerisset(&c->conn_timeout) && timercmp(&c->conn_timeout, &now, <)) { + errno = ETIMEDOUT; c->state = CLOSED; - return; + if(c->recv) + c->recv(c, NULL, 0); + continue; + } + + if(timerisset(&c->rtrx_timeout) && timercmp(&c->rtrx_timeout, &now, <)) { + retransmit(c); } - if(c->rtrx_timeout.tv_sec && timercmp(&c->rtrx_timeout, &now, <)) { - // TODO: retransmit stuff; + if(c->poll && buffer_free(&c->sndbuf) && (c->state == ESTABLISHED || c->state == CLOSE_WAIT)) + c->poll(c, buffer_free(&c->sndbuf)); + + 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; } + + struct timeval diff; + timersub(&next, &now, &diff); + return diff; } struct utcp *utcp_init(utcp_accept_t accept, utcp_pre_accept_t pre_accept, utcp_send_t send, void *priv) { @@ -841,12 +1124,17 @@ struct utcp *utcp_init(utcp_accept_t accept, utcp_pre_accept_t pre_accept, utcp_ if(!utcp) return NULL; + if(!send) { + errno = EFAULT; + return NULL; + } + utcp->accept = accept; utcp->pre_accept = pre_accept; utcp->send = send; utcp->priv = priv; - utcp->gap = -1; utcp->mtu = 1000; + utcp->timeout = 60; return utcp; } @@ -854,7 +1142,71 @@ 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]); + buffer_exit(&utcp->connections[i]->sndbuf); + free(utcp->connections[i]); + } + free(utcp->connections); free(utcp); } + +uint16_t utcp_get_mtu(struct utcp *utcp) { + return utcp->mtu; +} + +void utcp_set_mtu(struct utcp *utcp, uint16_t mtu) { + // TODO: handle overhead of the header + utcp->mtu = mtu; +} + +int utcp_get_user_timeout(struct utcp *u) { + return u->timeout; +} + +void utcp_set_user_timeout(struct utcp *u, int timeout) { + u->timeout = timeout; +} + +size_t utcp_get_sndbuf(struct utcp_connection *c) { + return c->sndbuf.maxsize; +} + +size_t utcp_get_sndbuf_free(struct utcp_connection *c) { + return buffer_free(&c->sndbuf); +} + +void utcp_set_sndbuf(struct utcp_connection *c, size_t size) { + c->sndbuf.maxsize = size; + if(c->sndbuf.maxsize != size) + c->sndbuf.maxsize = -1; +} + +bool utcp_get_nodelay(struct utcp_connection *c) { + return c->nodelay; +} + +void utcp_set_nodelay(struct utcp_connection *c, bool nodelay) { + c->nodelay = nodelay; +} + +bool utcp_get_keepalive(struct utcp_connection *c) { + return c->keepalive; +} + +void utcp_set_keepalive(struct utcp_connection *c, bool keepalive) { + c->keepalive = 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; +}