git.meshlink.io Git - meshlink/blob - 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 }