Implement retransmits of important updates

ffd/CMakeLists.txt

@@ -8,6 +8,7 @@ add_executable(ffd
   queue.c
   send.c
   tlv.c
+  update.c
   util.c
 )
 target_link_libraries(ffd rt)

ffd/announce.c

@@ -123,7 +123,7 @@ static inline void seqno_update(ffd_announce_t *announce) {
 			next = cur;
 
 			fprintf(stderr, "debug: update matches seqno request, forwarding\n");
-			ffd_send_update(NULL, req->neigh, announce, false);
+			ffd_update_enqueue(&announce->node, announce->type, announce->key, req->neigh, true);
 			ffd_neigh_unref(req->neigh);
 			free(req);
 		}
@@ -145,7 +145,7 @@ void ffd_announce_update(ffd_announce_t *announce) {
 	if (((announce->last_metric == 0xffff) != (announce->metric.metric == 0xffff))
 	    || diff <= -1024 || diff >= 384) {
 		fprintf(stderr, "info: announce metric has changed significantly, sending updates\n");
-		ffd_send_update(NULL, NULL, announce, false);
+		ffd_update_enqueue(&announce->node, announce->type, announce->key, NULL, announce->metric.metric == 0xffff);
 	}
 
 	if (announce->selected)
@@ -269,11 +269,13 @@ bool ffd_announce_seqno_request(ffd_announce_t *announce, ffd_neigh_t *neigh, ui
 	return false;
 }
 
-void ffd_announce_seqno_request_free_list(ffd_announce_t *announce) {
-	ffd_seqno_req_t *req, *next;
-	for (req = announce->seqno_req_list; req; req = next) {
-		next = req->next;
+void ffd_announce_free(ffd_announce_t *announce) {
+	ffd_seqno_req_t *req, *req_next;
+	for (req = announce->seqno_req_list; req; req = req_next) {
+		req_next = req->next;
 		ffd_neigh_unref(req->neigh);
 		free(req);
 	}
+
+	free(announce);
 }

ffd/ffd.c

@@ -52,7 +52,7 @@ static char *mesh = "bat0";
 int sockfd;
 struct timespec now;
 
-static ffd_queue_head *tasks;
+static ffd_queue_t *tasks = NULL;
 
 ffd_node_id_t self;
 ffd_iface_t *iface_list = NULL;
@@ -397,14 +397,14 @@ static void handle_tlv_seqno_req(const ffd_tlv_seqno_req_t *tlv_req, size_t len,
 
 	if ((int16_t)(seqno-announce->selected->metric_seqno.seqno) <= 0) {
 		fprintf(stderr, "debug: received seqno request, seqno already ok\n");
-		ffd_send_update(NULL, neigh, announce, false);
+		ffd_update_enqueue(&announce->node, announce->type, announce->key, neigh, false);
 		return;
 	}
 
 	if (!announce->selected->neigh) {
 		fprintf(stderr, "debug: received seqno request, incrementing seqno\n");
 		announce->selected->metric_seqno.seqno++;
-		ffd_send_update(NULL, neigh, announce, false);
+		ffd_update_enqueue(&announce->node, announce->type, announce->key, neigh, false);
 		return;
 	}
 
@@ -533,8 +533,7 @@ static void maintenance(void) {
 		if (!announce->nexthop_list) {
 			*cur = *next;
 			next = cur;
-			ffd_announce_seqno_request_free_list(announce);
-			free(announce);
+			ffd_announce_free(announce);
 
 			continue;
 		}
@@ -591,6 +590,16 @@ static void register_periodic_tasks(void) {
 }
 
 
+static inline int timeout_min(int a, int b) {
+	if (a < 0)
+		return b;
+	else if (b < 0)
+		return a;
+	else
+		return min(a, b);
+}
+
+
 int main() {
 	if (!check_config())
 		return 1;
@@ -606,10 +615,11 @@ int main() {
 
 	while (true) {
 		ffd_queue_run(&tasks);
+		ffd_update_run();
 
 		struct pollfd fds[1];
 
-		int timeout = 10*ffd_queue_timeout(&tasks);
+		int timeout = 10*timeout_min(ffd_queue_timeout(&tasks), ffd_update_timeout());
 		if (timeout < 0)
 			timeout = -1;
 

ffd/ffd.h

@@ -51,6 +51,8 @@
 
 #define FFD_MAINTENANCE_INTERVAL FFD_HELLO_INTERVAL
 
+#define FFD_DELAY (FFD_HELLO_INTERVAL/2)
+#define FFD_URGENT_DELAY 20
 
 #define FFD_UPDATE_WITH_DATA 0x01
 
@@ -203,7 +205,11 @@ ffd_announce_t* ffd_announce_get(const ffd_node_id_t *node, uint16_t type, uint1
 ffd_nexthop_t* ffd_announce_nexthop_find(const ffd_announce_t *announce, ffd_neigh_t *neigh);
 ffd_nexthop_t* ffd_announce_nexthop_new(ffd_announce_t *announce, ffd_neigh_t *neigh);
 bool ffd_announce_seqno_request(ffd_announce_t *announce, ffd_neigh_t *neigh, uint16_t seqno);
-void ffd_announce_seqno_request_free_list(ffd_announce_t *announce);
+void ffd_announce_free(ffd_announce_t *announce);
+
+void ffd_update_enqueue(const ffd_node_id_t *node, uint16_t type, uint16_t key, ffd_neigh_t *neigh, bool urgent);
+int ffd_update_timeout(void);
+void ffd_update_run(void);
 
 void ffd_send_ack(ffd_neigh_t *neigh, uint16_t nonce);
 void ffd_send_hellos(void);

ffd/neigh.c

@@ -55,6 +55,11 @@ void ffd_neigh_unref_list(ffd_neigh_t *neigh) {
 	}
 }
 
+void ffd_neigh_unref(ffd_neigh_t *neigh) {
+	if (!(--neigh->ref) && !neigh->iface)
+		free(neigh);
+}
+
 uint16_t ffd_neigh_get_rxcost(const ffd_neigh_t *neigh) {
 	if (!neigh->hello_log || !neigh->hello_interval || !neigh->iface)
 		return 0xffff;

ffd/neigh.h

@@ -44,10 +44,6 @@ static inline void ffd_neigh_ref(ffd_neigh_t *neigh) {
 	neigh->ref++;
 }
 
-static inline void ffd_neigh_unref(ffd_neigh_t *neigh) {
-	if (!(--neigh->ref) && !neigh->iface)
-		free(neigh);
-}
 
 ffd_neigh_t* ffd_neigh_get(ffd_iface_t *iface, const eth_addr_t *addr);
 void ffd_neigh_unref(ffd_neigh_t *neigh);

ffd/queue.c

@@ -27,11 +27,11 @@
 #include "queue.h"
 
 
-void ffd_queue_put(ffd_queue_head **queue, ffd_queue_cb cb, const struct timespec *timeout, void *arg) {
+void ffd_queue_put(ffd_queue_t **queue, ffd_queue_cb cb, const struct timespec *timeout, void *arg) {
 	while (*queue && timespec_after(timeout, &(*queue)->timeout))
 		queue = &(*queue)->next;
 
-	ffd_queue_head *entry = malloc(sizeof(ffd_queue_head));
+	ffd_queue_t *entry = malloc(sizeof(ffd_queue_t));
 
 	entry->cb = cb;
 	entry->timeout = *timeout;

ffd/queue.h

@@ -32,26 +32,32 @@
 
 typedef void (*ffd_queue_cb)(const struct timespec *timeout, void *arg);
 
-typedef struct _ffd_queue_head {
-	struct _ffd_queue_head *next;
+struct _ffd_queue_t {
+	ffd_queue_t *next;
 	ffd_queue_cb cb;
 	struct timespec timeout;
 	void *arg;
-} ffd_queue_head;
+};
 
 
-void ffd_queue_put(ffd_queue_head **queue, ffd_queue_cb cb, const struct timespec *timeout, void *arg);
+void ffd_queue_put(ffd_queue_t **queue, ffd_queue_cb cb, const struct timespec *timeout, void *arg);
 
 
-static inline void ffd_queue_put_delayed(ffd_queue_head **queue, ffd_queue_cb cb, const struct timespec *timeout, int delay, void *arg) {
+static inline void ffd_queue_put_delayed(ffd_queue_t **queue, ffd_queue_cb cb, const struct timespec *timeout, int delay, void *arg) {
 	struct timespec timeout_delayed = *timeout;
 	add_interval(&timeout_delayed, delay);
 	ffd_queue_put(queue, cb, &timeout_delayed, arg);
 }
 
-static inline void ffd_queue_run(ffd_queue_head **queue) {
+static inline void ffd_queue_drop(ffd_queue_t **queue) {
+	ffd_queue_t *entry = *queue;
+	*queue = entry->next;
+	free(entry);
+}
+
+static inline void ffd_queue_run(ffd_queue_t **queue) {
 	while (*queue && timespec_after(&now, &(*queue)->timeout)) {
-		ffd_queue_head *entry = *queue;
+		ffd_queue_t *entry = *queue;
 		*queue = (*queue)->next;
 
 		entry->cb(&entry->timeout, entry->arg);
@@ -59,7 +65,7 @@ static inline void ffd_queue_run(ffd_queue_head **queue) {
 	}
 }
 
-static inline int ffd_queue_timeout(ffd_queue_head *const *queue) {
+static inline int ffd_queue_timeout(ffd_queue_t *const *queue) {
 	if (!*queue)
 		return -1;
 

ffd/types.h

@@ -41,6 +41,6 @@ typedef struct __attribute__((packed)) _eth_addr_t {
 
 
 typedef struct _ffd_packet_t ffd_packet_t;
-
+typedef struct _ffd_queue_t ffd_queue_t;
 
 #endif /* _FFD_TYPES_H_ */

ffd/update.c

@@ -0,0 +1,116 @@
+/*
+  Copyright (c) 2012, Matthias Schiffer <mschiffer@universe-factory.net>
+  All rights reserved.
+
+  Redistribution and use in source and binary forms, with or without
+  modification, are permitted provided that the following conditions are met:
+
+    1. Redistributions of source code must retain the above copyright notice,
+       this list of conditions and the following disclaimer.
+    2. Redistributions in binary form must reproduce the above copyright notice,
+       this list of conditions and the following disclaimer in the documentation
+       and/or other materials provided with the distribution.
+
+  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+*/
+
+
+#include "ffd.h"
+#include "neigh.h"
+#include "queue.h"
+
+#include <stdio.h>
+
+
+static ffd_queue_t *pending_updates = NULL;
+
+
+typedef struct _update_arg {
+	ffd_node_id_t node;
+	uint16_t type;
+	uint16_t key;
+
+	ffd_neigh_t *neigh;
+	unsigned ref;
+} update_arg;
+
+
+static inline update_arg* update_new(const ffd_node_id_t *node, uint16_t type, uint16_t key, ffd_neigh_t *neigh) {
+	update_arg *arg = malloc(sizeof(update_arg));
+
+	arg->node = *node;
+	arg->type = type;
+	arg->key = key;
+
+	if (neigh)
+		ffd_neigh_ref(neigh);
+	arg->neigh = neigh;
+
+	arg->ref = 0;
+
+	return arg;
+}
+
+static inline update_arg* update_ref(update_arg *arg) {
+	arg->ref++;
+	return arg;
+}
+
+static inline void update_unref(update_arg *arg) {
+	if (!--arg->ref) {
+		if (arg->neigh)
+			ffd_neigh_unref(arg->neigh);
+
+		free(arg);
+	}
+}
+
+void ffd_update_enqueue(const ffd_node_id_t *node, uint16_t type, uint16_t key, ffd_neigh_t *neigh, bool urgent) {
+	if (neigh)
+		ffd_neigh_ref(neigh);
+
+	update_arg *arg = update_new(node, type, key, neigh);
+
+	if (urgent) {
+		ffd_queue_put_delayed(&pending_updates, NULL, &now, FFD_URGENT_DELAY, update_ref(arg));
+		ffd_queue_put_delayed(&pending_updates, NULL, &now, 2*FFD_URGENT_DELAY, update_ref(arg));
+		ffd_queue_put_delayed(&pending_updates, NULL, &now, 3*FFD_URGENT_DELAY, update_ref(arg));
+		ffd_queue_put_delayed(&pending_updates, NULL, &now, 4*FFD_URGENT_DELAY, update_ref(arg));
+		ffd_queue_put_delayed(&pending_updates, NULL, &now, 5*FFD_URGENT_DELAY, update_ref(arg));
+	}
+	else {
+		ffd_queue_put_delayed(&pending_updates, NULL, &now, FFD_DELAY, update_ref(arg));
+	}
+}
+
+int ffd_update_timeout(void) {
+	return ffd_queue_timeout(&pending_updates);
+}
+
+void ffd_update_run(void) {
+	while (!ffd_update_timeout()) {
+		update_arg *arg = pending_updates->arg;
+
+		ffd_neigh_t *neigh = arg->neigh;
+		ffd_announce_t *announce = ffd_announce_find(&arg->node, arg->type, arg->key);
+
+		fprintf(stderr, "debug: sending scheduled update.\n");
+
+		if (announce)
+			ffd_send_update(NULL, neigh, announce, false);
+		else
+			ffd_send_retract(neigh, arg->node, arg->type, arg->key);
+
+		update_unref(arg);
+		ffd_queue_drop(&pending_updates);
+	}
+}