Add timeout to task queue

6 files changed, 107 insertions, 34 deletions

diff --git a/src/Makefile b/src/Makefile
index caa6f9f..a10ef63 100644
--- a/src/Makefile
+++ b/src/Makefile
@@ -1,11 +1,13 @@
 all : fastd
 
-fastd : fastd.o method_null.o task.o handshake.o
+fastd : fastd.o method_null.o task.o handshake.o queue.o
+	$(CC) -o $@ $^ -lrt
 
 fastd.o : fastd.c fastd.h task.h queue.h handshake.h
-handshake.o : handshake.c packet.h handshake.h
-method_null.o : method_null.c fastd.h task.h
+handshake.o : handshake.c packet.h handshake.h queue.h
+method_null.o : method_null.c fastd.h task.h queue.h
 task.o : task.c task.h queue.h
+queue.o : queue.c queue.h
 
 clean :
 	rm -f fastd fastd.o method_null.o task.o
diff --git a/src/fastd.c b/src/fastd.c
index b76216c..b5036b4 100644
--- a/src/fastd.c
+++ b/src/fastd.c
@@ -99,7 +99,8 @@ static void configure(fastd_context *ctx, fastd_config *conf) {
 	ctx->peers = NULL;
 	fastd_peer **current_peer = &ctx->peers;
 
-	for (fastd_peer_config *peer_conf = conf->peers; peer_conf; peer_conf = peer_conf->next) {
+	fastd_peer_config *peer_conf;
+	for (peer_conf = conf->peers; peer_conf; peer_conf = peer_conf->next) {
 		*current_peer = malloc(sizeof(fastd_peer));
 		(*current_peer)->next = NULL;
 		(*current_peer)->config = peer_conf;
@@ -187,7 +188,7 @@ static void handle_input(fastd_context *ctx) {
 	fds[1].fd = ctx->sockfd;
 	fds[1].events = POLLIN;
 
-	int ret = poll(fds, 2, -1);
+	int ret = poll(fds, 2, fastd_task_timeout(ctx));
 	if (ret < 0)
 		exit_errno(ctx, "poll");
 
diff --git a/src/queue.c b/src/queue.c
new file mode 100644
index 0000000..79f4e96
--- /dev/null
+++ b/src/queue.c
@@ -0,0 +1,89 @@
+/*
+  Copyright (c) 2012, Matthias Schiffer <mschiffer@universe-factory.net>
+  Partly based on QuickTun Copyright (c) 2010, Ivo Smits <Ivo@UCIS.nl>.
+  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 "queue.h"
+
+#include <stdint.h>
+
+
+static inline int after(const struct timespec *tp1, const struct timespec *tp2) {
+	return (tp1->tv_sec > tp2->tv_sec ||
+		(tp1->tv_sec == tp2->tv_sec && tp1->tv_nsec > tp2->tv_nsec));
+}
+
+void fastd_queue_put(fastd_queue *queue, void *data, int timeout) {
+	fastd_queue_entry *entry = malloc(sizeof(fastd_queue_entry));
+	entry->data = data;
+	entry->timeout = (struct timespec){ 0, 0 };
+
+	if (timeout) {
+		clock_gettime(CLOCK_MONOTONIC, &entry->timeout);
+
+		entry->timeout.tv_sec += timeout/1000;
+		entry->timeout.tv_nsec += (timeout%1000)*1e6;
+
+		if (entry->timeout.tv_nsec > 1e9) {
+			entry->timeout.tv_sec++;
+			entry->timeout.tv_nsec -= 1e9;
+		}
+	}
+
+	fastd_queue_entry **current;
+	for (current = &queue->head;; current = &(*current)->next) {
+		if (!(*current) || after(&(*current)->timeout, &entry->timeout)) {
+			entry->next = *current;
+			*current = entry;
+			break;
+		}
+	}
+}
+
+void* fastd_queue_get(fastd_queue *queue) {
+	if (!queue->head || fastd_queue_timeout(queue) > 0)
+		return NULL;
+
+	fastd_queue_entry *entry = queue->head;
+	queue->head = entry->next;
+
+	void *data = entry->data;
+	free(entry);
+	return data;
+}
+
+int fastd_queue_timeout(fastd_queue *queue) {
+	if (!queue->head)
+		return -1;
+
+	struct timespec tp;
+	clock_gettime(CLOCK_MONOTONIC, &tp);
+
+	int64_t diff_msec = ((int64_t)(queue->head->timeout.tv_sec-tp.tv_sec))*1000 - (queue->head->timeout.tv_nsec-tp.tv_nsec)/1e6;
+	if (diff_msec < 0)
+		return 0;
+	else
+		return (int)diff_msec;
+}
diff --git a/src/queue.h b/src/queue.h
index a196153..e26946b 100644
--- a/src/queue.h
+++ b/src/queue.h
@@ -29,6 +29,7 @@
 #define _FASTD_QUEUE_H_
 
 #include <stdlib.h>
+#include <time.h>
 
 
 typedef struct _fastd_queue_entry fastd_queue_entry;
@@ -36,39 +37,16 @@ typedef struct _fastd_queue_entry fastd_queue_entry;
 struct _fastd_queue_entry {
 	fastd_queue_entry *next;
 	void *data;
+	struct timespec timeout;
 };
 
 typedef struct _fastd_queue {
 	fastd_queue_entry *head;
-	fastd_queue_entry *tail;
 } fastd_queue;
 
 
-static inline void fastd_queue_put(fastd_queue *queue, void *data) {
-	fastd_queue_entry *entry = malloc(sizeof(fastd_queue_entry));
-	entry->next = NULL;
-	entry->data = data;
-
-	if (queue->tail)
-		queue->tail->next = entry;
-	else
-		queue->head = entry;
-
-	queue->tail = entry;
-}
-
-static inline void* fastd_queue_get(fastd_queue *queue) {
-	if (!queue->head)
-		return NULL;
-
-	fastd_queue_entry *entry = queue->head;
-	queue->head = entry->next;
-	if (!queue->head)
-		queue->tail = NULL;
-
-	void *data = entry->data;
-	free(entry);
-	return data;
-}
+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);
 
 #endif /* _FASTD_QUEUE_H_ */
diff --git a/src/task.c b/src/task.c
index f3ccfd7..0367cb4 100644
--- a/src/task.c
+++ b/src/task.c
@@ -41,7 +41,7 @@ static void fastd_task_put_send_type(fastd_context *ctx, const fastd_peer *peer,
 	task->packet_type = packet_type;
 	task->buffer = buffer;
 
-	fastd_queue_put(&ctx->task_queue, task);
+	fastd_queue_put(&ctx->task_queue, task, 0);
 }
 
 void fastd_task_put_send_handshake(fastd_context *ctx, const fastd_peer *peer, fastd_buffer buffer) {
@@ -59,5 +59,5 @@ void fastd_task_put_handle_recv(fastd_context *ctx, const fastd_peer *peer, fast
 	task->peer = peer;
 	task->buffer = buffer;
 
-	fastd_queue_put(&ctx->task_queue, task);
+	fastd_queue_put(&ctx->task_queue, task, 0);
 }
diff --git a/src/task.h b/src/task.h
index d816de7..731510b 100644
--- a/src/task.h
+++ b/src/task.h
@@ -60,6 +60,9 @@ typedef union _fastd_task {
 
 
 fastd_task* fastd_task_get(fastd_context *ctx);
+static inline int fastd_task_timeout(fastd_context *ctx) {	
+	return fastd_queue_timeout(&ctx->task_queue);
+}
 
 void fastd_task_put_send_handshake(fastd_context *ctx, const fastd_peer *peer, fastd_buffer buffer);