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