git.meshlink.io Git - meshlink-tiny/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-tiny.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(recv(loop->pipefd[0], &signum, 1, MSG_DONTWAIT) != 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 struct sockaddr_in loopback = {0};
 212
 213 static void pipe_init(event_loop_t *loop) {
 214         meshlink_handle_t *mesh = loop->data;
 215
 216         loop->pipefd[0] = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
 217         loop->pipefd[1] = loop->pipefd[0];
 218
 219         if(loop->pipefd[0] == -1) {
 220                 logger(mesh, MESHLINK_ERROR, "Could not create UDP socket");
 221                 return;
 222         }
 223
 224         loopback.sin_family = AF_INET;
 225         loopback.sin_port = 1024;
 226         loopback.sin_addr.s_addr = htonl(0x7F000001);
 227
 228         int result = bind(loop->pipefd[0], (struct sockaddr *)&loopback, sizeof(loopback));
 229
 230         if(result == -1) {
 231                 logger(mesh, MESHLINK_ERROR, "Could not bind UDP socket to localhost");
 232         }
 233
 234         if(result == 0) {
 235 #ifdef O_NONBLOCK
 236                 fcntl(loop->pipefd[0], F_SETFL, O_NONBLOCK);
 237 #endif
 238                 io_add(loop, &loop->signalio, signalio_handler, NULL, loop->pipefd[0], IO_READ);
 239         }
 240 }
 241
 242 static void pipe_exit(event_loop_t *loop) {
 243         io_del(loop, &loop->signalio);
 244
 245         close(loop->pipefd[0]);
 246
 247         loop->pipefd[0] = -1;
 248         loop->pipefd[1] = -1;
 249 }
 250
 251 void signal_trigger(event_loop_t *loop, signal_t *sig) {
 252         meshlink_handle_t *mesh = loop->data;
 253
 254 #ifdef HAVE_STDATOMIC_H
 255
 256         if(atomic_flag_test_and_set(&sig->set)) {
 257                 return;
 258         }
 259
 260 #endif
 261
 262         uint8_t signum = sig->signum;
 263         sendto(loop->pipefd[1], &signum, 1, MSG_DONTWAIT, (struct sockaddr *)&loopback, sizeof(loopback));
 264         return;
 265 }
 266
 267 void signal_add(event_loop_t *loop, signal_t *sig, signal_cb_t cb, void *data, uint8_t signum) {
 268         assert(!sig->cb);
 269
 270         sig->cb = cb;
 271         sig->data = data;
 272         sig->signum = signum;
 273         sig->node.data = sig;
 274
 275 #ifdef HAVE_STDATOMIC_H
 276         atomic_flag_clear(&sig->set);
 277 #endif
 278
 279         if(loop->pipefd[0] == -1) {
 280                 pipe_init(loop);
 281         }
 282
 283         if(!splay_insert_node(&loop->signals, &sig->node)) {
 284                 abort();
 285         }
 286 }
 287
 288 void signal_del(event_loop_t *loop, signal_t *sig) {
 289         assert(sig->cb);
 290
 291         loop->deletion = true;
 292
 293         splay_unlink_node(&loop->signals, &sig->node);
 294         sig->cb = NULL;
 295
 296         if(!loop->signals.count && loop->pipefd[0] != -1) {
 297                 pipe_exit(loop);
 298         }
 299 }
 300
 301 void idle_set(event_loop_t *loop, idle_cb_t cb, void *data) {
 302         loop->idle_cb = cb;
 303         loop->idle_data = data;
 304 }
 305
 306 static void check_bad_fds(event_loop_t *loop, meshlink_handle_t *mesh) {
 307         // Just call all registered callbacks and have them check their fds
 308
 309         do {
 310                 loop->deletion = false;
 311
 312                 for splay_each(io_t, io, &loop->ios) {
 313                         if(io->flags & IO_WRITE) {
 314                                 io->cb(loop, io->data, IO_WRITE);
 315                         }
 316
 317                         if(loop->deletion) {
 318                                 break;
 319                         }
 320
 321                         if(io->flags & IO_READ) {
 322                                 io->cb(loop, io->data, IO_READ);
 323                         }
 324
 325                         if(loop->deletion) {
 326                                 break;
 327                         }
 328                 }
 329         } while(loop->deletion);
 330
 331         // Rebuild the fdsets
 332
 333         fd_set old_readfds;
 334         fd_set old_writefds;
 335         memcpy(&old_readfds, &loop->readfds, sizeof(old_readfds));
 336         memcpy(&old_writefds, &loop->writefds, sizeof(old_writefds));
 337
 338         memset(&loop->readfds, 0, sizeof(loop->readfds));
 339         memset(&loop->writefds, 0, sizeof(loop->writefds));
 340
 341         for splay_each(io_t, io, &loop->ios) {
 342                 if(io->flags & IO_READ) {
 343                         FD_SET(io->fd, &loop->readfds);
 344                         io->cb(loop, io->data, IO_READ);
 345                 }
 346
 347                 if(io->flags & IO_WRITE) {
 348                         FD_SET(io->fd, &loop->writefds);
 349                         io->cb(loop, io->data, IO_WRITE);
 350                 }
 351         }
 352
 353         if(memcmp(&old_readfds, &loop->readfds, sizeof(old_readfds))) {
 354                 logger(mesh, MESHLINK_WARNING, "Incorrect readfds fixed");
 355         }
 356
 357         if(memcmp(&old_writefds, &loop->writefds, sizeof(old_writefds))) {
 358                 logger(mesh, MESHLINK_WARNING, "Incorrect writefds fixed");
 359         }
 360 }
 361
 362 bool event_loop_run(event_loop_t *loop, meshlink_handle_t *mesh) {
 363         assert(mesh);
 364
 365         fd_set readable;
 366         fd_set writable;
 367         int errors = 0;
 368
 369         while(loop->running) {
 370                 clock_gettime(EVENT_CLOCK, &loop->now);
 371                 struct timespec it, ts = {3600, 0};
 372
 373                 while(loop->timeouts.head) {
 374                         timeout_t *timeout = loop->timeouts.head->data;
 375
 376                         if(timespec_lt(&timeout->tv, &loop->now)) {
 377                                 timeout_disable(loop, timeout);
 378                                 timeout->cb(loop, timeout->data);
 379                         } else {
 380                                 timespec_sub(&timeout->tv, &loop->now, &ts);
 381                                 break;
 382                         }
 383                 }
 384
 385                 if(loop->idle_cb) {
 386                         it = loop->idle_cb(loop, loop->idle_data);
 387
 388                         if(it.tv_sec >= 0 && timespec_lt(&it, &ts)) {
 389                                 ts = it;
 390                         }
 391                 }
 392
 393                 memcpy(&readable, &loop->readfds, sizeof(readable));
 394                 memcpy(&writable, &loop->writefds, sizeof(writable));
 395
 396                 int fds = 0;
 397
 398                 if(loop->ios.tail) {
 399                         io_t *last = loop->ios.tail->data;
 400                         fds = last->fd + 1;
 401                 }
 402
 403                 // release mesh mutex during select
 404                 pthread_mutex_unlock(&mesh->mutex);
 405
 406 #ifdef HAVE_PSELECT
 407                 int n = pselect(fds, &readable, &writable, NULL, &ts, NULL);
 408 #else
 409                 struct timeval tv = {ts.tv_sec, ts.tv_nsec / 1000};
 410                 int n = select(fds, &readable, &writable, NULL, (struct timeval *)&tv);
 411 #endif
 412
 413                 if(pthread_mutex_lock(&mesh->mutex) != 0) {
 414                         abort();
 415                 }
 416
 417                 clock_gettime(EVENT_CLOCK, &loop->now);
 418
 419                 if(n < 0) {
 420                         if(sockwouldblock(errno)) {
 421                                 continue;
 422                         } else {
 423                                 errors++;
 424
 425                                 if(errors > 10) {
 426                                         logger(mesh, MESHLINK_ERROR, "Unrecoverable error from select(): %s", strerror(errno));
 427                                         return false;
 428                                 }
 429
 430                                 logger(mesh, MESHLINK_WARNING, "Error from select(), checking for bad fds: %s", strerror(errno));
 431                                 check_bad_fds(loop, mesh);
 432                                 continue;
 433                         }
 434                 }
 435
 436                 errors = 0;
 437
 438                 if(!n) {
 439                         continue;
 440                 }
 441
 442                 // Normally, splay_each allows the current node to be deleted. However,
 443                 // it can be that one io callback triggers the deletion of another io,
 444                 // so we have to detect this and break the loop.
 445
 446                 loop->deletion = false;
 447
 448                 for splay_each(io_t, io, &loop->ios) {
 449                         if(FD_ISSET(io->fd, &writable) && io->cb) {
 450                                 io->cb(loop, io->data, IO_WRITE);
 451                         }
 452
 453                         if(loop->deletion) {
 454                                 break;
 455                         }
 456
 457                         if(FD_ISSET(io->fd, &readable) && io->cb) {
 458                                 io->cb(loop, io->data, IO_READ);
 459                         }
 460
 461                         if(loop->deletion) {
 462                                 break;
 463                         }
 464                 }
 465         }
 466
 467         return true;
 468 }
 469
 470 void event_loop_start(event_loop_t *loop) {
 471         loop->running = true;
 472 }
 473
 474 void event_loop_stop(event_loop_t *loop) {
 475         loop->running = false;
 476 }
 477
 478 void event_loop_init(event_loop_t *loop) {
 479         loop->ios.compare = (splay_compare_t)io_compare;
 480         loop->timeouts.compare = (splay_compare_t)timeout_compare;
 481         loop->signals.compare = (splay_compare_t)signal_compare;
 482         loop->pipefd[0] = -1;
 483         loop->pipefd[1] = -1;
 484         clock_gettime(EVENT_CLOCK, &loop->now);
 485 }
 486
 487 void event_loop_exit(event_loop_t *loop) {
 488         assert(!loop->ios.count);
 489         assert(!loop->timeouts.count);
 490         assert(!loop->signals.count);
 491
 492         for splay_each(io_t, io, &loop->ios) {
 493                 splay_unlink_node(&loop->ios, splay_node);
 494         }
 495
 496         for splay_each(timeout_t, timeout, &loop->timeouts) {
 497                 splay_unlink_node(&loop->timeouts, splay_node);
 498         }
 499
 500         for splay_each(signal_t, signal, &loop->signals) {
 501                 splay_unlink_node(&loop->signals, splay_node);
 502         }
 503 }