1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
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);
}
|
