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

   1 /***
   2   This file is part of PulseAudio.
   3
   4   Copyright 2004-2006 Lennart Poettering
   5   Copyright 2006 Pierre Ossman <ossman@cendio.se> for Cendio AB
   6
   7   PulseAudio is free software; you can redistribute it and/or modify
   8   it under the terms of the GNU Lesser General Public License as
   9   published by the Free Software Foundation; either version 2.1 of the
  10   License, or (at your option) any later version.
  11
  12   PulseAudio is distributed in the hope that it will be useful, but
  13   WITHOUT ANY WARRANTY; without even the implied warranty of
  14   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  15   Lesser General Public License for more details.
  16
  17   You should have received a copy of the GNU Lesser General Public
  18   License along with PulseAudio; if not, write to the Free Software
  19   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
  20   USA.
  21 ***/
  22
  23 #ifdef HAVE_CONFIG_H
  24 #include <config.h>
  25 #endif
  26
  27 #include <sys/types.h>
  28 #include <stdio.h>
  29 #include <signal.h>
  30 #include <string.h>
  31 #include <errno.h>
  32
  33 #ifdef __linux__
  34 #include <sys/utsname.h>
  35 #endif
  36
  37 #ifdef HAVE_POLL_H
  38 #include <poll.h>
  39 #else
  40 #include <pulsecore/poll.h>
  41 #endif
  42
  43 #include <pulse/xmalloc.h>
  44 #include <pulse/timeval.h>
  45
  46 #include <pulsecore/core-error.h>
  47 #include <pulsecore/rtclock.h>
  48 #include <pulsecore/macro.h>
  49 #include <pulsecore/llist.h>
  50 #include <pulsecore/rtsig.h>
  51 #include <pulsecore/flist.h>
  52 #include <pulsecore/core-util.h>
  53
  54 #include <pulsecore/winsock.h>
  55
  56 #include "rtpoll.h"
  57
  58 struct pa_rtpoll {
  59     struct pollfd *pollfd, *pollfd2;
  60     unsigned n_pollfd_alloc, n_pollfd_used;
  61
  62     pa_bool_t timer_enabled;
  63     struct timeval next_elapse;
  64
  65     pa_bool_t scan_for_dead;
  66     pa_bool_t running, installed, rebuild_needed, quit;
  67
  68 #ifdef HAVE_PPOLL
  69     int rtsig;
  70     sigset_t sigset_unblocked;
  71     timer_t timer;
  72     pa_bool_t timer_armed;
  73 #ifdef __linux__
  74     pa_bool_t dont_use_ppoll;
  75 #endif
  76 #endif
  77
  78     PA_LLIST_HEAD(pa_rtpoll_item, items);
  79 };
  80
  81 struct pa_rtpoll_item {
  82     pa_rtpoll *rtpoll;
  83     pa_bool_t dead;
  84
  85     pa_rtpoll_priority_t priority;
  86
  87     struct pollfd *pollfd;
  88     unsigned n_pollfd;
  89
  90     int (*work_cb)(pa_rtpoll_item *i);
  91     int (*before_cb)(pa_rtpoll_item *i);
  92     void (*after_cb)(pa_rtpoll_item *i);
  93     void *userdata;
  94
  95     PA_LLIST_FIELDS(pa_rtpoll_item);
  96 };
  97
  98 PA_STATIC_FLIST_DECLARE(items, 0, pa_xfree);
  99
 100 static void signal_handler_noop(int s) { /* write(2, "signal\n", 7); */ }
 101
 102 pa_rtpoll *pa_rtpoll_new(void) {
 103     pa_rtpoll *p;
 104
 105     p = pa_xnew(pa_rtpoll, 1);
 106
 107 #ifdef HAVE_PPOLL
 108
 109 #ifdef __linux__
 110     /* ppoll is broken on Linux < 2.6.16 */
 111     p->dont_use_ppoll = FALSE;
 112
 113     {
 114         struct utsname u;
 115         unsigned major, minor, micro;
 116
 117         pa_assert_se(uname(&u) == 0);
 118
 119         if (sscanf(u.release, "%u.%u.%u", &major, &minor, &micro) != 3 ||
 120             (major < 2) ||
 121             (major == 2 && minor < 6) ||
 122             (major == 2 && minor == 6 && micro < 16))
 123
 124             p->dont_use_ppoll = TRUE;
 125     }
 126
 127 #endif
 128
 129     p->rtsig = -1;
 130     sigemptyset(&p->sigset_unblocked);
 131     p->timer = (timer_t) -1;
 132     p->timer_armed = FALSE;
 133
 134 #endif
 135
 136     p->n_pollfd_alloc = 32;
 137     p->pollfd = pa_xnew(struct pollfd, p->n_pollfd_alloc);
 138     p->pollfd2 = pa_xnew(struct pollfd, p->n_pollfd_alloc);
 139     p->n_pollfd_used = 0;
 140
 141     memset(&p->next_elapse, 0, sizeof(p->next_elapse));
 142     p->timer_enabled = FALSE;
 143
 144     p->running = FALSE;
 145     p->installed = FALSE;
 146     p->scan_for_dead = FALSE;
 147     p->rebuild_needed = FALSE;
 148     p->quit = FALSE;
 149
 150     PA_LLIST_HEAD_INIT(pa_rtpoll_item, p->items);
 151
 152     return p;
 153 }
 154
 155 void pa_rtpoll_install(pa_rtpoll *p) {
 156     pa_assert(p);
 157     pa_assert(!p->installed);
 158
 159     p->installed = 1;
 160
 161 #ifdef HAVE_PPOLL
 162 # ifdef __linux__
 163     if (p->dont_use_ppoll)
 164         return;
 165 # endif
 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     memset(&timeout, 0, sizeof(timeout));
 370
 371     /* Calculate timeout */
 372     if (wait && !p->quit && p->timer_enabled) {
 373         struct timeval now;
 374         pa_rtclock_get(&now);
 375
 376         if (pa_timeval_cmp(&p->next_elapse, &now) > 0)
 377             pa_timeval_add(&timeout, pa_timeval_diff(&p->next_elapse, &now));
 378     }
 379
 380     /* OK, now let's sleep */
 381 #ifdef HAVE_PPOLL
 382
 383 #ifdef __linux__
 384     if (!p->dont_use_ppoll)
 385 #endif
 386     {
 387         struct timespec ts;
 388         ts.tv_sec = timeout.tv_sec;
 389         ts.tv_nsec = timeout.tv_usec * 1000;
 390         r = ppoll(p->pollfd, p->n_pollfd_used, (!wait || p->quit || p->timer_enabled) ? &ts : NULL, p->rtsig < 0 ? NULL : &p->sigset_unblocked);
 391     }
 392 #ifdef __linux__
 393     else
 394 #endif
 395
 396 #endif
 397         r = poll(p->pollfd, p->n_pollfd_used, (!wait || p->quit || p->timer_enabled) ? (int) ((timeout.tv_sec*1000) + (timeout.tv_usec / 1000)) : -1);
 398
 399     if (r < 0) {
 400         if (errno == EAGAIN || errno == EINTR)
 401             r = 0;
 402         else
 403             pa_log_error("poll(): %s", pa_cstrerror(errno));
 404
 405         reset_all_revents(p);
 406     }
 407
 408     /* Let's tell everyone that we left the sleep */
 409     for (i = p->items; i && i->priority < PA_RTPOLL_NEVER; i = i->next) {
 410
 411         if (i->dead)
 412             continue;
 413
 414         if (!i->after_cb)
 415             continue;
 416
 417         i->after_cb(i);
 418     }
 419
 420 finish:
 421
 422     p->running = FALSE;
 423
 424     if (p->scan_for_dead) {
 425         pa_rtpoll_item *n;
 426
 427         p->scan_for_dead = FALSE;
 428
 429         for (i = p->items; i; i = n) {
 430             n = i->next;
 431
 432             if (i->dead)
 433                 rtpoll_item_destroy(i);
 434         }
 435     }
 436
 437     return r < 0 ? r : !p->quit;
 438 }
 439
 440 static void update_timer(pa_rtpoll *p) {
 441     pa_assert(p);
 442
 443 #ifdef HAVE_PPOLL
 444
 445 #ifdef __linux__
 446     if (!p->dont_use_ppoll) {
 447 #endif
 448
 449         if (p->timer == (timer_t) -1) {
 450             struct sigevent se;
 451
 452             memset(&se, 0, sizeof(se));
 453             se.sigev_notify = SIGEV_SIGNAL;
 454             se.sigev_signo = p->rtsig;
 455
 456             if (timer_create(CLOCK_MONOTONIC, &se, &p->timer) < 0)
 457                 if (timer_create(CLOCK_REALTIME, &se, &p->timer) < 0) {
 458                     pa_log_warn("Failed to allocate POSIX timer: %s", pa_cstrerror(errno));
 459                     p->timer = (timer_t) -1;
 460                 }
 461         }
 462
 463         if (p->timer != (timer_t) -1) {
 464             struct itimerspec its;
 465             struct timespec ts = { .tv_sec = 0, .tv_nsec = 0 };
 466             sigset_t ss;
 467
 468             if (p->timer_armed) {
 469                 /* First disarm timer */
 470                 memset(&its, 0, sizeof(its));
 471                 pa_assert_se(timer_settime(p->timer, TIMER_ABSTIME, &its, NULL) == 0);
 472
 473                 /* Remove a signal that might be waiting in the signal q */
 474                 pa_assert_se(sigemptyset(&ss) == 0);
 475                 pa_assert_se(sigaddset(&ss, p->rtsig) == 0);
 476                 sigtimedwait(&ss, NULL, &ts);
 477             }
 478
 479             /* And install the new timer */
 480             if (p->timer_enabled) {
 481                 memset(&its, 0, sizeof(its));
 482
 483                 its.it_value.tv_sec = p->next_elapse.tv_sec;
 484                 its.it_value.tv_nsec = p->next_elapse.tv_usec*1000;
 485
 486                 /* Make sure that 0,0 is not understood as
 487                  * "disarming" */
 488                 if (its.it_value.tv_sec == 0 && its.it_value.tv_nsec == 0)
 489                     its.it_value.tv_nsec = 1;
 490                 pa_assert_se(timer_settime(p->timer, TIMER_ABSTIME, &its, NULL) == 0);
 491             }
 492
 493             p->timer_armed = p->timer_enabled;
 494         }
 495
 496 #ifdef __linux__
 497     }
 498 #endif
 499
 500 #endif
 501 }
 502
 503 void pa_rtpoll_set_timer_absolute(pa_rtpoll *p, pa_usec_t usec) {
 504     pa_assert(p);
 505
 506     pa_timeval_store(&p->next_elapse, usec);
 507     p->timer_enabled = TRUE;
 508
 509     update_timer(p);
 510 }
 511
 512 void pa_rtpoll_set_timer_relative(pa_rtpoll *p, pa_usec_t usec) {
 513     pa_assert(p);
 514
 515     /* Scheduling a timeout for more than an hour is very very suspicious */
 516     pa_assert(usec <= PA_USEC_PER_SEC*60ULL*60ULL);
 517
 518     pa_rtclock_get(&p->next_elapse);
 519     pa_timeval_add(&p->next_elapse, usec);
 520     p->timer_enabled = TRUE;
 521
 522     update_timer(p);
 523 }
 524
 525 void pa_rtpoll_set_timer_disabled(pa_rtpoll *p) {
 526     pa_assert(p);
 527
 528     memset(&p->next_elapse, 0, sizeof(p->next_elapse));
 529     p->timer_enabled = FALSE;
 530
 531     update_timer(p);
 532 }
 533
 534 pa_rtpoll_item *pa_rtpoll_item_new(pa_rtpoll *p, pa_rtpoll_priority_t prio, unsigned n_fds) {
 535     pa_rtpoll_item *i, *j, *l = NULL;
 536
 537     pa_assert(p);
 538
 539     if (!(i = pa_flist_pop(PA_STATIC_FLIST_GET(items))))
 540         i = pa_xnew(pa_rtpoll_item, 1);
 541
 542     i->rtpoll = p;
 543     i->dead = FALSE;
 544     i->n_pollfd = n_fds;
 545     i->pollfd = NULL;
 546     i->priority = prio;
 547
 548     i->userdata = NULL;
 549     i->before_cb = NULL;
 550     i->after_cb = NULL;
 551     i->work_cb = NULL;
 552
 553     for (j = p->items; j; j = j->next) {
 554         if (prio <= j->priority)
 555             break;
 556
 557         l = j;
 558     }
 559
 560     PA_LLIST_INSERT_AFTER(pa_rtpoll_item, p->items, j ? j->prev : l, i);
 561
 562     if (n_fds > 0) {
 563         p->rebuild_needed = 1;
 564         p->n_pollfd_used += n_fds;
 565     }
 566
 567     return i;
 568 }
 569
 570 void pa_rtpoll_item_free(pa_rtpoll_item *i) {
 571     pa_assert(i);
 572
 573     if (i->rtpoll->running) {
 574         i->dead = TRUE;
 575         i->rtpoll->scan_for_dead = TRUE;
 576         return;
 577     }
 578
 579     rtpoll_item_destroy(i);
 580 }
 581
 582 struct pollfd *pa_rtpoll_item_get_pollfd(pa_rtpoll_item *i, unsigned *n_fds) {
 583     pa_assert(i);
 584
 585     if (i->n_pollfd > 0)
 586         if (i->rtpoll->rebuild_needed)
 587             rtpoll_rebuild(i->rtpoll);
 588
 589     if (n_fds)
 590         *n_fds = i->n_pollfd;
 591
 592     return i->pollfd;
 593 }
 594
 595 void pa_rtpoll_item_set_before_callback(pa_rtpoll_item *i, int (*before_cb)(pa_rtpoll_item *i)) {
 596     pa_assert(i);
 597     pa_assert(i->priority < PA_RTPOLL_NEVER);
 598
 599     i->before_cb = before_cb;
 600 }
 601
 602 void pa_rtpoll_item_set_after_callback(pa_rtpoll_item *i, void (*after_cb)(pa_rtpoll_item *i)) {
 603     pa_assert(i);
 604     pa_assert(i->priority < PA_RTPOLL_NEVER);
 605
 606     i->after_cb = after_cb;
 607 }
 608
 609 void pa_rtpoll_item_set_work_callback(pa_rtpoll_item *i, int (*work_cb)(pa_rtpoll_item *i)) {
 610     pa_assert(i);
 611     pa_assert(i->priority < PA_RTPOLL_NEVER);
 612
 613     i->work_cb = work_cb;
 614 }
 615
 616 void pa_rtpoll_item_set_userdata(pa_rtpoll_item *i, void *userdata) {
 617     pa_assert(i);
 618
 619     i->userdata = userdata;
 620 }
 621
 622 void* pa_rtpoll_item_get_userdata(pa_rtpoll_item *i) {
 623     pa_assert(i);
 624
 625     return i->userdata;
 626 }
 627
 628 static int fdsem_before(pa_rtpoll_item *i) {
 629
 630     if (pa_fdsem_before_poll(i->userdata) < 0)
 631         return 1; /* 1 means immediate restart of the loop */
 632
 633     return 0;
 634 }
 635
 636 static void fdsem_after(pa_rtpoll_item *i) {
 637     pa_assert(i);
 638
 639     pa_assert((i->pollfd[0].revents & ~POLLIN) == 0);
 640     pa_fdsem_after_poll(i->userdata);
 641 }
 642
 643 pa_rtpoll_item *pa_rtpoll_item_new_fdsem(pa_rtpoll *p, pa_rtpoll_priority_t prio, pa_fdsem *f) {
 644     pa_rtpoll_item *i;
 645     struct pollfd *pollfd;
 646
 647     pa_assert(p);
 648     pa_assert(f);
 649
 650     i = pa_rtpoll_item_new(p, prio, 1);
 651
 652     pollfd = pa_rtpoll_item_get_pollfd(i, NULL);
 653
 654     pollfd->fd = pa_fdsem_get(f);
 655     pollfd->events = POLLIN;
 656
 657     i->before_cb = fdsem_before;
 658     i->after_cb = fdsem_after;
 659     i->userdata = f;
 660
 661     return i;
 662 }
 663
 664 static int asyncmsgq_read_before(pa_rtpoll_item *i) {
 665     pa_assert(i);
 666
 667     if (pa_asyncmsgq_read_before_poll(i->userdata) < 0)
 668         return 1; /* 1 means immediate restart of the loop */
 669
 670     return 0;
 671 }
 672
 673 static void asyncmsgq_read_after(pa_rtpoll_item *i) {
 674     pa_assert(i);
 675
 676     pa_assert((i->pollfd[0].revents & ~POLLIN) == 0);
 677     pa_asyncmsgq_read_after_poll(i->userdata);
 678 }
 679
 680 static int asyncmsgq_read_work(pa_rtpoll_item *i) {
 681     pa_msgobject *object;
 682     int code;
 683     void *data;
 684     pa_memchunk chunk;
 685     int64_t offset;
 686
 687     pa_assert(i);
 688
 689     if (pa_asyncmsgq_get(i->userdata, &object, &code, &data, &offset, &chunk, 0) == 0) {
 690         int ret;
 691
 692         if (!object && code == PA_MESSAGE_SHUTDOWN) {
 693             pa_asyncmsgq_done(i->userdata, 0);
 694             pa_rtpoll_quit(i->rtpoll);
 695             return 1;
 696         }
 697
 698         ret = pa_asyncmsgq_dispatch(object, code, data, offset, &chunk);
 699         pa_asyncmsgq_done(i->userdata, ret);
 700         return 1;
 701     }
 702
 703     return 0;
 704 }
 705
 706 pa_rtpoll_item *pa_rtpoll_item_new_asyncmsgq_read(pa_rtpoll *p, pa_rtpoll_priority_t prio, pa_asyncmsgq *q) {
 707     pa_rtpoll_item *i;
 708     struct pollfd *pollfd;
 709
 710     pa_assert(p);
 711     pa_assert(q);
 712
 713     i = pa_rtpoll_item_new(p, prio, 1);
 714
 715     pollfd = pa_rtpoll_item_get_pollfd(i, NULL);
 716     pollfd->fd = pa_asyncmsgq_read_fd(q);
 717     pollfd->events = POLLIN;
 718
 719     i->before_cb = asyncmsgq_read_before;
 720     i->after_cb = asyncmsgq_read_after;
 721     i->work_cb = asyncmsgq_read_work;
 722     i->userdata = q;
 723
 724     return i;
 725 }
 726
 727 static int asyncmsgq_write_before(pa_rtpoll_item *i) {
 728     pa_assert(i);
 729
 730     pa_asyncmsgq_write_before_poll(i->userdata);
 731     return 0;
 732 }
 733
 734 static void asyncmsgq_write_after(pa_rtpoll_item *i) {
 735     pa_assert(i);
 736
 737     pa_assert((i->pollfd[0].revents & ~POLLIN) == 0);
 738     pa_asyncmsgq_write_after_poll(i->userdata);
 739 }
 740
 741 pa_rtpoll_item *pa_rtpoll_item_new_asyncmsgq_write(pa_rtpoll *p, pa_rtpoll_priority_t prio, pa_asyncmsgq *q) {
 742     pa_rtpoll_item *i;
 743     struct pollfd *pollfd;
 744
 745     pa_assert(p);
 746     pa_assert(q);
 747
 748     i = pa_rtpoll_item_new(p, prio, 1);
 749
 750     pollfd = pa_rtpoll_item_get_pollfd(i, NULL);
 751     pollfd->fd = pa_asyncmsgq_write_fd(q);
 752     pollfd->events = POLLIN;
 753
 754     i->before_cb = asyncmsgq_write_before;
 755     i->after_cb = asyncmsgq_write_after;
 756     i->work_cb = NULL;
 757     i->userdata = q;
 758
 759     return i;
 760 }
 761
 762 void pa_rtpoll_quit(pa_rtpoll *p) {
 763     pa_assert(p);
 764
 765     p->quit = TRUE;
 766 }