src/libsystemd-bus/sd-event.c

   1 /*-*- Mode: C; c-basic-offset: 8; indent-tabs-mode: nil -*-*/
   2
   3 /***
   4   This file is part of systemd.
   5
   6   Copyright 2013 Lennart Poettering
   7
   8   systemd is free software; you can redistribute it and/or modify it
   9   under the terms of the GNU Lesser General Public License as published by
  10   the Free Software Foundation; either version 2.1 of the License, or
  11   (at your option) any later version.
  12
  13   systemd is distributed in the hope that it will be useful, but
  14   WITHOUT ANY WARRANTY; without even the implied warranty of
  15   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  16   Lesser General Public License for more details.
  17
  18   You should have received a copy of the GNU Lesser General Public License
  19   along with systemd; If not, see <http://www.gnu.org/licenses/>.
  20 ***/
  21
  22 #include <sys/epoll.h>
  23 #include <sys/timerfd.h>
  24 #include <sys/wait.h>
  25
  26 #include "macro.h"
  27 #include "prioq.h"
  28 #include "hashmap.h"
  29 #include "util.h"
  30 #include "time-util.h"
  31 #include "sd-id128.h"
  32
  33 #include "sd-event.h"
  34
  35 #define EPOLL_QUEUE_MAX 64
  36 #define DEFAULT_ACCURACY_USEC (250 * USEC_PER_MSEC)
  37
  38 typedef enum EventSourceType {
  39         SOURCE_IO,
  40         SOURCE_MONOTONIC,
  41         SOURCE_REALTIME,
  42         SOURCE_SIGNAL,
  43         SOURCE_CHILD,
  44         SOURCE_DEFER,
  45         SOURCE_QUIT
  46 } EventSourceType;
  47
  48 struct sd_event_source {
  49         unsigned n_ref;
  50
  51         sd_event *event;
  52         void *userdata;
  53         sd_prepare_handler_t prepare;
  54
  55         EventSourceType type:4;
  56         int enabled:3;
  57         bool pending:1;
  58
  59         int priority;
  60         unsigned pending_index;
  61         unsigned prepare_index;
  62         unsigned pending_iteration;
  63         unsigned prepare_iteration;
  64
  65         union {
  66                 struct {
  67                         sd_io_handler_t callback;
  68                         int fd;
  69                         uint32_t events;
  70                         uint32_t revents;
  71                         bool registered:1;
  72                 } io;
  73                 struct {
  74                         sd_time_handler_t callback;
  75                         usec_t next, accuracy;
  76                         unsigned earliest_index;
  77                         unsigned latest_index;
  78                 } time;
  79                 struct {
  80                         sd_signal_handler_t callback;
  81                         struct signalfd_siginfo siginfo;
  82                         int sig;
  83                 } signal;
  84                 struct {
  85                         sd_child_handler_t callback;
  86                         siginfo_t siginfo;
  87                         pid_t pid;
  88                         int options;
  89                 } child;
  90                 struct {
  91                         sd_defer_handler_t callback;
  92                 } defer;
  93                 struct {
  94                         sd_quit_handler_t callback;
  95                         unsigned prioq_index;
  96                 } quit;
  97         };
  98 };
  99
 100 struct sd_event {
 101         unsigned n_ref;
 102
 103         int epoll_fd;
 104         int signal_fd;
 105         int realtime_fd;
 106         int monotonic_fd;
 107
 108         Prioq *pending;
 109         Prioq *prepare;
 110
 111         /* For both clocks we maintain two priority queues each, one
 112          * ordered for the earliest times the events may be
 113          * dispatched, and one ordered by the latest times they must
 114          * have been dispatched. The range between the top entries in
 115          * the two prioqs is the time window we can freely schedule
 116          * wakeups in */
 117         Prioq *monotonic_earliest;
 118         Prioq *monotonic_latest;
 119         Prioq *realtime_earliest;
 120         Prioq *realtime_latest;
 121
 122         usec_t realtime_next, monotonic_next;
 123         usec_t perturb;
 124
 125         sigset_t sigset;
 126         sd_event_source **signal_sources;
 127
 128         Hashmap *child_sources;
 129         unsigned n_enabled_child_sources;
 130
 131         Prioq *quit;
 132
 133         pid_t original_pid;
 134
 135         unsigned iteration;
 136         dual_timestamp timestamp;
 137         int state;
 138
 139         bool quit_requested:1;
 140         bool need_process_child:1;
 141 };
 142
 143 static int pending_prioq_compare(const void *a, const void *b) {
 144         const sd_event_source *x = a, *y = b;
 145
 146         assert(x->pending);
 147         assert(y->pending);
 148
 149         /* Enabled ones first */
 150         if (x->enabled != SD_EVENT_OFF && y->enabled == SD_EVENT_OFF)
 151                 return -1;
 152         if (x->enabled == SD_EVENT_OFF && y->enabled != SD_EVENT_OFF)
 153                 return 1;
 154
 155         /* Lower priority values first */
 156         if (x->priority < y->priority)
 157                 return -1;
 158         if (x->priority > y->priority)
 159                 return 1;
 160
 161         /* Older entries first */
 162         if (x->pending_iteration < y->pending_iteration)
 163                 return -1;
 164         if (x->pending_iteration > y->pending_iteration)
 165                 return 1;
 166
 167         /* Stability for the rest */
 168         if (x < y)
 169                 return -1;
 170         if (x > y)
 171                 return 1;
 172
 173         return 0;
 174 }
 175
 176 static int prepare_prioq_compare(const void *a, const void *b) {
 177         const sd_event_source *x = a, *y = b;
 178
 179         assert(x->prepare);
 180         assert(y->prepare);
 181
 182         /* Move most recently prepared ones last, so that we can stop
 183          * preparing as soon as we hit one that has already been
 184          * prepared in the current iteration */
 185         if (x->prepare_iteration < y->prepare_iteration)
 186                 return -1;
 187         if (x->prepare_iteration > y->prepare_iteration)
 188                 return 1;
 189
 190         /* Enabled ones first */
 191         if (x->enabled != SD_EVENT_OFF && y->enabled == SD_EVENT_OFF)
 192                 return -1;
 193         if (x->enabled == SD_EVENT_OFF && y->enabled != SD_EVENT_OFF)
 194                 return 1;
 195
 196         /* Lower priority values first */
 197         if (x->priority < y->priority)
 198                 return -1;
 199         if (x->priority > y->priority)
 200                 return 1;
 201
 202         /* Stability for the rest */
 203         if (x < y)
 204                 return -1;
 205         if (x > y)
 206                 return 1;
 207
 208         return 0;
 209 }
 210
 211 static int earliest_time_prioq_compare(const void *a, const void *b) {
 212         const sd_event_source *x = a, *y = b;
 213
 214         assert(x->type == SOURCE_MONOTONIC || x->type == SOURCE_REALTIME);
 215         assert(y->type == SOURCE_MONOTONIC || y->type == SOURCE_REALTIME);
 216
 217         /* Enabled ones first */
 218         if (x->enabled != SD_EVENT_OFF && y->enabled == SD_EVENT_OFF)
 219                 return -1;
 220         if (x->enabled == SD_EVENT_OFF && y->enabled != SD_EVENT_OFF)
 221                 return 1;
 222
 223         /* Move the pending ones to the end */
 224         if (!x->pending && y->pending)
 225                 return -1;
 226         if (x->pending && !y->pending)
 227                 return 1;
 228
 229         /* Order by time */
 230         if (x->time.next < y->time.next)
 231                 return -1;
 232         if (x->time.next > y->time.next)
 233                 return -1;
 234
 235         /* Stability for the rest */
 236         if (x < y)
 237                 return -1;
 238         if (x > y)
 239                 return 1;
 240
 241         return 0;
 242 }
 243
 244 static int latest_time_prioq_compare(const void *a, const void *b) {
 245         const sd_event_source *x = a, *y = b;
 246
 247         assert((x->type == SOURCE_MONOTONIC && y->type == SOURCE_MONOTONIC) ||
 248                (x->type == SOURCE_REALTIME && y->type == SOURCE_REALTIME));
 249
 250         /* Enabled ones first */
 251         if (x->enabled != SD_EVENT_OFF && y->enabled == SD_EVENT_OFF)
 252                 return -1;
 253         if (x->enabled == SD_EVENT_OFF && y->enabled != SD_EVENT_OFF)
 254                 return 1;
 255
 256         /* Move the pending ones to the end */
 257         if (!x->pending && y->pending)
 258                 return -1;
 259         if (x->pending && !y->pending)
 260                 return 1;
 261
 262         /* Order by time */
 263         if (x->time.next + x->time.accuracy < y->time.next + y->time.accuracy)
 264                 return -1;
 265         if (x->time.next + x->time.accuracy > y->time.next + y->time.accuracy)
 266                 return -1;
 267
 268         /* Stability for the rest */
 269         if (x < y)
 270                 return -1;
 271         if (x > y)
 272                 return 1;
 273
 274         return 0;
 275 }
 276
 277 static int quit_prioq_compare(const void *a, const void *b) {
 278         const sd_event_source *x = a, *y = b;
 279
 280         assert(x->type == SOURCE_QUIT);
 281         assert(y->type == SOURCE_QUIT);
 282
 283         /* Enabled ones first */
 284         if (x->enabled != SD_EVENT_OFF && y->enabled == SD_EVENT_OFF)
 285                 return -1;
 286         if (x->enabled == SD_EVENT_OFF && y->enabled != SD_EVENT_OFF)
 287                 return 1;
 288
 289         /* Lower priority values first */
 290         if (x->priority < y->priority)
 291                 return -1;
 292         if (x->priority > y->priority)
 293                 return 1;
 294
 295         /* Stability for the rest */
 296         if (x < y)
 297                 return -1;
 298         if (x > y)
 299                 return 1;
 300
 301         return 0;
 302 }
 303
 304 static void event_free(sd_event *e) {
 305         assert(e);
 306
 307         if (e->epoll_fd >= 0)
 308                 close_nointr_nofail(e->epoll_fd);
 309
 310         if (e->signal_fd >= 0)
 311                 close_nointr_nofail(e->signal_fd);
 312
 313         if (e->realtime_fd >= 0)
 314                 close_nointr_nofail(e->realtime_fd);
 315
 316         if (e->monotonic_fd >= 0)
 317                 close_nointr_nofail(e->monotonic_fd);
 318
 319         prioq_free(e->pending);
 320         prioq_free(e->prepare);
 321         prioq_free(e->monotonic_earliest);
 322         prioq_free(e->monotonic_latest);
 323         prioq_free(e->realtime_earliest);
 324         prioq_free(e->realtime_latest);
 325         prioq_free(e->quit);
 326
 327         free(e->signal_sources);
 328
 329         hashmap_free(e->child_sources);
 330         free(e);
 331 }
 332
 333 int sd_event_new(sd_event** ret) {
 334         sd_event *e;
 335         int r;
 336
 337         assert_return(ret, -EINVAL);
 338
 339         e = new0(sd_event, 1);
 340         if (!e)
 341                 return -ENOMEM;
 342
 343         e->n_ref = 1;
 344         e->signal_fd = e->realtime_fd = e->monotonic_fd = e->epoll_fd = -1;
 345         e->realtime_next = e->monotonic_next = (usec_t) -1;
 346         e->original_pid = getpid();
 347
 348         assert_se(sigemptyset(&e->sigset) == 0);
 349
 350         e->pending = prioq_new(pending_prioq_compare);
 351         if (!e->pending) {
 352                 r = -ENOMEM;
 353                 goto fail;
 354         }
 355
 356         e->epoll_fd = epoll_create1(EPOLL_CLOEXEC);
 357         if (e->epoll_fd < 0) {
 358                 r = -errno;
 359                 goto fail;
 360         }
 361
 362         *ret = e;
 363         return 0;
 364
 365 fail:
 366         event_free(e);
 367         return r;
 368 }
 369
 370 sd_event* sd_event_ref(sd_event *e) {
 371         assert_return(e, NULL);
 372
 373         assert(e->n_ref >= 1);
 374         e->n_ref++;
 375
 376         return e;
 377 }
 378
 379 sd_event* sd_event_unref(sd_event *e) {
 380         assert_return(e, NULL);
 381
 382         assert(e->n_ref >= 1);
 383         e->n_ref--;
 384
 385         if (e->n_ref <= 0)
 386                 event_free(e);
 387
 388         return NULL;
 389 }
 390
 391 static bool event_pid_changed(sd_event *e) {
 392         assert(e);
 393
 394         /* We don't support people creating am event loop and keeping
 395          * it around over a fork(). Let's complain. */
 396
 397         return e->original_pid != getpid();
 398 }
 399
 400 static int source_io_unregister(sd_event_source *s) {
 401         int r;
 402
 403         assert(s);
 404         assert(s->type == SOURCE_IO);
 405
 406         if (!s->io.registered)
 407                 return 0;
 408
 409         r = epoll_ctl(s->event->epoll_fd, EPOLL_CTL_DEL, s->io.fd, NULL);
 410         if (r < 0)
 411                 return -errno;
 412
 413         s->io.registered = false;
 414         return 0;
 415 }
 416
 417 static int source_io_register(
 418                 sd_event_source *s,
 419                 int enabled,
 420                 uint32_t events) {
 421
 422         struct epoll_event ev = {};
 423         int r;
 424
 425         assert(s);
 426         assert(s->type == SOURCE_IO);
 427         assert(enabled != SD_EVENT_OFF);
 428
 429         ev.events = events;
 430         ev.data.ptr = s;
 431
 432         if (enabled == SD_EVENT_ONESHOT)
 433                 ev.events |= EPOLLONESHOT;
 434
 435         if (s->io.registered)
 436                 r = epoll_ctl(s->event->epoll_fd, EPOLL_CTL_MOD, s->io.fd, &ev);
 437         else
 438                 r = epoll_ctl(s->event->epoll_fd, EPOLL_CTL_ADD, s->io.fd, &ev);
 439
 440         if (r < 0)
 441                 return -errno;
 442
 443         s->io.registered = true;
 444
 445         return 0;
 446 }
 447
 448 static void source_free(sd_event_source *s) {
 449         assert(s);
 450
 451         if (s->event) {
 452                 switch (s->type) {
 453
 454                 case SOURCE_IO:
 455                         if (s->io.fd >= 0)
 456                                 source_io_unregister(s);
 457
 458                         break;
 459
 460                 case SOURCE_MONOTONIC:
 461                         prioq_remove(s->event->monotonic_earliest, s, &s->time.earliest_index);
 462                         prioq_remove(s->event->monotonic_latest, s, &s->time.latest_index);
 463                         break;
 464
 465                 case SOURCE_REALTIME:
 466                         prioq_remove(s->event->realtime_earliest, s, &s->time.earliest_index);
 467                         prioq_remove(s->event->realtime_latest, s, &s->time.latest_index);
 468                         break;
 469
 470                 case SOURCE_SIGNAL:
 471                         if (s->signal.sig > 0) {
 472                                 if (s->signal.sig != SIGCHLD || s->event->n_enabled_child_sources == 0)
 473                                         assert_se(sigdelset(&s->event->sigset, s->signal.sig) == 0);
 474
 475                                 if (s->event->signal_sources)
 476                                         s->event->signal_sources[s->signal.sig] = NULL;
 477                         }
 478
 479                         break;
 480
 481                 case SOURCE_CHILD:
 482                         if (s->child.pid > 0) {
 483                                 if (s->enabled != SD_EVENT_OFF) {
 484                                         assert(s->event->n_enabled_child_sources > 0);
 485                                         s->event->n_enabled_child_sources--;
 486                                 }
 487
 488                                 if (!s->event->signal_sources || !s->event->signal_sources[SIGCHLD])
 489                                         assert_se(sigdelset(&s->event->sigset, SIGCHLD) == 0);
 490
 491                                 hashmap_remove(s->event->child_sources, INT_TO_PTR(s->child.pid));
 492                         }
 493
 494                         break;
 495
 496                 case SOURCE_DEFER:
 497                         /* nothing */
 498                         break;
 499
 500                 case SOURCE_QUIT:
 501                         prioq_remove(s->event->quit, s, &s->quit.prioq_index);
 502                         break;
 503                 }
 504
 505                 if (s->pending)
 506                         prioq_remove(s->event->pending, s, &s->pending_index);
 507
 508                 if (s->prepare)
 509                         prioq_remove(s->event->prepare, s, &s->prepare_index);
 510
 511                 sd_event_unref(s->event);
 512         }
 513
 514         free(s);
 515 }
 516
 517 static int source_set_pending(sd_event_source *s, bool b) {
 518         int r;
 519
 520         assert(s);
 521         assert(s->type != SOURCE_QUIT);
 522
 523         if (s->pending == b)
 524                 return 0;
 525
 526         s->pending = b;
 527
 528         if (b) {
 529                 s->pending_iteration = s->event->iteration;
 530
 531                 r = prioq_put(s->event->pending, s, &s->pending_index);
 532                 if (r < 0) {
 533                         s->pending = false;
 534                         return r;
 535                 }
 536         } else
 537                 assert_se(prioq_remove(s->event->pending, s, &s->pending_index));
 538
 539         return 0;
 540 }
 541
 542 static sd_event_source *source_new(sd_event *e, EventSourceType type) {
 543         sd_event_source *s;
 544
 545         assert(e);
 546
 547         s = new0(sd_event_source, 1);
 548         if (!s)
 549                 return NULL;
 550
 551         s->n_ref = 1;
 552         s->event = sd_event_ref(e);
 553         s->type = type;
 554         s->pending_index = s->prepare_index = PRIOQ_IDX_NULL;
 555
 556         return s;
 557 }
 558
 559 int sd_event_add_io(
 560                 sd_event *e,
 561                 int fd,
 562                 uint32_t events,
 563                 sd_io_handler_t callback,
 564                 void *userdata,
 565                 sd_event_source **ret) {
 566
 567         sd_event_source *s;
 568         int r;
 569
 570         assert_return(e, -EINVAL);
 571         assert_return(fd >= 0, -EINVAL);
 572         assert_return(!(events & ~(EPOLLIN|EPOLLOUT|EPOLLRDHUP|EPOLLPRI|EPOLLERR|EPOLLHUP)), -EINVAL);
 573         assert_return(callback, -EINVAL);
 574         assert_return(ret, -EINVAL);
 575         assert_return(e->state != SD_EVENT_FINISHED, -ESTALE);
 576         assert_return(!event_pid_changed(e), -ECHILD);
 577
 578         s = source_new(e, SOURCE_IO);
 579         if (!s)
 580                 return -ENOMEM;
 581
 582         s->io.fd = fd;
 583         s->io.events = events;
 584         s->io.callback = callback;
 585         s->userdata = userdata;
 586         s->enabled = SD_EVENT_ON;
 587
 588         r = source_io_register(s, s->enabled, events);
 589         if (r < 0) {
 590                 source_free(s);
 591                 return -errno;
 592         }
 593
 594         *ret = s;
 595         return 0;
 596 }
 597
 598 static int event_setup_timer_fd(
 599                 sd_event *e,
 600                 EventSourceType type,
 601                 int *timer_fd,
 602                 clockid_t id) {
 603
 604         struct epoll_event ev = {};
 605         int r, fd;
 606         sd_id128_t bootid;
 607
 608         assert(e);
 609         assert(timer_fd);
 610
 611         if (_likely_(*timer_fd >= 0))
 612                 return 0;
 613
 614         fd = timerfd_create(id, TFD_NONBLOCK|TFD_CLOEXEC);
 615         if (fd < 0)
 616                 return -errno;
 617
 618         ev.events = EPOLLIN;
 619         ev.data.ptr = INT_TO_PTR(type);
 620
 621         r = epoll_ctl(e->epoll_fd, EPOLL_CTL_ADD, fd, &ev);
 622         if (r < 0) {
 623                 close_nointr_nofail(fd);
 624                 return -errno;
 625         }
 626
 627         /* When we sleep for longer, we try to realign the wakeup to
 628            the same time wihtin each second, so that events all across
 629            the system can be coalesced into a single CPU
 630            wakeup. However, let's take some system-specific randomness
 631            for this value, so that in a network of systems with synced
 632            clocks timer events are distributed a bit. Here, we
 633            calculate a perturbation usec offset from the boot ID. */
 634
 635         if (sd_id128_get_boot(&bootid) >= 0)
 636                 e->perturb = (bootid.qwords[0] ^ bootid.qwords[1]) % USEC_PER_SEC;
 637
 638         *timer_fd = fd;
 639         return 0;
 640 }
 641
 642 static int event_add_time_internal(
 643                 sd_event *e,
 644                 EventSourceType type,
 645                 int *timer_fd,
 646                 clockid_t id,
 647                 Prioq **earliest,
 648                 Prioq **latest,
 649                 uint64_t usec,
 650                 uint64_t accuracy,
 651                 sd_time_handler_t callback,
 652                 void *userdata,
 653                 sd_event_source **ret) {
 654
 655         sd_event_source *s;
 656         int r;
 657
 658         assert_return(e, -EINVAL);
 659         assert_return(callback, -EINVAL);
 660         assert_return(ret, -EINVAL);
 661         assert_return(usec != (uint64_t) -1, -EINVAL);
 662         assert_return(accuracy != (uint64_t) -1, -EINVAL);
 663         assert_return(e->state != SD_EVENT_FINISHED, -ESTALE);
 664         assert_return(!event_pid_changed(e), -ECHILD);
 665
 666         assert(timer_fd);
 667         assert(earliest);
 668         assert(latest);
 669
 670         if (!*earliest) {
 671                 *earliest = prioq_new(earliest_time_prioq_compare);
 672                 if (!*earliest)
 673                         return -ENOMEM;
 674         }
 675
 676         if (!*latest) {
 677                 *latest = prioq_new(latest_time_prioq_compare);
 678                 if (!*latest)
 679                         return -ENOMEM;
 680         }
 681
 682         if (*timer_fd < 0) {
 683                 r = event_setup_timer_fd(e, type, timer_fd, id);
 684                 if (r < 0)
 685                         return r;
 686         }
 687
 688         s = source_new(e, type);
 689         if (!s)
 690                 return -ENOMEM;
 691
 692         s->time.next = usec;
 693         s->time.accuracy = accuracy == 0 ? DEFAULT_ACCURACY_USEC : accuracy;
 694         s->time.callback = callback;
 695         s->time.earliest_index = s->time.latest_index = PRIOQ_IDX_NULL;
 696         s->userdata = userdata;
 697         s->enabled = SD_EVENT_ONESHOT;
 698
 699         r = prioq_put(*earliest, s, &s->time.earliest_index);
 700         if (r < 0)
 701                 goto fail;
 702
 703         r = prioq_put(*latest, s, &s->time.latest_index);
 704         if (r < 0)
 705                 goto fail;
 706
 707         *ret = s;
 708         return 0;
 709
 710 fail:
 711         source_free(s);
 712         return r;
 713 }
 714
 715 int sd_event_add_monotonic(sd_event *e, uint64_t usec, uint64_t accuracy, sd_time_handler_t callback, void *userdata, sd_event_source **ret) {
 716         return event_add_time_internal(e, SOURCE_MONOTONIC, &e->monotonic_fd, CLOCK_MONOTONIC, &e->monotonic_earliest, &e->monotonic_latest, usec, accuracy, callback, userdata, ret);
 717 }
 718
 719 int sd_event_add_realtime(sd_event *e, uint64_t usec, uint64_t accuracy, sd_time_handler_t callback, void *userdata, sd_event_source **ret) {
 720         return event_add_time_internal(e, SOURCE_REALTIME, &e->realtime_fd, CLOCK_REALTIME, &e->realtime_earliest, &e->monotonic_latest, usec, accuracy, callback, userdata, ret);
 721 }
 722
 723 static int event_update_signal_fd(sd_event *e) {
 724         struct epoll_event ev = {};
 725         bool add_to_epoll;
 726         int r;
 727
 728         assert(e);
 729
 730         add_to_epoll = e->signal_fd < 0;
 731
 732         r = signalfd(e->signal_fd, &e->sigset, SFD_NONBLOCK|SFD_CLOEXEC);
 733         if (r < 0)
 734                 return -errno;
 735
 736         e->signal_fd = r;
 737
 738         if (!add_to_epoll)
 739                 return 0;
 740
 741         ev.events = EPOLLIN;
 742         ev.data.ptr = INT_TO_PTR(SOURCE_SIGNAL);
 743
 744         r = epoll_ctl(e->epoll_fd, EPOLL_CTL_ADD, e->signal_fd, &ev);
 745         if (r < 0) {
 746                 close_nointr_nofail(e->signal_fd);
 747                 e->signal_fd = -1;
 748
 749                 return -errno;
 750         }
 751
 752         return 0;
 753 }
 754
 755 int sd_event_add_signal(
 756                 sd_event *e,
 757                 int sig,
 758                 sd_signal_handler_t callback,
 759                 void *userdata,
 760                 sd_event_source **ret) {
 761
 762         sd_event_source *s;
 763         int r;
 764
 765         assert_return(e, -EINVAL);
 766         assert_return(sig > 0, -EINVAL);
 767         assert_return(sig < _NSIG, -EINVAL);
 768         assert_return(callback, -EINVAL);
 769         assert_return(ret, -EINVAL);
 770         assert_return(e->state != SD_EVENT_FINISHED, -ESTALE);
 771         assert_return(!event_pid_changed(e), -ECHILD);
 772
 773         if (!e->signal_sources) {
 774                 e->signal_sources = new0(sd_event_source*, _NSIG);
 775                 if (!e->signal_sources)
 776                         return -ENOMEM;
 777         } else if (e->signal_sources[sig])
 778                 return -EBUSY;
 779
 780         s = source_new(e, SOURCE_SIGNAL);
 781         if (!s)
 782                 return -ENOMEM;
 783
 784         s->signal.sig = sig;
 785         s->signal.callback = callback;
 786         s->userdata = userdata;
 787         s->enabled = SD_EVENT_ON;
 788
 789         e->signal_sources[sig] = s;
 790         assert_se(sigaddset(&e->sigset, sig) == 0);
 791
 792         if (sig != SIGCHLD || e->n_enabled_child_sources == 0) {
 793                 r = event_update_signal_fd(e);
 794                 if (r < 0) {
 795                         source_free(s);
 796                         return r;
 797                 }
 798         }
 799
 800         *ret = s;
 801         return 0;
 802 }
 803
 804 int sd_event_add_child(
 805                 sd_event *e,
 806                 pid_t pid,
 807                 int options,
 808                 sd_child_handler_t callback,
 809                 void *userdata,
 810                 sd_event_source **ret) {
 811
 812         sd_event_source *s;
 813         int r;
 814
 815         assert_return(e, -EINVAL);
 816         assert_return(pid > 1, -EINVAL);
 817         assert_return(!(options & ~(WEXITED|WSTOPPED|WCONTINUED)), -EINVAL);
 818         assert_return(options != 0, -EINVAL);
 819         assert_return(callback, -EINVAL);
 820         assert_return(ret, -EINVAL);
 821         assert_return(e->state != SD_EVENT_FINISHED, -ESTALE);
 822         assert_return(!event_pid_changed(e), -ECHILD);
 823
 824         r = hashmap_ensure_allocated(&e->child_sources, trivial_hash_func, trivial_compare_func);
 825         if (r < 0)
 826                 return r;
 827
 828         if (hashmap_contains(e->child_sources, INT_TO_PTR(pid)))
 829                 return -EBUSY;
 830
 831         s = source_new(e, SOURCE_CHILD);
 832         if (!s)
 833                 return -ENOMEM;
 834
 835         s->child.pid = pid;
 836         s->child.options = options;
 837         s->child.callback = callback;
 838         s->userdata = userdata;
 839         s->enabled = SD_EVENT_ONESHOT;
 840
 841         r = hashmap_put(e->child_sources, INT_TO_PTR(pid), s);
 842         if (r < 0) {
 843                 source_free(s);
 844                 return r;
 845         }
 846
 847         e->n_enabled_child_sources ++;
 848
 849         assert_se(sigaddset(&e->sigset, SIGCHLD) == 0);
 850
 851         if (!e->signal_sources || !e->signal_sources[SIGCHLD]) {
 852                 r = event_update_signal_fd(e);
 853                 if (r < 0) {
 854                         source_free(s);
 855                         return -errno;
 856                 }
 857         }
 858
 859         e->need_process_child = true;
 860
 861         *ret = s;
 862         return 0;
 863 }
 864
 865 int sd_event_add_defer(
 866                 sd_event *e,
 867                 sd_defer_handler_t callback,
 868                 void *userdata,
 869                 sd_event_source **ret) {
 870
 871         sd_event_source *s;
 872         int r;
 873
 874         assert_return(e, -EINVAL);
 875         assert_return(callback, -EINVAL);
 876         assert_return(ret, -EINVAL);
 877         assert_return(e->state != SD_EVENT_FINISHED, -ESTALE);
 878         assert_return(!event_pid_changed(e), -ECHILD);
 879
 880         s = source_new(e, SOURCE_DEFER);
 881         if (!s)
 882                 return -ENOMEM;
 883
 884         s->defer.callback = callback;
 885         s->userdata = userdata;
 886         s->enabled = SD_EVENT_ONESHOT;
 887
 888         r = source_set_pending(s, true);
 889         if (r < 0) {
 890                 source_free(s);
 891                 return r;
 892         }
 893
 894         *ret = s;
 895         return 0;
 896 }
 897
 898 int sd_event_add_quit(
 899                 sd_event *e,
 900                 sd_quit_handler_t callback,
 901                 void *userdata,
 902                 sd_event_source **ret) {
 903
 904         sd_event_source *s;
 905         int r;
 906
 907         assert_return(e, -EINVAL);
 908         assert_return(callback, -EINVAL);
 909         assert_return(ret, -EINVAL);
 910         assert_return(e->state != SD_EVENT_FINISHED, -ESTALE);
 911         assert_return(!event_pid_changed(e), -ECHILD);
 912
 913         if (!e->quit) {
 914                 e->quit = prioq_new(quit_prioq_compare);
 915                 if (!e->quit)
 916                         return -ENOMEM;
 917         }
 918
 919         s = source_new(e, SOURCE_QUIT);
 920         if (!s)
 921                 return -ENOMEM;
 922
 923         s->quit.callback = callback;
 924         s->userdata = userdata;
 925         s->quit.prioq_index = PRIOQ_IDX_NULL;
 926         s->enabled = SD_EVENT_ONESHOT;
 927
 928         r = prioq_put(s->event->quit, s, &s->quit.prioq_index);
 929         if (r < 0) {
 930                 source_free(s);
 931                 return r;
 932         }
 933
 934         *ret = s;
 935         return 0;
 936 }
 937
 938 sd_event_source* sd_event_source_ref(sd_event_source *s) {
 939         assert_return(s, NULL);
 940
 941         assert(s->n_ref >= 1);
 942         s->n_ref++;
 943
 944         return s;
 945 }
 946
 947 sd_event_source* sd_event_source_unref(sd_event_source *s) {
 948         assert_return(s, NULL);
 949
 950         assert(s->n_ref >= 1);
 951         s->n_ref--;
 952
 953         if (s->n_ref <= 0)
 954                 source_free(s);
 955
 956         return NULL;
 957 }
 958
 959 sd_event *sd_event_get(sd_event_source *s) {
 960         assert_return(s, NULL);
 961
 962         return s->event;
 963 }
 964
 965 int sd_event_source_get_pending(sd_event_source *s) {
 966         assert_return(s, -EINVAL);
 967         assert_return(s->type != SOURCE_QUIT, -EDOM);
 968         assert_return(s->event->state != SD_EVENT_FINISHED, -ESTALE);
 969         assert_return(!event_pid_changed(s->event), -ECHILD);
 970
 971         return s->pending;
 972 }
 973
 974 int sd_event_source_get_io_fd(sd_event_source *s) {
 975         assert_return(s, -EINVAL);
 976         assert_return(s->type == SOURCE_IO, -EDOM);
 977         assert_return(!event_pid_changed(s->event), -ECHILD);
 978
 979         return s->io.fd;
 980 }
 981
 982 int sd_event_source_get_io_events(sd_event_source *s, uint32_t* events) {
 983         assert_return(s, -EINVAL);
 984         assert_return(events, -EINVAL);
 985         assert_return(s->type == SOURCE_IO, -EDOM);
 986         assert_return(!event_pid_changed(s->event), -ECHILD);
 987
 988         *events = s->io.events;
 989         return 0;
 990 }
 991
 992 int sd_event_source_set_io_events(sd_event_source *s, uint32_t events) {
 993         int r;
 994
 995         assert_return(s, -EINVAL);
 996         assert_return(s->type == SOURCE_IO, -EDOM);
 997         assert_return(!(events & ~(EPOLLIN|EPOLLOUT|EPOLLRDHUP|EPOLLPRI|EPOLLERR|EPOLLHUP)), -EINVAL);
 998         assert_return(s->event->state != SD_EVENT_FINISHED, -ESTALE);
 999         assert_return(!event_pid_changed(s->event), -ECHILD);
1000
1001         if (s->io.events == events)
1002                 return 0;
1003
1004         if (s->enabled != SD_EVENT_OFF) {
1005                 r = source_io_register(s, s->enabled, events);
1006                 if (r < 0)
1007                         return r;
1008         }
1009
1010         s->io.events = events;
1011
1012         return 0;
1013 }
1014
1015 int sd_event_source_get_io_revents(sd_event_source *s, uint32_t* revents) {
1016         assert_return(s, -EINVAL);
1017         assert_return(revents, -EINVAL);
1018         assert_return(s->type == SOURCE_IO, -EDOM);
1019         assert_return(s->pending, -ENODATA);
1020         assert_return(!event_pid_changed(s->event), -ECHILD);
1021
1022         *revents = s->io.revents;
1023         return 0;
1024 }
1025
1026 int sd_event_source_get_signal(sd_event_source *s) {
1027         assert_return(s, -EINVAL);
1028         assert_return(s->type == SOURCE_SIGNAL, -EDOM);
1029         assert_return(!event_pid_changed(s->event), -ECHILD);
1030
1031         return s->signal.sig;
1032 }
1033
1034 int sd_event_source_get_priority(sd_event_source *s, int *priority) {
1035         assert_return(s, -EINVAL);
1036         assert_return(!event_pid_changed(s->event), -ECHILD);
1037
1038         return s->priority;
1039 }
1040
1041 int sd_event_source_set_priority(sd_event_source *s, int priority) {
1042         assert_return(s, -EINVAL);
1043         assert_return(s->event->state != SD_EVENT_FINISHED, -ESTALE);
1044         assert_return(!event_pid_changed(s->event), -ECHILD);
1045
1046         if (s->priority == priority)
1047                 return 0;
1048
1049         s->priority = priority;
1050
1051         if (s->pending)
1052                 prioq_reshuffle(s->event->pending, s, &s->pending_index);
1053
1054         if (s->prepare)
1055                 prioq_reshuffle(s->event->prepare, s, &s->prepare_index);
1056
1057         if (s->type == SOURCE_QUIT)
1058                 prioq_reshuffle(s->event->quit, s, &s->quit.prioq_index);
1059
1060         return 0;
1061 }
1062
1063 int sd_event_source_get_enabled(sd_event_source *s, int *m) {
1064         assert_return(s, -EINVAL);
1065         assert_return(m, -EINVAL);
1066         assert_return(!event_pid_changed(s->event), -ECHILD);
1067
1068         *m = s->enabled;
1069         return 0;
1070 }
1071
1072 int sd_event_source_set_enabled(sd_event_source *s, int m) {
1073         int r;
1074
1075         assert_return(s, -EINVAL);
1076         assert_return(m == SD_EVENT_OFF || m == SD_EVENT_ON || m == SD_EVENT_ONESHOT, -EINVAL);
1077         assert_return(s->event->state != SD_EVENT_FINISHED, -ESTALE);
1078         assert_return(!event_pid_changed(s->event), -ECHILD);
1079
1080         if (s->enabled == m)
1081                 return 0;
1082
1083         if (m == SD_EVENT_OFF) {
1084
1085                 switch (s->type) {
1086
1087                 case SOURCE_IO:
1088                         r = source_io_unregister(s);
1089                         if (r < 0)
1090                                 return r;
1091
1092                         s->enabled = m;
1093                         break;
1094
1095                 case SOURCE_MONOTONIC:
1096                         s->enabled = m;
1097                         prioq_reshuffle(s->event->monotonic_earliest, s, &s->time.earliest_index);
1098                         prioq_reshuffle(s->event->monotonic_latest, s, &s->time.latest_index);
1099                         break;
1100
1101                 case SOURCE_REALTIME:
1102                         s->enabled = m;
1103                         prioq_reshuffle(s->event->realtime_earliest, s, &s->time.earliest_index);
1104                         prioq_reshuffle(s->event->realtime_latest, s, &s->time.latest_index);
1105                         break;
1106
1107                 case SOURCE_SIGNAL:
1108                         s->enabled = m;
1109                         if (s->signal.sig != SIGCHLD || s->event->n_enabled_child_sources == 0) {
1110                                 assert_se(sigdelset(&s->event->sigset, s->signal.sig) == 0);
1111                                 event_update_signal_fd(s->event);
1112                         }
1113
1114                         break;
1115
1116                 case SOURCE_CHILD:
1117                         s->enabled = m;
1118
1119                         assert(s->event->n_enabled_child_sources > 0);
1120                         s->event->n_enabled_child_sources--;
1121
1122                         if (!s->event->signal_sources || !s->event->signal_sources[SIGCHLD]) {
1123                                 assert_se(sigdelset(&s->event->sigset, SIGCHLD) == 0);
1124                                 event_update_signal_fd(s->event);
1125                         }
1126
1127                         break;
1128
1129                 case SOURCE_QUIT:
1130                         s->enabled = m;
1131                         prioq_reshuffle(s->event->quit, s, &s->quit.prioq_index);
1132                         break;
1133
1134                 case SOURCE_DEFER:
1135                         s->enabled = m;
1136                         break;
1137                 }
1138
1139         } else {
1140                 switch (s->type) {
1141
1142                 case SOURCE_IO:
1143                         r = source_io_register(s, m, s->io.events);
1144                         if (r < 0)
1145                                 return r;
1146
1147                         s->enabled = m;
1148                         break;
1149
1150                 case SOURCE_MONOTONIC:
1151                         s->enabled = m;
1152                         prioq_reshuffle(s->event->monotonic_earliest, s, &s->time.earliest_index);
1153                         prioq_reshuffle(s->event->monotonic_latest, s, &s->time.latest_index);
1154                         break;
1155
1156                 case SOURCE_REALTIME:
1157                         s->enabled = m;
1158                         prioq_reshuffle(s->event->realtime_earliest, s, &s->time.earliest_index);
1159                         prioq_reshuffle(s->event->realtime_latest, s, &s->time.latest_index);
1160                         break;
1161
1162                 case SOURCE_SIGNAL:
1163                         s->enabled = m;
1164
1165                         if (s->signal.sig != SIGCHLD || s->event->n_enabled_child_sources == 0)  {
1166                                 assert_se(sigaddset(&s->event->sigset, s->signal.sig) == 0);
1167                                 event_update_signal_fd(s->event);
1168                         }
1169                         break;
1170
1171                 case SOURCE_CHILD:
1172                         s->enabled = m;
1173
1174                         if (s->enabled == SD_EVENT_OFF) {
1175                                 s->event->n_enabled_child_sources++;
1176
1177                                 if (!s->event->signal_sources || !s->event->signal_sources[SIGCHLD]) {
1178                                         assert_se(sigaddset(&s->event->sigset, SIGCHLD) == 0);
1179                                         event_update_signal_fd(s->event);
1180                                 }
1181                         }
1182                         break;
1183
1184                 case SOURCE_QUIT:
1185                         s->enabled = m;
1186                         prioq_reshuffle(s->event->quit, s, &s->quit.prioq_index);
1187                         break;
1188
1189                 case SOURCE_DEFER:
1190                         s->enabled = m;
1191                         break;
1192                 }
1193         }
1194
1195         if (s->pending)
1196                 prioq_reshuffle(s->event->pending, s, &s->pending_index);
1197
1198         if (s->prepare)
1199                 prioq_reshuffle(s->event->prepare, s, &s->prepare_index);
1200
1201         return 0;
1202 }
1203
1204 int sd_event_source_get_time(sd_event_source *s, uint64_t *usec) {
1205         assert_return(s, -EINVAL);
1206         assert_return(usec, -EINVAL);
1207         assert_return(s->type == SOURCE_REALTIME || s->type == SOURCE_MONOTONIC, -EDOM);
1208         assert_return(!event_pid_changed(s->event), -ECHILD);
1209
1210         *usec = s->time.next;
1211         return 0;
1212 }
1213
1214 int sd_event_source_set_time(sd_event_source *s, uint64_t usec) {
1215         assert_return(s, -EINVAL);
1216         assert_return(usec != (uint64_t) -1, -EINVAL);
1217         assert_return(s->type == SOURCE_REALTIME || s->type == SOURCE_MONOTONIC, -EDOM);
1218         assert_return(s->event->state != SD_EVENT_FINISHED, -ESTALE);
1219         assert_return(!event_pid_changed(s->event), -ECHILD);
1220
1221         if (s->time.next == usec)
1222                 return 0;
1223
1224         s->time.next = usec;
1225
1226         if (s->type == SOURCE_REALTIME) {
1227                 prioq_reshuffle(s->event->realtime_earliest, s, &s->time.earliest_index);
1228                 prioq_reshuffle(s->event->realtime_latest, s, &s->time.latest_index);
1229         } else {
1230                 prioq_reshuffle(s->event->monotonic_earliest, s, &s->time.earliest_index);
1231                 prioq_reshuffle(s->event->monotonic_latest, s, &s->time.latest_index);
1232         }
1233
1234         return 0;
1235 }
1236
1237 int sd_event_source_get_time_accuracy(sd_event_source *s, uint64_t *usec) {
1238         assert_return(s, -EINVAL);
1239         assert_return(usec, -EINVAL);
1240         assert_return(s->type == SOURCE_REALTIME || s->type == SOURCE_MONOTONIC, -EDOM);
1241         assert_return(!event_pid_changed(s->event), -ECHILD);
1242
1243         *usec = s->time.accuracy;
1244         return 0;
1245 }
1246
1247 int sd_event_source_set_time_accuracy(sd_event_source *s, uint64_t usec) {
1248         assert_return(s, -EINVAL);
1249         assert_return(usec != (uint64_t) -1, -EINVAL);
1250         assert_return(s->type == SOURCE_REALTIME || s->type == SOURCE_MONOTONIC, -EDOM);
1251         assert_return(s->event->state != SD_EVENT_FINISHED, -ESTALE);
1252         assert_return(!event_pid_changed(s->event), -ECHILD);
1253
1254         if (usec == 0)
1255                 usec = DEFAULT_ACCURACY_USEC;
1256
1257         if (s->time.accuracy == usec)
1258                 return 0;
1259
1260         s->time.accuracy = usec;
1261
1262         if (s->type == SOURCE_REALTIME)
1263                 prioq_reshuffle(s->event->realtime_latest, s, &s->time.latest_index);
1264         else
1265                 prioq_reshuffle(s->event->monotonic_latest, s, &s->time.latest_index);
1266
1267         return 0;
1268 }
1269
1270 int sd_event_source_get_child_pid(sd_event_source *s, pid_t *pid) {
1271         assert_return(s, -EINVAL);
1272         assert_return(pid, -EINVAL);
1273         assert_return(s->type == SOURCE_CHILD, -EDOM);
1274         assert_return(!event_pid_changed(s->event), -ECHILD);
1275
1276         *pid = s->child.pid;
1277         return 0;
1278 }
1279
1280 int sd_event_source_set_prepare(sd_event_source *s, sd_prepare_handler_t callback) {
1281         int r;
1282
1283         assert_return(s, -EINVAL);
1284         assert_return(s->type != SOURCE_QUIT, -EDOM);
1285         assert_return(s->event->state != SD_EVENT_FINISHED, -ESTALE);
1286         assert_return(!event_pid_changed(s->event), -ECHILD);
1287
1288         if (s->prepare == callback)
1289                 return 0;
1290
1291         if (callback && s->prepare) {
1292                 s->prepare = callback;
1293                 return 0;
1294         }
1295
1296         r = prioq_ensure_allocated(&s->event->prepare, prepare_prioq_compare);
1297         if (r < 0)
1298                 return r;
1299
1300         s->prepare = callback;
1301
1302         if (callback) {
1303                 r = prioq_put(s->event->prepare, s, &s->prepare_index);
1304                 if (r < 0)
1305                         return r;
1306         } else
1307                 prioq_remove(s->event->prepare, s, &s->prepare_index);
1308
1309         return 0;
1310 }
1311
1312 void* sd_event_source_get_userdata(sd_event_source *s) {
1313         assert_return(s, NULL);
1314
1315         return s->userdata;
1316 }
1317
1318 static usec_t sleep_between(sd_event *e, usec_t a, usec_t b) {
1319         usec_t c;
1320         assert(e);
1321         assert(a <= b);
1322
1323         if (a <= 0)
1324                 return 0;
1325
1326         if (b <= a + 1)
1327                 return a;
1328
1329         /*
1330           Find a good time to wake up again between times a and b. We
1331           have two goals here:
1332
1333           a) We want to wake up as seldom as possible, hence prefer
1334              later times over earlier times.
1335
1336           b) But if we have to wake up, then let's make sure to
1337              dispatch as much as possible on the entire system.
1338
1339           We implement this by waking up everywhere at the same time
1340           within any given second if we can, synchronised via the
1341           perturbation value determined from the boot ID. If we can't,
1342           then we try to find the same spot in every a 250ms
1343           step. Otherwise, we pick the last possible time to wake up.
1344         */
1345
1346         c = (b / USEC_PER_SEC) * USEC_PER_SEC + e->perturb;
1347         if (c >= b) {
1348                 if (_unlikely_(c < USEC_PER_SEC))
1349                         return b;
1350
1351                 c -= USEC_PER_SEC;
1352         }
1353
1354         if (c >= a)
1355                 return c;
1356
1357         c = (b / (USEC_PER_MSEC*250)) * (USEC_PER_MSEC*250) + (e->perturb % (USEC_PER_MSEC*250));
1358         if (c >= b) {
1359                 if (_unlikely_(c < USEC_PER_MSEC*250))
1360                         return b;
1361
1362                 c -= USEC_PER_MSEC*250;
1363         }
1364
1365         if (c >= a)
1366                 return c;
1367
1368         return b;
1369 }
1370
1371 static int event_arm_timer(
1372                 sd_event *e,
1373                 int timer_fd,
1374                 Prioq *earliest,
1375                 Prioq *latest,
1376                 usec_t *next) {
1377
1378         struct itimerspec its = {};
1379         sd_event_source *a, *b;
1380         usec_t t;
1381         int r;
1382
1383         assert_se(e);
1384         assert_se(next);
1385
1386         a = prioq_peek(earliest);
1387         if (!a || a->enabled == SD_EVENT_OFF) {
1388
1389                 if (*next == (usec_t) -1)
1390                         return 0;
1391
1392                 /* disarm */
1393                 r = timerfd_settime(timer_fd, TFD_TIMER_ABSTIME, &its, NULL);
1394                 if (r < 0)
1395                         return r;
1396
1397                 *next = (usec_t) -1;
1398
1399                 return 0;
1400         }
1401
1402         b = prioq_peek(latest);
1403         assert_se(b && b->enabled != SD_EVENT_OFF);
1404
1405         t = sleep_between(e, a->time.next, b->time.next + b->time.accuracy);
1406         if (*next == t)
1407                 return 0;
1408
1409         assert_se(timer_fd >= 0);
1410
1411         if (t == 0) {
1412                 /* We don' want to disarm here, just mean some time looooong ago. */
1413                 its.it_value.tv_sec = 0;
1414                 its.it_value.tv_nsec = 1;
1415         } else
1416                 timespec_store(&its.it_value, t);
1417
1418         r = timerfd_settime(timer_fd, TFD_TIMER_ABSTIME, &its, NULL);
1419         if (r < 0)
1420                 return r;
1421
1422         *next = t;
1423         return 0;
1424 }
1425
1426 static int process_io(sd_event *e, sd_event_source *s, uint32_t events) {
1427         assert(e);
1428         assert(s);
1429         assert(s->type == SOURCE_IO);
1430
1431         s->io.revents = events;
1432
1433         /*
1434            If this is a oneshot event source, then we added it to the
1435            epoll with EPOLLONESHOT, hence we know it's not registered
1436            anymore. We can save a syscall here...
1437         */
1438
1439         if (s->enabled == SD_EVENT_ONESHOT)
1440                 s->io.registered = false;
1441
1442         return source_set_pending(s, true);
1443 }
1444
1445 static int flush_timer(sd_event *e, int fd, uint32_t events, usec_t *next) {
1446         uint64_t x;
1447         ssize_t ss;
1448
1449         assert(e);
1450         assert(fd >= 0);
1451         assert(next);
1452
1453         assert_return(events == EPOLLIN, -EIO);
1454
1455         ss = read(fd, &x, sizeof(x));
1456         if (ss < 0) {
1457                 if (errno == EAGAIN || errno == EINTR)
1458                         return 0;
1459
1460                 return -errno;
1461         }
1462
1463         if (ss != sizeof(x))
1464                 return -EIO;
1465
1466         *next = (usec_t) -1;
1467
1468         return 0;
1469 }
1470
1471 static int process_timer(
1472                 sd_event *e,
1473                 usec_t n,
1474                 Prioq *earliest,
1475                 Prioq *latest) {
1476
1477         sd_event_source *s;
1478         int r;
1479
1480         assert(e);
1481
1482         for (;;) {
1483                 s = prioq_peek(earliest);
1484                 if (!s ||
1485                     s->time.next > n ||
1486                     s->enabled == SD_EVENT_OFF ||
1487                     s->pending)
1488                         break;
1489
1490                 r = source_set_pending(s, true);
1491                 if (r < 0)
1492                         return r;
1493
1494                 prioq_reshuffle(earliest, s, &s->time.earliest_index);
1495                 prioq_reshuffle(latest, s, &s->time.latest_index);
1496         }
1497
1498         return 0;
1499 }
1500
1501 static int process_child(sd_event *e) {
1502         sd_event_source *s;
1503         Iterator i;
1504         int r;
1505
1506         assert(e);
1507
1508         e->need_process_child = false;
1509
1510         /*
1511            So, this is ugly. We iteratively invoke waitid() with P_PID
1512            + WNOHANG for each PID we wait for, instead of using
1513            P_ALL. This is because we only want to get child
1514            information of very specific child processes, and not all
1515            of them. We might not have processed the SIGCHLD even of a
1516            previous invocation and we don't want to maintain a
1517            unbounded *per-child* event queue, hence we really don't
1518            want anything flushed out of the kernel's queue that we
1519            don't care about. Since this is O(n) this means that if you
1520            have a lot of processes you probably want to handle SIGCHLD
1521            yourself.
1522         */
1523
1524         HASHMAP_FOREACH(s, e->child_sources, i) {
1525                 assert(s->type == SOURCE_CHILD);
1526
1527                 if (s->pending)
1528                         continue;
1529
1530                 if (s->enabled == SD_EVENT_OFF)
1531                         continue;
1532
1533                 zero(s->child.siginfo);
1534                 r = waitid(P_PID, s->child.pid, &s->child.siginfo, WNOHANG|s->child.options);
1535                 if (r < 0)
1536                         return -errno;
1537
1538                 if (s->child.siginfo.si_pid != 0) {
1539                         r = source_set_pending(s, true);
1540                         if (r < 0)
1541                                 return r;
1542                 }
1543         }
1544
1545         return 0;
1546 }
1547
1548 static int process_signal(sd_event *e, uint32_t events) {
1549         bool read_one = false;
1550         int r;
1551
1552         assert(e);
1553         assert(e->signal_sources);
1554
1555         assert_return(events == EPOLLIN, -EIO);
1556
1557         for (;;) {
1558                 struct signalfd_siginfo si;
1559                 ssize_t ss;
1560                 sd_event_source *s;
1561
1562                 ss = read(e->signal_fd, &si, sizeof(si));
1563                 if (ss < 0) {
1564                         if (errno == EAGAIN || errno == EINTR)
1565                                 return read_one;
1566
1567                         return -errno;
1568                 }
1569
1570                 if (ss != sizeof(si))
1571                         return -EIO;
1572
1573                 read_one = true;
1574
1575                 s = e->signal_sources[si.ssi_signo];
1576                 if (si.ssi_signo == SIGCHLD) {
1577                         r = process_child(e);
1578                         if (r < 0)
1579                                 return r;
1580                         if (r > 0 || !s)
1581                                 continue;
1582                 } else
1583                         if (!s)
1584                                 return -EIO;
1585
1586                 s->signal.siginfo = si;
1587                 r = source_set_pending(s, true);
1588                 if (r < 0)
1589                         return r;
1590         }
1591
1592
1593         return 0;
1594 }
1595
1596 static int source_dispatch(sd_event_source *s) {
1597         int r = 0;
1598
1599         assert(s);
1600         assert(s->pending || s->type == SOURCE_QUIT);
1601
1602         if (s->type != SOURCE_DEFER && s->type != SOURCE_QUIT) {
1603                 r = source_set_pending(s, false);
1604                 if (r < 0)
1605                         return r;
1606         }
1607
1608         if (s->enabled == SD_EVENT_ONESHOT) {
1609                 r = sd_event_source_set_enabled(s, SD_EVENT_OFF);
1610                 if (r < 0)
1611                         return r;
1612         }
1613
1614         switch (s->type) {
1615
1616         case SOURCE_IO:
1617                 r = s->io.callback(s, s->io.fd, s->io.revents, s->userdata);
1618                 break;
1619
1620         case SOURCE_MONOTONIC:
1621                 r = s->time.callback(s, s->time.next, s->userdata);
1622                 break;
1623
1624         case SOURCE_REALTIME:
1625                 r = s->time.callback(s, s->time.next, s->userdata);
1626                 break;
1627
1628         case SOURCE_SIGNAL:
1629                 r = s->signal.callback(s, &s->signal.siginfo, s->userdata);
1630                 break;
1631
1632         case SOURCE_CHILD:
1633                 r = s->child.callback(s, &s->child.siginfo, s->userdata);
1634                 break;
1635
1636         case SOURCE_DEFER:
1637                 r = s->defer.callback(s, s->userdata);
1638                 break;
1639
1640         case SOURCE_QUIT:
1641                 r = s->quit.callback(s, s->userdata);
1642                 break;
1643         }
1644
1645         return r;
1646 }
1647
1648 static int event_prepare(sd_event *e) {
1649         int r;
1650
1651         assert(e);
1652
1653         for (;;) {
1654                 sd_event_source *s;
1655
1656                 s = prioq_peek(e->prepare);
1657                 if (!s || s->prepare_iteration == e->iteration || s->enabled == SD_EVENT_OFF)
1658                         break;
1659
1660                 s->prepare_iteration = e->iteration;
1661                 r = prioq_reshuffle(e->prepare, s, &s->prepare_index);
1662                 if (r < 0)
1663                         return r;
1664
1665                 assert(s->prepare);
1666                 r = s->prepare(s, s->userdata);
1667                 if (r < 0)
1668                         return r;
1669
1670         }
1671
1672         return 0;
1673 }
1674
1675 static int dispatch_quit(sd_event *e) {
1676         sd_event_source *p;
1677         int r;
1678
1679         assert(e);
1680
1681         p = prioq_peek(e->quit);
1682         if (!p || p->enabled == SD_EVENT_OFF) {
1683                 e->state = SD_EVENT_FINISHED;
1684                 return 0;
1685         }
1686
1687         sd_event_ref(e);
1688         e->iteration++;
1689         e->state = SD_EVENT_QUITTING;
1690
1691         r = source_dispatch(p);
1692
1693         e->state = SD_EVENT_PASSIVE;
1694         sd_event_unref(e);
1695
1696         return r;
1697 }
1698
1699 static sd_event_source* event_next_pending(sd_event *e) {
1700         sd_event_source *p;
1701
1702         assert(e);
1703
1704         p = prioq_peek(e->pending);
1705         if (!p)
1706                 return NULL;
1707
1708         if (p->enabled == SD_EVENT_OFF)
1709                 return NULL;
1710
1711         return p;
1712 }
1713
1714 int sd_event_run(sd_event *e, uint64_t timeout) {
1715         struct epoll_event ev_queue[EPOLL_QUEUE_MAX];
1716         sd_event_source *p;
1717         int r, i, m;
1718
1719         assert_return(e, -EINVAL);
1720         assert_return(!event_pid_changed(e), -ECHILD);
1721         assert_return(e->state != SD_EVENT_FINISHED, -ESTALE);
1722         assert_return(e->state == SD_EVENT_PASSIVE, -EBUSY);
1723
1724         if (e->quit_requested)
1725                 return dispatch_quit(e);
1726
1727         sd_event_ref(e);
1728         e->iteration++;
1729         e->state = SD_EVENT_RUNNING;
1730
1731         r = event_prepare(e);
1732         if (r < 0)
1733                 goto finish;
1734
1735         if (event_next_pending(e) || e->need_process_child)
1736                 timeout = 0;
1737
1738         if (timeout > 0) {
1739                 r = event_arm_timer(e, e->monotonic_fd, e->monotonic_earliest, e->monotonic_latest, &e->monotonic_next);
1740                 if (r < 0)
1741                         goto finish;
1742
1743                 r = event_arm_timer(e, e->realtime_fd, e->realtime_earliest, e->realtime_latest, &e->realtime_next);
1744                 if (r < 0)
1745                         goto finish;
1746         }
1747
1748         m = epoll_wait(e->epoll_fd, ev_queue, EPOLL_QUEUE_MAX,
1749                        timeout == (uint64_t) -1 ? -1 : (int) ((timeout + USEC_PER_MSEC - 1) / USEC_PER_MSEC));
1750         if (m < 0) {
1751                 r = errno == EAGAIN || errno == EINTR ? 0 : -errno;
1752                 goto finish;
1753         }
1754
1755         dual_timestamp_get(&e->timestamp);
1756
1757         for (i = 0; i < m; i++) {
1758
1759                 if (ev_queue[i].data.ptr == INT_TO_PTR(SOURCE_MONOTONIC))
1760                         r = flush_timer(e, e->monotonic_fd, ev_queue[i].events, &e->monotonic_next);
1761                 else if (ev_queue[i].data.ptr == INT_TO_PTR(SOURCE_REALTIME))
1762                         r = flush_timer(e, e->realtime_fd, ev_queue[i].events, &e->realtime_next);
1763                 else if (ev_queue[i].data.ptr == INT_TO_PTR(SOURCE_SIGNAL))
1764                         r = process_signal(e, ev_queue[i].events);
1765                 else
1766                         r = process_io(e, ev_queue[i].data.ptr, ev_queue[i].events);
1767
1768                 if (r < 0)
1769                         goto finish;
1770         }
1771
1772         r = process_timer(e, e->timestamp.monotonic, e->monotonic_earliest, e->monotonic_latest);
1773         if (r < 0)
1774                 goto finish;
1775
1776         r = process_timer(e, e->timestamp.realtime, e->realtime_earliest, e->realtime_latest);
1777         if (r < 0)
1778                 goto finish;
1779
1780         if (e->need_process_child) {
1781                 r = process_child(e);
1782                 if (r < 0)
1783                         goto finish;
1784         }
1785
1786         p = event_next_pending(e);
1787         if (!p) {
1788                 r = 0;
1789                 goto finish;
1790         }
1791
1792         r = source_dispatch(p);
1793
1794 finish:
1795         e->state = SD_EVENT_PASSIVE;
1796         sd_event_unref(e);
1797
1798         return r;
1799 }
1800
1801 int sd_event_loop(sd_event *e) {
1802         int r;
1803
1804         assert_return(e, -EINVAL);
1805         assert_return(!event_pid_changed(e), -ECHILD);
1806         assert_return(e->state == SD_EVENT_PASSIVE, -EBUSY);
1807
1808         sd_event_ref(e);
1809
1810         while (e->state != SD_EVENT_FINISHED) {
1811                 r = sd_event_run(e, (uint64_t) -1);
1812                 if (r < 0)
1813                         goto finish;
1814         }
1815
1816         r = 0;
1817
1818 finish:
1819         sd_event_unref(e);
1820         return r;
1821 }
1822
1823 int sd_event_get_state(sd_event *e) {
1824         assert_return(e, -EINVAL);
1825         assert_return(!event_pid_changed(e), -ECHILD);
1826
1827         return e->state;
1828 }
1829
1830 int sd_event_get_quit(sd_event *e) {
1831         assert_return(e, -EINVAL);
1832         assert_return(!event_pid_changed(e), -ECHILD);
1833
1834         return e->quit_requested;
1835 }
1836
1837 int sd_event_request_quit(sd_event *e) {
1838         assert_return(e, -EINVAL);
1839         assert_return(e->state != SD_EVENT_FINISHED, -ESTALE);
1840         assert_return(!event_pid_changed(e), -ECHILD);
1841
1842         e->quit_requested = true;
1843         return 0;
1844 }
1845
1846 int sd_event_get_now_realtime(sd_event *e, uint64_t *usec) {
1847         assert_return(e, -EINVAL);
1848         assert_return(usec, -EINVAL);
1849         assert_return(dual_timestamp_is_set(&e->timestamp), -ENODATA);
1850         assert_return(!event_pid_changed(e), -ECHILD);
1851
1852         *usec = e->timestamp.realtime;
1853         return 0;
1854 }
1855
1856 int sd_event_get_now_monotonic(sd_event *e, uint64_t *usec) {
1857         assert_return(e, -EINVAL);
1858         assert_return(usec, -EINVAL);
1859         assert_return(dual_timestamp_is_set(&e->timestamp), -ENODATA);
1860         assert_return(!event_pid_changed(e), -ECHILD);
1861
1862         *usec = e->timestamp.monotonic;
1863         return 0;
1864 }