Use task queue for peer resets and keepalives

This makes it unnecessary to iterate over all peers for maintenance, and
desynchronizes different peers' keepalives.

8 files changed, 98 insertions, 65 deletions

diff --git a/src/build.h.in b/src/build.h.in
index 5f9c868..9b4727e 100644
--- a/src/build.h.in
+++ b/src/build.h.in
@@ -115,7 +115,7 @@
 #define MAINTENANCE_INTERVAL 10000	/* 10 seconds */
 
 /** The time after which a keepalive should be sent */
-#define KEEPALIVE_TIMEOUT 15000		/* 15 seconds */
+#define KEEPALIVE_TIMEOUT 20000		/* 20 seconds */
 
 /** The time after with a peer is reset if no traffic is received from it */
 #define PEER_STALE_TIME 90000		/* 90 seconds */
diff --git a/src/fastd.h b/src/fastd.h
index 9d44b21..bc5433c 100644
--- a/src/fastd.h
+++ b/src/fastd.h
@@ -533,6 +533,17 @@ static inline bool fastd_timed_out(fastd_timeout_t timeout) {
 	return timeout <= ctx.now;
 }
 
+/** Returns the minimum of two fastd_timeout_t values */
+static inline fastd_timeout_t fastd_timeout_min(fastd_timeout_t a, fastd_timeout_t b) {
+	return (a < b) ? a : b;
+}
+
+/** Updates a timeout, ensuring it can only increase */
+static inline void fastd_timeout_advance(fastd_timeout_t *a, fastd_timeout_t v) {
+	if (*a < v)
+		*a = v;
+}
+
 /** Updates the current time */
 static inline void fastd_update_time(void) {
 	struct timespec ts;
diff --git a/src/peer.c b/src/peer.c
index 3b66bae..9f26803 100644
--- a/src/peer.c
+++ b/src/peer.c
@@ -314,6 +314,7 @@ static void reset_peer(fastd_peer_t *peer) {
 	VECTOR_RESIZE(ctx.eth_addrs, VECTOR_LEN(ctx.eth_addrs)-deleted);
 
 	fastd_peer_unschedule_handshake(peer);
+	fastd_task_unschedule(&peer->task);
 
 	fastd_peer_hashtable_remove(peer);
 
@@ -363,6 +364,15 @@ void fastd_peer_handle_resolve(fastd_peer_t *peer, fastd_remote_t *remote, size_
 		init_handshake(peer);
 }
 
+/** Schedules the peer maintenance task (or removes the schduled task if there's nothing to do) */
+void schedule_peer_task(fastd_peer_t *peer) {
+	fastd_task_unschedule(&peer->task);
+
+	fastd_timeout_t timeout = fastd_timeout_min(peer->reset_timeout, peer->keepalive_timeout);
+	if (timeout != fastd_timeout_inv)
+		fastd_task_schedule(&peer->task, TASK_TYPE_PEER, timeout);
+}
+
 /** Initializes a peer */
 static void setup_peer(fastd_peer_t *peer) {
 	if (VECTOR_LEN(peer->remotes) == 0) {
@@ -397,6 +407,14 @@ static void setup_peer(fastd_peer_t *peer) {
 	peer->verify_valid_timeout = ctx.now;
 #endif
 
+	peer->reset_timeout = fastd_timeout_inv;
+	peer->keepalive_timeout = fastd_timeout_inv;
+
+	if (fastd_peer_is_dynamic(peer)) {
+		peer->reset_timeout = ctx.now;
+		schedule_peer_task(peer);
+	}
+
 	if (!fastd_peer_is_enabled(peer))
 		/* Keep the peer in STATE_INACTIVE */
 		return;
@@ -864,14 +882,19 @@ bool fastd_peer_set_established(fastd_peer_t *peer) {
 
 	if (!peer->iface) {
 		peer->iface = fastd_iface_open(peer);
-		if (peer->iface)
-			on_up(peer, false);
-		else
+		if (!peer->iface)
 			return false;
+
+		on_up(peer, false);
 	}
 
 	peer->state = STATE_ESTABLISHED;
 	peer->established = ctx.now;
+	fastd_peer_seen(peer);
+	fastd_peer_clear_keepalive(peer);
+
+	schedule_peer_task(peer);
+
 	on_establish(peer);
 	pr_info("connection with %P established.", peer);
 
@@ -938,36 +961,29 @@ bool fastd_peer_find_by_eth_addr(const fastd_eth_addr_t addr, fastd_peer_t **pee
    \li If no data was received from the peer for some time, it is reset.
    \li If no data was sent to the peer for some time, a keepalive is sent.
  */
-static bool maintain_peer(fastd_peer_t *peer) {
-	if (fastd_peer_is_dynamic(peer) || fastd_peer_is_established(peer)) {
-		/* check for peer timeout */
-		if (fastd_timed_out(peer->timeout)) {
-#ifdef WITH_DYNAMIC_PEERS
-			if (fastd_peer_is_dynamic(peer) &&
-			    fastd_timed_out(peer->verify_timeout) &&
-			    fastd_timed_out(peer->verify_valid_timeout)) {
-				fastd_peer_delete(peer);
-				return false;
-			}
-#endif
+void fastd_peer_handle_task(fastd_task_t *task) {
+	fastd_peer_t *peer = container_of(task, fastd_peer_t, task);
 
-			if (fastd_peer_is_established(peer))
-				fastd_peer_reset(peer);
-			return true;
-		}
+	if (!fastd_peer_is_dynamic(peer) && !fastd_peer_is_established(peer))
+		return;
 
-		/* check for keepalive timeout */
-		if (!fastd_peer_is_established(peer))
-			return true;
+	/* check for peer timeout */
+	if (fastd_timed_out(peer->reset_timeout)) {
+		if (fastd_peer_is_dynamic(peer))
+			fastd_peer_delete(peer);
+		else
+			fastd_peer_reset(peer);
 
-		if (!fastd_timed_out(peer->keepalive_timeout))
-			return true;
+		return;
+	}
 
+	/* check for keepalive timeout */
+	if (fastd_timed_out(peer->keepalive_timeout)) {
 		pr_debug2("sending keepalive to %P", peer);
 		conf.protocol->send(peer, fastd_buffer_alloc(0, conf.min_encrypt_head_space, conf.min_encrypt_tail_space));
 	}
 
-	return true;
+	schedule_peer_task(peer);
 }
 
 /** Removes all time-outed MAC addresses from \e ctx.eth_addrs */
@@ -988,17 +1004,6 @@ void fastd_peer_eth_addr_cleanup(void) {
 	VECTOR_RESIZE(ctx.eth_addrs, VECTOR_LEN(ctx.eth_addrs)-deleted);
 }
 
-/** Performs periodic maintenance tasks for peers */
-void fastd_peer_maintenance(void) {
-	size_t i;
-	for (i = 0; i < VECTOR_LEN(ctx.peers);) {
-		fastd_peer_t *peer = VECTOR_INDEX(ctx.peers, i);
-
-		if (maintain_peer(peer))
-			i++;
-	}
-}
-
 /** Resets all peers */
 void fastd_peer_reset_all(void) {
 	size_t i;
diff --git a/src/peer.h b/src/peer.h
index a9f48ab..27894f5 100644
--- a/src/peer.h
+++ b/src/peer.h
@@ -33,7 +33,6 @@
 #pragma once
 
 #include "fastd.h"
-#include "pqueue.h"
 
 
 /** The state of a peer */
@@ -97,7 +96,8 @@ struct fastd_peer {
 	fastd_timeout_t establish_handshake_timeout;	/**< A timeout during which all handshakes for this peer will be ignored after a new connection has been established */
 	int64_t established;				/**< The time this peer connection has been established */
 
-	fastd_timeout_t timeout;			/**< The timeout after which the peer is reset */
+	fastd_task_t task;				/**< Task queue entry for periodic maintenance tasks */
+	fastd_timeout_t reset_timeout;			/**< The timeout after which the peer is reset */
 	fastd_timeout_t keepalive_timeout;		/**< The timeout after which a keepalive is sent to the peer */
 
 	fastd_stats_t stats;				/**< Traffic statistics */
@@ -133,20 +133,6 @@ bool fastd_peer_address_equal(const fastd_peer_address_t *addr1, const fastd_pee
 void fastd_peer_address_simplify(fastd_peer_address_t *addr);
 void fastd_peer_address_widen(fastd_peer_address_t *addr);
 
-/** Returns the port of a fastd_peer_address_t (in network byte order) */
-static inline uint16_t fastd_peer_address_get_port(const fastd_peer_address_t *addr) {
-	switch (addr->sa.sa_family) {
-	case AF_INET:
-		return addr->in.sin_port;
-
-	case AF_INET6:
-		return addr->in6.sin6_port;
-
-	default:
-		return 0;
-	}
-}
-
 bool fastd_peer_add(fastd_peer_t *peer);
 void fastd_peer_reset(fastd_peer_t *peer);
 void fastd_peer_delete(fastd_peer_t *peer);
@@ -164,6 +150,29 @@ fastd_peer_t * fastd_peer_find_by_id(uint64_t id);
 void fastd_peer_set_shell_env(fastd_shell_env_t *env, const fastd_peer_t *peer, const fastd_peer_address_t *local_addr, const fastd_peer_address_t *peer_addr);
 void fastd_peer_exec_shell_command(const fastd_shell_command_t *command, const fastd_peer_t *peer, const fastd_peer_address_t *local_addr, const fastd_peer_address_t *peer_addr, bool sync);
 
+void fastd_peer_eth_addr_add(fastd_peer_t *peer, fastd_eth_addr_t addr);
+bool fastd_peer_find_by_eth_addr(const fastd_eth_addr_t addr, fastd_peer_t **peer);
+
+void fastd_peer_handle_handshake_task(fastd_task_t *task);
+void fastd_peer_handle_task(fastd_task_t *task);
+void fastd_peer_eth_addr_cleanup(void);
+void fastd_peer_reset_all(void);
+
+
+/** Returns the port of a fastd_peer_address_t (in network byte order) */
+static inline uint16_t fastd_peer_address_get_port(const fastd_peer_address_t *addr) {
+	switch (addr->sa.sa_family) {
+	case AF_INET:
+		return addr->in.sin_port;
+
+	case AF_INET6:
+		return addr->in6.sin6_port;
+
+	default:
+		return 0;
+	}
+}
+
 /**
    Schedules a handshake with the default delay and jitter
 
@@ -182,11 +191,15 @@ static inline void fastd_peer_unschedule_handshake(fastd_peer_t *peer) {
 /** Call to signal that there is currently an asychronous on-verify command running for the peer */
 static inline void fastd_peer_set_verifying(fastd_peer_t *peer) {
 	peer->verify_timeout = ctx.now + MIN_VERIFY_INTERVAL;
+
+	fastd_timeout_advance(&peer->reset_timeout, peer->verify_timeout);
 }
 
 /** Marks the peer verification as successful or failed */
 static inline void fastd_peer_set_verified(fastd_peer_t *peer, bool ok) {
 	peer->verify_valid_timeout = ctx.now + (ok ? VERIFY_VALID_TIME : 0);
+
+	fastd_timeout_advance(&peer->reset_timeout, peer->verify_valid_timeout);
 }
 #endif
 
@@ -243,7 +256,12 @@ static inline bool fastd_peer_is_established(const fastd_peer_t *peer) {
 
 /** Signals that a valid packet was received from the peer */
 static inline void fastd_peer_seen(fastd_peer_t *peer) {
-	peer->timeout = ctx.now + PEER_STALE_TIME;
+	peer->reset_timeout = ctx.now + PEER_STALE_TIME;
+}
+
+/** Resets the keepalive timeout */
+static inline void fastd_peer_clear_keepalive(fastd_peer_t *peer) {
+	peer->keepalive_timeout = ctx.now + KEEPALIVE_TIMEOUT;
 }
 
 /** Checks if a peer uses dynamic sockets (which means that each connection attempt uses a new socket) */
@@ -267,14 +285,6 @@ static inline bool fastd_eth_addr_is_unicast(fastd_eth_addr_t addr) {
 	return ((addr.data[0] & 1) == 0);
 }
 
-void fastd_peer_eth_addr_add(fastd_peer_t *peer, fastd_eth_addr_t addr);
-bool fastd_peer_find_by_eth_addr(const fastd_eth_addr_t addr, fastd_peer_t **peer);
-
-void fastd_peer_handle_handshake_task(fastd_task_t *task);
-void fastd_peer_eth_addr_cleanup(void);
-void fastd_peer_maintenance(void);
-void fastd_peer_reset_all(void);
-
 /** Adds statistics for a single packet of a given size */
 static inline void fastd_stats_add(UNUSED fastd_peer_t *peer, UNUSED fastd_stat_type_t stat, UNUSED size_t bytes) {
 #ifdef WITH_STATUS_SOCKET
diff --git a/src/protocols/ec25519_fhmqvc/ec25519_fhmqvc.c b/src/protocols/ec25519_fhmqvc/ec25519_fhmqvc.c
index 0696bf0..7f9abe5 100644
--- a/src/protocols/ec25519_fhmqvc/ec25519_fhmqvc.c
+++ b/src/protocols/ec25519_fhmqvc/ec25519_fhmqvc.c
@@ -192,7 +192,7 @@ static void session_send(fastd_peer_t *peer, fastd_buffer_t buffer, protocol_ses
 	}
 
 	fastd_send(peer->sock, &peer->local_address, &peer->address, peer, send_buffer, stat_size);
-	peer->keepalive_timeout = ctx.now + KEEPALIVE_TIMEOUT;
+	fastd_peer_clear_keepalive(peer);
 }
 
 /** Encrypts and sends a packet to a peer */
diff --git a/src/protocols/ec25519_fhmqvc/handshake.c b/src/protocols/ec25519_fhmqvc/handshake.c
index 0609355..245b6f4 100644
--- a/src/protocols/ec25519_fhmqvc/handshake.c
+++ b/src/protocols/ec25519_fhmqvc/handshake.c
@@ -163,7 +163,6 @@ static bool establish(fastd_peer_t *peer, const fastd_method_info_t *method, fas
 	}
 
 	peer->establish_handshake_timeout = ctx.now + MIN_HANDSHAKE_INTERVAL;
-	fastd_peer_seen(peer);
 
 	pr_verbose("new session with %P established using method `%s'%s.", peer, method->name, salt ? "" : " (compat mode)");
 
@@ -586,6 +585,7 @@ static fastd_peer_t * add_dynamic(fastd_socket_t *sock, const fastd_peer_address
 	/* Ugly hack */
 	peer->protocol_state->last_serial--;
 
+	/* Performs further peer initialization */
 	fastd_peer_reset(peer);
 
 	return peer;
diff --git a/src/task.c b/src/task.c
index 2fe79c8..e1dc77d 100644
--- a/src/task.c
+++ b/src/task.c
@@ -36,7 +36,6 @@
 /** Performs periodic maintenance tasks */
 static inline void maintenance(void) {
 	fastd_socket_handle_binds();
-	fastd_peer_maintenance();
 	fastd_peer_eth_addr_cleanup();
 
 	fastd_task_reschedule_relative(&ctx.next_maintenance, MAINTENANCE_INTERVAL);
@@ -56,6 +55,10 @@ static void handle_task(void) {
 		fastd_peer_handle_handshake_task(task);
 		break;
 
+	case TASK_TYPE_PEER:
+		fastd_peer_handle_task(task);
+		break;
+
 	default:
 		exit_bug("unknown task type");
 	}
diff --git a/src/types.h b/src/types.h
index 5ba6481..edc84f8 100644
--- a/src/types.h
+++ b/src/types.h
@@ -90,12 +90,16 @@ typedef enum fastd_task_type {
 	TASK_TYPE_UNSPEC = 0,	/**< Unspecified task type */
 	TASK_TYPE_MAINTENANCE,	/**< Scheduled maintenance */
 	TASK_TYPE_HANDSHAKE,	/**< Scheduled handshake */
+	TASK_TYPE_PEER,		/**< Peer maintenance */
 } fastd_task_type_t;
 
 
 /** A timestamp used as a timeout */
 typedef int64_t fastd_timeout_t;
 
+/** Invalid timestamp */
+static const fastd_timeout_t fastd_timeout_inv = INT64_MAX;
+
 
 typedef struct fastd_buffer fastd_buffer_t;
 typedef struct fastd_poll_fd fastd_poll_fd_t;