]> git.meshlink.io Git - meshlink/blob - src/event.c
Use multicast to respond to mDNS requests.
[meshlink] / src / event.c
1 /*
2     event.c -- I/O, timeout and signal event handling
3     Copyright (C) 2014-2017 Guus Sliepen <guus@meshlink.io>
4
5     This program is free software; you can redistribute it and/or modify
6     it under the terms of the GNU General Public License as published by
7     the Free Software Foundation; either version 2 of the License, or
8     (at your option) any later version.
9
10     This program is distributed in the hope that it will be useful,
11     but WITHOUT ANY WARRANTY; without even the implied warranty of
12     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13     GNU General Public License for more details.
14
15     You should have received a copy of the GNU General Public License along
16     with this program; if not, write to the Free Software Foundation, Inc.,
17     51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
18 */
19
20 #include "system.h"
21
22 #include "dropin.h"
23 #include "event.h"
24 #include "logger.h"
25 #include "meshlink.h"
26 #include "net.h"
27 #include "splay_tree.h"
28 #include "utils.h"
29 #include "xalloc.h"
30
31 #ifndef EVENT_CLOCK
32 #if defined(CLOCK_MONOTONIC_RAW) && defined(__x86_64__)
33 #define EVENT_CLOCK CLOCK_MONOTONIC_RAW
34 #else
35 #define EVENT_CLOCK CLOCK_MONOTONIC
36 #endif
37 #endif
38
39 static void timespec_add(const struct timespec *a, const struct timespec *b, struct timespec *r) {
40         r->tv_sec = a->tv_sec + b->tv_sec;
41         r->tv_nsec = a->tv_nsec + b->tv_nsec;
42
43         if(r->tv_nsec > 1000000000) {
44                 r->tv_sec++, r->tv_nsec -= 1000000000;
45         }
46 }
47
48 static void timespec_sub(const struct timespec *a, const struct timespec *b, struct timespec *r) {
49         r->tv_sec = a->tv_sec - b->tv_sec;
50         r->tv_nsec = a->tv_nsec - b->tv_nsec;
51
52         if(r->tv_nsec < 0) {
53                 r->tv_sec--, r->tv_nsec += 1000000000;
54         }
55 }
56
57 static bool timespec_lt(const struct timespec *a, const struct timespec *b) {
58         if(a->tv_sec == b->tv_sec) {
59                 return a->tv_nsec < b->tv_nsec;
60         } else {
61                 return a->tv_sec < b->tv_sec;
62         }
63 }
64
65 static void timespec_clear(struct timespec *a) {
66         a->tv_sec = 0;
67 }
68
69 static int io_compare(const io_t *a, const io_t *b) {
70         return a->fd - b->fd;
71 }
72
73 static int timeout_compare(const timeout_t *a, const timeout_t *b) {
74         if(a->tv.tv_sec < b->tv.tv_sec) {
75                 return -1;
76         } else if(a->tv.tv_sec > b->tv.tv_sec) {
77                 return 1;
78         } else if(a->tv.tv_nsec < b->tv.tv_nsec) {
79                 return -1;
80         } else if(a->tv.tv_nsec > b->tv.tv_nsec) {
81                 return 1;
82         } else if(a < b) {
83                 return -1;
84         } else if(a > b) {
85                 return 1;
86         } else {
87                 return 0;
88         }
89 }
90
91 void io_add(event_loop_t *loop, io_t *io, io_cb_t cb, void *data, int fd, int flags) {
92         assert(!io->cb);
93
94         io->fd = fd;
95         io->cb = cb;
96         io->data = data;
97         io->node.data = io;
98
99         io_set(loop, io, flags);
100
101         splay_node_t *node = splay_insert_node(&loop->ios, &io->node);
102         assert(node);
103         (void)node;
104 }
105
106 void io_set(event_loop_t *loop, io_t *io, int flags) {
107         assert(io->cb);
108
109         io->flags = flags;
110
111         if(flags & IO_READ) {
112                 FD_SET(io->fd, &loop->readfds);
113         } else {
114                 FD_CLR(io->fd, &loop->readfds);
115         }
116
117         if(flags & IO_WRITE) {
118                 FD_SET(io->fd, &loop->writefds);
119         } else {
120                 FD_CLR(io->fd, &loop->writefds);
121         }
122 }
123
124 void io_del(event_loop_t *loop, io_t *io) {
125         assert(io->cb);
126
127         loop->deletion = true;
128
129         io_set(loop, io, 0);
130
131         splay_unlink_node(&loop->ios, &io->node);
132         io->cb = NULL;
133 }
134
135 void timeout_add(event_loop_t *loop, timeout_t *timeout, timeout_cb_t cb, void *data, struct timespec *tv) {
136         timeout->cb = cb;
137         timeout->data = data;
138
139         timeout_set(loop, timeout, tv);
140 }
141
142 void timeout_set(event_loop_t *loop, timeout_t *timeout, struct timespec *tv) {
143         assert(timeout->cb);
144
145         if(timeout->node.data) {
146                 splay_unlink_node(&loop->timeouts, &timeout->node);
147         } else {
148                 timeout->node.data = timeout;
149         }
150
151         if(!loop->now.tv_sec) {
152                 clock_gettime(EVENT_CLOCK, &loop->now);
153         }
154
155         timespec_add(&loop->now, tv, &timeout->tv);
156
157         if(!splay_insert_node(&loop->timeouts, &timeout->node)) {
158                 abort();
159         }
160
161         loop->deletion = true;
162 }
163
164 static void timeout_disable(event_loop_t *loop, timeout_t *timeout) {
165         if(timeout->node.data) {
166                 splay_unlink_node(&loop->timeouts, &timeout->node);
167                 timeout->node.data = NULL;
168         }
169
170         timespec_clear(&timeout->tv);
171 }
172
173 void timeout_del(event_loop_t *loop, timeout_t *timeout) {
174         if(!timeout->cb) {
175                 return;
176         }
177
178         if(timeout->node.data) {
179                 timeout_disable(loop, timeout);
180         }
181
182         timeout->cb = NULL;
183         loop->deletion = true;
184 }
185
186 static int signal_compare(const signal_t *a, const signal_t *b) {
187         return (int)a->signum - (int)b->signum;
188 }
189
190 static void signalio_handler(event_loop_t *loop, void *data, int flags) {
191         (void)data;
192         (void)flags;
193         unsigned char signum;
194
195         if(read(loop->pipefd[0], &signum, 1) != 1) {
196                 return;
197         }
198
199         signal_t *sig = splay_search(&loop->signals, &(signal_t) {
200                 .signum = signum
201         });
202
203         if(sig) {
204 #ifdef HAVE_STDATOMIC_H
205                 atomic_flag_clear(&sig->set);
206 #endif
207                 sig->cb(loop, sig->data);
208         }
209 }
210
211 static void pipe_init(event_loop_t *loop) {
212         int result = pipe(loop->pipefd);
213         assert(result == 0);
214
215         if(result == 0) {
216 #ifdef O_NONBLOCK
217                 fcntl(loop->pipefd[0], F_SETFL, O_NONBLOCK);
218                 fcntl(loop->pipefd[1], F_SETFL, O_NONBLOCK);
219 #endif
220                 io_add(loop, &loop->signalio, signalio_handler, NULL, loop->pipefd[0], IO_READ);
221         }
222 }
223
224 static void pipe_exit(event_loop_t *loop) {
225         io_del(loop, &loop->signalio);
226
227         close(loop->pipefd[0]);
228         close(loop->pipefd[1]);
229
230         loop->pipefd[0] = -1;
231         loop->pipefd[1] = -1;
232 }
233
234 void signal_trigger(event_loop_t *loop, signal_t *sig) {
235 #ifdef HAVE_STDATOMIC_H
236
237         if(atomic_flag_test_and_set(&sig->set)) {
238                 return;
239         }
240
241 #endif
242
243         uint8_t signum = sig->signum;
244         write(loop->pipefd[1], &signum, 1);
245         return;
246 }
247
248 void signal_add(event_loop_t *loop, signal_t *sig, signal_cb_t cb, void *data, uint8_t signum) {
249         assert(!sig->cb);
250
251         sig->cb = cb;
252         sig->data = data;
253         sig->signum = signum;
254         sig->node.data = sig;
255
256 #ifdef HAVE_STDATOMIC_H
257         atomic_flag_clear(&sig->set);
258 #endif
259
260         if(loop->pipefd[0] == -1) {
261                 pipe_init(loop);
262         }
263
264         if(!splay_insert_node(&loop->signals, &sig->node)) {
265                 abort();
266         }
267 }
268
269 void signal_del(event_loop_t *loop, signal_t *sig) {
270         assert(sig->cb);
271
272         loop->deletion = true;
273
274         splay_unlink_node(&loop->signals, &sig->node);
275         sig->cb = NULL;
276
277         if(!loop->signals.count && loop->pipefd[0] != -1) {
278                 pipe_exit(loop);
279         }
280 }
281
282 void idle_set(event_loop_t *loop, idle_cb_t cb, void *data) {
283         loop->idle_cb = cb;
284         loop->idle_data = data;
285 }
286
287 static void check_bad_fds(event_loop_t *loop, meshlink_handle_t *mesh) {
288         // Just call all registered callbacks and have them check their fds
289
290         do {
291                 loop->deletion = false;
292
293                 for splay_each(io_t, io, &loop->ios) {
294                         if(io->flags & IO_WRITE) {
295                                 io->cb(loop, io->data, IO_WRITE);
296                         }
297
298                         if(loop->deletion) {
299                                 break;
300                         }
301
302                         if(io->flags & IO_READ) {
303                                 io->cb(loop, io->data, IO_READ);
304                         }
305
306                         if(loop->deletion) {
307                                 break;
308                         }
309                 }
310         } while(loop->deletion);
311
312         // Rebuild the fdsets
313
314         fd_set old_readfds;
315         fd_set old_writefds;
316         memcpy(&old_readfds, &loop->readfds, sizeof(old_readfds));
317         memcpy(&old_writefds, &loop->writefds, sizeof(old_writefds));
318
319         memset(&loop->readfds, 0, sizeof(loop->readfds));
320         memset(&loop->writefds, 0, sizeof(loop->writefds));
321
322         for splay_each(io_t, io, &loop->ios) {
323                 if(io->flags & IO_READ) {
324                         FD_SET(io->fd, &loop->readfds);
325                         io->cb(loop, io->data, IO_READ);
326                 }
327
328                 if(io->flags & IO_WRITE) {
329                         FD_SET(io->fd, &loop->writefds);
330                         io->cb(loop, io->data, IO_WRITE);
331                 }
332         }
333
334         if(memcmp(&old_readfds, &loop->readfds, sizeof(old_readfds))) {
335                 logger(mesh, MESHLINK_WARNING, "Incorrect readfds fixed");
336         }
337
338         if(memcmp(&old_writefds, &loop->writefds, sizeof(old_writefds))) {
339                 logger(mesh, MESHLINK_WARNING, "Incorrect writefds fixed");
340         }
341 }
342
343 bool event_loop_run(event_loop_t *loop, meshlink_handle_t *mesh) {
344         assert(mesh);
345
346         fd_set readable;
347         fd_set writable;
348         int errors = 0;
349
350         while(loop->running) {
351                 clock_gettime(EVENT_CLOCK, &loop->now);
352                 struct timespec it, ts = {3600, 0};
353
354                 while(loop->timeouts.head) {
355                         timeout_t *timeout = loop->timeouts.head->data;
356
357                         if(timespec_lt(&timeout->tv, &loop->now)) {
358                                 timeout_disable(loop, timeout);
359                                 timeout->cb(loop, timeout->data);
360                         } else {
361                                 timespec_sub(&timeout->tv, &loop->now, &ts);
362                                 break;
363                         }
364                 }
365
366                 if(loop->idle_cb) {
367                         it = loop->idle_cb(loop, loop->idle_data);
368
369                         if(it.tv_sec >= 0 && timespec_lt(&it, &ts)) {
370                                 ts = it;
371                         }
372                 }
373
374                 memcpy(&readable, &loop->readfds, sizeof(readable));
375                 memcpy(&writable, &loop->writefds, sizeof(writable));
376
377                 int fds = 0;
378
379                 if(loop->ios.tail) {
380                         io_t *last = loop->ios.tail->data;
381                         fds = last->fd + 1;
382                 }
383
384                 // release mesh mutex during select
385                 pthread_mutex_unlock(&mesh->mutex);
386
387 #ifdef HAVE_PSELECT
388                 int n = pselect(fds, &readable, &writable, NULL, &ts, NULL);
389 #else
390                 struct timeval tv = {ts.tv_sec, ts.tv_nsec / 1000};
391                 int n = select(fds, &readable, &writable, NULL, (struct timeval *)&tv);
392 #endif
393
394                 if(pthread_mutex_lock(&mesh->mutex) != 0) {
395                         abort();
396                 }
397
398                 clock_gettime(EVENT_CLOCK, &loop->now);
399
400                 if(n < 0) {
401                         if(sockwouldblock(errno)) {
402                                 continue;
403                         } else {
404                                 errors++;
405
406                                 if(errors > 10) {
407                                         logger(mesh, MESHLINK_ERROR, "Unrecoverable error from select(): %s", strerror(errno));
408                                         return false;
409                                 }
410
411                                 logger(mesh, MESHLINK_WARNING, "Error from select(), checking for bad fds: %s", strerror(errno));
412                                 check_bad_fds(loop, mesh);
413                                 continue;
414                         }
415                 }
416
417                 errors = 0;
418
419                 if(!n) {
420                         continue;
421                 }
422
423                 // Normally, splay_each allows the current node to be deleted. However,
424                 // it can be that one io callback triggers the deletion of another io,
425                 // so we have to detect this and break the loop.
426
427                 loop->deletion = false;
428
429                 for splay_each(io_t, io, &loop->ios) {
430                         if(FD_ISSET(io->fd, &writable) && io->cb) {
431                                 io->cb(loop, io->data, IO_WRITE);
432                         }
433
434                         if(loop->deletion) {
435                                 break;
436                         }
437
438                         if(FD_ISSET(io->fd, &readable) && io->cb) {
439                                 io->cb(loop, io->data, IO_READ);
440                         }
441
442                         if(loop->deletion) {
443                                 break;
444                         }
445                 }
446         }
447
448         return true;
449 }
450
451 void event_loop_start(event_loop_t *loop) {
452         loop->running = true;
453 }
454
455 void event_loop_stop(event_loop_t *loop) {
456         loop->running = false;
457 }
458
459 void event_loop_init(event_loop_t *loop) {
460         loop->ios.compare = (splay_compare_t)io_compare;
461         loop->timeouts.compare = (splay_compare_t)timeout_compare;
462         loop->signals.compare = (splay_compare_t)signal_compare;
463         loop->pipefd[0] = -1;
464         loop->pipefd[1] = -1;
465         clock_gettime(EVENT_CLOCK, &loop->now);
466 }
467
468 void event_loop_exit(event_loop_t *loop) {
469         assert(!loop->ios.count);
470         assert(!loop->timeouts.count);
471         assert(!loop->signals.count);
472
473         for splay_each(io_t, io, &loop->ios) {
474                 splay_unlink_node(&loop->ios, splay_node);
475         }
476
477         for splay_each(timeout_t, timeout, &loop->timeouts) {
478                 splay_unlink_node(&loop->timeouts, splay_node);
479         }
480
481         for splay_each(signal_t, signal, &loop->signals) {
482                 splay_unlink_node(&loop->signals, splay_node);
483         }
484 }