Delete pending tasks for deleted peers

6 files changed, 56 insertions, 2 deletions

diff --git a/src/fastd.c b/src/fastd.c
index 0f52435..f6635d3 100644
--- a/src/fastd.c
+++ b/src/fastd.c
@@ -260,7 +260,7 @@ static void init_peers(fastd_context *ctx) {
 static void handle_tasks(fastd_context *ctx) {
 	fastd_task *task;
 	while ((task = fastd_task_get(ctx)) != NULL) {
-		switch (task->type) {
+		switch (task->any.type) {
 		case TASK_SEND:
 			if (task->send.peer) {
 				struct msghdr msg;
diff --git a/src/peer.c b/src/peer.c
index 721deb6..93cb465 100644
--- a/src/peer.c
+++ b/src/peer.c
@@ -133,6 +133,8 @@ void fastd_peer_delete(fastd_context *ctx, fastd_peer *peer) {
 		}
 	}
 
+	fastd_task_delete_peer(ctx, peer);
+
 	free(peer);
 }
 
diff --git a/src/queue.c b/src/queue.c
index 87ca255..e0e25e1 100644
--- a/src/queue.c
+++ b/src/queue.c
@@ -87,3 +87,17 @@ int fastd_queue_timeout(fastd_queue *queue) {
 	else
 		return (int)diff_msec;
 }
+
+void fastd_queue_filter(fastd_queue *queue, bool (*pred)(void*, void*), void *extra) {
+	fastd_queue_entry **entry;
+	for (entry = &queue->head; *entry; ) {
+		if (!pred((*entry)->data, extra)) {
+			fastd_queue_entry *cur = *entry;
+			*entry = cur->next;
+			free(cur);
+		}
+		else {
+			entry = &(*entry)->next;
+		}
+	}
+}
diff --git a/src/queue.h b/src/queue.h
index e26946b..f132d1d 100644
--- a/src/queue.h
+++ b/src/queue.h
@@ -28,6 +28,7 @@
 #ifndef _FASTD_QUEUE_H_
 #define _FASTD_QUEUE_H_
 
+#include <stdbool.h>
 #include <stdlib.h>
 #include <time.h>
 
@@ -48,5 +49,6 @@ typedef struct _fastd_queue {
 void fastd_queue_put(fastd_queue *queue, void *data, int timeout);
 void* fastd_queue_get(fastd_queue *queue);
 int fastd_queue_timeout(fastd_queue *queue);
+void fastd_queue_filter(fastd_queue *queue, bool (*pred)(void*, void*), void *extra);
 
 #endif /* _FASTD_QUEUE_H_ */
diff --git a/src/task.c b/src/task.c
index 0d2dd6e..b565f1b 100644
--- a/src/task.c
+++ b/src/task.c
@@ -70,3 +70,32 @@ void fastd_task_schedule_handshake(fastd_context *ctx, fastd_peer *peer, int tim
 
 	fastd_queue_put(&ctx->task_queue, task, timeout);
 }
+
+static bool delete_task(void *data, void *extra) {
+	fastd_task *task = data;
+	fastd_peer *peer = extra;
+
+	if (task->any.peer != peer)
+		return true;
+
+	switch (task->any.type) {
+	case TASK_SEND:
+		fastd_buffer_free(task->send.buffer);
+		break;
+
+	case TASK_HANDLE_RECV:
+		fastd_buffer_free(task->handle_recv.buffer);
+		break;
+
+	case TASK_HANDSHAKE:
+		break;
+	}
+
+	free(task);
+
+	return false;
+}
+
+void fastd_task_delete_peer(fastd_context *ctx, fastd_peer *peer) {
+	fastd_queue_filter(&ctx->task_queue, delete_task, peer);
+}
diff --git a/src/task.h b/src/task.h
index fbf2c63..b65e992 100644
--- a/src/task.h
+++ b/src/task.h
@@ -39,6 +39,11 @@ typedef enum _fastd_task_type {
 	TASK_HANDSHAKE,
 } fastd_task_type;
 
+typedef struct _fastd_task_any {
+	fastd_task_type type;
+	fastd_peer *peer;
+} fastd_task_any;
+
 typedef struct _fastd_task_send {
 	fastd_task_type type;
 	fastd_peer *peer;
@@ -59,7 +64,7 @@ typedef struct _fastd_task_handshake {
 } fastd_task_handshake;
 
 typedef union _fastd_task {
-	fastd_task_type type;
+	fastd_task_any any;
 	fastd_task_send send;
 	fastd_task_handle_recv handle_recv;
 	fastd_task_handshake handshake;
@@ -78,4 +83,6 @@ void fastd_task_put_handle_recv(fastd_context *ctx, fastd_peer *peer, fastd_buff
 
 void fastd_task_schedule_handshake(fastd_context *ctx, fastd_peer *peer, int timeout);
 
+void fastd_task_delete_peer(fastd_context *ctx, fastd_peer *peer);
+
 #endif /* _FASTD_TASK_H_ */