code.delx.au - pulseaudio/blob - src/pulsecore/rtpoll.c

   1 /* $Id$ */
   2
   3 /***
   4   This file is part of PulseAudio.
   5
   6   Copyright 2004-2006 Lennart Poettering
   7   Copyright 2006 Pierre Ossman <ossman@cendio.se> for Cendio AB
   8
   9   PulseAudio is free software; you can redistribute it and/or modify
  10   it under the terms of the GNU Lesser General Public License as
  11   published by the Free Software Foundation; either version 2.1 of the
  12   License, or (at your option) any later version.
  13
  14   PulseAudio is distributed in the hope that it will be useful, but
  15   WITHOUT ANY WARRANTY; without even the implied warranty of
  16   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  17   Lesser General Public License for more details.
  18
  19   You should have received a copy of the GNU Lesser General Public
  20   License along with PulseAudio; if not, write to the Free Software
  21   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
  22   USA.
  23 ***/
  24
  25 #ifdef HAVE_CONFIG_H
  26 #include <config.h>
  27 #endif
  28
  29 #include <sys/types.h>
  30 #include <stdio.h>
  31 #include <signal.h>
  32 #include <string.h>
  33 #include <errno.h>
  34
  35 #ifdef __linux__
  36 #include <sys/utsname.h>
  37 #endif
  38
  39 #ifdef HAVE_POLL_H
  40 #include <poll.h>
  41 #else
  42 #include <pulsecore/poll.h>
  43 #endif
  44
  45 #include <pulse/xmalloc.h>
  46 #include <pulse/timeval.h>
  47
  48 #include <pulsecore/core-error.h>
  49 #include <pulsecore/rtclock.h>
  50 #include <pulsecore/macro.h>
  51 #include <pulsecore/llist.h>
  52 #include <pulsecore/rtsig.h>
  53 #include <pulsecore/flist.h>
  54 #include <pulsecore/core-util.h>
  55
  56 #include <pulsecore/winsock.h>
  57
  58 #include "rtpoll.h"
  59
  60 struct pa_rtpoll {
  61     struct pollfd *pollfd, *pollfd2;
  62     unsigned n_pollfd_alloc, n_pollfd_used;
  63
  64     pa_bool_t timer_enabled;
  65     struct timeval next_elapse;
  66     pa_usec_t period;
  67
  68     pa_bool_t scan_for_dead;
  69     pa_bool_t running, installed, rebuild_needed, quit;
  70
  71 #ifdef HAVE_PPOLL
  72     int rtsig;
  73     sigset_t sigset_unblocked;
  74     timer_t timer;
  75 #ifdef __linux__
  76     pa_bool_t dont_use_ppoll;
  77 #endif
  78 #endif
  79
  80     PA_LLIST_HEAD(pa_rtpoll_item, items);
  81 };
  82
  83 struct pa_rtpoll_item {
  84     pa_rtpoll *rtpoll;
  85     pa_bool_t dead;
  86
  87     pa_rtpoll_priority_t priority;
  88
  89     struct pollfd *pollfd;
  90     unsigned n_pollfd;
  91
  92     int (*work_cb)(pa_rtpoll_item *i);
  93     int (*before_cb)(pa_rtpoll_item *i);
  94     void (*after_cb)(pa_rtpoll_item *i);
  95     void *userdata;
  96
  97     PA_LLIST_FIELDS(pa_rtpoll_item);
  98 };
  99
 100 PA_STATIC_FLIST_DECLARE(items, 0, pa_xfree);
 101
 102 static void signal_handler_noop(int s) { }
 103
 104 pa_rtpoll *pa_rtpoll_new(void) {
 105     pa_rtpoll *p;
 106
 107     p = pa_xnew(pa_rtpoll, 1);
 108
 109 #ifdef HAVE_PPOLL
 110
 111 #ifdef __linux__
 112     /* ppoll is broken on Linux < 2.6.16 */
 113     p->dont_use_ppoll = FALSE;
 114
 115     {
 116         struct utsname u;
 117         unsigned major, minor, micro;
 118
 119         pa_assert_se(uname(&u) == 0);
 120
 121         if (sscanf(u.release, "%u.%u.%u", &major, &minor, &micro) != 3 ||
 122             (major < 2) ||
 123             (major == 2 && minor < 6) ||
 124             (major == 2 && minor == 6 && micro < 16))
 125
 126             p->dont_use_ppoll = TRUE;
 127     }
 128
 129 #endif
 130
 131     p->rtsig = -1;
 132     sigemptyset(&p->sigset_unblocked);
 133     p->timer = (timer_t) -1;
 134
 135 #endif
 136
 137     p->n_pollfd_alloc = 32;
 138     p->pollfd = pa_xnew(struct pollfd, p->n_pollfd_alloc);
 139     p->pollfd2 = pa_xnew(struct pollfd, p->n_pollfd_alloc);
 140     p->n_pollfd_used = 0;
 141
 142     p->period = 0;
 143     memset(&p->next_elapse, 0, sizeof(p->next_elapse));
 144     p->timer_enabled = FALSE;
 145
 146     p->running = FALSE;
 147     p->installed = FALSE;
 148     p->scan_for_dead = FALSE;
 149     p->rebuild_needed = FALSE;
 150     p->quit = FALSE;
 151
 152     PA_LLIST_HEAD_INIT(pa_rtpoll_item, p->items);
 153
 154     return p;
 155 }
 156
 157 void pa_rtpoll_install(pa_rtpoll *p) {
 158     pa_assert(p);
 159     pa_assert(!p->installed);
 160
 161     p->installed = 1;
 162
 163 #ifdef HAVE_PPOLL
 164     if (p->dont_use_ppoll)
 165         return;
 166
 167     if ((p->rtsig = pa_rtsig_get_for_thread()) < 0) {
 168         pa_log_warn("Failed to reserve POSIX realtime signal.");
 169         return;
 170     }
 171
 172     pa_log_debug("Acquired POSIX realtime signal %s", pa_sig2str(p->rtsig));
 173
 174     {
 175         sigset_t ss;
 176         struct sigaction sa;
 177
 178         pa_assert_se(sigemptyset(&ss) == 0);
 179         pa_assert_se(sigaddset(&ss, p->rtsig) == 0);
 180         pa_assert_se(pthread_sigmask(SIG_BLOCK, &ss, &p->sigset_unblocked) == 0);
 181         pa_assert_se(sigdelset(&p->sigset_unblocked, p->rtsig) == 0);
 182
 183         memset(&sa, 0, sizeof(sa));
 184         sa.sa_handler = signal_handler_noop;
 185         pa_assert_se(sigemptyset(&sa.sa_mask) == 0);
 186
 187         pa_assert_se(sigaction(p->rtsig, &sa, NULL) == 0);
 188
 189         /* We never reset the signal handler. Why should we? */
 190     }
 191
 192 #endif
 193 }
 194
 195 static void rtpoll_rebuild(pa_rtpoll *p) {
 196
 197     struct pollfd *e, *t;
 198     pa_rtpoll_item *i;
 199     int ra = 0;
 200
 201     pa_assert(p);
 202
 203     p->rebuild_needed = FALSE;
 204
 205     if (p->n_pollfd_used > p->n_pollfd_alloc) {
 206         /* Hmm, we have to allocate some more space */
 207         p->n_pollfd_alloc = p->n_pollfd_used * 2;
 208         p->pollfd2 = pa_xrealloc(p->pollfd2, p->n_pollfd_alloc * sizeof(struct pollfd));
 209         ra = 1;
 210     }
 211
 212     e = p->pollfd2;
 213
 214     for (i = p->items; i; i = i->next) {
 215
 216         if (i->n_pollfd > 0)  {
 217             size_t l = i->n_pollfd * sizeof(struct pollfd);
 218
 219             if (i->pollfd)
 220                 memcpy(e, i->pollfd, l);
 221             else
 222                 memset(e, 0, l);
 223
 224             i->pollfd = e;
 225         } else
 226             i->pollfd = NULL;
 227
 228         e += i->n_pollfd;
 229     }
 230
 231     pa_assert((unsigned) (e - p->pollfd2) == p->n_pollfd_used);
 232     t = p->pollfd;
 233     p->pollfd = p->pollfd2;
 234     p->pollfd2 = t;
 235
 236     if (ra)
 237         p->pollfd2 = pa_xrealloc(p->pollfd2, p->n_pollfd_alloc * sizeof(struct pollfd));
 238
 239 }
 240
 241 static void rtpoll_item_destroy(pa_rtpoll_item *i) {
 242     pa_rtpoll *p;
 243
 244     pa_assert(i);
 245
 246     p = i->rtpoll;
 247
 248     PA_LLIST_REMOVE(pa_rtpoll_item, p->items, i);
 249
 250     p->n_pollfd_used -= i->n_pollfd;
 251
 252     if (pa_flist_push(PA_STATIC_FLIST_GET(items), i) < 0)
 253         pa_xfree(i);
 254
 255     p->rebuild_needed = TRUE;
 256 }
 257
 258 void pa_rtpoll_free(pa_rtpoll *p) {
 259     pa_assert(p);
 260
 261     while (p->items)
 262         rtpoll_item_destroy(p->items);
 263
 264     pa_xfree(p->pollfd);
 265     pa_xfree(p->pollfd2);
 266
 267 #ifdef HAVE_PPOLL
 268     if (p->timer != (timer_t) -1)
 269         timer_delete(p->timer);
 270 #endif
 271
 272     pa_xfree(p);
 273 }
 274
 275 static void reset_revents(pa_rtpoll_item *i) {
 276     struct pollfd *f;
 277     unsigned n;
 278
 279     pa_assert(i);
 280
 281     if (!(f = pa_rtpoll_item_get_pollfd(i, &n)))
 282         return;
 283
 284     for (; n > 0; n--)
 285         f[n-1].revents = 0;
 286 }
 287
 288 static void reset_all_revents(pa_rtpoll *p) {
 289     pa_rtpoll_item *i;
 290
 291     pa_assert(p);
 292
 293     for (i = p->items; i; i = i->next) {
 294
 295         if (i->dead)
 296             continue;
 297
 298         reset_revents(i);
 299     }
 300 }
 301
 302 int pa_rtpoll_run(pa_rtpoll *p, pa_bool_t wait) {
 303     pa_rtpoll_item *i;
 304     int r = 0;
 305     struct timeval timeout;
 306
 307     pa_assert(p);
 308     pa_assert(!p->running);
 309     pa_assert(p->installed);
 310
 311     p->running = TRUE;
 312
 313     /* First, let's do some work */
 314     for (i = p->items; i && i->priority < PA_RTPOLL_NEVER; i = i->next) {
 315         int k;
 316
 317         if (i->dead)
 318             continue;
 319
 320         if (!i->work_cb)
 321             continue;
 322
 323         if (p->quit)
 324             goto finish;
 325
 326         if ((k = i->work_cb(i)) != 0) {
 327             if (k < 0)
 328                 r = k;
 329
 330             goto finish;
 331         }
 332     }
 333
 334     /* Now let's prepare for entering the sleep */
 335     for (i = p->items; i && i->priority < PA_RTPOLL_NEVER; i = i->next) {
 336         int k = 0;
 337
 338         if (i->dead)
 339             continue;
 340
 341         if (!i->before_cb)
 342             continue;
 343
 344         if (p->quit || (k = i->before_cb(i)) != 0) {
 345
 346             /* Hmm, this one doesn't let us enter the poll, so rewind everything */
 347
 348             for (i = i->prev; i; i = i->prev) {
 349
 350                 if (i->dead)
 351                     continue;
 352
 353                 if (!i->after_cb)
 354                     continue;
 355
 356                 i->after_cb(i);
 357             }
 358
 359             if (k < 0)
 360                 r = k;
 361
 362             goto finish;
 363         }
 364     }
 365
 366     if (p->rebuild_needed)
 367         rtpoll_rebuild(p);
 368
 369     /* Calculate timeout */
 370     if (!wait || p->quit) {
 371         timeout.tv_sec = 0;
 372         timeout.tv_usec = 0;
 373     } else if (p->timer_enabled) {
 374         struct timeval now;
 375         pa_rtclock_get(&now);
 376
 377         memset(&timeout, 0, sizeof(timeout));
 378         if (pa_timeval_cmp(&p->next_elapse, &now) > 0)
 379             pa_timeval_add(&timeout, pa_timeval_diff(&p->next_elapse, &now));
 380     }
 381
 382     /* OK, now let's sleep */
 383 #ifdef HAVE_PPOLL
 384
 385 #ifdef __linux__
 386     if (!p->dont_use_ppoll)
 387 #endif
 388     {
 389         struct timespec ts;
 390         ts.tv_sec = timeout.tv_sec;
 391         ts.tv_nsec = timeout.tv_usec * 1000;
 392         r = ppoll(p->pollfd, p->n_pollfd_used, p->timer_enabled ? &ts : NULL, p->rtsig < 0 ? NULL : &p->sigset_unblocked);
 393     }
 394 #ifdef __linux__
 395     else
 396 #endif
 397
 398 #endif
 399         r = poll(p->pollfd, p->n_pollfd_used, p->timer_enabled ? (timeout.tv_sec*1000) + (timeout.tv_usec / 1000) : -1);
 400
 401     if (r < 0) {
 402         if (errno == EAGAIN || errno == EINTR)
 403             r = 0;
 404         else
 405             pa_log_error("poll(): %s", pa_cstrerror(errno));
 406
 407         reset_all_revents(p);
 408     }
 409
 410     if (p->timer_enabled) {
 411         if (p->period > 0) {
 412             struct timeval now;
 413             pa_rtclock_get(&now);
 414
 415             pa_timeval_add(&p->next_elapse, p->period);
 416
 417             /* Guarantee that the next timeout will happen in the future */
 418             if (pa_timeval_cmp(&p->next_elapse, &now) < 0)
 419                 pa_timeval_add(&p->next_elapse, (pa_timeval_diff(&now, &p->next_elapse) / p->period + 1) * p->period);
 420
 421         } else
 422             p->timer_enabled = FALSE;
 423     }
 424
 425     /* Let's tell everyone that we left the sleep */
 426     for (i = p->items; i && i->priority < PA_RTPOLL_NEVER; i = i->next) {
 427
 428         if (i->dead)
 429             continue;
 430
 431         if (!i->after_cb)
 432             continue;
 433
 434         i->after_cb(i);
 435     }
 436
 437 finish:
 438
 439     p->running = FALSE;
 440
 441     if (p->scan_for_dead) {
 442         pa_rtpoll_item *n;
 443
 444         p->scan_for_dead = FALSE;
 445
 446         for (i = p->items; i; i = n) {
 447             n = i->next;
 448
 449             if (i->dead)
 450                 rtpoll_item_destroy(i);
 451         }
 452     }
 453
 454     return r < 0 ? r : !p->quit;
 455 }
 456
 457 static void update_timer(pa_rtpoll *p) {
 458     pa_assert(p);
 459
 460 #ifdef HAVE_PPOLL
 461
 462 #ifdef __linux__
 463     if (!p->dont_use_ppoll) {
 464 #endif
 465
 466         if (p->timer == (timer_t) -1) {
 467             struct sigevent se;
 468
 469             memset(&se, 0, sizeof(se));
 470             se.sigev_notify = SIGEV_SIGNAL;
 471             se.sigev_signo = p->rtsig;
 472
 473             if (timer_create(CLOCK_MONOTONIC, &se, &p->timer) < 0)
 474                 if (timer_create(CLOCK_REALTIME, &se, &p->timer) < 0) {
 475                     pa_log_warn("Failed to allocate POSIX timer: %s", pa_cstrerror(errno));
 476                     p->timer = (timer_t) -1;
 477                 }
 478         }
 479
 480         if (p->timer != (timer_t) -1) {
 481             struct itimerspec its;
 482             memset(&its, 0, sizeof(its));
 483
 484             if (p->timer_enabled) {
 485                 its.it_value.tv_sec = p->next_elapse.tv_sec;
 486                 its.it_value.tv_nsec = p->next_elapse.tv_usec*1000;
 487
 488                 /* Make sure that 0,0 is not understood as
 489                  * "disarming" */
 490                 if (its.it_value.tv_sec == 0)
 491                     its.it_value.tv_nsec = 1;
 492
 493                 if (p->period > 0) {
 494                     struct timeval tv;
 495                     pa_timeval_store(&tv, p->period);
 496                     its.it_interval.tv_sec = tv.tv_sec;
 497                     its.it_interval.tv_nsec = tv.tv_usec*1000;
 498                 }
 499             }
 500
 501             pa_assert_se(timer_settime(p->timer, TIMER_ABSTIME, &its, NULL) == 0);
 502         }
 503
 504 #ifdef __linux__
 505     }
 506 #endif
 507
 508 #endif
 509 }
 510
 511 void pa_rtpoll_set_timer_absolute(pa_rtpoll *p, const struct timeval *ts) {
 512     pa_assert(p);
 513     pa_assert(ts);
 514
 515     p->next_elapse = *ts;
 516     p->period = 0;
 517     p->timer_enabled = TRUE;
 518
 519     update_timer(p);
 520 }
 521
 522 void pa_rtpoll_set_timer_periodic(pa_rtpoll *p, pa_usec_t usec) {
 523     pa_assert(p);
 524
 525     p->period = usec;
 526     pa_rtclock_get(&p->next_elapse);
 527     pa_timeval_add(&p->next_elapse, usec);
 528     p->timer_enabled = TRUE;
 529
 530     update_timer(p);
 531 }
 532
 533 void pa_rtpoll_set_timer_relative(pa_rtpoll *p, pa_usec_t usec) {
 534     pa_assert(p);
 535
 536     p->period = 0;
 537     pa_rtclock_get(&p->next_elapse);
 538     pa_timeval_add(&p->next_elapse, usec);
 539     p->timer_enabled = TRUE;
 540
 541     update_timer(p);
 542 }
 543
 544 void pa_rtpoll_set_timer_disabled(pa_rtpoll *p) {
 545     pa_assert(p);
 546
 547     p->period = 0;
 548     memset(&p->next_elapse, 0, sizeof(p->next_elapse));
 549     p->timer_enabled = FALSE;
 550
 551     update_timer(p);
 552 }
 553
 554 pa_rtpoll_item *pa_rtpoll_item_new(pa_rtpoll *p, pa_rtpoll_priority_t prio, unsigned n_fds) {
 555     pa_rtpoll_item *i, *j, *l = NULL;
 556
 557     pa_assert(p);
 558
 559     if (!(i = pa_flist_pop(PA_STATIC_FLIST_GET(items))))
 560         i = pa_xnew(pa_rtpoll_item, 1);
 561
 562     i->rtpoll = p;
 563     i->dead = FALSE;
 564     i->n_pollfd = n_fds;
 565     i->pollfd = NULL;
 566     i->priority = prio;
 567
 568     i->userdata = NULL;
 569     i->before_cb = NULL;
 570     i->after_cb = NULL;
 571     i->work_cb = NULL;
 572
 573     for (j = p->items; j; j = j->next) {
 574         if (prio <= j->priority)
 575             break;
 576
 577         l = j;
 578     }
 579
 580     PA_LLIST_INSERT_AFTER(pa_rtpoll_item, p->items, j ? j->prev : l, i);
 581
 582     if (n_fds > 0) {
 583         p->rebuild_needed = 1;
 584         p->n_pollfd_used += n_fds;
 585     }
 586
 587     return i;
 588 }
 589
 590 void pa_rtpoll_item_free(pa_rtpoll_item *i) {
 591     pa_assert(i);
 592
 593     if (i->rtpoll->running) {
 594         i->dead = TRUE;
 595         i->rtpoll->scan_for_dead = TRUE;
 596         return;
 597     }
 598
 599     rtpoll_item_destroy(i);
 600 }
 601
 602 struct pollfd *pa_rtpoll_item_get_pollfd(pa_rtpoll_item *i, unsigned *n_fds) {
 603     pa_assert(i);
 604
 605     if (i->n_pollfd > 0)
 606         if (i->rtpoll->rebuild_needed)
 607             rtpoll_rebuild(i->rtpoll);
 608
 609     if (n_fds)
 610         *n_fds = i->n_pollfd;
 611
 612     return i->pollfd;
 613 }
 614
 615 void pa_rtpoll_item_set_before_callback(pa_rtpoll_item *i, int (*before_cb)(pa_rtpoll_item *i)) {
 616     pa_assert(i);
 617     pa_assert(i->priority < PA_RTPOLL_NEVER);
 618
 619     i->before_cb = before_cb;
 620 }
 621
 622 void pa_rtpoll_item_set_after_callback(pa_rtpoll_item *i, void (*after_cb)(pa_rtpoll_item *i)) {
 623     pa_assert(i);
 624     pa_assert(i->priority < PA_RTPOLL_NEVER);
 625
 626     i->after_cb = after_cb;
 627 }
 628
 629 void pa_rtpoll_item_set_work_callback(pa_rtpoll_item *i, int (*work_cb)(pa_rtpoll_item *i)) {
 630     pa_assert(i);
 631     pa_assert(i->priority < PA_RTPOLL_NEVER);
 632
 633     i->work_cb = work_cb;
 634 }
 635
 636 void pa_rtpoll_item_set_userdata(pa_rtpoll_item *i, void *userdata) {
 637     pa_assert(i);
 638
 639     i->userdata = userdata;
 640 }
 641
 642 void* pa_rtpoll_item_get_userdata(pa_rtpoll_item *i) {
 643     pa_assert(i);
 644
 645     return i->userdata;
 646 }
 647
 648 static int fdsem_before(pa_rtpoll_item *i) {
 649
 650     if (pa_fdsem_before_poll(i->userdata) < 0)
 651         return 1; /* 1 means immediate restart of the loop */
 652
 653     return 0;
 654 }
 655
 656 static void fdsem_after(pa_rtpoll_item *i) {
 657     pa_assert(i);
 658
 659     pa_assert((i->pollfd[0].revents & ~POLLIN) == 0);
 660     pa_fdsem_after_poll(i->userdata);
 661 }
 662
 663 pa_rtpoll_item *pa_rtpoll_item_new_fdsem(pa_rtpoll *p, pa_rtpoll_priority_t prio, pa_fdsem *f) {
 664     pa_rtpoll_item *i;
 665     struct pollfd *pollfd;
 666
 667     pa_assert(p);
 668     pa_assert(f);
 669
 670     i = pa_rtpoll_item_new(p, prio, 1);
 671
 672     pollfd = pa_rtpoll_item_get_pollfd(i, NULL);
 673
 674     pollfd->fd = pa_fdsem_get(f);
 675     pollfd->events = POLLIN;
 676
 677     i->before_cb = fdsem_before;
 678     i->after_cb = fdsem_after;
 679     i->userdata = f;
 680
 681     return i;
 682 }
 683
 684 static int asyncmsgq_before(pa_rtpoll_item *i) {
 685     pa_assert(i);
 686
 687     if (pa_asyncmsgq_before_poll(i->userdata) < 0)
 688         return 1; /* 1 means immediate restart of the loop */
 689
 690     return 0;
 691 }
 692
 693 static void asyncmsgq_after(pa_rtpoll_item *i) {
 694     pa_assert(i);
 695
 696     pa_assert((i->pollfd[0].revents & ~POLLIN) == 0);
 697     pa_asyncmsgq_after_poll(i->userdata);
 698 }
 699
 700 static int asyncmsgq_work(pa_rtpoll_item *i) {
 701     pa_msgobject *object;
 702     int code;
 703     void *data;
 704     pa_memchunk chunk;
 705     int64_t offset;
 706
 707     pa_assert(i);
 708
 709     if (pa_asyncmsgq_get(i->userdata, &object, &code, &data, &offset, &chunk, 0) == 0) {
 710         int ret;
 711
 712         if (!object && code == PA_MESSAGE_SHUTDOWN) {
 713             pa_asyncmsgq_done(i->userdata, 0);
 714             pa_rtpoll_quit(i->rtpoll);
 715             return 1;
 716         }
 717
 718         ret = pa_asyncmsgq_dispatch(object, code, data, offset, &chunk);
 719         pa_asyncmsgq_done(i->userdata, ret);
 720         return 1;
 721     }
 722
 723     return 0;
 724 }
 725
 726 pa_rtpoll_item *pa_rtpoll_item_new_asyncmsgq(pa_rtpoll *p, pa_rtpoll_priority_t prio, pa_asyncmsgq *q) {
 727     pa_rtpoll_item *i;
 728     struct pollfd *pollfd;
 729
 730     pa_assert(p);
 731     pa_assert(q);
 732
 733     i = pa_rtpoll_item_new(p, prio, 1);
 734
 735     pollfd = pa_rtpoll_item_get_pollfd(i, NULL);
 736     pollfd->fd = pa_asyncmsgq_get_fd(q);
 737     pollfd->events = POLLIN;
 738
 739     i->before_cb = asyncmsgq_before;
 740     i->after_cb = asyncmsgq_after;
 741     i->work_cb = asyncmsgq_work;
 742     i->userdata = q;
 743
 744     return i;
 745 }
 746
 747 void pa_rtpoll_quit(pa_rtpoll *p) {
 748     pa_assert(p);
 749
 750     p->quit = TRUE;
 751 }