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 <string.h>
  30 #include <errno.h>
  31
  32 #include <pulse/xmalloc.h>
  33 #include <pulse/timeval.h>
  34
  35 #include <pulsecore/poll.h>
  36 #include <pulsecore/core-error.h>
  37 #include <pulsecore/core-rtclock.h>
  38 #include <pulsecore/macro.h>
  39 #include <pulsecore/llist.h>
  40 #include <pulsecore/flist.h>
  41 #include <pulsecore/core-util.h>
  42 #include <pulsecore/ratelimit.h>
  43 #include <pulse/rtclock.h>
  44
  45 #include "rtpoll.h"
  46
  47 /* #define DEBUG_TIMING */
  48
  49 struct pa_rtpoll {
  50     struct pollfd *pollfd, *pollfd2;
  51     unsigned n_pollfd_alloc, n_pollfd_used;
  52
  53     struct timeval next_elapse;
  54     pa_bool_t timer_enabled:1;
  55
  56     pa_bool_t scan_for_dead:1;
  57     pa_bool_t running:1;
  58     pa_bool_t rebuild_needed:1;
  59     pa_bool_t quit:1;
  60     pa_bool_t timer_elapsed:1;
  61
  62 #ifdef DEBUG_TIMING
  63     pa_usec_t timestamp;
  64     pa_usec_t slept, awake;
  65 #endif
  66
  67     PA_LLIST_HEAD(pa_rtpoll_item, items);
  68 };
  69
  70 struct pa_rtpoll_item {
  71     pa_rtpoll *rtpoll;
  72     pa_bool_t dead;
  73
  74     pa_rtpoll_priority_t priority;
  75
  76     struct pollfd *pollfd;
  77     unsigned n_pollfd;
  78
  79     int (*work_cb)(pa_rtpoll_item *i);
  80     int (*before_cb)(pa_rtpoll_item *i);
  81     void (*after_cb)(pa_rtpoll_item *i);
  82     void *userdata;
  83
  84     PA_LLIST_FIELDS(pa_rtpoll_item);
  85 };
  86
  87 PA_STATIC_FLIST_DECLARE(items, 0, pa_xfree);
  88
  89 pa_rtpoll *pa_rtpoll_new(void) {
  90     pa_rtpoll *p;
  91
  92     p = pa_xnew0(pa_rtpoll, 1);
  93
  94     p->n_pollfd_alloc = 32;
  95     p->pollfd = pa_xnew(struct pollfd, p->n_pollfd_alloc);
  96     p->pollfd2 = pa_xnew(struct pollfd, p->n_pollfd_alloc);
  97
  98 #ifdef DEBUG_TIMING
  99     p->timestamp = pa_rtclock_now();
 100 #endif
 101
 102     return p;
 103 }
 104
 105 static void rtpoll_rebuild(pa_rtpoll *p) {
 106
 107     struct pollfd *e, *t;
 108     pa_rtpoll_item *i;
 109     int ra = 0;
 110
 111     pa_assert(p);
 112
 113     p->rebuild_needed = FALSE;
 114
 115     if (p->n_pollfd_used > p->n_pollfd_alloc) {
 116         /* Hmm, we have to allocate some more space */
 117         p->n_pollfd_alloc = p->n_pollfd_used * 2;
 118         p->pollfd2 = pa_xrealloc(p->pollfd2, p->n_pollfd_alloc * sizeof(struct pollfd));
 119         ra = 1;
 120     }
 121
 122     e = p->pollfd2;
 123
 124     for (i = p->items; i; i = i->next) {
 125
 126         if (i->n_pollfd > 0) {
 127             size_t l = i->n_pollfd * sizeof(struct pollfd);
 128
 129             if (i->pollfd)
 130                 memcpy(e, i->pollfd, l);
 131             else
 132                 memset(e, 0, l);
 133
 134             i->pollfd = e;
 135         } else
 136             i->pollfd = NULL;
 137
 138         e += i->n_pollfd;
 139     }
 140
 141     pa_assert((unsigned) (e - p->pollfd2) == p->n_pollfd_used);
 142     t = p->pollfd;
 143     p->pollfd = p->pollfd2;
 144     p->pollfd2 = t;
 145
 146     if (ra)
 147         p->pollfd2 = pa_xrealloc(p->pollfd2, p->n_pollfd_alloc * sizeof(struct pollfd));
 148 }
 149
 150 static void rtpoll_item_destroy(pa_rtpoll_item *i) {
 151     pa_rtpoll *p;
 152
 153     pa_assert(i);
 154
 155     p = i->rtpoll;
 156
 157     PA_LLIST_REMOVE(pa_rtpoll_item, p->items, i);
 158
 159     p->n_pollfd_used -= i->n_pollfd;
 160
 161     if (pa_flist_push(PA_STATIC_FLIST_GET(items), i) < 0)
 162         pa_xfree(i);
 163
 164     p->rebuild_needed = TRUE;
 165 }
 166
 167 void pa_rtpoll_free(pa_rtpoll *p) {
 168     pa_assert(p);
 169
 170     while (p->items)
 171         rtpoll_item_destroy(p->items);
 172
 173     pa_xfree(p->pollfd);
 174     pa_xfree(p->pollfd2);
 175
 176     pa_xfree(p);
 177 }
 178
 179 static void reset_revents(pa_rtpoll_item *i) {
 180     struct pollfd *f;
 181     unsigned n;
 182
 183     pa_assert(i);
 184
 185     if (!(f = pa_rtpoll_item_get_pollfd(i, &n)))
 186         return;
 187
 188     for (; n > 0; n--)
 189         f[n-1].revents = 0;
 190 }
 191
 192 static void reset_all_revents(pa_rtpoll *p) {
 193     pa_rtpoll_item *i;
 194
 195     pa_assert(p);
 196
 197     for (i = p->items; i; i = i->next) {
 198
 199         if (i->dead)
 200             continue;
 201
 202         reset_revents(i);
 203     }
 204 }
 205
 206 int pa_rtpoll_run(pa_rtpoll *p, pa_bool_t wait_op) {
 207     pa_rtpoll_item *i;
 208     int r = 0;
 209     struct timeval timeout;
 210
 211     pa_assert(p);
 212     pa_assert(!p->running);
 213
 214 #ifdef DEBUG_TIMING
 215     pa_log("rtpoll_run");
 216 #endif
 217
 218     p->running = TRUE;
 219     p->timer_elapsed = FALSE;
 220
 221     /* First, let's do some work */
 222     for (i = p->items; i && i->priority < PA_RTPOLL_NEVER; i = i->next) {
 223         int k;
 224
 225         if (i->dead)
 226             continue;
 227
 228         if (!i->work_cb)
 229             continue;
 230
 231         if (p->quit) {
 232 #ifdef DEBUG_TIMING
 233             pa_log("rtpoll finish");
 234 #endif
 235             goto finish;
 236         }
 237
 238         if ((k = i->work_cb(i)) != 0) {
 239             if (k < 0)
 240                 r = k;
 241 #ifdef DEBUG_TIMING
 242             pa_log("rtpoll finish");
 243 #endif
 244             goto finish;
 245         }
 246     }
 247
 248     /* Now let's prepare for entering the sleep */
 249     for (i = p->items; i && i->priority < PA_RTPOLL_NEVER; i = i->next) {
 250         int k = 0;
 251
 252         if (i->dead)
 253             continue;
 254
 255         if (!i->before_cb)
 256             continue;
 257
 258         if (p->quit || (k = i->before_cb(i)) != 0) {
 259
 260             /* Hmm, this one doesn't let us enter the poll, so rewind everything */
 261
 262             for (i = i->prev; i; i = i->prev) {
 263
 264                 if (i->dead)
 265                     continue;
 266
 267                 if (!i->after_cb)
 268                     continue;
 269
 270                 i->after_cb(i);
 271             }
 272
 273             if (k < 0)
 274                 r = k;
 275 #ifdef DEBUG_TIMING
 276             pa_log("rtpoll finish");
 277 #endif
 278             goto finish;
 279         }
 280     }
 281
 282     if (p->rebuild_needed)
 283         rtpoll_rebuild(p);
 284
 285     pa_zero(timeout);
 286
 287     /* Calculate timeout */
 288     if (wait_op && !p->quit && p->timer_enabled) {
 289         struct timeval now;
 290         pa_rtclock_get(&now);
 291
 292         if (pa_timeval_cmp(&p->next_elapse, &now) > 0)
 293             pa_timeval_add(&timeout, pa_timeval_diff(&p->next_elapse, &now));
 294     }
 295
 296 #ifdef DEBUG_TIMING
 297     {
 298         pa_usec_t now = pa_rtclock_now();
 299         p->awake = now - p->timestamp;
 300         p->timestamp = now;
 301         if (!wait_op || p->quit || p->timer_enabled)
 302             pa_log("poll timeout: %d ms ",(int) ((timeout.tv_sec*1000) + (timeout.tv_usec / 1000)));
 303         else
 304             pa_log("poll timeout is ZERO");
 305     }
 306 #endif
 307
 308     /* OK, now let's sleep */
 309 #ifdef HAVE_PPOLL
 310     {
 311         struct timespec ts;
 312         ts.tv_sec = timeout.tv_sec;
 313         ts.tv_nsec = timeout.tv_usec * 1000;
 314         r = ppoll(p->pollfd, p->n_pollfd_used, (!wait_op || p->quit || p->timer_enabled) ? &ts : NULL, NULL);
 315     }
 316 #else
 317     r = pa_poll(p->pollfd, p->n_pollfd_used, (!wait_op || p->quit || p->timer_enabled) ? (int) ((timeout.tv_sec*1000) + (timeout.tv_usec / 1000)) : -1);
 318 #endif
 319
 320     p->timer_elapsed = r == 0;
 321
 322 #ifdef DEBUG_TIMING
 323     {
 324         pa_usec_t now = pa_rtclock_now();
 325         p->slept = now - p->timestamp;
 326         p->timestamp = now;
 327
 328         pa_log("Process time %llu ms; sleep time %llu ms",
 329                (unsigned long long) (p->awake / PA_USEC_PER_MSEC),
 330                (unsigned long long) (p->slept / PA_USEC_PER_MSEC));
 331     }
 332 #endif
 333
 334     if (r < 0) {
 335         if (errno == EAGAIN || errno == EINTR)
 336             r = 0;
 337         else
 338             pa_log_error("poll(): %s", pa_cstrerror(errno));
 339
 340         reset_all_revents(p);
 341     }
 342
 343     /* Let's tell everyone that we left the sleep */
 344     for (i = p->items; i && i->priority < PA_RTPOLL_NEVER; i = i->next) {
 345
 346         if (i->dead)
 347             continue;
 348
 349         if (!i->after_cb)
 350             continue;
 351
 352         i->after_cb(i);
 353     }
 354
 355 finish:
 356
 357     p->running = FALSE;
 358
 359     if (p->scan_for_dead) {
 360         pa_rtpoll_item *n;
 361
 362         p->scan_for_dead = FALSE;
 363
 364         for (i = p->items; i; i = n) {
 365             n = i->next;
 366
 367             if (i->dead)
 368                 rtpoll_item_destroy(i);
 369         }
 370     }
 371
 372     return r < 0 ? r : !p->quit;
 373 }
 374
 375 void pa_rtpoll_set_timer_absolute(pa_rtpoll *p, pa_usec_t usec) {
 376     pa_assert(p);
 377
 378     pa_timeval_store(&p->next_elapse, usec);
 379     p->timer_enabled = TRUE;
 380 }
 381
 382 void pa_rtpoll_set_timer_relative(pa_rtpoll *p, pa_usec_t usec) {
 383     pa_assert(p);
 384
 385     /* Scheduling a timeout for more than an hour is very very suspicious */
 386     pa_assert(usec <= PA_USEC_PER_SEC*60ULL*60ULL);
 387
 388     pa_rtclock_get(&p->next_elapse);
 389     pa_timeval_add(&p->next_elapse, usec);
 390     p->timer_enabled = TRUE;
 391 }
 392
 393 void pa_rtpoll_set_timer_disabled(pa_rtpoll *p) {
 394     pa_assert(p);
 395
 396     memset(&p->next_elapse, 0, sizeof(p->next_elapse));
 397     p->timer_enabled = FALSE;
 398 }
 399
 400 pa_rtpoll_item *pa_rtpoll_item_new(pa_rtpoll *p, pa_rtpoll_priority_t prio, unsigned n_fds) {
 401     pa_rtpoll_item *i, *j, *l = NULL;
 402
 403     pa_assert(p);
 404
 405     if (!(i = pa_flist_pop(PA_STATIC_FLIST_GET(items))))
 406         i = pa_xnew(pa_rtpoll_item, 1);
 407
 408     i->rtpoll = p;
 409     i->dead = FALSE;
 410     i->n_pollfd = n_fds;
 411     i->pollfd = NULL;
 412     i->priority = prio;
 413
 414     i->userdata = NULL;
 415     i->before_cb = NULL;
 416     i->after_cb = NULL;
 417     i->work_cb = NULL;
 418
 419     for (j = p->items; j; j = j->next) {
 420         if (prio <= j->priority)
 421             break;
 422
 423         l = j;
 424     }
 425
 426     PA_LLIST_INSERT_AFTER(pa_rtpoll_item, p->items, j ? j->prev : l, i);
 427
 428     if (n_fds > 0) {
 429         p->rebuild_needed = 1;
 430         p->n_pollfd_used += n_fds;
 431     }
 432
 433     return i;
 434 }
 435
 436 void pa_rtpoll_item_free(pa_rtpoll_item *i) {
 437     pa_assert(i);
 438
 439     if (i->rtpoll->running) {
 440         i->dead = TRUE;
 441         i->rtpoll->scan_for_dead = TRUE;
 442         return;
 443     }
 444
 445     rtpoll_item_destroy(i);
 446 }
 447
 448 struct pollfd *pa_rtpoll_item_get_pollfd(pa_rtpoll_item *i, unsigned *n_fds) {
 449     pa_assert(i);
 450
 451     if (i->n_pollfd > 0)
 452         if (i->rtpoll->rebuild_needed)
 453             rtpoll_rebuild(i->rtpoll);
 454
 455     if (n_fds)
 456         *n_fds = i->n_pollfd;
 457
 458     return i->pollfd;
 459 }
 460
 461 void pa_rtpoll_item_set_before_callback(pa_rtpoll_item *i, int (*before_cb)(pa_rtpoll_item *i)) {
 462     pa_assert(i);
 463     pa_assert(i->priority < PA_RTPOLL_NEVER);
 464
 465     i->before_cb = before_cb;
 466 }
 467
 468 void pa_rtpoll_item_set_after_callback(pa_rtpoll_item *i, void (*after_cb)(pa_rtpoll_item *i)) {
 469     pa_assert(i);
 470     pa_assert(i->priority < PA_RTPOLL_NEVER);
 471
 472     i->after_cb = after_cb;
 473 }
 474
 475 void pa_rtpoll_item_set_work_callback(pa_rtpoll_item *i, int (*work_cb)(pa_rtpoll_item *i)) {
 476     pa_assert(i);
 477     pa_assert(i->priority < PA_RTPOLL_NEVER);
 478
 479     i->work_cb = work_cb;
 480 }
 481
 482 void pa_rtpoll_item_set_userdata(pa_rtpoll_item *i, void *userdata) {
 483     pa_assert(i);
 484
 485     i->userdata = userdata;
 486 }
 487
 488 void* pa_rtpoll_item_get_userdata(pa_rtpoll_item *i) {
 489     pa_assert(i);
 490
 491     return i->userdata;
 492 }
 493
 494 static int fdsem_before(pa_rtpoll_item *i) {
 495
 496     if (pa_fdsem_before_poll(i->userdata) < 0)
 497         return 1; /* 1 means immediate restart of the loop */
 498
 499     return 0;
 500 }
 501
 502 static void fdsem_after(pa_rtpoll_item *i) {
 503     pa_assert(i);
 504
 505     pa_assert((i->pollfd[0].revents & ~POLLIN) == 0);
 506     pa_fdsem_after_poll(i->userdata);
 507 }
 508
 509 pa_rtpoll_item *pa_rtpoll_item_new_fdsem(pa_rtpoll *p, pa_rtpoll_priority_t prio, pa_fdsem *f) {
 510     pa_rtpoll_item *i;
 511     struct pollfd *pollfd;
 512
 513     pa_assert(p);
 514     pa_assert(f);
 515
 516     i = pa_rtpoll_item_new(p, prio, 1);
 517
 518     pollfd = pa_rtpoll_item_get_pollfd(i, NULL);
 519
 520     pollfd->fd = pa_fdsem_get(f);
 521     pollfd->events = POLLIN;
 522
 523     i->before_cb = fdsem_before;
 524     i->after_cb = fdsem_after;
 525     i->userdata = f;
 526
 527     return i;
 528 }
 529
 530 static int asyncmsgq_read_before(pa_rtpoll_item *i) {
 531     pa_assert(i);
 532
 533     if (pa_asyncmsgq_read_before_poll(i->userdata) < 0)
 534         return 1; /* 1 means immediate restart of the loop */
 535
 536     return 0;
 537 }
 538
 539 static void asyncmsgq_read_after(pa_rtpoll_item *i) {
 540     pa_assert(i);
 541
 542     pa_assert((i->pollfd[0].revents & ~POLLIN) == 0);
 543     pa_asyncmsgq_read_after_poll(i->userdata);
 544 }
 545
 546 static int asyncmsgq_read_work(pa_rtpoll_item *i) {
 547     pa_msgobject *object;
 548     int code;
 549     void *data;
 550     pa_memchunk chunk;
 551     int64_t offset;
 552
 553     pa_assert(i);
 554
 555     if (pa_asyncmsgq_get(i->userdata, &object, &code, &data, &offset, &chunk, 0) == 0) {
 556         int ret;
 557
 558         if (!object && code == PA_MESSAGE_SHUTDOWN) {
 559             pa_asyncmsgq_done(i->userdata, 0);
 560             pa_rtpoll_quit(i->rtpoll);
 561             return 1;
 562         }
 563
 564         ret = pa_asyncmsgq_dispatch(object, code, data, offset, &chunk);
 565         pa_asyncmsgq_done(i->userdata, ret);
 566         return 1;
 567     }
 568
 569     return 0;
 570 }
 571
 572 pa_rtpoll_item *pa_rtpoll_item_new_asyncmsgq_read(pa_rtpoll *p, pa_rtpoll_priority_t prio, pa_asyncmsgq *q) {
 573     pa_rtpoll_item *i;
 574     struct pollfd *pollfd;
 575
 576     pa_assert(p);
 577     pa_assert(q);
 578
 579     i = pa_rtpoll_item_new(p, prio, 1);
 580
 581     pollfd = pa_rtpoll_item_get_pollfd(i, NULL);
 582     pollfd->fd = pa_asyncmsgq_read_fd(q);
 583     pollfd->events = POLLIN;
 584
 585     i->before_cb = asyncmsgq_read_before;
 586     i->after_cb = asyncmsgq_read_after;
 587     i->work_cb = asyncmsgq_read_work;
 588     i->userdata = q;
 589
 590     return i;
 591 }
 592
 593 static int asyncmsgq_write_before(pa_rtpoll_item *i) {
 594     pa_assert(i);
 595
 596     pa_asyncmsgq_write_before_poll(i->userdata);
 597     return 0;
 598 }
 599
 600 static void asyncmsgq_write_after(pa_rtpoll_item *i) {
 601     pa_assert(i);
 602
 603     pa_assert((i->pollfd[0].revents & ~POLLIN) == 0);
 604     pa_asyncmsgq_write_after_poll(i->userdata);
 605 }
 606
 607 pa_rtpoll_item *pa_rtpoll_item_new_asyncmsgq_write(pa_rtpoll *p, pa_rtpoll_priority_t prio, pa_asyncmsgq *q) {
 608     pa_rtpoll_item *i;
 609     struct pollfd *pollfd;
 610
 611     pa_assert(p);
 612     pa_assert(q);
 613
 614     i = pa_rtpoll_item_new(p, prio, 1);
 615
 616     pollfd = pa_rtpoll_item_get_pollfd(i, NULL);
 617     pollfd->fd = pa_asyncmsgq_write_fd(q);
 618     pollfd->events = POLLIN;
 619
 620     i->before_cb = asyncmsgq_write_before;
 621     i->after_cb = asyncmsgq_write_after;
 622     i->work_cb = NULL;
 623     i->userdata = q;
 624
 625     return i;
 626 }
 627
 628 void pa_rtpoll_quit(pa_rtpoll *p) {
 629     pa_assert(p);
 630
 631     p->quit = TRUE;
 632 }
 633
 634 pa_bool_t pa_rtpoll_timer_elapsed(pa_rtpoll *p) {
 635     pa_assert(p);
 636
 637     return p->timer_elapsed;
 638 }