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
  67     pa_bool_t scan_for_dead;
  68     pa_bool_t running, installed, rebuild_needed, quit;
  69
  70 #ifdef HAVE_PPOLL
  71     int rtsig;
  72     sigset_t sigset_unblocked;
  73     timer_t timer;
  74     pa_bool_t timer_armed;
  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) { /* write(2, "signal\n", 7); */ }
 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     p->timer_armed = FALSE;
 135
 136 #endif
 137
 138     p->n_pollfd_alloc = 32;
 139     p->pollfd = pa_xnew(struct pollfd, p->n_pollfd_alloc);
 140     p->pollfd2 = pa_xnew(struct pollfd, p->n_pollfd_alloc);
 141     p->n_pollfd_used = 0;
 142
 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 # ifdef __linux__
 165     if (p->dont_use_ppoll)
 166         return;
 167 # endif
 168
 169     if ((p->rtsig = pa_rtsig_get_for_thread()) < 0) {
 170         pa_log_warn("Failed to reserve POSIX realtime signal.");
 171         return;
 172     }
 173
 174     pa_log_debug("Acquired POSIX realtime signal %s", pa_sig2str(p->rtsig));
 175
 176     {
 177         sigset_t ss;
 178         struct sigaction sa;
 179
 180         pa_assert_se(sigemptyset(&ss) == 0);
 181         pa_assert_se(sigaddset(&ss, p->rtsig) == 0);
 182         pa_assert_se(pthread_sigmask(SIG_BLOCK, &ss, &p->sigset_unblocked) == 0);
 183         pa_assert_se(sigdelset(&p->sigset_unblocked, p->rtsig) == 0);
 184
 185         memset(&sa, 0, sizeof(sa));
 186         sa.sa_handler = signal_handler_noop;
 187         pa_assert_se(sigemptyset(&sa.sa_mask) == 0);
 188
 189         pa_assert_se(sigaction(p->rtsig, &sa, NULL) == 0);
 190
 191         /* We never reset the signal handler. Why should we? */
 192     }
 193
 194 #endif
 195 }
 196
 197 static void rtpoll_rebuild(pa_rtpoll *p) {
 198
 199     struct pollfd *e, *t;
 200     pa_rtpoll_item *i;
 201     int ra = 0;
 202
 203     pa_assert(p);
 204
 205     p->rebuild_needed = FALSE;
 206
 207     if (p->n_pollfd_used > p->n_pollfd_alloc) {
 208         /* Hmm, we have to allocate some more space */
 209         p->n_pollfd_alloc = p->n_pollfd_used * 2;
 210         p->pollfd2 = pa_xrealloc(p->pollfd2, p->n_pollfd_alloc * sizeof(struct pollfd));
 211         ra = 1;
 212     }
 213
 214     e = p->pollfd2;
 215
 216     for (i = p->items; i; i = i->next) {
 217
 218         if (i->n_pollfd > 0)  {
 219             size_t l = i->n_pollfd * sizeof(struct pollfd);
 220
 221             if (i->pollfd)
 222                 memcpy(e, i->pollfd, l);
 223             else
 224                 memset(e, 0, l);
 225
 226             i->pollfd = e;
 227         } else
 228             i->pollfd = NULL;
 229
 230         e += i->n_pollfd;
 231     }
 232
 233     pa_assert((unsigned) (e - p->pollfd2) == p->n_pollfd_used);
 234     t = p->pollfd;
 235     p->pollfd = p->pollfd2;
 236     p->pollfd2 = t;
 237
 238     if (ra)
 239         p->pollfd2 = pa_xrealloc(p->pollfd2, p->n_pollfd_alloc * sizeof(struct pollfd));
 240
 241 }
 242
 243 static void rtpoll_item_destroy(pa_rtpoll_item *i) {
 244     pa_rtpoll *p;
 245
 246     pa_assert(i);
 247
 248     p = i->rtpoll;
 249
 250     PA_LLIST_REMOVE(pa_rtpoll_item, p->items, i);
 251
 252     p->n_pollfd_used -= i->n_pollfd;
 253
 254     if (pa_flist_push(PA_STATIC_FLIST_GET(items), i) < 0)
 255         pa_xfree(i);
 256
 257     p->rebuild_needed = TRUE;
 258 }
 259
 260 void pa_rtpoll_free(pa_rtpoll *p) {
 261     pa_assert(p);
 262
 263     while (p->items)
 264         rtpoll_item_destroy(p->items);
 265
 266     pa_xfree(p->pollfd);
 267     pa_xfree(p->pollfd2);
 268
 269 #ifdef HAVE_PPOLL
 270     if (p->timer != (timer_t) -1)
 271         timer_delete(p->timer);
 272 #endif
 273
 274     pa_xfree(p);
 275 }
 276
 277 static void reset_revents(pa_rtpoll_item *i) {
 278     struct pollfd *f;
 279     unsigned n;
 280
 281     pa_assert(i);
 282
 283     if (!(f = pa_rtpoll_item_get_pollfd(i, &n)))
 284         return;
 285
 286     for (; n > 0; n--)
 287         f[n-1].revents = 0;
 288 }
 289
 290 static void reset_all_revents(pa_rtpoll *p) {
 291     pa_rtpoll_item *i;
 292
 293     pa_assert(p);
 294
 295     for (i = p->items; i; i = i->next) {
 296
 297         if (i->dead)
 298             continue;
 299
 300         reset_revents(i);
 301     }
 302 }
 303
 304 int pa_rtpoll_run(pa_rtpoll *p, pa_bool_t wait) {
 305     pa_rtpoll_item *i;
 306     int r = 0;
 307     struct timeval timeout;
 308
 309     pa_assert(p);
 310     pa_assert(!p->running);
 311     pa_assert(p->installed);
 312
 313     p->running = TRUE;
 314
 315     /* First, let's do some work */
 316     for (i = p->items; i && i->priority < PA_RTPOLL_NEVER; i = i->next) {
 317         int k;
 318
 319         if (i->dead)
 320             continue;
 321
 322         if (!i->work_cb)
 323             continue;
 324
 325         if (p->quit)
 326             goto finish;
 327
 328         if ((k = i->work_cb(i)) != 0) {
 329             if (k < 0)
 330                 r = k;
 331
 332             goto finish;
 333         }
 334     }
 335
 336     /* Now let's prepare for entering the sleep */
 337     for (i = p->items; i && i->priority < PA_RTPOLL_NEVER; i = i->next) {
 338         int k = 0;
 339
 340         if (i->dead)
 341             continue;
 342
 343         if (!i->before_cb)
 344             continue;
 345
 346         if (p->quit || (k = i->before_cb(i)) != 0) {
 347
 348             /* Hmm, this one doesn't let us enter the poll, so rewind everything */
 349
 350             for (i = i->prev; i; i = i->prev) {
 351
 352                 if (i->dead)
 353                     continue;
 354
 355                 if (!i->after_cb)
 356                     continue;
 357
 358                 i->after_cb(i);
 359             }
 360
 361             if (k < 0)
 362                 r = k;
 363
 364             goto finish;
 365         }
 366     }
 367
 368     if (p->rebuild_needed)
 369         rtpoll_rebuild(p);
 370
 371     memset(&timeout, 0, sizeof(timeout));
 372
 373     /* Calculate timeout */
 374     if (wait && !p->quit && p->timer_enabled) {
 375         struct timeval now;
 376         pa_rtclock_get(&now);
 377
 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, (!wait || p->quit || 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, (!wait || p->quit || 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     /* Let's tell everyone that we left the sleep */
 411     for (i = p->items; i && i->priority < PA_RTPOLL_NEVER; i = i->next) {
 412
 413         if (i->dead)
 414             continue;
 415
 416         if (!i->after_cb)
 417             continue;
 418
 419         i->after_cb(i);
 420     }
 421
 422 finish:
 423
 424     p->running = FALSE;
 425
 426     if (p->scan_for_dead) {
 427         pa_rtpoll_item *n;
 428
 429         p->scan_for_dead = FALSE;
 430
 431         for (i = p->items; i; i = n) {
 432             n = i->next;
 433
 434             if (i->dead)
 435                 rtpoll_item_destroy(i);
 436         }
 437     }
 438
 439     return r < 0 ? r : !p->quit;
 440 }
 441
 442 static void update_timer(pa_rtpoll *p) {
 443     pa_assert(p);
 444
 445 #ifdef HAVE_PPOLL
 446
 447 #ifdef __linux__
 448     if (!p->dont_use_ppoll) {
 449 #endif
 450
 451         if (p->timer == (timer_t) -1) {
 452             struct sigevent se;
 453
 454             memset(&se, 0, sizeof(se));
 455             se.sigev_notify = SIGEV_SIGNAL;
 456             se.sigev_signo = p->rtsig;
 457
 458             if (timer_create(CLOCK_MONOTONIC, &se, &p->timer) < 0)
 459                 if (timer_create(CLOCK_REALTIME, &se, &p->timer) < 0) {
 460                     pa_log_warn("Failed to allocate POSIX timer: %s", pa_cstrerror(errno));
 461                     p->timer = (timer_t) -1;
 462                 }
 463         }
 464
 465         if (p->timer != (timer_t) -1) {
 466             struct itimerspec its;
 467             struct timespec ts = { .tv_sec = 0, .tv_nsec = 0 };
 468             sigset_t ss;
 469
 470             if (p->timer_armed) {
 471                 /* First disarm timer */
 472                 memset(&its, 0, sizeof(its));
 473                 pa_assert_se(timer_settime(p->timer, TIMER_ABSTIME, &its, NULL) == 0);
 474
 475                 /* Remove a signal that might be waiting in the signal q */
 476                 pa_assert_se(sigemptyset(&ss) == 0);
 477                 pa_assert_se(sigaddset(&ss, p->rtsig) == 0);
 478                 sigtimedwait(&ss, NULL, &ts);
 479             }
 480
 481             /* And install the new timer */
 482             if (p->timer_enabled) {
 483                 memset(&its, 0, sizeof(its));
 484
 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 && its.it_value.tv_nsec == 0)
 491                     its.it_value.tv_nsec = 1;
 492                 pa_assert_se(timer_settime(p->timer, TIMER_ABSTIME, &its, NULL) == 0);
 493             }
 494
 495             p->timer_armed = p->timer_enabled;
 496         }
 497
 498 #ifdef __linux__
 499     }
 500 #endif
 501
 502 #endif
 503 }
 504
 505 void pa_rtpoll_set_timer_absolute(pa_rtpoll *p, pa_usec_t usec) {
 506     pa_assert(p);
 507
 508     pa_timeval_store(&p->next_elapse, usec);
 509     p->timer_enabled = TRUE;
 510
 511     update_timer(p);
 512 }
 513
 514 void pa_rtpoll_set_timer_relative(pa_rtpoll *p, pa_usec_t usec) {
 515     pa_assert(p);
 516
 517     pa_rtclock_get(&p->next_elapse);
 518     pa_timeval_add(&p->next_elapse, usec);
 519     p->timer_enabled = TRUE;
 520
 521     update_timer(p);
 522 }
 523
 524 void pa_rtpoll_set_timer_disabled(pa_rtpoll *p) {
 525     pa_assert(p);
 526
 527     memset(&p->next_elapse, 0, sizeof(p->next_elapse));
 528     p->timer_enabled = FALSE;
 529
 530     update_timer(p);
 531 }
 532
 533 pa_rtpoll_item *pa_rtpoll_item_new(pa_rtpoll *p, pa_rtpoll_priority_t prio, unsigned n_fds) {
 534     pa_rtpoll_item *i, *j, *l = NULL;
 535
 536     pa_assert(p);
 537
 538     if (!(i = pa_flist_pop(PA_STATIC_FLIST_GET(items))))
 539         i = pa_xnew(pa_rtpoll_item, 1);
 540
 541     i->rtpoll = p;
 542     i->dead = FALSE;
 543     i->n_pollfd = n_fds;
 544     i->pollfd = NULL;
 545     i->priority = prio;
 546
 547     i->userdata = NULL;
 548     i->before_cb = NULL;
 549     i->after_cb = NULL;
 550     i->work_cb = NULL;
 551
 552     for (j = p->items; j; j = j->next) {
 553         if (prio <= j->priority)
 554             break;
 555
 556         l = j;
 557     }
 558
 559     PA_LLIST_INSERT_AFTER(pa_rtpoll_item, p->items, j ? j->prev : l, i);
 560
 561     if (n_fds > 0) {
 562         p->rebuild_needed = 1;
 563         p->n_pollfd_used += n_fds;
 564     }
 565
 566     return i;
 567 }
 568
 569 void pa_rtpoll_item_free(pa_rtpoll_item *i) {
 570     pa_assert(i);
 571
 572     if (i->rtpoll->running) {
 573         i->dead = TRUE;
 574         i->rtpoll->scan_for_dead = TRUE;
 575         return;
 576     }
 577
 578     rtpoll_item_destroy(i);
 579 }
 580
 581 struct pollfd *pa_rtpoll_item_get_pollfd(pa_rtpoll_item *i, unsigned *n_fds) {
 582     pa_assert(i);
 583
 584     if (i->n_pollfd > 0)
 585         if (i->rtpoll->rebuild_needed)
 586             rtpoll_rebuild(i->rtpoll);
 587
 588     if (n_fds)
 589         *n_fds = i->n_pollfd;
 590
 591     return i->pollfd;
 592 }
 593
 594 void pa_rtpoll_item_set_before_callback(pa_rtpoll_item *i, int (*before_cb)(pa_rtpoll_item *i)) {
 595     pa_assert(i);
 596     pa_assert(i->priority < PA_RTPOLL_NEVER);
 597
 598     i->before_cb = before_cb;
 599 }
 600
 601 void pa_rtpoll_item_set_after_callback(pa_rtpoll_item *i, void (*after_cb)(pa_rtpoll_item *i)) {
 602     pa_assert(i);
 603     pa_assert(i->priority < PA_RTPOLL_NEVER);
 604
 605     i->after_cb = after_cb;
 606 }
 607
 608 void pa_rtpoll_item_set_work_callback(pa_rtpoll_item *i, int (*work_cb)(pa_rtpoll_item *i)) {
 609     pa_assert(i);
 610     pa_assert(i->priority < PA_RTPOLL_NEVER);
 611
 612     i->work_cb = work_cb;
 613 }
 614
 615 void pa_rtpoll_item_set_userdata(pa_rtpoll_item *i, void *userdata) {
 616     pa_assert(i);
 617
 618     i->userdata = userdata;
 619 }
 620
 621 void* pa_rtpoll_item_get_userdata(pa_rtpoll_item *i) {
 622     pa_assert(i);
 623
 624     return i->userdata;
 625 }
 626
 627 static int fdsem_before(pa_rtpoll_item *i) {
 628
 629     if (pa_fdsem_before_poll(i->userdata) < 0)
 630         return 1; /* 1 means immediate restart of the loop */
 631
 632     return 0;
 633 }
 634
 635 static void fdsem_after(pa_rtpoll_item *i) {
 636     pa_assert(i);
 637
 638     pa_assert((i->pollfd[0].revents & ~POLLIN) == 0);
 639     pa_fdsem_after_poll(i->userdata);
 640 }
 641
 642 pa_rtpoll_item *pa_rtpoll_item_new_fdsem(pa_rtpoll *p, pa_rtpoll_priority_t prio, pa_fdsem *f) {
 643     pa_rtpoll_item *i;
 644     struct pollfd *pollfd;
 645
 646     pa_assert(p);
 647     pa_assert(f);
 648
 649     i = pa_rtpoll_item_new(p, prio, 1);
 650
 651     pollfd = pa_rtpoll_item_get_pollfd(i, NULL);
 652
 653     pollfd->fd = pa_fdsem_get(f);
 654     pollfd->events = POLLIN;
 655
 656     i->before_cb = fdsem_before;
 657     i->after_cb = fdsem_after;
 658     i->userdata = f;
 659
 660     return i;
 661 }
 662
 663 static int asyncmsgq_read_before(pa_rtpoll_item *i) {
 664     pa_assert(i);
 665
 666     if (pa_asyncmsgq_read_before_poll(i->userdata) < 0)
 667         return 1; /* 1 means immediate restart of the loop */
 668
 669     return 0;
 670 }
 671
 672 static void asyncmsgq_read_after(pa_rtpoll_item *i) {
 673     pa_assert(i);
 674
 675     pa_assert((i->pollfd[0].revents & ~POLLIN) == 0);
 676     pa_asyncmsgq_read_after_poll(i->userdata);
 677 }
 678
 679 static int asyncmsgq_read_work(pa_rtpoll_item *i) {
 680     pa_msgobject *object;
 681     int code;
 682     void *data;
 683     pa_memchunk chunk;
 684     int64_t offset;
 685
 686     pa_assert(i);
 687
 688     if (pa_asyncmsgq_get(i->userdata, &object, &code, &data, &offset, &chunk, 0) == 0) {
 689         int ret;
 690
 691         if (!object && code == PA_MESSAGE_SHUTDOWN) {
 692             pa_asyncmsgq_done(i->userdata, 0);
 693             pa_rtpoll_quit(i->rtpoll);
 694             return 1;
 695         }
 696
 697         ret = pa_asyncmsgq_dispatch(object, code, data, offset, &chunk);
 698         pa_asyncmsgq_done(i->userdata, ret);
 699         return 1;
 700     }
 701
 702     return 0;
 703 }
 704
 705 pa_rtpoll_item *pa_rtpoll_item_new_asyncmsgq_read(pa_rtpoll *p, pa_rtpoll_priority_t prio, pa_asyncmsgq *q) {
 706     pa_rtpoll_item *i;
 707     struct pollfd *pollfd;
 708
 709     pa_assert(p);
 710     pa_assert(q);
 711
 712     i = pa_rtpoll_item_new(p, prio, 1);
 713
 714     pollfd = pa_rtpoll_item_get_pollfd(i, NULL);
 715     pollfd->fd = pa_asyncmsgq_read_fd(q);
 716     pollfd->events = POLLIN;
 717
 718     i->before_cb = asyncmsgq_read_before;
 719     i->after_cb = asyncmsgq_read_after;
 720     i->work_cb = asyncmsgq_read_work;
 721     i->userdata = q;
 722
 723     return i;
 724 }
 725
 726 static int asyncmsgq_write_before(pa_rtpoll_item *i) {
 727     pa_assert(i);
 728
 729     pa_asyncmsgq_write_before_poll(i->userdata);
 730     return 0;
 731 }
 732
 733 static void asyncmsgq_write_after(pa_rtpoll_item *i) {
 734     pa_assert(i);
 735
 736     pa_assert((i->pollfd[0].revents & ~POLLIN) == 0);
 737     pa_asyncmsgq_write_after_poll(i->userdata);
 738 }
 739
 740 pa_rtpoll_item *pa_rtpoll_item_new_asyncmsgq_write(pa_rtpoll *p, pa_rtpoll_priority_t prio, pa_asyncmsgq *q) {
 741     pa_rtpoll_item *i;
 742     struct pollfd *pollfd;
 743
 744     pa_assert(p);
 745     pa_assert(q);
 746
 747     i = pa_rtpoll_item_new(p, prio, 1);
 748
 749     pollfd = pa_rtpoll_item_get_pollfd(i, NULL);
 750     pollfd->fd = pa_asyncmsgq_write_fd(q);
 751     pollfd->events = POLLIN;
 752
 753     i->before_cb = asyncmsgq_write_before;
 754     i->after_cb = asyncmsgq_write_after;
 755     i->work_cb = NULL;
 756     i->userdata = q;
 757
 758     return i;
 759 }
 760
 761 void pa_rtpoll_quit(pa_rtpoll *p) {
 762     pa_assert(p);
 763
 764     p->quit = TRUE;
 765 }