1 files changed, 120 insertions, 0 deletions
diff --git a/src/timeout.c b/src/timeout.c
new file mode 100644
index 0000000..fc7acd2
--- /dev/null
+++ b/src/timeout.c
@@ -0,0 +1,120 @@
+/* timeout.c - timerfd based fiber wakeups
+ *
+ * Copyright (C) 2010 Timo Teräs <timo.teras@iki.fi>
+ * All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 or later as
+ * published by the Free Software Foundation.
+ *
+ * See http://www.gnu.org/ for details.
+ */
+
+#include <libtf/fiber.h>
+#include <libtf/io.h>
+#include <sys/timerfd.h>
+
+struct tf_timeout_fiber {
+	struct tf_timeout_manager	manager;
+	struct tf_fd			fd;
+	tf_mtime_t			programmed;
+};
+
+static tf_mtime_t tf_mtime_cached;
+
+tf_mtime_t tf_mtime_now(void)
+{
+	return tf_mtime_cached;
+}
+
+tf_mtime_t tf_mtime_update(void)
+{
+	struct timespec ts;
+
+	clock_gettime(CLOCK_MONOTONIC, &ts);
+	tf_mtime_cached = ts.tv_sec * 1000 + ts.tv_nsec / 1000000;
+
+	return tf_mtime_cached;
+}
+
+static void tf_timeout_process_heap(struct tf_timeout_fiber *t)
+{
+	struct tf_heap_head *heap = &t->manager.heap;
+	struct tf_heap_node *node;
+	tf_mtime_t now = tf_mtime_update();
+	tf_mtime_diff_t timeout, value;
+
+	while (!tf_heap_empty(heap)) {
+		value = tf_heap_get_value(heap);
+		timeout = tf_mtime_diff(value, now);
+		if (timeout > 0) {
+			if (t->programmed != value) {
+				struct itimerspec its = {
+					.it_value.tv_sec  = timeout / 1000,
+					.it_value.tv_nsec = (timeout % 1000) * 1000000,
+				};
+				t->programmed = value;
+				timerfd_settime(t->fd.fd, 0, &its, NULL);
+			}
+			break;	
+		}
+
+		node = tf_heap_get_node(heap);
+		tf_heap_delete(node, heap);
+		tf_fiber_wakeup(container_of(node, struct tf_fiber, timeout.heap_node));
+	}
+}
+
+static void tf_timeout_worker(void *ctx)
+{
+	do {
+		tf_ifc_process_unordered();
+		tf_timeout_process_heap(ctx);
+	} while (tf_fiber_schedule() == 0);
+}
+
+struct tf_timeout_fiber *tf_timeout_fiber_create(void)
+{
+	struct tf_timeout_fiber *f;
+
+	f = tf_fiber_create(tf_timeout_worker, sizeof(struct tf_timeout_fiber));
+	if (f == NULL)
+		return f;
+
+	if (tf_open_fd(&f->fd,
+		       timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK|TFD_CLOEXEC),
+		       TF_FD_READ | TF_FD_ALREADY_NONBLOCKING | TF_FD_AUTOCLOSE) != 0) {
+		tf_fiber_put(f);
+		return NULL;
+	}
+	f->fd.fiber = container_of((void *) f, struct tf_fiber, data);
+	tf_fiber_run(tf_fiber_get(f));
+
+	return f;
+}
+
+static void tf_timeout_adjust_ifc(void *fiber, struct tf_ifc *ifc)
+{
+	struct tf_timeout_fiber *t = fiber;
+	struct tf_timeout_client *c = container_of(ifc, struct tf_timeout_client, ifc);
+	tf_mtime_t val;
+
+	val = c->value;
+	if (val == 0)
+		tf_heap_delete(&c->heap_node, &t->manager.heap);
+	else
+		tf_heap_change(&c->heap_node, &t->manager.heap, val);
+	c->latched = val;
+}
+
+void tf_timeout_adjust(struct tf_timeout_client *tc)
+{
+	tf_ifc_queue(tc->manager, &tc->ifc, tf_timeout_adjust_ifc);
+}
+
+void tf_timeout_delete(struct tf_timeout_client *tc)
+{
+	tc->value = 0;
+	tc->latched = 0;
+	tf_ifc_complete(tc->manager, &tc->ifc, tf_timeout_adjust_ifc);
+}