From: Guus Sliepen <guus@meshlink.io>
Date: Sat, 9 Aug 2014 16:37:10 +0000 (+0200)
Subject: Start implementing timeout handling.
X-Git-Url: https://git.meshlink.io/?a=commitdiff_plain;h=bf061dc83602d9ec71591c4e138cc9cd583bcfad;p=utcp

Start implementing timeout handling.

- Initial connection timeout (1 minute by default)
- Retransmission of SYN, SYNACK and normal data packets
---

diff --git a/test.c b/test.c
index 4742308..10ebc9a 100644
--- a/test.c
+++ b/test.c
@@ -19,15 +19,16 @@ bool running = true;
 
 int do_recv(struct utcp_connection *c, const void *data, size_t len) {
 	if(!data || !len) {
-		if(errno)
+		if(errno) {
 			fprintf(stderr, "Error: %s\n", strerror(errno));
-		else {
+			dir = 0;
+		} else {
 			dir &= ~2;
 			fprintf(stderr, "Connection closed by peer\n");
 		}
 		return 0;
 	}
-	return write(0, data, len);
+	return write(1, data, len);
 }
 
 void do_accept(struct utcp_connection *nc, uint16_t port) {
@@ -76,6 +77,8 @@ int main(int argc, char *argv[]) {
 	if(!u)
 		return 1;
 
+	utcp_set_connection_timeout(u, 10);
+
 	if(!server)
 		c = utcp_connect(u, 1, do_recv, NULL);
 
@@ -85,13 +88,10 @@ int main(int argc, char *argv[]) {
 	};
 
 	char buf[1024];
+	int timeout = utcp_timeout(u);
 
 	while(dir) {
-		int r = poll(fds, 2, 1000);
-		if(!r) {
-			utcp_timeout(u);
-			continue;
-		}
+		poll(fds, 2, timeout);
 
 		if(fds[0].revents) {
 			int len = read(0, buf, sizeof buf);
@@ -120,6 +120,8 @@ int main(int argc, char *argv[]) {
 					connected = true;
 			utcp_recv(u, buf, len);
 		}
+
+		timeout = utcp_timeout(u);
 	};
 
 	utcp_close(c);
diff --git a/utcp.c b/utcp.c
index d9ba952..15f6374 100644
--- a/utcp.c
+++ b/utcp.c
@@ -122,15 +122,17 @@ struct utcp {
 	utcp_send_t send;
 
 	uint16_t mtu;
+	int timeout;
 
 	struct utcp_connection **connections;
 	int nconnections;
 	int nallocated;
-	int gap;
 };
 
 static void set_state(struct utcp_connection *c, enum state state) {
 	c->state = state;
+	if(state == ESTABLISHED)
+		timerclear(&c->conn_timeout);
 	fprintf(stderr, "%p new state: %s\n", c->utcp, strstate[state]);
 }
 
@@ -163,7 +165,7 @@ static void print_packet(const void *pkt, size_t len) {
 	fprintf(stderr, "\n");
 }
 
-static void list_connections(struct utcp *utcp) {
+static inline void list_connections(struct utcp *utcp) {
 	fprintf(stderr, "%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]);
@@ -289,7 +291,8 @@ struct utcp_connection *utcp_connect(struct utcp *utcp, uint16_t dst, utcp_recv_
 
 	utcp->send(utcp, &hdr, sizeof hdr);
 
-	// Set timeout?
+	gettimeofday(&c->conn_timeout, NULL);
+	c->conn_timeout.tv_sec += utcp->timeout;
 
 	return c;
 }
@@ -379,7 +382,7 @@ ssize_t utcp_send(struct utcp_connection *c, const void *data, size_t len) {
 		c->utcp->send(c->utcp, &pkt, sizeof pkt.hdr + seglen);
 	}
 
-	fprintf(stderr, "len=%u\n", len);
+	fprintf(stderr, "len=%zu\n", len);
 	return len;
 }
 
@@ -389,8 +392,8 @@ static void swap_ports(struct hdr *hdr) {
 	hdr->dst = tmp;
 }
 
-static int16_t seqdiff(uint16_t a, uint16_t b) {
-	return a -b;
+static int32_t seqdiff(uint32_t a, uint32_t b) {
+	return a - b;
 }
 
 int utcp_recv(struct utcp *utcp, const void *data, size_t len) {
@@ -477,7 +480,8 @@ int utcp_recv(struct utcp *utcp, const void *data, size_t len) {
 				return 0;
 			set_state(c, CLOSED);
 			errno = ECONNREFUSED;
-			c->recv(c, NULL, 0);
+			if(c->recv)
+				c->recv(c, NULL, 0);
 			return 0;
 		}
 		if(hdr.ctl & SYN) {
@@ -545,7 +549,8 @@ int utcp_recv(struct utcp *utcp, const void *data, size_t len) {
 		case CLOSE_WAIT:
 			set_state(c, CLOSED);
 			errno = ECONNRESET;
-			c->recv(c, NULL, 0);
+			if(c->recv)
+				c->recv(c, NULL, 0);
 			break;
 		case CLOSING:
 		case LAST_ACK:
@@ -572,7 +577,8 @@ int utcp_recv(struct utcp *utcp, const void *data, size_t len) {
 		case TIME_WAIT:
 			set_state(c, CLOSED);
 			errno = ECONNRESET;
-			c->recv(c, NULL, 0);
+			if(c->recv)
+				c->recv(c, NULL, 0);
 			goto reset;
 			break;
 		default:
@@ -709,7 +715,8 @@ ack_and_drop:
 	utcp->send(utcp, &hdr, sizeof hdr);
 	if(c->state == CLOSE_WAIT || c->state == TIME_WAIT) {
 		errno = 0;
-		c->recv(c, NULL, 0);
+		if(c->recv)
+			c->recv(c, NULL, 0);
 	}
 	return 0;
 }
@@ -826,37 +833,113 @@ int utcp_abort(struct utcp_connection *c) {
 	return 0;
 }
 
-void utcp_timeout(struct utcp *utcp) {
+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[c->utcp->mtu];
+	} pkt;
+
+	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;
+			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;
+			utcp->send(utcp, &pkt, sizeof pkt.hdr);
+			break;
+
+		case ESTABLISHED:
+			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(len > utcp->mtu)
+				len = utcp->mtu;
+			memcpy(pkt.data, c->sndbuf, len);
+			utcp->send(utcp, &pkt, sizeof pkt.hdr + len);
+			break;
+
+		default:
+			// TODO: implement
+			abort();
+	}
+}
+
+/* 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.
+ */
+int 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) {
+				fprintf(stderr, "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(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);
+	if(diff.tv_sec < 0)
+		return 0;
+	return diff.tv_sec * 1000 + diff.tv_usec / 1000;
 }
 
 struct utcp *utcp_init(utcp_accept_t accept, utcp_pre_accept_t pre_accept, utcp_send_t send, void *priv) {
@@ -864,12 +947,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;
 }
@@ -881,3 +969,9 @@ void utcp_exit(struct utcp *utcp) {
 		free_connection(utcp->connections[i]);
 	free(utcp);
 }
+
+int utcp_set_connection_timeout(struct utcp *u, int timeout) {
+	int prev = u->timeout;
+	u->timeout = timeout;
+	return prev;
+}
diff --git a/utcp.h b/utcp.h
index 9c80c45..3e069a0 100644
--- a/utcp.h
+++ b/utcp.h
@@ -53,6 +53,7 @@ extern int utcp_recv(struct utcp *utcp, const void *data, size_t len);
 extern int utcp_close(struct utcp_connection *connection);
 extern int utcp_abort(struct utcp_connection *connection);
 extern int utcp_shutdown(struct utcp_connection *connection, int how);
-extern void utcp_timeout(struct utcp *utcp);
+extern int utcp_timeout(struct utcp *utcp);
+extern int utcp_set_connection_timeout(struct utcp *utcp, int seconds);
 
 #endif