From 23b95bf1a15322c2f471b80c06cb65d9b2d2a282 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Timo=20Ter=C3=A4s?= Date: Fri, 2 Jul 2010 20:23:07 +0300 Subject: libtf: major redesign started 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 ;) --- src/timeout.c | 120 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 120 insertions(+) create mode 100644 src/timeout.c (limited to 'src/timeout.c') 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 + * 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 +#include +#include + +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); +} -- cgit v1.2.3