libtf: major redesign startedHEADmaster

the idea is to make libtf completely multi-threaded. meaning each
fiber can be running concurrently in separate thread. quite a bit
of framework is added for this and some atomic helpers are already
introduced. however, io polling is busy polling now (will be soon
in own thread) and timeouts are still more or less broken. oh, and
the multithreading core is not there yet. basically we are currently
mostly broken ;)

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);
+}