Keepalive pings in icmpd

4 files changed, 72 insertions, 9 deletions

diff --git a/clientd.c b/clientd.c
index 2462669..0f8bde1 100644
--- a/clientd.c
+++ b/clientd.c
@@ -14,7 +14,9 @@
 #include "util.h"
 
 int main(void) {
-	struct icmpd *d = icmpd_create_client(inet_addr("127.0.0.1"));
+	const char *remote_addr = "127.0.0.1";
+	// const char *remote_addr = "198.211.118.67";
+	struct icmpd *d = icmpd_create_client(inet_addr(remote_addr));
 	int d_fd = icmpd_get_select_fd(d);
 
 	while (true) {
diff --git a/icmpd.c b/icmpd.c
index 1cfc051..fc47daa 100644
--- a/icmpd.c
+++ b/icmpd.c
@@ -2,6 +2,7 @@
 #include <stdlib.h>
 #include <stdbool.h>
 #include <string.h>
+#include <limits.h>
 #include <fcntl.h>
 #include <unistd.h>
 #include <errno.h>
@@ -13,6 +14,9 @@
 #include "icmp_client.h"
 
 
+#define KEEPALIVE_DELAY 5000  // maximum time to send nothing (milliseconds)
+
+
 struct icmpd {
 	// CONSTANTS
 
@@ -31,6 +35,10 @@ struct icmpd {
 	// server: seqnum is the sequence number of the last-received ECHO
 	// value is -1 if not set
 	int seqnum;
+
+	// client: timestamp of when last message was sent
+	// value is -1 if no message sent yet
+	int64_t last_send_stamp;
 };
 
 // Global state
@@ -137,7 +145,7 @@ static void* thread_entry(void *arg) {
 	size_t recvqu_cap = 8, recvqu_len = 0;
 	struct recvqu_item *recvqu = malloc(recvqu_cap * sizeof(struct recvqu_item));
 
-	// Server messages submitted for transmission while no non-ponged client
+	// Server messages submitted for transmission while no outstanding client
 	// ping is available.
 	size_t sendqu_cap = 8, sendqu_len = 0;
 	struct sendqu_item *sendqu = malloc(sendqu_cap * sizeof(struct sendqu_item));
@@ -149,17 +157,33 @@ static void* thread_entry(void *arg) {
 		if (sock_server >= 0) FD_SET(sock_server, &inset);
 		if (sock_client >= 0) FD_SET(sock_client, &inset);
 
-		int nfds = thread_in;
-		if (sock_server > nfds) nfds = sock_server;
-		if (sock_client > nfds) nfds = sock_client;
+		int nfds = maxi(maxi(thread_in, sock_server), sock_client) + 1;
+
+		int64_t now = gettimestamp();
+		int64_t wait_interval = INT64_MAX;  // microseconds
+
+		for (size_t i = 0; i < conns_len; i++) {
+			if (conns[i]->isserver) continue;
+
+			mt_mutex_lock(&conns[i]->mut);
+			int64_t stamp = conns[i]->last_send_stamp;
+			mt_mutex_unlock(&conns[i]->mut);
 
-		int ret = select(nfds + 1, &inset, NULL, NULL, NULL);
+			if (stamp + KEEPALIVE_DELAY * 1000 - now < wait_interval) {
+				wait_interval = stamp + KEEPALIVE_DELAY * 1000 - now;
+			}
+		}
+
+		struct timeval timeout_tv;
+		timeout_tv.tv_sec = wait_interval / 1000000;
+		timeout_tv.tv_usec = wait_interval % 1000000;
+
+		int ret = select(nfds, &inset, NULL, NULL, wait_interval == INT64_MAX ? NULL : &timeout_tv);
 		if (ret < 0) {
 			if (errno == EINTR) continue;
 			perror("select");
 			assert(false);
 		}
-		assert(ret > 0);
 
 		if (FD_ISSET(thread_in, &inset)) {
 			struct msg_in msg = recv_message(thread_in);
@@ -293,6 +317,7 @@ static void* thread_entry(void *arg) {
 						mt_mutex_lock(&d->mut);
 						client_increment_seqnum(d);
 						int seqnum = d->seqnum;
+						d->last_send_stamp = gettimestamp();
 						mt_mutex_unlock(&d->mut);
 
 						int ret = icmp_client_send(
@@ -403,6 +428,7 @@ static void* thread_entry(void *arg) {
 				mt_mutex_lock(&d->mut);
 				client_increment_seqnum(d);
 				int seqnum = d->seqnum;
+				d->last_send_stamp = gettimestamp();
 				mt_mutex_unlock(&d->mut);
 
 				int ret = icmp_client_send(sock_client, d->other_addr, seqnum, NULL, 0);
@@ -411,6 +437,29 @@ static void* thread_entry(void *arg) {
 				}
 			}
 		}
+
+		now = gettimestamp();
+
+		for (size_t i = 0; i < conns_len; i++) {
+			if (conns[i]->isserver) continue;
+
+			mt_mutex_lock(&conns[i]->mut);
+			int64_t stamp = conns[i]->last_send_stamp;
+			mt_mutex_unlock(&conns[i]->mut);
+
+			if (now - stamp >= KEEPALIVE_DELAY) {
+				mt_mutex_lock(&conns[i]->mut);
+				client_increment_seqnum(conns[i]);
+				int seqnum = conns[i]->seqnum;
+				conns[i]->last_send_stamp = gettimestamp();
+				mt_mutex_unlock(&conns[i]->mut);
+
+				int ret = icmp_client_send(sock_client, conns[i]->other_addr, seqnum, NULL, 0);
+				if (ret < 0) {
+					perror("icmp_client_send");
+				}
+			}
+		}
 	}
 
 	return NULL;
@@ -449,6 +498,7 @@ static struct icmpd* icmpd_create_base(int id, bool isserver, uint32_t other_add
 	d->id = id;
 	d->seqnum = -1;
 	d->outstanding = false;
+	d->last_send_stamp = gettimestamp();
 
 	int pp[2];
 	assert(pipe(pp) == 0);
@@ -483,7 +533,7 @@ void icmpd_server_set_outstanding(struct icmpd *d, int seqnum) {
 }
 
 bool icmpd_peek(struct icmpd *d) {
-#if 0
+#if 1
 	send_message(host_out, MSG_PEEK, &d, sizeof d);
 	struct msg_in msg = recv_message(host_in);
 	assert(msg.hdr.type == MSG_PEEK_ANS);
@@ -492,6 +542,9 @@ bool icmpd_peek(struct icmpd *d) {
 	free(msg.data);
 	return ret;
 #else
+	// This version has the same fallacy as just selecting on icmpd_get_select_fd() directly,
+	// i.e. that select(2) can return when there's nothing to read
+
 	fd_set inset;
 	FD_ZERO(&inset);
 	FD_SET(d->signal_out, &inset);
diff --git a/util.c b/util.c
index 0f3f6bc..9c7e1a3 100644
--- a/util.c
+++ b/util.c
@@ -4,6 +4,7 @@
 #include <unistd.h>
 #include <errno.h>
 #include <assert.h>
+#include <sys/time.h>
 #include "util.h"
 
 
@@ -69,3 +70,9 @@ ssize_t writeall(int fd, const void *data, size_t length) {
 int maxi(int a, int b) {
 	return a > b ? a : b;
 }
+
+int64_t gettimestamp(void) {
+	struct timeval tv;
+	gettimeofday(&tv, NULL);
+	return tv.tv_sec * 1000000LL + tv.tv_usec;
+}
diff --git a/util.h b/util.h
index 3e228f9..88ad996 100644
--- a/util.h
+++ b/util.h
@@ -1,6 +1,7 @@
 #pragma once
 
 #include <stdlib.h>
+#include <stdint.h>
 #include <unistd.h>
 
 
@@ -8,5 +9,5 @@ int uniqid(void);
 void xxd(const void *buf, size_t length);
 ssize_t readall(int fd, void *data, size_t length);
 ssize_t writeall(int fd, const void *data, size_t length);
-
 int maxi(int a, int b);
+int64_t gettimestamp(void);